Merge pull request #1675 from kodustech/fix/query-perf-and-indexes

fix(pool,queries,locks): pool exhaustion mitigations from the 2026-08 incident audit

Author: Wellington01

apps/worker/src/cron/analytics-classifier.cron.ts

@@ -2,19 +2,26 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Cron, CronExpression } from '@nestjs/schedule';
 
 import { PullRequestClassifierService } from '@libs/ee/analytics-warehouse';
+import { DistributedLockService } from '@libs/core/workflow/infrastructure/distributed-lock.service';
+
+const LOCK_KEY = 'CRON:ANALYTICS_CLASSIFIER';
+// 25min TTL for a 30min-tick cron. Auto-clears on crashes without
+// letting a runaway hold the slot into the next tick.
+const LOCK_TTL_MS = 25 * 60 * 1000;
 
 /**
  * Cron wrapper that drives `PullRequestClassifierService` on a schedule.
  * Classifies unclassified PRs via LLM and fills `analytics.pull_request_types`
  * so bug-ratio and other "by type" aggregates have ground truth.
  *
  * Tunable via `ANALYTICS_CLASSIFIER_CRON` (standard cron expression).
- * Default = every 15 minutes. Disable with `ANALYTICS_CLASSIFIER_DISABLED=true`.
+ * Default = every 30 minutes. Disable with `ANALYTICS_CLASSIFIER_DISABLED=true`.
  *
- * Concurrency: same in-memory mutex trick as `AnalyticsIngestionCron`.
- * Two ticks overlapping would mostly be wasted work (both picking the
- * same unclassified rows) — the upserts are idempotent but the LLM
- * cost would double.
+ * Concurrency: 15 worker replicas in prod. Without a distributed lock
+ * every replica would call the LLM on the same unclassified rows every
+ * 30min — the upserts are idempotent so the DB stays correct, but the
+ * OpenAI spend multiplies by 15. Local `running` mutex still catches
+ * same-node reentry.
  */
 @Injectable()
 export class AnalyticsClassifierCron {
@@ -23,6 +30,7 @@ export class AnalyticsClassifierCron {
 
     constructor(
         private readonly classifier: PullRequestClassifierService,
+        private readonly distributedLockService: DistributedLockService,
     ) {}
 
     // `||` so that docker-compose's `${VAR:-}` empty-string fallthrough
@@ -43,6 +51,22 @@ export class AnalyticsClassifierCron {
             return;
         }
 
+        const lock = await this.distributedLockService
+            .acquire(LOCK_KEY, { ttl: LOCK_TTL_MS })
+            .catch((err: unknown) => {
+                this.logger.warn(
+                    `analytics classifier lock acquire threw: ${err instanceof Error ? err.message : String(err)}`,
+                );
+                return null;
+            });
+
+        if (!lock) {
+            this.logger.log(
+                'skipping analytics classifier — another replica holds the lock',
+            );
+            return;
+        }
+
         this.running = true;
         const start = Date.now();
         try {
@@ -59,6 +83,11 @@ export class AnalyticsClassifierCron {
             );
         } finally {
             this.running = false;
+            await lock.release().catch((err: unknown) => {
+                this.logger.warn(
+                    `analytics classifier lock release failed: ${err instanceof Error ? err.message : String(err)}`,
+                );
+            });
         }
     }
 }

apps/worker/src/cron/analytics-ingestion.cron.ts

@@ -6,17 +6,28 @@ import {
     PullRequestIngestionService,
     ReviewOperationalIngestionService,
 } from '@libs/ee/analytics-warehouse';
+import { DistributedLockService } from '@libs/core/workflow/infrastructure/distributed-lock.service';
+
+const LOCK_KEY = 'CRON:ANALYTICS_INGESTION';
+// 25min — the run typically takes 5–15min; the tick is every 30min so
+// this covers the slowest observed pass while still auto-clearing if
+// a worker crashes mid-run.
+const LOCK_TTL_MS = 25 * 60 * 1000;
 
 /**
  * Cron wrapper that drives cockpit warehouse ingestion on a schedule.
  * Interval is tunable via `ANALYTICS_INGESTION_CRON` (standard cron
- * expression). Default = every 15 minutes.
+ * expression). Default = every 30 minutes.
  *
- * Concurrency: a second instance landing while one is still running
- * would cause transaction contention, not correctness issues — UPSERTs
- * and the per-PR DELETE/INSERT children run inside a single tx per
- * batch, and the watermark is idempotent. We keep a local in-memory
- * guard as a cheap mutex so we don't stack up runs on a single node.
+ * Concurrency: the worker deployment runs 15 replicas in prod
+ * (`worker_desired_count = 15`), so an in-process `running` mutex is
+ * not enough — without a distributed lock every replica would run the
+ * warehouse ingestion (LLM calls + DELETE/INSERT children in tx per
+ * batch) simultaneously every 30min. That produces 15× the OpenAI
+ * spend on classification + burns warehouse write capacity for no
+ * benefit. The local `running` flag is still kept as a cheap short-
+ * circuit against reentry on the same node (e.g. the boot spawn
+ * overlapping the first cron tick).
  */
 @Injectable()
 export class AnalyticsIngestionCron implements OnApplicationBootstrap {
@@ -27,6 +38,7 @@ export class AnalyticsIngestionCron implements OnApplicationBootstrap {
         private readonly ingestion: PullRequestIngestionService,
         private readonly feedbackIngestion: FeedbackIngestionService,
         private readonly reviewOperationalIngestion: ReviewOperationalIngestionService,
+        private readonly distributedLockService: DistributedLockService,
     ) {}
 
     onApplicationBootstrap(): void {
@@ -66,6 +78,23 @@ export class AnalyticsIngestionCron implements OnApplicationBootstrap {
             return;
         }
 
+        // Cross-replica gate: only one worker across the fleet runs.
+        const lock = await this.distributedLockService
+            .acquire(LOCK_KEY, { ttl: LOCK_TTL_MS })
+            .catch((err: unknown) => {
+                this.logger.warn(
+                    `analytics ingestion (${trigger}) lock acquire threw: ${err instanceof Error ? err.message : String(err)}`,
+                );
+                return null;
+            });
+
+        if (!lock) {
+            this.logger.log(
+                `skipping analytics ingestion (${trigger}) — another replica holds the lock`,
+            );
+            return;
+        }
+
         this.running = true;
         const start = Date.now();
         try {
@@ -109,6 +138,11 @@ export class AnalyticsIngestionCron implements OnApplicationBootstrap {
             }
         } finally {
             this.running = false;
+            await lock.release().catch((err: unknown) => {
+                this.logger.warn(
+                    `analytics ingestion lock release failed: ${err instanceof Error ? err.message : String(err)}`,
+                );
+            });
         }
     }
 }

libs/centralized-config/infrastructure/adapters/listeners/centralized-config-sync.listener.spec.ts

@@ -7,6 +7,7 @@ import {
     ICentralizedConfigService,
 } from '@libs/centralized-config/domain/contracts/CentralizedConfigService.contract';
 import { PullRequestClosedEvent } from '@libs/core/domain/events/pull-request-closed.event';
+import { DistributedLockService } from '@libs/core/workflow/infrastructure/distributed-lock.service';
 import { CentralizedConfigSyncListener } from './centralized-config-sync.listener';
 
 describe('CentralizedConfigSyncListener', () => {
@@ -34,10 +35,19 @@ describe('CentralizedConfigSyncListener', () => {
             removeStaleKodyRules: jest.fn(),
         };
 
+    // Default: lock acquired (sync runs). Individual tests can override
+    // by calling `distributedLockServiceMock.acquire.mockResolvedValueOnce(null)`.
+    const distributedLockServiceMock = {
+        acquire: jest.fn(),
+    };
+
     beforeEach(async () => {
         centralizedConfigSyncUseCaseMock.execute.mockReset();
         centralizedConfigPrServiceMock.handleTrackedPullRequestClose.mockReset();
         jest.clearAllMocks();
+        distributedLockServiceMock.acquire.mockResolvedValue({
+            release: jest.fn().mockResolvedValue(undefined),
+        });
 
         const module: TestingModule = await Test.createTestingModule({
             providers: [
@@ -54,6 +64,10 @@ describe('CentralizedConfigSyncListener', () => {
                     provide: CENTRALIZED_CONFIG_SERVICE_TOKEN,
                     useValue: centralizedConfigServiceMock,
                 },
+                {
+                    provide: DistributedLockService,
+                    useValue: distributedLockServiceMock,
+                },
             ],
         }).compile();
 

libs/centralized-config/infrastructure/adapters/listeners/centralized-config-sync.listener.ts

@@ -6,9 +6,15 @@ import {
     ICentralizedConfigService,
 } from '@libs/centralized-config/domain/contracts/CentralizedConfigService.contract';
 import { PullRequestClosedEvent } from '@libs/core/domain/events/pull-request-closed.event';
+import { DistributedLockService } from '@libs/core/workflow/infrastructure/distributed-lock.service';
 import { Inject, Injectable } from '@nestjs/common';
 import { OnEvent } from '@nestjs/event-emitter';
 
+// Sync involves ~15 GitHub calls + ~45 pg queries for a monorepo with
+// centralized config enabled. 5min covers the largest observed run;
+// TTL exists so a crashed handler doesn't leave the lock stuck.
+const CENTRALIZED_SYNC_LOCK_TTL_MS = 5 * 60 * 1000;
+
 @Injectable()
 export class CentralizedConfigSyncListener {
     private readonly logger = createLogger(CentralizedConfigSyncListener.name);
@@ -18,6 +24,7 @@ export class CentralizedConfigSyncListener {
         private readonly centralizedConfigPrService: CentralizedConfigPrService,
         @Inject(CENTRALIZED_CONFIG_SERVICE_TOKEN)
         private readonly centralizedConfigService: ICentralizedConfigService,
+        private readonly distributedLockService: DistributedLockService,
     ) {}
 
     @OnEvent('pull-request.closed')
@@ -88,9 +95,38 @@ export class CentralizedConfigSyncListener {
             return;
         }
 
-        await this.centralizedConfigSyncUseCase.execute({
-            organizationAndTeamData: event.organizationAndTeamData,
-            repository: event.repository,
+        // Cross-process idempotency via pg_try_advisory_lock. The
+        // pull-request.closed event reaches every process hosting this
+        // listener (API + worker, plus CrossProcessEventsBridge re-emit)
+        // — without a shared claim, each replica runs the ~45 pg + 15
+        // GitHub call sync pipeline for the same merge. First acquirer
+        // wins; the others get null and skip. Same pattern used by
+        // KodyRulesSyncListener.
+        const lockKey = `CENTRALIZED_CONFIG:SYNC:${event.organizationAndTeamData?.organizationId}:${event.repository.id}:${event.pullRequestNumber}`;
+        const lock = await this.distributedLockService.acquire(lockKey, {
+            ttl: CENTRALIZED_SYNC_LOCK_TTL_MS,
         });
+
+        if (!lock) {
+            this.logger.log({
+                message:
+                    'Centralized sync already claimed by another process for this merge — skipping duplicate run',
+                context: CentralizedConfigSyncListener.name,
+                metadata: {
+                    lockKey,
+                    pullRequestNumber: event.pullRequestNumber,
+                },
+            });
+            return;
+        }
+
+        try {
+            await this.centralizedConfigSyncUseCase.execute({
+                organizationAndTeamData: event.organizationAndTeamData,
+                repository: event.repository,
+            });
+        } finally {
+            await lock.release();
+        }
     }
 }

libs/core/context-resolution/infrastructure/adapters/services/context-resolution.service.ts

@@ -7,10 +7,6 @@ import {
     IIntegrationConfigService,
     INTEGRATION_CONFIG_SERVICE_TOKEN,
 } from '@libs/integrations/domain/integrationConfigs/contracts/integration-config.service.contracts';
-import {
-    IIntegrationService,
-    INTEGRATION_SERVICE_TOKEN,
-} from '@libs/integrations/domain/integrations/contracts/integration.service.contracts';
 import {
     IParametersService,
     PARAMETERS_SERVICE_TOKEN,
@@ -21,8 +17,6 @@ export class ContextResolutionService implements IContextResolutionService {
     constructor(
         @Inject(PARAMETERS_SERVICE_TOKEN)
         private readonly parametersService: IParametersService,
-        @Inject(INTEGRATION_SERVICE_TOKEN)
-        private readonly integrationService: IIntegrationService,
         @Inject(INTEGRATION_CONFIG_SERVICE_TOKEN)
         private readonly integrationConfigService: IIntegrationConfigService,
     ) {}
@@ -31,44 +25,33 @@ export class ContextResolutionService implements IContextResolutionService {
         organizationId: string,
         repositoryId: string,
     ): Promise<string> {
-        // 1. Fetch all active integrations for the organization
-        const integrations = await this.integrationService.find({
-            organization: { uuid: organizationId },
-            status: true,
-        });
+        // Prior version fetched active integrations first, then issued
+        // one `find({ integration: { uuid } })` per integration inside a
+        // sequential loop — a textbook 1+N on the code-review hot path.
+        // The repository method below collapses it to a single JOIN
+        // query over integration_configs → integrations → organization.
+        const integrationConfigs =
+            await this.integrationConfigService.findByOrganizationAndConfigKey(
+                organizationId,
+                IntegrationConfigKey.REPOSITORIES,
+            );
 
-        if (!integrations || integrations.length === 0) {
+        if (!integrationConfigs || integrationConfigs.length === 0) {
             throw new Error('No active integrations found for organization');
         }
 
-        // 2. For each integration, fetch integration configs with REPOSITORIES key
-        for (const integration of integrations) {
-            const integrationConfigs = await this.integrationConfigService.find(
-                {
-                    integration: { uuid: integration.uuid },
-                    configKey: IntegrationConfigKey.REPOSITORIES,
-                },
-            );
+        for (const config of integrationConfigs) {
+            const repositories = config.configValue;
 
-            if (!integrationConfigs || integrationConfigs.length === 0) {
-                continue;
-            }
+            if (Array.isArray(repositories)) {
+                const foundRepository = repositories.find(
+                    (repo: any) =>
+                        repo.id === repositoryId ||
+                        repo.id === repositoryId.toString(),
+                );
 
-            // 3. Search the repository list for one with the same id
-            for (const config of integrationConfigs) {
-                const repositories = config.configValue;
-
-                if (Array.isArray(repositories)) {
-                    const foundRepository = repositories.find(
-                        (repo: any) =>
-                            repo.id === repositoryId ||
-                            repo.id === repositoryId.toString(),
-                    );
-
-                    if (foundRepository) {
-                        // 4. Return the teamId from this integration config
-                        return config?.team?.uuid;
-                    }
+                if (foundRepository) {
+                    return config?.team?.uuid;
                 }
             }
         }

libs/core/infrastructure/config/axios/microservices/azureRepos.axios.ts

@@ -1,11 +1,19 @@
 import axios, { AxiosInstance } from 'axios';
 
+import { INTEGRATION_REQUEST_TIMEOUT_MS } from '@libs/core/infrastructure/http/integration-timeouts';
+
 export class AxiosAzureReposService {
     private axiosInstance: AxiosInstance;
 
     constructor({ tenantId = '', organization = '' }) {
         this.axiosInstance = axios.create({
             baseURL: process.env.KODUS_SERVICE_AZURE_REPOS,
+            // axios default is 0 = infinite. Without this, a stalled
+            // upstream microservice would keep the caller's HTTP handler
+            // (and its pool connection) hanging until the global undici
+            // 10-minute ceiling. Matches the pattern already used in
+            // sibling microservice clients (license.axios, mcpManager.axios).
+            timeout: INTEGRATION_REQUEST_TIMEOUT_MS,
             headers: {
                 'Content-Type': 'application/json',
                 'x-tenant-id': tenantId,