perf: serialize heavy database workloads

backend/src/analysis/heavyWorkAdmission.test.ts

@@ -0,0 +1,53 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { withHeavyWorkAdmission } from './heavyWorkAdmission.js';
+
+function fakeAdmission(sequence: boolean[]) {
+  const calls: string[] = [];
+  const client = {
+    async query(sql: string) {
+      if (sql.includes('pg_try_advisory_lock')) {
+        calls.push('try');
+        return { rows: [{ acquired: sequence.shift() ?? true }] };
+      }
+      calls.push('unlock');
+      return { rows: [{ unlocked: true }] };
+    },
+    release(error?: Error) {
+      calls.push(error ? 'destroy' : 'release');
+    },
+  };
+  return {
+    calls,
+    pool: { async connect() { calls.push('connect'); return client; } },
+  };
+}
+
+test('heavy work admission waits, runs once, and releases the session claim', async () => {
+  const fake = fakeAdmission([false, true]);
+  const logs: string[] = [];
+  const value = await withHeavyWorkAdmission({
+    pool: fake.pool as never,
+    workload: 'chart-refresh:ukmesh',
+    retryMs: 1,
+    log: { log: (line) => logs.push(String(line)), warn: (line) => logs.push(String(line)) },
+    task: async () => {
+      fake.calls.push('task');
+      return 42;
+    },
+  });
+  assert.equal(value, 42);
+  assert.deepEqual(fake.calls, ['connect', 'try', 'try', 'task', 'unlock', 'release']);
+  assert.ok(logs.some((line) => line.includes('waiting')));
+  assert.ok(logs.some((line) => line.includes('released')));
+});
+
+test('heavy work admission unlocks when the protected task fails', async () => {
+  const fake = fakeAdmission([true]);
+  await assert.rejects(withHeavyWorkAdmission({
+    pool: fake.pool as never,
+    workload: 'path-learning:all',
+    task: async () => { throw new Error('publication failed'); },
+  }), /publication failed/);
+  assert.deepEqual(fake.calls, ['connect', 'try', 'unlock', 'release']);
+});

backend/src/analysis/heavyWorkAdmission.ts

@@ -0,0 +1,96 @@
+import type { Pool, PoolClient, QueryResultRow } from 'pg';
+
+// One cluster-wide session advisory lock serialises the three analytical jobs
+// that have the highest database hold time. PostgreSQL owns the claim, so a
+// crashed process or severed connection releases it without a stale lock row.
+const HEAVY_WORK_LOCK_CLASS = 1_433_214_840;
+const HEAVY_WORK_LOCK_KEY = 1_463_235_684;
+const DEFAULT_RETRY_MS = 1_000;
+
+type AdmissionClient = Pick<PoolClient, 'query' | 'release'>;
+type AdmissionPool = Pick<Pool, 'connect'>;
+
+type HeavyWorkAdmissionOptions<T> = {
+  pool: AdmissionPool;
+  workload: string;
+  task: () => Promise<T>;
+  signal?: AbortSignal;
+  retryMs?: number;
+  log?: Pick<Console, 'log' | 'warn'>;
+};
+
+function sleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) return Promise.reject(signal.reason ?? new Error('HEAVY_WORK_ADMISSION_ABORTED'));
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(resolve, ms);
+    const onAbort = () => {
+      clearTimeout(timer);
+      reject(signal?.reason ?? new Error('HEAVY_WORK_ADMISSION_ABORTED'));
+    };
+    signal?.addEventListener('abort', onAbort, { once: true });
+    if (signal) {
+      const cleanup = () => signal.removeEventListener('abort', onAbort);
+      setTimeout(cleanup, ms);
+    }
+  });
+}
+
+async function tryAcquire(client: AdmissionClient): Promise<boolean> {
+  const result = await client.query<{ acquired: boolean }>(
+    'SELECT pg_try_advisory_lock($1, $2) AS acquired',
+    [HEAVY_WORK_LOCK_CLASS, HEAVY_WORK_LOCK_KEY],
+  );
+  return result.rows[0]?.acquired === true;
+}
+
+export async function withHeavyWorkAdmission<T>(
+  options: HeavyWorkAdmissionOptions<T>,
+): Promise<T> {
+  if (!/^[a-z0-9][a-z0-9:_-]{0,95}$/i.test(options.workload)) {
+    throw new Error('INVALID_HEAVY_WORKLOAD_NAME');
+  }
+  const retryMs = Math.max(1, Math.min(30_000, Math.trunc(options.retryMs ?? DEFAULT_RETRY_MS)));
+  const logger = options.log ?? console;
+  const client = await options.pool.connect() as AdmissionClient;
+  const waitingStartedAt = Date.now();
+  let acquired = false;
+  let waitingLogged = false;
+  try {
+    while (!acquired) {
+      options.signal?.throwIfAborted();
+      acquired = await tryAcquire(client);
+      if (acquired) break;
+      if (!waitingLogged) {
+        logger.log(`[db-admission] workload=${options.workload} waiting`);
+        waitingLogged = true;
+      }
+      await sleep(retryMs, options.signal);
+    }
+    const acquiredAt = Date.now();
+    logger.log(
+      `[db-admission] workload=${options.workload} acquired waitMs=${acquiredAt - waitingStartedAt}`,
+    );
+    try {
+      return await options.task();
+    } finally {
+      const unlock = await client.query<QueryResultRow>(
+        'SELECT pg_advisory_unlock($1, $2) AS unlocked',
+        [HEAVY_WORK_LOCK_CLASS, HEAVY_WORK_LOCK_KEY],
+      ).catch((error: unknown) => {
+        logger.warn(
+          `[db-admission] workload=${options.workload} unlock failed: ${error instanceof Error ? error.message : String(error)}`,
+        );
+        return null;
+      });
+      const unlocked = unlock?.rows[0]?.['unlocked'] === true;
+      logger.log(
+        `[db-admission] workload=${options.workload} released durationMs=${Date.now() - acquiredAt} unlocked=${unlocked}`,
+      );
+      acquired = !unlocked;
+    }
+  } finally {
+    // If explicit unlock failed, closing this dedicated session is the
+    // authoritative stale-claim reclamation path.
+    client.release(acquired ? new Error('HEAVY_WORK_ADMISSION_RELEASE') : undefined);
+  }
+}

backend/src/api/routes.ts

@@ -86,6 +86,7 @@ import {
 import { assertUniqueRouteRegistry } from './routeRegistry.js';
 import { assertContractCoverage } from './contracts.js';
 import { ApiInputError, wrapAsyncHandlers } from './errors.js';
+import { withHeavyWorkAdmission } from '../analysis/heavyWorkAdmission.js';
 
 const router = Router();
 // Anonymous cross-network aggregation is not a public API capability. Operator
@@ -232,6 +233,7 @@ registerStatsRoutes(router, {
   analyticsQuery,
   getPublicVisibilityGeneration,
   maskDecodedPathNodes,
+  runHeavyWork: (workload, task) => withHeavyWorkAdmission({ pool, workload, task }),
 });
 registerTelemetryRoutes(router, { query });
 registerSpamRoutes(router, { expensiveLimiter: EXPENSIVE_LIMITER });

backend/src/api/routes/stats.ts

@@ -49,6 +49,7 @@ type StatsRouteDeps = {
   analyticsQuery: QueryFn;
   getPublicVisibilityGeneration: () => Promise<number>;
   maskDecodedPathNodes: MaskDecodedPathNodesFn;
+  runHeavyWork: <T>(workload: string, task: () => Promise<T>) => Promise<T>;
 };
 
 export function registerStatsRoutes(router: Router, deps: StatsRouteDeps): void {
@@ -75,6 +76,7 @@ export function registerStatsRoutes(router: Router, deps: StatsRouteDeps): void
     },
     getPublicVisibilityGeneration: deps.getPublicVisibilityGeneration,
     maskDecodedPathNodes: deps.maskDecodedPathNodes,
+    runHeavyWork: deps.runHeavyWork,
   });
 
   chartsWarmup.register(service.startChartsWarmup);

backend/src/path-learning/rebuild.ts

@@ -1,5 +1,5 @@
 import { createHash } from 'node:crypto';
-import { getPublicVisibilityGeneration, pool, query } from '../db/index.js';
+import { analyticsQuery, getPublicVisibilityGeneration, pool } from '../db/index.js';
 import {
   analysisGeneration,
   beginAnalysisRun,
@@ -21,6 +21,7 @@ import {
   PATH_LEARNING_DELTA_DEFINITIONS,
   publishPathLearningDelta,
 } from './deltaPublication.js';
+import { withHeavyWorkAdmission } from '../analysis/heavyWorkAdmission.js';
 
 type LearningNode = {
   node_id: string;
@@ -379,6 +380,35 @@ async function publishPathLearningRowsDelta(
   },
   analysisRun: AnalysisRunHandle,
   heartbeat: AnalysisRunHeartbeat,
+): Promise<{ skipped: boolean; upserted: number; deleted: number }> {
+  return withHeavyWorkAdmission({
+    pool,
+    workload: `path-learning:${network}`,
+    signal: heartbeat.signal,
+    task: () => publishPathLearningRowsDeltaUnderAdmission(
+      network,
+      datasets,
+      calibration,
+      metadata,
+      analysisRun,
+      heartbeat,
+    ),
+  });
+}
+
+async function publishPathLearningRowsDeltaUnderAdmission(
+  network: string,
+  datasets: object[][],
+  calibration: PathLearningCalibration,
+  metadata: {
+    inputHash: string;
+    modelHash: string;
+    privacyGeneration: number;
+    windowStart: Date;
+    windowEnd: Date;
+  },
+  analysisRun: AnalysisRunHandle,
+  heartbeat: AnalysisRunHeartbeat,
 ): Promise<{ skipped: boolean; upserted: number; deleted: number }> {
   const client = await pool.connect();
   try {
@@ -460,7 +490,7 @@ async function publishPathLearningRowsDelta(
 export async function rebuildPathLearningModels(): Promise<void> {
   const windowEnd = new Date();
   const windowStart = new Date(windowEnd.getTime() - 30 * 24 * 60 * 60_000);
-  const networksResult = await query<{ network: string }>(
+  const networksResult = await analyticsQuery<{ network: string }>(
     `SELECT DISTINCT network
      FROM packets
      WHERE network IS NOT NULL
@@ -560,7 +590,7 @@ async function rebuildNetworkUnderLease(
     : [run.windowStart, run.windowEnd, MAX_TRAINING_PACKETS];
   const linkParams: unknown[] = sourceNetwork ? [sourceNetwork, MAX_LEARNING_LINKS + 1] : [MAX_LEARNING_LINKS + 1];
 
-  const packetsResult = await query<LearningPacket>(
+  const packetsResult = await analyticsQuery<LearningPacket>(
     `SELECT DISTINCT ON (packet_hash, rx_node_id, src_node_id, path_hashes)
             time, rx_node_id, src_node_id, path_hashes
        FROM packets
@@ -591,7 +621,7 @@ async function rebuildNetworkUnderLease(
     });
   }
 
-  const nodesResult = await query<LearningNode>(
+  const nodesResult = await analyticsQuery<LearningNode>(
     `SELECT node_id, lat, lon, elevation_m, iata
      FROM nodes
      WHERE lat IS NOT NULL
@@ -610,7 +640,7 @@ async function rebuildNetworkUnderLease(
 
   const pathHashIndex = buildNodePathHashIndex(nodesResult.rows);
 
-  const linksResult = await query<LearningLink>(
+  const linksResult = await analyticsQuery<LearningLink>(
     `SELECT nl.node_a_id, nl.node_b_id, nl.itm_path_loss_db, nl.count_a_to_b, nl.count_b_to_a
      FROM node_links nl
      JOIN nodes a ON a.node_id = nl.node_a_id

backend/src/platform/config/database.test.ts

@@ -50,7 +50,7 @@ test('database configuration rejects invalid pool, timeout, and schema settings
 });
 
 test('analytics queries honor a longer configured timeout while retaining a safe floor', () => {
-  assert.equal(analyticsStatementTimeoutMs(30_000), 300_000);
+  assert.equal(analyticsStatementTimeoutMs(30_000), 900_000);
   assert.equal(analyticsStatementTimeoutMs(900_000), 900_000);
   assert.equal(analyticsStatementTimeoutMs(0), 0);
 });

backend/src/platform/config/database.ts

@@ -38,7 +38,7 @@ export function loadDatabaseConfig(env: NodeJS.ProcessEnv) {
 }
 
 export function analyticsStatementTimeoutMs(configuredTimeoutMs: number): number {
-  return configuredTimeoutMs === 0 ? 0 : Math.max(300_000, configuredTimeoutMs);
+  return configuredTimeoutMs === 0 ? 0 : Math.max(900_000, configuredTimeoutMs);
 }
 
 export const databaseConfig = loadDatabaseConfig(process.env);

backend/src/stats/statsService.test.ts

@@ -42,6 +42,7 @@ test('completed canonical charts are reused while observer-scoped charts are nev
   let regionSummaryCalls = 0;
   let snapshotLoads = 0;
   let snapshotSaves = 0;
+  let admittedChartRefreshes = 0;
   const repository = {
     loadChartSnapshot: async () => {
       snapshotLoads += 1;
@@ -71,6 +72,11 @@ test('completed canonical charts are reused while observer-scoped charts are nev
     repository,
     getPublicVisibilityGeneration: async () => 1,
     maskDecodedPathNodes: () => [],
+    runHeavyWork: async (workload, task) => {
+      assert.equal(workload, 'chart-refresh:ukmesh');
+      admittedChartRefreshes += 1;
+      return task();
+    },
   });
 
   await service.getCharts('ukmesh', undefined);
@@ -80,6 +86,7 @@ test('completed canonical charts are reused while observer-scoped charts are nev
   assert.equal(chartsCache.size, 1);
   assert.equal(snapshotLoads, 1);
   assert.equal(snapshotSaves, 1);
+  assert.equal(admittedChartRefreshes, 1);
 
   await service.getCharts('ukmesh', 'A'.repeat(64));
   await service.getCharts('ukmesh', 'A'.repeat(64));
@@ -87,6 +94,7 @@ test('completed canonical charts are reused while observer-scoped charts are nev
   assert.equal(chartsCache.size, 1);
   assert.equal(snapshotLoads, 1);
   assert.equal(snapshotSaves, 1);
+  assert.equal(admittedChartRefreshes, 1);
 });
 
 test('a valid durable chart snapshot serves a cold process without analytical queries', async () => {

backend/src/stats/statsService.ts

@@ -33,6 +33,7 @@ type StatsServiceDeps = {
   repository: StatsRepository;
   getPublicVisibilityGeneration: () => Promise<number>;
   maskDecodedPathNodes: MaskDecodedPathNodesFn;
+  runHeavyWork?: <T>(workload: string, task: () => Promise<T>) => Promise<T>;
 };
 
 const CHANNEL_TRAFFIC_CACHE_TTL_MS = 60 * 60_000;
@@ -102,6 +103,7 @@ export function createStatsService(deps: StatsServiceDeps) {
     repository,
     getPublicVisibilityGeneration,
     maskDecodedPathNodes,
+    runHeavyWork = async (_workload, task) => task(),
   } = deps;
 
   const PAYLOAD_LABELS: Record<number, string> = {
@@ -521,10 +523,9 @@ export function createStatsService(deps: StatsServiceDeps) {
           // Chart refresh can still proceed if the lightweight warmup failed.
         });
       }
-      const data = await computeChartsData(
-        network,
-        undefined,
-        visibilityGeneration,
+      const data = await runHeavyWork(
+        `chart-refresh:${scope}`,
+        () => computeChartsData(network, undefined, visibilityGeneration),
       );
       const validated = validateChartSnapshotPayload(
         data,

backend/src/workers/path-history.ts

@@ -6,6 +6,7 @@ import {
   refreshRecentPathEvidence,
   upsertPathHistoryCache,
   type PathHistorySegmentRow,
+  pool,
 } from '../db/index.js';
 import {
   createMultiObserverPathBatchResolver,
@@ -35,6 +36,7 @@ import {
   resumablePathHistoryCheckpoint,
   type PathHistoryCheckpoint,
 } from './pathHistoryCheckpoint.js';
+import { withHeavyWorkAdmission } from '../analysis/heavyWorkAdmission.js';
 
 const RETRY_INTERVAL_MS = pathHistoryRetryIntervalMs(process.env['PATH_HISTORY_RETRY_INTERVAL_MS']);
 const WINDOW_HOURS = 168;
@@ -407,6 +409,22 @@ async function refreshAll(tag: 'initial' | 'scheduled'): Promise<boolean> {
     return true;
   }
   isRunning = true;
+  try {
+    return await withHeavyWorkAdmission({
+      pool,
+      workload: `path-history:${tag}`,
+      task: () => refreshAllUnderAdmission(tag),
+    });
+  } catch (err) {
+    observeWorkerOutcome('path_history', 'refresh', 'failure');
+    console.error(`[path-history] ${tag} refresh failed`, (err as Error).message);
+    return true;
+  } finally {
+    isRunning = false;
+  }
+}
+
+async function refreshAllUnderAdmission(tag: 'initial' | 'scheduled'): Promise<boolean> {
   try {
     let publicEvidenceUpdates = 0;
     let testEvidenceUpdates = 0;
@@ -430,8 +448,6 @@ async function refreshAll(tag: 'initial' | 'scheduled'): Promise<boolean> {
     observeWorkerOutcome('path_history', 'refresh', 'failure');
     console.error(`[path-history] ${tag} refresh failed`, (err as Error).message);
     return true;
-  } finally {
-    isRunning = false;
   }
 }