From 13fb33bb5111dbd52b59ef71e52732034d7c6783 Mon Sep 17 00:00:00 2001 From: gadgethd <111318106+gadgethd@users.noreply.github.com> Date: Sun, 9 Aug 2026 16:23:41 +0000 Subject: [PATCH] perf: serialize heavy database workloads --- .../src/analysis/heavyWorkAdmission.test.ts | 53 ++++++++++ backend/src/analysis/heavyWorkAdmission.ts | 96 +++++++++++++++++++ backend/src/api/routes.ts | 2 + backend/src/api/routes/stats.ts | 2 + backend/src/path-learning/rebuild.ts | 40 +++++++- backend/src/platform/config/database.test.ts | 2 +- backend/src/platform/config/database.ts | 2 +- backend/src/stats/statsService.test.ts | 8 ++ backend/src/stats/statsService.ts | 9 +- backend/src/workers/path-history.ts | 20 +++- 10 files changed, 221 insertions(+), 13 deletions(-) create mode 100644 backend/src/analysis/heavyWorkAdmission.test.ts create mode 100644 backend/src/analysis/heavyWorkAdmission.ts diff --git a/backend/src/analysis/heavyWorkAdmission.test.ts b/backend/src/analysis/heavyWorkAdmission.test.ts new file mode 100644 index 0000000..fefc7dd --- /dev/null +++ b/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']); +}); diff --git a/backend/src/analysis/heavyWorkAdmission.ts b/backend/src/analysis/heavyWorkAdmission.ts new file mode 100644 index 0000000..e7084bc --- /dev/null +++ b/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; +type AdmissionPool = Pick; + +type HeavyWorkAdmissionOptions = { + pool: AdmissionPool; + workload: string; + task: () => Promise; + signal?: AbortSignal; + retryMs?: number; + log?: Pick; +}; + +function sleep(ms: number, signal?: AbortSignal): Promise { + 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 { + 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( + options: HeavyWorkAdmissionOptions, +): Promise { + 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( + '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); + } +} diff --git a/backend/src/api/routes.ts b/backend/src/api/routes.ts index e46ce9c..52b0a69 100644 --- a/backend/src/api/routes.ts +++ b/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 }); diff --git a/backend/src/api/routes/stats.ts b/backend/src/api/routes/stats.ts index 32e9c46..f9c16df 100644 --- a/backend/src/api/routes/stats.ts +++ b/backend/src/api/routes/stats.ts @@ -49,6 +49,7 @@ type StatsRouteDeps = { analyticsQuery: QueryFn; getPublicVisibilityGeneration: () => Promise; maskDecodedPathNodes: MaskDecodedPathNodesFn; + runHeavyWork: (workload: string, task: () => Promise) => Promise; }; 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); diff --git a/backend/src/path-learning/rebuild.ts b/backend/src/path-learning/rebuild.ts index d9e5bd7..22697bb 100644 --- a/backend/src/path-learning/rebuild.ts +++ b/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 { 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( + const packetsResult = await analyticsQuery( `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( + const nodesResult = await analyticsQuery( `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( + const linksResult = await analyticsQuery( `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 diff --git a/backend/src/platform/config/database.test.ts b/backend/src/platform/config/database.test.ts index 9a5a06b..ce7cfa6 100644 --- a/backend/src/platform/config/database.test.ts +++ b/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); }); diff --git a/backend/src/platform/config/database.ts b/backend/src/platform/config/database.ts index 7e0de71..98415fa 100644 --- a/backend/src/platform/config/database.ts +++ b/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); diff --git a/backend/src/stats/statsService.test.ts b/backend/src/stats/statsService.test.ts index bc25b29..5c7f032 100644 --- a/backend/src/stats/statsService.test.ts +++ b/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 () => { diff --git a/backend/src/stats/statsService.ts b/backend/src/stats/statsService.ts index 3357016..81a00a6 100644 --- a/backend/src/stats/statsService.ts +++ b/backend/src/stats/statsService.ts @@ -33,6 +33,7 @@ type StatsServiceDeps = { repository: StatsRepository; getPublicVisibilityGeneration: () => Promise; maskDecodedPathNodes: MaskDecodedPathNodesFn; + runHeavyWork?: (workload: string, task: () => Promise) => Promise; }; 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 = { @@ -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, diff --git a/backend/src/workers/path-history.ts b/backend/src/workers/path-history.ts index 331f110..abe4727 100644 --- a/backend/src/workers/path-history.ts +++ b/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 { 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 { try { let publicEvidenceUpdates = 0; let testEvidenceUpdates = 0; @@ -430,8 +448,6 @@ async function refreshAll(tag: 'initial' | 'scheduled'): Promise { observeWorkerOutcome('path_history', 'refresh', 'failure'); console.error(`[path-history] ${tag} refresh failed`, (err as Error).message); return true; - } finally { - isRunning = false; } }