From e82d86bfebe34db8757036316b8997162aaa5b10 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:01:15 -0700 Subject: [PATCH] fix(server): gate periodic heartbeat recovery to one chain in flight (#11699) Bind periodic recovery admission to lifecycle tracking. Keep one physical pass active through cleanup, report skipped ticks and long-held ownership, and preserve direct scheduled delivery. Full build, typecheck, 32 focused tests, complete CI, and exact-head Greptile5/5 pass. Co-Authored-By: Senior Engineer Fable Co-Authored-By: Dotta Co-Authored-By: Devin Foley Co-Authored-By: Claude Fable 5 Co-Authored-By: Paperclip --- doc/native-workspace-finalization-recovery.md | 10 ++ server/src/index.ts | 13 +- server/src/lib/single-flight.test.ts | 163 ++++++++++++++++++ server/src/lib/single-flight.ts | 51 ++++++ 4 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 server/src/lib/single-flight.test.ts create mode 100644 server/src/lib/single-flight.ts diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 8b9c004282c..428c32b3fe4 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -1,5 +1,15 @@ # Native workspace finalization ownership and recovery +Periodic heartbeat recovery runs one pass at a time in each server process. +A tick that arrives during a slow pass reports its age and starts no duplicate. +A later tick can start a fresh pass after the current work settles. Shutdown +and idle sleep still track the actual recovery promise through cleanup. Direct +run operations and scheduled work delivery keep their existing paths. A pass +that remains active for three scheduler intervals produces an error log with its +age. A waiting timeout does not admit another pass: the actual work must settle +before ownership is released. A stuck operation requires investigation and +recovery at the operation or provider boundary. + A native provider can fail before it submits a result while its workspace still contains unpublished edits. A prepared workspace reference is not an export receipt. Terminal provider failure pins the exact bound sandbox and records a diff --git a/server/src/index.ts b/server/src/index.ts index 584e9e67697..e83d603e2aa 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -38,6 +38,7 @@ import { chatCompletionDeliveryService } from "./services/chat-completion-delive // instrumentationReady before opening DB connections or constructing the // HTTP server, so trace coverage does not depend on incidental timing. import { instrumentationReady, shutdownInstrumentation } from "./instrumentation.js"; +import { createTrackedSingleFlight } from "./lib/single-flight.js"; import { sentryReady, shutdownSentry, captureException } from "./sentry.js"; import { setSentryRunLinkResolver } from "./sentry-links.js"; import { createSentryRunLinkResolver } from "./services/sentry-run-links.js"; @@ -1209,6 +1210,16 @@ async function startServerWithDatabaseTeardown( }); heartbeatSchedulerInFlight.add(tracked); }; + const runPeriodicRecovery = createTrackedSingleFlight({ + trackWork: trackHeartbeatSchedulerWork, + onSkipped: (inFlightForMs) => { + if (inFlightForMs >= 3 * config.heartbeatSchedulerIntervalMs) { + logger.error({ inFlightForMs }, "periodic heartbeat recovery remains in flight; duplicate pass suppressed"); + } else { + logger.warn({ inFlightForMs }, "skipped periodic heartbeat recovery tick; previous chain still in flight"); + } + }, + }); const waitForHeartbeatSchedulerIdle = async () => { while (heartbeatSchedulerInFlight.size > 0) { await Promise.allSettled([...heartbeatSchedulerInFlight]); @@ -1887,7 +1898,7 @@ async function startServerWithDatabaseTeardown( if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. - trackHeartbeatSchedulerWork(heartbeat + runPeriodicRecovery(() => heartbeat .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { diff --git a/server/src/lib/single-flight.test.ts b/server/src/lib/single-flight.test.ts new file mode 100644 index 00000000000..06dabc92485 --- /dev/null +++ b/server/src/lib/single-flight.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it, vi } from "vitest"; +import { createSingleFlight, createTrackedSingleFlight } from "./single-flight.js"; +import { beginIdleTrackedWork, idleWorkSnapshot } from "../services/task-admission.js"; + +function deferred() { + let resolve!: () => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +describe("createSingleFlight", () => { + it("runs work when idle and reports busy until it settles", async () => { + const gate = createSingleFlight(); + const gap = deferred(); + + const first = gate.run(() => gap.promise); + expect(first.started).toBe(true); + expect(gate.busy).toBe(true); + + gap.resolve(); + await (first as { settled: Promise }).settled; + expect(gate.busy).toBe(false); + }); + + it("skips a tick while the previous invocation is still in flight", async () => { + const clock = { nowMs: 1_000 }; + const gate = createSingleFlight(() => clock.nowMs); + const gap = deferred(); + const work = vi.fn(() => gap.promise); + + const first = gate.run(work); + clock.nowMs = 31_000; + const second = gate.run(work); + + expect(second).toEqual({ started: false, inFlightForMs: 30_000 }); + expect(work).toHaveBeenCalledTimes(1); + + gap.resolve(); + await (first as { settled: Promise }).settled; + }); + + it("admits new work after the previous invocation resolves", async () => { + const gate = createSingleFlight(); + const gap = deferred(); + + const first = gate.run(() => gap.promise); + gap.resolve(); + await (first as { settled: Promise }).settled; + + const second = gate.run(async () => {}); + expect(second.started).toBe(true); + await (second as { settled: Promise }).settled; + }); + + it("releases the gate when work rejects and never rejects the settled promise", async () => { + const gate = createSingleFlight(); + const gap = deferred(); + + const first = gate.run(() => gap.promise); + gap.reject(new Error("chain failed")); + await expect((first as { settled: Promise }).settled).resolves.toBeUndefined(); + expect(gate.busy).toBe(false); + + const second = gate.run(async () => {}); + expect(second.started).toBe(true); + await (second as { settled: Promise }).settled; + }); + + it("propagates a synchronous throw and leaves the gate free", async () => { + const gate = createSingleFlight(); + + expect(() => + gate.run(() => { + throw new Error("sync failure"); + }), + ).toThrow("sync failure"); + expect(gate.busy).toBe(false); + + const second = gate.run(async () => {}); + expect(second.started).toBe(true); + await (second as { settled: Promise }).settled; + }); + + it("keeps the owned pass active after a caller stops waiting", async () => { + const gate = createSingleFlight(), work = deferred(), observer = deferred(); + const first = gate.run(() => work.promise); + if (!first.started) throw new Error("initial pass was not admitted"); + const stopped = Promise.race([first.settled, observer.promise]); + observer.reject(new Error("observer stopped")); + await expect(stopped).rejects.toThrow("observer stopped"); + const duplicate = vi.fn(async () => undefined); + expect(gate.run(duplicate).started).toBe(false); + expect(duplicate).not.toHaveBeenCalled(); + expect(gate.busy).toBe(true); + work.resolve(); + await first.settled; + expect(gate.busy).toBe(false); + }); + + it("keeps cleanup joined after the main operation completes", async () => { + const gate = createSingleFlight(), operation = deferred(), cleanup = deferred(); + const first = gate.run(() => operation.promise.finally(() => cleanup.promise)); + if (!first.started) throw new Error("initial pass was not admitted"); + operation.resolve(); + await Promise.resolve(); + expect(gate.busy).toBe(true); + const duplicate = vi.fn(async () => undefined); + expect(gate.run(duplicate).started).toBe(false); + expect(duplicate).not.toHaveBeenCalled(); + cleanup.resolve(); + await first.settled; + expect(gate.busy).toBe(false); + const next = gate.run(duplicate); + expect(next.started).toBe(true); + if (next.started) await next.settled; + expect(duplicate).toHaveBeenCalledTimes(1); + }); + + it("tracks one periodic pass through scheduler shutdown and idle sleep", async () => { + vi.useFakeTimers(); + const before = idleWorkSnapshot().active; + const operation = deferred(), cleanup = deferred(); + const tracked: Promise[] = []; + const trackWork = vi.fn((work: Promise) => { + const finish = beginIdleTrackedWork(); + tracked.push(work.finally(finish)); + }); + const onSkipped = vi.fn(); + const tick = createTrackedSingleFlight({ trackWork, onSkipped }); + const heartbeat = vi.fn(() => operation.promise.finally(() => cleanup.promise)); + const timer = setInterval(() => tick(heartbeat), 30_000); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(heartbeat).toHaveBeenCalledOnce(); + expect(trackWork).toHaveBeenCalledOnce(); + expect(onSkipped).toHaveBeenCalledExactlyOnceWith(30_000); + clearInterval(timer); + let shutdownComplete = false; + const shutdown = Promise.all(tracked).then(() => { shutdownComplete = true; }); + await Promise.resolve(); + expect(shutdownComplete).toBe(false); + expect(idleWorkSnapshot().active).toBe(before + 1); + operation.resolve(); + await Promise.resolve(); + expect(shutdownComplete).toBe(false); + expect(idleWorkSnapshot().active).toBe(before + 1); + cleanup.resolve(); + await shutdown; + expect(shutdownComplete).toBe(true); + expect(idleWorkSnapshot().active).toBe(before); + } finally { + clearInterval(timer); + operation.resolve(); + cleanup.resolve(); + await Promise.all(tracked); + vi.useRealTimers(); + } + }); +}); diff --git a/server/src/lib/single-flight.ts b/server/src/lib/single-flight.ts new file mode 100644 index 00000000000..68dc1b035f4 --- /dev/null +++ b/server/src/lib/single-flight.ts @@ -0,0 +1,51 @@ +export interface SingleFlightSkip { + started: false; + inFlightForMs: number; +} + +export interface SingleFlightStart { + started: true; + settled: Promise; +} + +export type SingleFlightResult = SingleFlightSkip | SingleFlightStart; + +export function createSingleFlight(now: () => number = Date.now) { + let inFlight: Promise | null = null; + let startedAtMs = 0; + + return { + get busy(): boolean { + return inFlight !== null; + }, + run(work: () => Promise): SingleFlightResult { + if (inFlight) { + return { started: false, inFlightForMs: now() - startedAtMs }; + } + startedAtMs = now(); + const settled = Promise.resolve(work()) + .then( + () => undefined, + () => undefined, + ) + .finally(() => { + inFlight = null; + }); + inFlight = settled; + return { started: true, settled }; + }, + }; +} + +export function createTrackedSingleFlight(options: { + trackWork: (work: Promise) => void; + onSkipped: (inFlightForMs: number) => void; + now?: () => number; +}) { + const gate = createSingleFlight(options.now); + return (work: () => Promise) => { + const tick = gate.run(work); + if (tick.started) options.trackWork(tick.settled); + else options.onSkipped(tick.inFlightForMs); + }; +}