fix(heartbeat): single-flight the periodic recovery chain (BLO-30203)

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<T>() {
+  let resolve!: (value: T) => void;
+  let reject!: (err: unknown) => void;
+  const promise = new Promise<T>((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<string>();
+    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<string>();
+
+    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<string>();
+
+    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<string>();
+
+    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;
+  });
+});

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<StartedServer> {
   // 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<Promise<void>>();
   const trackHeartbeatSchedulerWork = (work: Promise<unknown>) => {
     let tracked: Promise<void>;
@@ -1604,7 +1618,11 @@ export async function startServer(): Promise<StartedServer> {
 
           // 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<StartedServer> {
             .catch((err) => {
               logger.error({ err }, "periodic heartbeat recovery failed");
             }));
+          if (recoveryPass) trackHeartbeatSchedulerWork(recoveryPass);
         }
       })();
     }, config.heartbeatSchedulerIntervalMs);

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(): <T>(task: () => Promise<T>) => Promise<T> | null {
+  let inFlight = false;
+  return <T>(task: () => Promise<T>): Promise<T> | null => {
+    if (inFlight) return null;
+    inFlight = true;
+    let started: Promise<T>;
+    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;
+    });
+  };
+}