From 20501d11f785a1226592438c8b65ffbdebfafafc Mon Sep 17 00:00:00 2001 From: CTO Date: Mon, 14 Sep 2026 06:33:38 +0000 Subject: [PATCH] fix(heartbeat): single-flight the periodic recovery chain (BLO-30203) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker tier climbs to its 6 GiB --max-old-space-size ceiling and SIGABRTs (exit 134). The API tier runs the same image with the same Node flags and is memory-flat, so the retainer is in worker-only code. setInterval fires the scheduler tick every 30 s and does not wait for the previous callback. The recovery chain registered in that tick had no single-flight latch, so copies overlap. Each copy retains its own hydrated snapshot of the issue graph: reconcileStrandedAssignedIssues selects every stranded issue with no LIMIT, and collectIssueGraphLiveness scans the whole issues table twice per pass, once including `description`. The file already carries this exact argument for crashReconcileSweepInFlight (index.ts:1069-1081) — it was simply never applied to this chain. Skipping a tick is safe: every stage is an idempotent sweep, so a skipped tick reconciles on the next one. --- server/src/__tests__/single-flight.test.ts | 104 +++++++++++++++++++++ server/src/index.ts | 21 ++++- server/src/services/single-flight.ts | 33 +++++++ 3 files changed, 157 insertions(+), 1 deletion(-) create mode 100644 server/src/__tests__/single-flight.test.ts create mode 100644 server/src/services/single-flight.ts diff --git a/server/src/__tests__/single-flight.test.ts b/server/src/__tests__/single-flight.test.ts new file mode 100644 index 000000000000..4905e4d2ccf0 --- /dev/null +++ b/server/src/__tests__/single-flight.test.ts @@ -0,0 +1,104 @@ +// Tests for the single-flight latch that stops the periodic heartbeat +// recovery chain from running concurrently with itself (BLO-30203). +// +// The failure this guards against is not an incorrect result — every stage of +// the chain is an idempotent sweep — it is retention. `setInterval` fires every +// 30 s and the chain routinely takes longer, so without the latch N copies are +// alive at once and each holds its own full hydrated snapshot of the issue +// graph. That is what drove the worker into its 6 GiB heap ceiling. + +import { describe, it, expect } from "vitest"; +import { createSingleFlight } from "../services/single-flight.js"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +describe("createSingleFlight", () => { + it("declines a second start while the first is still settling", async () => { + const runOnce = createSingleFlight(); + const gate = deferred(); + let starts = 0; + + const first = runOnce(() => { + starts += 1; + return gate.promise; + }); + const second = runOnce(() => { + starts += 1; + return Promise.resolve("second"); + }); + + // The declined call must not invoke the task at all — invoking it and + // discarding the promise would still do the DB work and still retain the + // snapshot, which is the entire cost this latch exists to avoid. + expect(starts).toBe(1); + expect(first).not.toBeNull(); + expect(second).toBeNull(); + + gate.resolve("first"); + await expect(first).resolves.toBe("first"); + }); + + it("re-opens once the in-flight pass settles", async () => { + const runOnce = createSingleFlight(); + const gate = deferred(); + + const first = runOnce(() => gate.promise); + expect(runOnce(() => Promise.resolve("blocked"))).toBeNull(); + + gate.resolve("first"); + await first; + + const third = runOnce(() => Promise.resolve("third")); + expect(third).not.toBeNull(); + await expect(third).resolves.toBe("third"); + }); + + it("re-opens when the pass rejects, so one failure cannot wedge the sweep", async () => { + const runOnce = createSingleFlight(); + const gate = deferred(); + + const first = runOnce(() => gate.promise); + gate.reject(new Error("pass failed")); + await expect(first).rejects.toThrow("pass failed"); + + const next = runOnce(() => Promise.resolve("recovered")); + expect(next).not.toBeNull(); + await expect(next).resolves.toBe("recovered"); + }); + + it("re-opens when the task throws synchronously", async () => { + const runOnce = createSingleFlight(); + + expect(() => + runOnce(() => { + throw new Error("sync boom"); + }), + ).toThrow("sync boom"); + + const next = runOnce(() => Promise.resolve("recovered")); + expect(next).not.toBeNull(); + await expect(next).resolves.toBe("recovered"); + }); + + it("keeps separate latches independent", async () => { + const runA = createSingleFlight(); + const runB = createSingleFlight(); + const gate = deferred(); + + const a = runA(() => gate.promise); + expect(runA(() => Promise.resolve("a2"))).toBeNull(); + // A different sweep must not be blocked by A's in-flight pass. + expect(runB(() => Promise.resolve("b1"))).not.toBeNull(); + + gate.resolve("a1"); + await a; + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 966a343514dd..6d86dd7432f4 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -68,6 +68,7 @@ import { summarizeStrandedRecoveryHandBackPass } from "./services/recovery/servi import { buildRuntimeApiCandidateUrls, choosePrimaryRuntimeApiUrl } from "./runtime-api.js"; import { createPluginWorkerManager } from "./services/plugin-worker-manager.js"; import { createApiTierPluginWorkerManagerStub } from "./services/plugin-worker-manager-stub.js"; +import { createSingleFlight } from "./services/single-flight.js"; import { createStorageServiceFromConfig } from "./storage/index.js"; import { printStartupBanner } from "./startup-banner.js"; import { getBoardClaimWarningUrl, initializeBoardClaimChallenge } from "./board-claim.js"; @@ -1080,6 +1081,19 @@ export async function startServer(): Promise { // crashed run and handing that lock to the retry — exactly the interleaving // the in-tick `await` was added to remove. let crashReconcileSweepInFlight = false; + // Single-flight latch for the periodic recovery chain (BLO-30203). Same + // argument as the latch above, which was only ever applied to the + // crash-reconcile pair: `setInterval` does not wait for the previous + // callback, and this chain is far slower than one tick. Its first stage + // alone (`reconcileStrandedAssignedIssues`) selects every stranded issue + // with no LIMIT and then issues ~5 awaited queries per candidate, and the + // liveness classifier it calls scans the whole `issues` table twice per + // pass. Overlapping copies therefore each retain a full hydrated snapshot + // of the issue graph, which is how the worker reaches the 6 GiB + // --max-old-space-size ceiling and SIGABRTs. Skipping a tick while the + // previous pass is still running is safe: every stage is an idempotent + // sweep, so a skipped tick reconciles on the next one. + const runRecoveryChainOnce = createSingleFlight(); const heartbeatSchedulerInFlight = new Set>(); const trackHeartbeatSchedulerWork = (work: Promise) => { let tracked: Promise; @@ -1604,7 +1618,11 @@ export async function startServer(): Promise { // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. - trackHeartbeatSchedulerWork(heartbeat + // BLO-30203: single-flighted. Still registered with + // trackHeartbeatSchedulerWork when it actually starts, so shutdown + // drains it, and deliberately NOT awaited by the passes above so + // they keep their own cadence. + const recoveryPass = runRecoveryChainOnce(() => heartbeat .resumeRunningExternalRuntimeRuns() .then(() => heartbeat.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 })) .then(() => heartbeat.promoteDueScheduledRetries()) @@ -1700,6 +1718,7 @@ export async function startServer(): Promise { .catch((err) => { logger.error({ err }, "periodic heartbeat recovery failed"); })); + if (recoveryPass) trackHeartbeatSchedulerWork(recoveryPass); } })(); }, config.heartbeatSchedulerIntervalMs); diff --git a/server/src/services/single-flight.ts b/server/src/services/single-flight.ts new file mode 100644 index 000000000000..a98051ca87c2 --- /dev/null +++ b/server/src/services/single-flight.ts @@ -0,0 +1,33 @@ +// Single-flight latch for periodic sweeps driven by `setInterval` (BLO-30203). +// +// `setInterval` starts the next callback on schedule regardless of whether the +// previous one has settled, so a sweep that outlives its own interval runs +// concurrently with itself and each overlapping copy retains its own working +// set. For the heartbeat recovery chain that working set is a full hydrated +// snapshot of the issue graph, which is how the worker walked up to its 6 GiB +// --max-old-space-size ceiling and SIGABRTed roughly every two days. +// +// `createSingleFlight()` returns a runner that starts the task only when no +// previous invocation is still settling, and returns `null` when it declines. +// Callers that need to observe the pass (to drain it on shutdown, say) should +// branch on that null rather than assuming a promise. +export function createSingleFlight(): (task: () => Promise) => Promise | null { + let inFlight = false; + return (task: () => Promise): Promise | null => { + if (inFlight) return null; + inFlight = true; + let started: Promise; + try { + started = task(); + } catch (err) { + // A task that throws synchronously never produces a promise to settle, + // so release here or the latch would be stuck closed for the process + // lifetime — a silent, permanent stall of the sweep. + inFlight = false; + throw err; + } + return started.finally(() => { + inFlight = false; + }); + }; +}