diff --git a/src/cli/index.ts b/src/cli/index.ts index 40ad1dca3d..56abb5d05a 100755 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -363,11 +363,12 @@ async function handleStart(options: { block?: boolean } = {}) { shutdownStartedAt = now; console.log("\nπŸ›‘ Shutting down opencodex proxy..."); void (async () => { + let shutdownSucceeded = false; try { - await drainAndShutdown(server, config.shutdownTimeoutMs ?? 5000); + shutdownSucceeded = await drainAndShutdown(server, config.shutdownTimeoutMs ?? 5000); } finally { const restored = syncCleanup(); // idempotent (cleaned-guard); also re-run by process.on("exit") - process.exit(restored ? 0 : 1); + process.exit(restored && shutdownSucceeded ? 0 : 1); } })(); }; diff --git a/src/config/paths.ts b/src/config/paths.ts index b8b494ecd7..4c351a6ae8 100644 --- a/src/config/paths.ts +++ b/src/config/paths.ts @@ -1,7 +1,7 @@ import { chmodSync, existsSync } from "node:fs"; import { homedir } from "node:os"; import { join, resolve } from "node:path"; -import { hardenSecretDir } from "../lib/windows-secret-acl"; +import { hardenSecretDirAsync, windowsSecretAclApplies } from "../lib/windows-secret-acl"; import { assertNotRealHomeUnderTest } from "../lib/test-home-guard"; /** @@ -14,6 +14,7 @@ export function expandUserPath(raw: string): string { return raw; } let resolvedConfigDirCache: { raw: string | undefined; path: string } | null = null; +const configDirHardeningFlights = new Map>(); export function getConfigDir(): string { const raw = process.env["OPENCODEX_HOME"]?.trim() || undefined; @@ -34,7 +35,21 @@ export function hardenConfigDir(): void { assertNotRealHomeUnderTest(dir); if (!existsSync(dir)) return; try { chmodSync(dir, 0o700); } catch { /* best-effort */ } - if (process.platform === "win32") { - hardenSecretDir(dir, { required: false }); + if (windowsSecretAclApplies() && !configDirHardeningFlights.has(dir)) { + // This is an optional read-path harden. Waiting synchronously here used to stop the Bun + // event loop (including /healthz) for the full icacls timeout. Required mutation paths keep + // their own awaited/fail-closed hardening; ordinary config reads only start one soft flight. + const flight = hardenSecretDirAsync(dir, { required: false }) + .then(() => undefined) + .catch(() => undefined) + .finally(() => { + if (configDirHardeningFlights.get(dir) === flight) configDirHardeningFlights.delete(dir); + }); + configDirHardeningFlights.set(dir, flight); } } + +/** Test-only: settle optional config-directory hardening without exposing it to production callers. */ +export async function flushConfigDirHardeningForTests(): Promise { + await Promise.all([...configDirHardeningFlights.values()]); +} diff --git a/src/lib/bounded-subprocess.ts b/src/lib/bounded-subprocess.ts new file mode 100644 index 0000000000..543a74ab40 --- /dev/null +++ b/src/lib/bounded-subprocess.ts @@ -0,0 +1,36 @@ +export interface KillableSubprocess { + exited: Promise; + kill(): unknown; + unref?(): unknown; +} + +export interface BoundedSubprocessExit { + exitCode: number | null; + timedOut: boolean; +} + +/** Kill at the deadline and abandon immediately; late exit/rejection remains observed. */ +export function waitForSubprocessExit( + proc: KillableSubprocess, + timeoutMs: number, +): Promise { + return new Promise(resolve => { + let settled = false; + let timer: ReturnType | undefined; + const finish = (result: BoundedSubprocessExit): void => { + if (settled) return; + settled = true; + if (timer !== undefined) clearTimeout(timer); + resolve(result); + }; + timer = setTimeout(() => { + try { proc.kill(); } catch { /* already exited */ } + try { proc.unref?.(); } catch { /* abandonment is still authoritative */ } + finish({ exitCode: null, timedOut: true }); + }, Math.max(1, timeoutMs)); + void proc.exited.then( + exitCode => finish({ exitCode, timedOut: false }), + () => finish({ exitCode: null, timedOut: false }), + ); + }); +} diff --git a/src/lib/windows-secret-acl.ts b/src/lib/windows-secret-acl.ts index 04ecb25ea0..dbf1f06b79 100644 --- a/src/lib/windows-secret-acl.ts +++ b/src/lib/windows-secret-acl.ts @@ -31,6 +31,7 @@ import { existsSync, statSync } from "node:fs"; import { env, platform } from "node:process"; +import { waitForSubprocessExit } from "./bounded-subprocess"; import { resolveTrustedWindowsIcaclsExe } from "./windows-elevation"; import { cachedCurrentWindowsIdentity, @@ -216,6 +217,11 @@ export interface HardenResult { export interface HardenOptions { required: boolean; + /** + * Explicit total budget for this harden call. Shutdown recovery uses a reduced + * caller-owned slice instead of opening the normal 30-second window. + */ + deadlineMs?: number; /** * Optional timeout-memo key distinct from `targetPath` (issue #612). * Atomic writers mint a fresh `.tmp` path per write; keying the timeout cache by the @@ -257,7 +263,11 @@ const HARDEN_DEADLINE_MIN_MS = 1_000; const HARDEN_DEADLINE_MAX_MS = 60_000; /** Resolve the total harden budget once per call (env mutation cannot change it midway). */ -function resolveHardenDeadlineMs(): number { +function resolveHardenDeadlineMs(overrideMs?: number): number { + if (overrideMs !== undefined) { + if (!Number.isSafeInteger(overrideMs) || overrideMs <= 0) return 1; + return Math.min(HARDEN_DEADLINE_MAX_MS, overrideMs); + } const raw = env["OPENCODEX_ACL_TIMEOUT_MS"]?.trim(); if (!raw) return HARDEN_DEADLINE_DEFAULT_MS; const parsed = Number(raw); @@ -328,27 +338,16 @@ function defaultIcaclsRunner(args: string[], timeoutMs: number): IcaclsResult { /** * Async icacls runner (#612): yields the event loop while waiting for the child. - * Timeout provenance is recorded by our timer (async Subprocess has no exitedDueToTimeout); - * we still await process exit before classifying so settlement is confirmed. + * Async Subprocess has no exitedDueToTimeout, so the shared settlement helper + * classifies the deadline and abandons a child that does not settle after kill. */ async function defaultAsyncIcaclsRunner(args: string[], timeoutMs: number): Promise { const proc = trySpawnIcacls(args); if (!proc) return spawnFailedResult(); - let timedOutByUs = false; - const timer = setTimeout(() => { - timedOutByUs = true; - try { proc.kill(); } catch { /* already exited */ } - }, Math.max(1, timeoutMs)); - let exitCode: number | null = null; - try { - exitCode = await proc.exited; - } finally { - clearTimeout(timer); - } - const stdout = proc.stdout + const { exitCode, timedOut } = await waitForSubprocessExit(proc, timeoutMs); + const stdout = !timedOut && proc.stdout ? await new Response(proc.stdout).text().catch(() => "") : ""; - const timedOut = timedOutByUs; return { success: !timedOut && exitCode === 0, exitCode: timedOut ? null : exitCode, @@ -357,6 +356,24 @@ async function defaultAsyncIcaclsRunner(args: string[], timeoutMs: number): Prom }; } +function awaitAsyncIcaclsRunner(args: string[], timeoutMs: number): Promise { + return new Promise(resolve => { + let settled = false; + let timer: ReturnType | undefined; + const finish = (result: IcaclsResult): void => { + if (settled) return; + settled = true; + if (timer !== undefined) clearTimeout(timer); + resolve(result); + }; + timer = setTimeout( + () => finish({ success: false, exitCode: null, timedOut: true, stdout: "" }), + Math.max(1, timeoutMs), + ); + void asyncIcaclsRunner(args, timeoutMs).then(finish, () => finish(spawnFailedResult())); + }); +} + let icaclsRunner: IcaclsRunner = defaultIcaclsRunner; let asyncIcaclsRunner: AsyncIcaclsRunner = defaultAsyncIcaclsRunner; let platformOverride: string | null = null; @@ -537,12 +554,14 @@ function shouldVerifyExistingAcl(): boolean { return env["OPENCODEX_ACL_VERIFY_EXISTING"] === "1"; } -function existingAclAlreadyCompliant(targetPath: string, directory: boolean): boolean { +function existingAclAlreadyCompliant(targetPath: string, directory: boolean, deadline: number): boolean { if (!shouldVerifyExistingAcl()) return false; const identity = cachedCurrentWindowsIdentity(); if (!identity) return false; try { - const result = icaclsRunner([targetPath], resolveHardenDeadlineMs()); + const remaining = deadline - nowFn(); + if (remaining <= 0) return false; + const result = icaclsRunner([targetPath], remaining); return result.success && existingAclIsCompliant(targetPath, directory, result.stdout, identity.name); } catch { return false; @@ -552,12 +571,15 @@ function existingAclAlreadyCompliant(targetPath: string, directory: boolean): bo async function existingAclAlreadyCompliantAsync( targetPath: string, directory: boolean, + deadline: number, ): Promise { if (!shouldVerifyExistingAcl()) return false; const identity = cachedCurrentWindowsIdentity(); if (!identity) return false; try { - const result = await asyncIcaclsRunner([targetPath], resolveHardenDeadlineMs()); + const remaining = deadline - nowFn(); + if (remaining <= 0) return false; + const result = await awaitAsyncIcaclsRunner([targetPath], remaining); return result.success && existingAclIsCompliant(targetPath, directory, result.stdout, identity.name); } catch { return false; @@ -618,7 +640,7 @@ async function runIcaclsAsync(targetPath: string, directory: boolean, deadline: if (remaining <= 0) { throw icaclsError(step, { success: false, exitCode: null, timedOut: true, stdout: "" }); } - return asyncIcaclsRunner(args, remaining); + return awaitAsyncIcaclsRunner(args, remaining); }; const runOrThrow = async (step: string, args: string[]): Promise => { const result = await run(step, args); @@ -751,7 +773,7 @@ async function describeAclStateAfterTimeoutAsync(targetPath: string, deadline: n for (const sid of BROAD_SIDS) { const remaining = deadline - nowFn(); if (remaining <= 0) return "ACL state unverified (budget exhausted)"; - const found = await asyncIcaclsRunner([targetPath, "/findsid", sid], remaining); + const found = await awaitAsyncIcaclsRunner([targetPath, "/findsid", sid], remaining); if (!found.success) return "ACL state unverified (probe failed)"; if (found.stdout.includes(targetPath)) return "broad ACL grants still present"; } @@ -788,7 +810,8 @@ function hardenEntry( if (!existsSync(targetPath)) { cache.delete(targetPath); return { ok: true }; } if (effectivePlatform() !== "win32") return { ok: true }; if (memoSatisfied(cache, targetPath)) return { ok: true }; - if (existingAclAlreadyCompliant(targetPath, directory)) return { ok: true }; + const deadline = nowFn() + resolveHardenDeadlineMs(opts.deadlineMs); + if (existingAclAlreadyCompliant(targetPath, directory, deadline)) return { ok: true }; const memoKey = timeoutMemoKey(targetPath, opts); const timeoutMemoError = timeoutMemoErrorIfBlocked(memoKey, opts); if (timeoutMemoError) { @@ -796,7 +819,6 @@ function hardenEntry( return { ok: false, diagnostics: timeoutMemoError.message }; } - const deadline = nowFn() + resolveHardenDeadlineMs(); let lastErr: unknown; for (let attempt = 0; attempt < 2; attempt++) { if (attempt > 0 && deadline - nowFn() <= 0) break; // retry only while budget remains @@ -841,7 +863,8 @@ async function hardenEntryAsync( if (!existsSync(targetPath)) { cache.delete(targetPath); return { ok: true }; } if (effectivePlatform() !== "win32") return { ok: true }; if (memoSatisfied(cache, targetPath)) return { ok: true }; - if (await existingAclAlreadyCompliantAsync(targetPath, directory)) return { ok: true }; + const deadline = nowFn() + resolveHardenDeadlineMs(opts.deadlineMs); + if (await existingAclAlreadyCompliantAsync(targetPath, directory, deadline)) return { ok: true }; const memoKey = timeoutMemoKey(targetPath, opts); const timeoutMemoError = timeoutMemoErrorIfBlocked(memoKey, opts); if (timeoutMemoError) { @@ -849,7 +872,6 @@ async function hardenEntryAsync( return { ok: false, diagnostics: timeoutMemoError.message }; } - const deadline = nowFn() + resolveHardenDeadlineMs(); let lastErr: unknown; for (let attempt = 0; attempt < 2; attempt++) { if (attempt > 0 && deadline - nowFn() <= 0) break; diff --git a/src/lib/windows-user-principal.ts b/src/lib/windows-user-principal.ts index 1064ec88cc..69083218c0 100644 --- a/src/lib/windows-user-principal.ts +++ b/src/lib/windows-user-principal.ts @@ -22,6 +22,7 @@ import { existsSync } from "node:fs"; import { win32 as windowsPath } from "node:path"; +import { waitForSubprocessExit } from "./bounded-subprocess"; import { resolveTrustedWindowsPowerShellExe, @@ -143,18 +144,8 @@ async function defaultAsyncWindowsPrincipalRunner( stderr: "ignore", windowsHide: true, }); - let timedOut = false; - const timer = setTimeout(() => { - timedOut = true; - try { proc.kill(); } catch { /* already exited */ } - }, Math.max(1, timeoutMs)); - let exitCode: number | null = null; - try { - exitCode = await proc.exited; - } finally { - clearTimeout(timer); - } - const stdout = proc.stdout + const { exitCode, timedOut } = await waitForSubprocessExit(proc, timeoutMs); + const stdout = !timedOut && proc.stdout ? await new Response(proc.stdout).text().catch(() => "") : ""; return { @@ -314,11 +305,11 @@ export async function resolveCurrentWindowsPrincipalAsync(timeoutMs: number): Pr return `*${cachedIdentity.sid}`; })(); asyncLookupInFlight = lookup; - try { - return await lookup; - } finally { - if (asyncLookupInFlight === lookup) asyncLookupInFlight = null; - } + void lookup.then( + () => { if (asyncLookupInFlight === lookup) asyncLookupInFlight = null; }, + () => { if (asyncLookupInFlight === lookup) asyncLookupInFlight = null; }, + ); + return waitForExistingLookup(lookup, timeoutMs); } /** Test seam: replace the sync resolver process and clear its successful cache. */ diff --git a/src/responses/spill-store.ts b/src/responses/spill-store.ts index 825a3da3a3..f61d475c5a 100644 --- a/src/responses/spill-store.ts +++ b/src/responses/spill-store.ts @@ -15,9 +15,17 @@ import { writeSync, } from "node:fs"; import { createHash, randomBytes } from "node:crypto"; -import { join } from "node:path"; +import { dirname, join } from "node:path"; import { getConfigDir } from "../config"; -import { forgetEphemeralSecretPath, forgetHardenedSecretPath, hardenSecretDir, hardenSecretPath } from "../lib/windows-secret-acl"; +import { + forgetEphemeralSecretPath, + forgetHardenedSecretPath, + hardenSecretDir, + hardenSecretDirAsync, + hardenSecretPath, + hardenSecretPathAsync, + windowsSecretAclApplies, +} from "../lib/windows-secret-acl"; import { isValidProviderContinuationOwner } from "./provider-continuation"; import type { OcxProviderContinuationState } from "../types"; @@ -110,11 +118,47 @@ export interface ResponseSpillIoForTest { let spillIoForTest: ResponseSpillIoForTest | null = null; let spillGeneration = 0; +let spillNowOverride: (() => number) | null = null; + +interface ResponseSpillWriteOptions { + retryTimedOutOnce?: boolean; + /** Total caller-owned ACL budget shared by every harden in this publication. */ + aclBudgetMs?: number; + publicationControl?: ResponseSpillPublicationControl; +} + +export interface ResponseSpillPublicationControl { + superseded: boolean; + tempPath: string | null; + destinationPath: string | null; +} + +interface SpillAclBudget { + deadline: number; + perCallMs: number; +} + +export function createResponseSpillPublicationControl(): ResponseSpillPublicationControl { + return { superseded: false, tempPath: null, destinationPath: null }; +} + +export function markResponseSpillPublicationSuperseded(control: ResponseSpillPublicationControl): void { + control.superseded = true; +} export function setSpillIoForTest(io: ResponseSpillIoForTest | null): void { spillIoForTest = io; } +/** Test-only: inject the spill deadline clock. */ +export function setResponseSpillNowForTests(now: (() => number) | null): void { + spillNowOverride = now; +} + +function spillNow(): number { + return spillNowOverride?.() ?? Date.now(); +} + function record(event: "write" | "fsync" | "close" | "harden" | "publish" | "dir-fsync" | "stub-swap"): void { spillIoForTest?.record?.(event); } @@ -175,16 +219,62 @@ function canUseExclusiveCopyFallback(error: unknown): boolean { .some(code => isErrno(error, code)); } -function harden(path: string, mode: number): void { +function spillAclBudget(totalMs: number | undefined): SpillAclBudget | undefined { + if (totalMs === undefined) return undefined; + const bounded = Math.max(1, Math.floor(totalMs)); + return { deadline: spillNow() + bounded, perCallMs: Math.max(1, Math.floor(bounded / 2)) }; +} + +function nextSpillHardenDeadlineMs(budget: SpillAclBudget | undefined): number | undefined { + if (!budget) return undefined; + const remaining = budget.deadline - spillNow(); + if (remaining <= 0) { + throw Object.assign(new Error("Response spill ACL budget exhausted"), { code: "ETIMEDOUT" }); + } + return Math.min(budget.perCallMs, remaining); +} + +function harden(path: string, mode: number, budget?: SpillAclBudget): void { + const aclApplies = budget ? windowsSecretAclApplies() : process.platform === "win32"; + try { + chmodSync(path, mode); + } catch { + if (!aclApplies) throw new Error("Response spill permission hardening failed"); + } + if (aclApplies) { + const deadlineMs = nextSpillHardenDeadlineMs(budget); + const options = { + required: true, + ...(deadlineMs !== undefined ? { deadlineMs } : {}), + }; + const result = mode === 0o700 + ? hardenSecretDir(path, options) + : hardenSecretPath(path, options); + if (!result.ok) throw new Error("Response spill permission hardening failed"); + } +} + +async function hardenAsync( + path: string, + mode: number, + budget: SpillAclBudget, + retryTimedOutOnce = false, +): Promise { try { chmodSync(path, mode); } catch { - if (process.platform !== "win32") throw new Error("Response spill permission hardening failed"); + if (!windowsSecretAclApplies()) throw new Error("Response spill permission hardening failed"); } - if (process.platform === "win32") { + if (windowsSecretAclApplies()) { + const deadlineMs = nextSpillHardenDeadlineMs(budget); + const options = { + required: true, + retryTimedOutOnce, + ...(deadlineMs !== undefined ? { deadlineMs } : {}), + }; const result = mode === 0o700 - ? hardenSecretDir(path, { required: true }) - : hardenSecretPath(path, { required: true }); + ? await hardenSecretDirAsync(path, options) + : await hardenSecretPathAsync(path, options); if (!result.ok) throw new Error("Response spill permission hardening failed"); } } @@ -231,7 +321,94 @@ function unlinkEphemeral(path: string): void { unlink(path, true); } -function publishNoReplace(tempPath: string, destinationPath: string): void { +function supersededPublicationError(): NodeJS.ErrnoException { + return Object.assign(new Error("Response spill publication superseded"), { code: "ECANCELED" }); +} + +function throwIfPublicationSuperseded(control: ResponseSpillPublicationControl | undefined): void { + if (control?.superseded) throw supersededPublicationError(); +} + +function clearOwnedPath( + control: ResponseSpillPublicationControl, + key: "tempPath" | "destinationPath", + ephemeral: boolean, +): unknown { + const path = control[key]; + if (!path) return null; + try { + if (ephemeral) unlinkEphemeral(path); + else unlink(path); + control[key] = null; + return null; + } catch (error) { + if (isErrno(error, "ENOENT")) { + control[key] = null; + return null; + } + return error; + } +} + +/** Claim and remove every path still owned by an abandoned async publication. */ +export function cleanupSupersededResponseSpillPublication( + control: ResponseSpillPublicationControl, +): Error | null { + control.superseded = true; + const ownedDir = control.destinationPath + ? dirname(control.destinationPath) + : control.tempPath + ? dirname(control.tempPath) + : null; + const destinationError = clearOwnedPath(control, "destinationPath", false); + const tempError = clearOwnedPath(control, "tempPath", true); + if (ownedDir) fsyncDirectoryBestEffort(ownedDir); + const cleanupError = destinationError ?? tempError; + return cleanupError ? responseSpillWriteError(cleanupError) : null; +} + +function publishNoReplace( + tempPath: string, + destinationPath: string, + budget?: SpillAclBudget, +): void { + try { + if (spillIoForTest?.link) spillIoForTest.link(tempPath, destinationPath); + else linkSync(tempPath, destinationPath); + } catch (error) { + if (isErrno(error, "EEXIST")) throw error; + if (!canUseExclusiveCopyFallback(error)) throw error; + let copied = false; + try { + if (spillIoForTest?.copyFileExcl) spillIoForTest.copyFileExcl(tempPath, destinationPath); + else copyFileSync(tempPath, destinationPath, constants.COPYFILE_EXCL); + copied = true; + harden(destinationPath, 0o600, budget); + const copyFd = openSync(destinationPath, "r"); + try { + if (spillIoForTest?.fsync) spillIoForTest.fsync(copyFd); + else fsyncSync(copyFd); + } finally { + closeSync(copyFd); + } + } catch (copyError) { + if (copied) { + try { unlink(destinationPath); } catch { /* startup GC reclaims an incomplete publication */ } + } + throw copyError; + } + } + record("publish"); +} + +async function publishNoReplaceAsync( + tempPath: string, + destinationPath: string, + budget: SpillAclBudget, + retryTimedOutOnce: boolean, + publicationControl?: ResponseSpillPublicationControl, +): Promise { + throwIfPublicationSuperseded(publicationControl); try { if (spillIoForTest?.link) spillIoForTest.link(tempPath, destinationPath); else linkSync(tempPath, destinationPath); @@ -243,7 +420,8 @@ function publishNoReplace(tempPath: string, destinationPath: string): void { if (spillIoForTest?.copyFileExcl) spillIoForTest.copyFileExcl(tempPath, destinationPath); else copyFileSync(tempPath, destinationPath, constants.COPYFILE_EXCL); copied = true; - harden(destinationPath, 0o600); + await hardenAsync(destinationPath, 0o600, budget, retryTimedOutOnce); + throwIfPublicationSuperseded(publicationControl); const copyFd = openSync(destinationPath, "r"); try { if (spillIoForTest?.fsync) spillIoForTest.fsync(copyFd); @@ -258,9 +436,48 @@ function publishNoReplace(tempPath: string, destinationPath: string): void { throw copyError; } } + throwIfPublicationSuperseded(publicationControl); record("publish"); } +function serializedSpill( + responseId: string, + state: Omit, +): { + bytes: Buffer; + digest: string; + idDigest: string; + contentDigest: string; +} { + const payload: ResponseSpillPayload = { + version: 1, + responseId, + createdAt: state.createdAt, + ...(state.clientThreadId ? { clientThreadId: state.clientThreadId } : {}), + items: state.items, + ...(state.providerOutputStart !== undefined ? { providerOutputStart: state.providerOutputStart } : {}), + ...(state.providers ? { providers: state.providers } : {}), + }; + const serialized = JSON.stringify(payload); + if (serialized === undefined) throw new Error("Response spill serialization failed"); + const bytes = Buffer.from(serialized, "utf8"); + const digest = sha256(bytes); + return { + bytes, + digest, + idDigest: sha256(responseId).slice(0, 12), + contentDigest: digest.slice(0, 24), + }; +} + +function responseSpillWriteError(cause: unknown): NodeJS.ErrnoException { + const error = new Error("Response spill write failed", { cause }) as NodeJS.ErrnoException; + if (cause && typeof cause === "object" && "code" in cause) { + error.code = String((cause as { code?: unknown }).code); + } + return error; +} + function validSpillRef(ref: ResponseSpillRef): boolean { return ref.version === 1 && OWNED_SPILL_NAME.test(ref.fileName) @@ -302,28 +519,16 @@ function validPayload(value: unknown, responseId: string): value is ResponseSpil export function writeResponseSpillDurably( responseId: string, state: Omit, + options: ResponseSpillWriteOptions = {}, ): ResponseSpillRef { let tempPath: string | null = null; let fd: number | null = null; try { - const payload: ResponseSpillPayload = { - version: 1, - responseId, - createdAt: state.createdAt, - ...(state.clientThreadId ? { clientThreadId: state.clientThreadId } : {}), - items: state.items, - ...(state.providerOutputStart !== undefined ? { providerOutputStart: state.providerOutputStart } : {}), - ...(state.providers ? { providers: state.providers } : {}), - }; - const serialized = JSON.stringify(payload); - if (serialized === undefined) throw new Error("Response spill serialization failed"); - const bytes = Buffer.from(serialized, "utf8"); - const digest = sha256(bytes); - const idDigest = sha256(responseId).slice(0, 12); - const contentDigest = digest.slice(0, 24); + const aclBudget = spillAclBudget(options.aclBudgetMs); + const { bytes, digest, idDigest, contentDigest } = serializedSpill(responseId, state); const dir = responseSpillDirectory(); mkdirSync(dir, { recursive: true, mode: 0o700 }); - harden(dir, 0o700); + harden(dir, 0o700, aclBudget); tempPath = join(dir, `.response-spill.${process.pid}.${randomBytes(8).toString("hex")}.tmp`); fd = openSync(tempPath, "wx", 0o600); @@ -331,7 +536,7 @@ export function writeResponseSpillDurably( fsyncFile(fd); closeFile(fd); fd = null; - harden(tempPath, 0o600); + harden(tempPath, 0o600, aclBudget); record("harden"); const publishTempPath = tempPath; @@ -341,7 +546,7 @@ export function writeResponseSpillDurably( if (!OWNED_SPILL_NAME.test(fileName)) throw new Error("Response spill name allocation failed"); const destinationPath = join(dir, fileName); try { - publishNoReplace(publishTempPath, destinationPath); + publishNoReplace(publishTempPath, destinationPath, aclBudget); fsyncDirectoryBestEffort(dir); unlinkEphemeral(publishTempPath); tempPath = null; @@ -352,14 +557,114 @@ export function writeResponseSpillDurably( } } throw new Error("Response spill publication retries exhausted"); - } catch { + } catch (cause) { if (fd !== null) { try { closeSync(fd); } catch { /* best effort */ } } if (tempPath) { try { unlinkEphemeral(tempPath); } catch { /* best effort */ } } - throw new Error("Response spill write failed"); + throw responseSpillWriteError(cause); + } +} + +/** + * Windows runtime counterpart of `writeResponseSpillDurably`. + * + * The filesystem publication contract stays identical, but required NTFS ACL subprocesses are + * awaited through Bun.spawn instead of Bun.spawnSync. State ownership and serialization remain in + * `state.ts`; callers must compare the resident generation again before installing the returned + * reference because another response can replace it while ACL hardening is pending. + */ +export async function writeResponseSpillDurablyAsync( + responseId: string, + state: Omit, + options: ResponseSpillWriteOptions & { aclBudgetMs: number }, +): Promise { + const publicationControl = options.publicationControl; + const aclBudget = spillAclBudget(options.aclBudgetMs); + if (!aclBudget) throw new Error("Response spill async ACL budget is required"); + let tempPath: string | null = null; + let fd: number | null = null; + try { + throwIfPublicationSuperseded(publicationControl); + const { bytes, digest, idDigest, contentDigest } = serializedSpill(responseId, state); + const dir = responseSpillDirectory(); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + await hardenAsync(dir, 0o700, aclBudget, options.retryTimedOutOnce === true); + throwIfPublicationSuperseded(publicationControl); + + tempPath = join(dir, `.response-spill.${process.pid}.${randomBytes(8).toString("hex")}.tmp`); + fd = openSync(tempPath, "wx", 0o600); + if (publicationControl) publicationControl.tempPath = tempPath; + writeAll(fd, bytes); + fsyncFile(fd); + closeFile(fd); + fd = null; + await hardenAsync(tempPath, 0o600, aclBudget, options.retryTimedOutOnce === true); + throwIfPublicationSuperseded(publicationControl); + record("harden"); + const publishTempPath = tempPath; + + for (let attempt = 0; attempt < RESPONSE_SPILL_PUBLISH_RETRIES; attempt++) { + throwIfPublicationSuperseded(publicationControl); + spillGeneration += 1; + const fileName = `${sanitizeResponseId(responseId)}.${idDigest}.${contentDigest}.${spillGeneration}.${bytes.byteLength}.spill.json`; + if (!OWNED_SPILL_NAME.test(fileName)) throw new Error("Response spill name allocation failed"); + const destinationPath = join(dir, fileName); + if (publicationControl) publicationControl.destinationPath = destinationPath; + try { + throwIfPublicationSuperseded(publicationControl); + await publishNoReplaceAsync( + publishTempPath, + destinationPath, + aclBudget, + options.retryTimedOutOnce === true, + publicationControl, + ); + throwIfPublicationSuperseded(publicationControl); + fsyncDirectoryBestEffort(dir); + unlinkEphemeral(publishTempPath); + tempPath = null; + if (publicationControl) { + publicationControl.tempPath = null; + publicationControl.destinationPath = null; + } + return { version: 1, fileName, digest, payloadBytes: bytes.byteLength }; + } catch (error) { + if (publicationControl?.superseded) throw error; + if (publicationControl) publicationControl.destinationPath = null; + if (isErrno(error, "EEXIST")) continue; + throw error; + } + } + throw new Error("Response spill publication retries exhausted"); + } catch (cause) { + if (fd !== null) { + try { closeSync(fd); } catch { /* best effort */ } + } + if (tempPath) { + try { + unlinkEphemeral(tempPath); + if (publicationControl?.tempPath === tempPath) publicationControl.tempPath = null; + } catch (error) { + if (isErrno(error, "ENOENT") && publicationControl?.tempPath === tempPath) { + publicationControl.tempPath = null; + } + } + } + if (publicationControl) { + const destinationPath = publicationControl.destinationPath; + if (destinationPath) { + try { + unlink(destinationPath); + publicationControl.destinationPath = null; + } catch (error) { + if (isErrno(error, "ENOENT")) publicationControl.destinationPath = null; + } + } + } + throw responseSpillWriteError(cause); } } diff --git a/src/responses/state.ts b/src/responses/state.ts index 940de10e69..35540a0ee7 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -3,16 +3,23 @@ import { uptime } from "node:os"; import { dirname, join } from "node:path"; import { atomicWriteFileAsync, getConfigDir, resolveWriteTarget } from "../config"; import { enforceAppOwnedMemoryBudget, type RetainedStoreSnapshot } from "../lib/app-owned-memory"; +import { windowsSecretAclApplies } from "../lib/windows-secret-acl"; import type { OcxProviderContinuationState } from "../types"; import { + cleanupSupersededResponseSpillPublication, + createResponseSpillPublicationControl, deleteResponseSpill, + MAX_RESPONSE_SPILL_PAYLOAD_BYTES, noteStubSwapForTest, readResponseSpill, recoverOrphanedResponseSpills, responseSpillDirectory, responseSpillPayloadCap, + markResponseSpillPublicationSuperseded, + type ResponseSpillPublicationControl, type ResponseSpillRef, writeResponseSpillDurably, + writeResponseSpillDurablyAsync, } from "./spill-store"; const MAX_STORED_RESPONSES = 1_000; @@ -51,6 +58,10 @@ const PERIODIC_TEMP_MAX_CLEANUPS = 64; const PERIODIC_TEMP_SCAN_DEADLINE_MS = 25; const RESPONSE_STATE_TEMP_NAME = /^responses-state\.json\.ocx\.(\d+)\.(\d+)\.tmp$/; const MAX_SNAPSHOT_REWRITE_ATTEMPTS = 4; +const RESPONSE_SPILL_SHUTDOWN_BUDGET_MS = 5_000; +const RESPONSE_SPILL_SHUTDOWN_FALLBACK_RESERVE_MS = 4_000; +const RESPONSE_SPILL_ASYNC_ACL_ATTEMPT_BUDGET_MS = 30_000; +const RESPONSE_SPILL_SHUTDOWN_TERMINALIZATION_MAX_PASSES = MAX_STORED_RESPONSES + 1; interface ResidentResponseState { kind: "resident"; @@ -160,6 +171,424 @@ const pendingSpillUnlinks: ResponseSpillRef[] = []; // structured 400 β€” bounded-loss, never silent corruption or unbounded disk. const PENDING_SPILL_UNLINKS_MAX = 128; +/** + * Windows keeps the candidate replayable while required ACL hardening runs off the event loop. + * Pending bytes are pinned, not evictable; cap them below the process-owned 512 MiB ceiling so an + * icacls outage cannot turn the serialized queue into an unbounded resident backlog. + */ +const MAX_PENDING_RESPONSE_SPILL_BYTES = MAX_RESPONSE_SPILL_PAYLOAD_BYTES; + +interface PendingResponseSpill { + id: string; + candidate: ResidentResponseState | null; + supersededSpill?: ResponseSpillRef; + directAdmission: boolean; + running: boolean; + cancelled: boolean; + released: boolean; + sizeBytes: number; + publicationControl: ResponseSpillPublicationControl; +} + +const pendingResponseSpills = new Set(); +const pendingResponseSpillById = new Map(); +let pendingResponseSpillBytes = 0; +let responseSpillPublicationTail: Promise = Promise.resolve(); +let responseSpillShutdownBudgetOverride: { totalMs: number; fallbackReserveMs: number } | null = null; +let responseSpillShutdownTerminalizationPassLimitOverride: number | null = null; +let responseSpillAsyncAclAttemptBudgetOverride: number | null = null; + +function deferSupersededSpill(ref: ResponseSpillRef | undefined): void { + if (!ref) return; + pendingSpillUnlinks.push(ref); + while (pendingSpillUnlinks.length > PENDING_SPILL_UNLINKS_MAX) { + deleteResponseSpill(pendingSpillUnlinks.shift()!); + } +} + +function releasePendingResponseSpill(job: PendingResponseSpill): void { + if (job.released) return; + job.released = true; + pendingResponseSpillBytes = Math.max(0, pendingResponseSpillBytes - job.sizeBytes); + pendingResponseSpills.delete(job); + if (pendingResponseSpillById.get(job.id) === job) pendingResponseSpillById.delete(job.id); + job.candidate = null; +} + +function cancelPendingResponseSpill(id: string): ResponseSpillRef | undefined { + const job = pendingResponseSpillById.get(id); + if (!job) return undefined; + pendingResponseSpillById.delete(id); + job.cancelled = true; + markResponseSpillPublicationSuperseded(job.publicationControl); + const superseded = job.supersededSpill; + // A queued job has not captured the candidate in an async frame yet, so release it now. + // A running job retains its accounting until settlement and will discard its stale file. + if (!job.running) releasePendingResponseSpill(job); + return superseded; +} + +function isAclTimeout(error: unknown): boolean { + return !!error && typeof error === "object" && "code" in error + && String((error as { code?: unknown }).code) === "ETIMEDOUT"; +} + +function spillPayloadForResident(candidate: ResidentResponseState): Parameters[1] { + return { + createdAt: candidate.createdAt, + ...(candidate.clientThreadId ? { clientThreadId: candidate.clientThreadId } : {}), + items: candidate.items, + ...(candidate.providerOutputStart !== undefined ? { providerOutputStart: candidate.providerOutputStart } : {}), + ...(candidate.providers ? { providers: candidate.providers } : {}), + }; +} + +async function runPendingResponseSpill(job: PendingResponseSpill): Promise { + if (job.cancelled || !job.candidate) return; + job.running = true; + const candidate = job.candidate; + let ref: ResponseSpillRef | null = null; + try { + const state = spillPayloadForResident(candidate); + try { + ref = await writeResponseSpillDurablyAsync(job.id, state, { + aclBudgetMs: responseSpillAsyncAclAttemptBudgetMs(), + publicationControl: job.publicationControl, + }); + } catch (error) { + if (!isAclTimeout(error)) throw error; + // The ACL helper permits exactly one caller-owned recovery budget. The resident generation + // remains replayable during both attempts, so a transient timeout never becomes a tombstone. + ref = await writeResponseSpillDurablyAsync(job.id, state, { + aclBudgetMs: responseSpillAsyncAclAttemptBudgetMs(), + retryTimedOutOnce: true, + publicationControl: job.publicationControl, + }); + } + if (ref.payloadBytes > responseSpillPayloadCap()) { + deleteResponseSpill(ref); + ref = null; + if (job.directAdmission) admissionCounters.oversizedDrops += 1; + throw Object.assign(new Error("Response spill payload exceeds replay ceiling"), { code: "EFBIG" }); + } + if (states.get(job.id) !== candidate || job.cancelled) { + deleteResponseSpill(ref); + ref = null; + return; + } + if (swapResidentForSpill(job.id, candidate, ref)) { + ref = null; + spillCounters.writes += 1; + if (job.directAdmission) admissionCounters.directSpills += 1; + deferSupersededSpill(job.supersededSpill); + } + } catch { + if (ref) deleteResponseSpill(ref); + if (states.get(job.id) === candidate && !job.cancelled) { + spillCounters.writeFailures += 1; + replaceWithSpillFailure(job.id, candidate); + deferSupersededSpill(job.supersededSpill); + } + } finally { + const cancelled = job.cancelled; + releasePendingResponseSpill(job); + recomputeOldestResident(); + if (!cancelled) { + schedulePersist(); + pruneResponses(); + enforceAppOwnedMemoryBudget(); + } + } +} + +function queuePendingResponseSpill( + id: string, + candidate: ResidentResponseState, + options: { supersededSpill?: ResponseSpillRef; directAdmission?: boolean } = {}, +): void { + const inheritedSpill = cancelPendingResponseSpill(id) ?? options.supersededSpill; + if (pendingResponseSpillBytes + candidate.sizeBytes > MAX_PENDING_RESPONSE_SPILL_BYTES) { + spillCounters.writeFailures += 1; + replaceWithSpillFailure(id, candidate); + deferSupersededSpill(inheritedSpill); + return; + } + const job: PendingResponseSpill = { + id, + candidate, + ...(inheritedSpill ? { supersededSpill: inheritedSpill } : {}), + directAdmission: options.directAdmission === true, + running: false, + cancelled: false, + released: false, + sizeBytes: candidate.sizeBytes, + publicationControl: createResponseSpillPublicationControl(), + }; + pendingResponseSpills.add(job); + pendingResponseSpillById.set(id, job); + pendingResponseSpillBytes += job.sizeBytes; + recomputeOldestResident(); + responseSpillPublicationTail = responseSpillPublicationTail + .then(() => runPendingResponseSpill(job), () => runPendingResponseSpill(job)); +} + +function replaceWithPendingResponseSpill( + id: string, + candidate: ResidentResponseState, + expected: StoredResponseState | undefined, + options: { directAdmission?: boolean } = {}, +): boolean { + const inheritedSpill = pendingResponseSpillById.get(id)?.supersededSpill + ?? (expected?.kind === "spill" ? expected.spill : undefined); + if (!replaceMapEntry(id, candidate, expected)) return false; + queuePendingResponseSpill(id, candidate, { + ...(inheritedSpill ? { supersededSpill: inheritedSpill } : {}), + directAdmission: options.directAdmission === true, + }); + return true; +} + +/** Test-only: settle every serialized Windows spill publication. */ +export async function flushPendingResponseSpillsForTests(): Promise { + await drainResponseSpillPublications(); +} + +/** Test-only: observe ordinary queue settlement without invoking shutdown fallback. */ +export async function awaitResponseSpillPublicationTailForTests(): Promise { + await responseSpillPublicationTail; +} + +/** Test-only: observe the bounded queue without exposing payloads. */ +export function pendingResponseSpillMetricsForTests(): { count: number; bytes: number } { + return { count: pendingResponseSpills.size, bytes: pendingResponseSpillBytes }; +} + +/** Test-only: shorten the shutdown drain/fallback budget (null restores production values). */ +export function setResponseSpillShutdownBudgetForTests( + budget: { totalMs: number; fallbackReserveMs: number } | null, +): void { + responseSpillShutdownBudgetOverride = budget; +} + +/** Test-only: shorten the ordinary async whole-attempt ACL budget. */ +export function setResponseSpillAsyncAclAttemptBudgetForTests(budgetMs: number | null): void { + responseSpillAsyncAclAttemptBudgetOverride = budgetMs; +} + +function responseSpillAsyncAclAttemptBudgetMs(): number { + return responseSpillAsyncAclAttemptBudgetOverride ?? RESPONSE_SPILL_ASYNC_ACL_ATTEMPT_BUDGET_MS; +} + +/** Test-only: lower the hard terminalization pass guard (null restores production). */ +export function setResponseSpillShutdownTerminalizationPassLimitForTests(limit: number | null): void { + responseSpillShutdownTerminalizationPassLimitOverride = limit; +} + +function responseSpillShutdownTerminalizationPassLimit(): number { + return responseSpillShutdownTerminalizationPassLimitOverride + ?? RESPONSE_SPILL_SHUTDOWN_TERMINALIZATION_MAX_PASSES; +} + +function responseSpillShutdownBudget(): { totalMs: number; fallbackReserveMs: number } { + return responseSpillShutdownBudgetOverride ?? { + totalMs: RESPONSE_SPILL_SHUTDOWN_BUDGET_MS, + fallbackReserveMs: RESPONSE_SPILL_SHUTDOWN_FALLBACK_RESERVE_MS, + }; +} + +function awaitResponseSpillTailUntil(observed: Promise, deadline: number): Promise { + const remaining = deadline - Date.now(); + if (remaining <= 0) return Promise.resolve(false); + return new Promise(resolve => { + let finished = false; + const finish = (settled: boolean): void => { + if (finished) return; + finished = true; + clearTimeout(timer); + resolve(settled); + }; + const timer = setTimeout(() => finish(false), remaining); + observed.then(() => finish(true), () => finish(true)); + }); +} + +function installShutdownFallbackSpill( + job: PendingResponseSpill, + candidate: ResidentResponseState, + aclBudgetMs: number, +): void { + let ref: ResponseSpillRef | null = null; + try { + ref = writeResponseSpillDurably(job.id, spillPayloadForResident(candidate), { aclBudgetMs }); + if (ref.payloadBytes > responseSpillPayloadCap()) { + deleteResponseSpill(ref); + ref = null; + if (job.directAdmission) admissionCounters.oversizedDrops += 1; + throw Object.assign(new Error("Response spill payload exceeds replay ceiling"), { code: "EFBIG" }); + } + if (states.get(job.id) !== candidate) { + deleteResponseSpill(ref); + ref = null; + return; + } + if (swapResidentForSpill(job.id, candidate, ref)) { + ref = null; + spillCounters.writes += 1; + if (job.directAdmission) admissionCounters.directSpills += 1; + deferSupersededSpill(job.supersededSpill); + } + } catch (error) { + if (ref) deleteResponseSpill(ref); + if (states.get(job.id) === candidate) { + spillCounters.writeFailures += 1; + replaceWithSpillFailure(job.id, candidate); + deferSupersededSpill(job.supersededSpill); + } + throw error; + } +} + +function terminalizeShutdownFallbackCandidate( + job: PendingResponseSpill, + candidate: ResidentResponseState, +): void { + if (states.get(job.id) !== candidate) return; + spillCounters.writeFailures += 1; + replaceWithSpillFailure(job.id, candidate); + deferSupersededSpill(job.supersededSpill); +} + +function pendingShutdownFallbackCandidates(): Array<{ + job: PendingResponseSpill; + candidate: ResidentResponseState; +}> { + return [...pendingResponseSpills] + .map(job => ({ job, candidate: job.candidate })) + .filter((entry): entry is { job: PendingResponseSpill; candidate: ResidentResponseState } => !!entry.candidate); +} + +function supersedeShutdownFallbackBatch( + pending: Array<{ job: PendingResponseSpill; candidate: ResidentResponseState }>, + failures: Error[], +): void { + for (const { job } of pending) { + job.cancelled = true; + markResponseSpillPublicationSuperseded(job.publicationControl); + } + for (const { job } of pending) { + const cleanupFailure = cleanupSupersededResponseSpillPublication(job.publicationControl); + if (cleanupFailure) failures.push(cleanupFailure); + releasePendingResponseSpill(job); + } +} + +function stopAtShutdownTerminalizationPassLimit( + pending: Array<{ job: PendingResponseSpill; candidate: ResidentResponseState }>, + failures: Error[], +): void { + failures.push(Object.assign(new Error("Response spill shutdown terminalization pass limit exceeded"), { code: "ELOOP" })); + supersedeShutdownFallbackBatch(pending, failures); + for (const { job, candidate } of pending) { + terminalizeShutdownFallbackCandidate(job, candidate); + } + for (const [id, state] of [...states]) { + if (state.kind !== "resident") continue; + spillCounters.writeFailures += 1; + replaceWithSpillFailure(id, state); + } + recomputeOldestResident(); + pruneResponses(); + enforceAppOwnedMemoryBudget(); +} + +function terminalizeExhaustedShutdownFallback( + initial: Array<{ job: PendingResponseSpill; candidate: ResidentResponseState }>, + failures: Error[], +): void { + let pending = initial; + let passes = 0; + const passLimit = responseSpillShutdownTerminalizationPassLimit(); + // Every pass replaces each captured resident with a tombstone. Pruning may expose + // another finite batch, but resident count strictly decreases until none can requeue. + while (pending.length > 0) { + if (passes >= passLimit) { + stopAtShutdownTerminalizationPassLimit(pending, failures); + return; + } + passes += 1; + supersedeShutdownFallbackBatch(pending, failures); + for (const { job, candidate } of pending) { + failures.push(Object.assign(new Error("Response spill shutdown fallback budget exhausted"), { code: "ETIMEDOUT" })); + terminalizeShutdownFallbackCandidate(job, candidate); + } + recomputeOldestResident(); + pruneResponses(); + enforceAppOwnedMemoryBudget(); + pending = pendingShutdownFallbackCandidates(); + } +} + +function fallbackPendingResponseSpills(reserveMs: number): Error[] { + const deadline = Date.now() + reserveMs; + const failures: Error[] = []; + for (;;) { + const pending = pendingShutdownFallbackCandidates(); + if (pending.length === 0) return failures; + if (Date.now() >= deadline) { + terminalizeExhaustedShutdownFallback(pending, failures); + return failures; + } + + supersedeShutdownFallbackBatch(pending, failures); + let reserveExhausted = false; + for (let index = 0; index < pending.length; index += 1) { + const { job, candidate } = pending[index]!; + if (states.get(job.id) !== candidate) continue; + const remaining = deadline - Date.now(); + if (remaining <= 0) { + reserveExhausted = true; + for (const exhausted of pending.slice(index)) { + failures.push(Object.assign(new Error("Response spill shutdown fallback budget exhausted"), { code: "ETIMEDOUT" })); + terminalizeShutdownFallbackCandidate(exhausted.job, exhausted.candidate); + } + break; + } + try { + installShutdownFallbackSpill(job, candidate, remaining); + } catch (error) { + failures.push(error instanceof Error ? error : new Error("Response spill shutdown fallback failed")); + } + } + recomputeOldestResident(); + pruneResponses(); + enforceAppOwnedMemoryBudget(); + if (reserveExhausted || Date.now() >= deadline) { + terminalizeExhaustedShutdownFallback(pendingShutdownFallbackCandidates(), failures); + return failures; + } + } +} + +async function drainResponseSpillPublications(): Promise { + const budget = responseSpillShutdownBudget(); + const fallbackReserveMs = Math.min(budget.totalMs, Math.max(1, budget.fallbackReserveMs)); + const drainDeadline = Date.now() + Math.max(0, budget.totalMs - fallbackReserveMs); + + for (;;) { + if (pendingResponseSpills.size === 0) return; + const observed = responseSpillPublicationTail; + const settled = await awaitResponseSpillTailUntil(observed, drainDeadline); + if (!settled) { + const failures = fallbackPendingResponseSpills(fallbackReserveMs); + if (failures.length > 0) { + throw new AggregateError(failures, "Response spill shutdown fallback incomplete"); + } + return; + } + if (observed === responseSpillPublicationTail) return; + } +} + function byteCap(): number { return byteCapOverride ?? MAX_STORED_RESPONSE_BYTES; } @@ -200,6 +629,7 @@ function recomputeOldestResident(): void { oldestResidentAt = null; for (const [id, state] of states) { if (state.kind !== "resident") continue; + if (pendingResponseSpillById.get(id)?.candidate === state) continue; if (oldestResidentAt !== null && state.createdAt >= oldestResidentAt) continue; oldestResidentId = id; oldestResidentAt = state.createdAt; @@ -248,6 +678,7 @@ function deleteOwnedSpills(entry: StoredResponseState): void { function deleteEntry(id: string, options: { deleteSpill?: boolean } = {}): void { const existing = states.get(id); if (!existing) return; + const supersededSpill = cancelPendingResponseSpill(id); storedResponseBytes -= existing.sizeBytes; if (existing.kind === "resident") { residentResponseBytes -= existing.sizeBytes; @@ -258,6 +689,7 @@ function deleteEntry(id: string, options: { deleteSpill?: boolean } = {}): void if (oldestResidentId === id) recomputeOldestResident(); stateRevision += 1; if (options.deleteSpill !== false) deleteOwnedSpills(existing); + if (options.deleteSpill !== false && supersededSpill) deleteResponseSpill(supersededSpill); } function replaceWithSpillFailure( @@ -362,11 +794,18 @@ function setResidentEntry(id: string, entry: ResidentInput): void { pruneResponses(); return; } + const pending = pendingResponseSpillById.get(id); + if (windowsSecretAclApplies() && (expected?.kind === "spill" || pending?.supersededSpill)) { + replaceWithPendingResponseSpill(id, candidate, expected); + pruneResponses(); + return; + } if (expected?.kind === "spill") { replaceSpillEntryAtomically(id, expected, candidate); pruneResponses(); return; } + if (windowsSecretAclApplies()) cancelPendingResponseSpill(id); if (!replaceMapEntry(id, candidate, expected)) return; pruneResponses(); } @@ -389,6 +828,10 @@ function admitOversizedCandidate( replaceWithSpillFailure(id, expected, { deferSpillUnlink: true }); return; } + if (windowsSecretAclApplies()) { + replaceWithPendingResponseSpill(id, candidate, expected, { directAdmission: true }); + return; + } try { const ref = writeResponseSpillDurably(id, { createdAt: candidate.createdAt, @@ -927,8 +1370,7 @@ function schedulePersist(): void { schedulePersistAt(snapshotPath()); } -/** Flush any pending debounced snapshot write (graceful shutdown / deterministic tests). */ -export async function flushResponseState(): Promise { +async function flushResponseSnapshot(): Promise { if (persistTimer) { await persistNow(pendingPersistPath ?? snapshotPath(), true); return; @@ -941,6 +1383,23 @@ export async function flushResponseState(): Promise { if (persistTimer) await persistNow(pendingPersistPath ?? snapshotPath(), true); } +/** Flush publications and snapshot state; report drain failure only after persistence completes. */ +export async function flushResponseState(): Promise { + const failures: unknown[] = []; + try { + await drainResponseSpillPublications(); + } catch (error) { + failures.push(error); + } + try { + await flushResponseSnapshot(); + } catch (error) { + failures.push(error); + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, "Response state shutdown flush incomplete"); +} + function inputItems(input: unknown): unknown[] { if (input === undefined) return []; if (Array.isArray(input)) return input; @@ -1048,7 +1507,11 @@ function pruneResponses(at = now()): void { // Unconditional RAM cap. Resident payloads demote durably; stubs/tombstones are // deleted only when even their bounded metadata cannot fit the override. while (storedResponseBytes > byteCap() && states.size > 0) { - const oldestResident = [...states].find(([, entry]) => entry.kind === "resident"); + const oldestResident = [...states].find(([id, entry]) => entry.kind === "resident" + && pendingResponseSpillById.get(id)?.candidate !== entry); + const hasPendingResident = !oldestResident && [...states].some(([id, entry]) => entry.kind === "resident" + && pendingResponseSpillById.get(id)?.candidate === entry); + if (hasPendingResident) break; const oldestId = oldestResident?.[0] ?? states.keys().next().value as string | undefined; if (!oldestId) break; const entry = states.get(oldestId)!; @@ -1056,6 +1519,10 @@ function pruneResponses(at = now()): void { deleteEntry(oldestId); continue; } + if (windowsSecretAclApplies()) { + queuePendingResponseSpill(oldestId, entry); + continue; + } try { const ref = writeResponseSpillDurably(oldestId, { createdAt: entry.createdAt, @@ -1143,11 +1610,18 @@ export function sweepAbandonedResponseStateTemps(): number { } export function responseContinuationRetainedStoreSnapshot(): RetainedStoreSnapshot { + let currentPendingBytes = 0; + for (const job of pendingResponseSpills) { + if (job.candidate && states.get(job.id) === job.candidate) currentPendingBytes += job.sizeBytes; + } + const detachedPendingBytes = Math.max(0, pendingResponseSpillBytes - currentPendingBytes); + const bytes = storedResponseBytes + detachedPendingBytes; + const evictableBytes = Math.max(0, residentResponseBytes - currentPendingBytes); return { count: states.size, - bytes: storedResponseBytes, - evictableBytes: residentResponseBytes, - pinnedBytes: Math.max(0, storedResponseBytes - residentResponseBytes), + bytes, + evictableBytes, + pinnedBytes: Math.max(0, bytes - evictableBytes), oldestAt: oldestResidentAt, }; } @@ -1157,6 +1631,11 @@ export function evictOldestResponseContinuationForBudget(): number { const id = oldestResidentId; const entry = states.get(id); if (!entry || entry.kind !== "resident") return 0; + if (windowsSecretAclApplies()) { + queuePendingResponseSpill(id, entry); + schedulePersist(); + return 0; + } try { const ref = writeResponseSpillDurably(id, { createdAt: entry.createdAt, @@ -1380,7 +1859,7 @@ export function responseStateMetrics(): ResponseStateMetrics { residentCount, spillStubCount, tombstoneCount, - totalBytes: storedResponseBytes, + totalBytes: responseContinuationRetainedStoreSnapshot().bytes, spillPayloadBytes, largestBytes, oldestAgeMs: states.size > 0 ? at - oldestCreatedAt : 0, @@ -1492,6 +1971,8 @@ export function clearResponseStateMemoryForTests(): void { persistTimer = null; } pendingPersistPath = null; + for (const id of [...pendingResponseSpillById.keys()]) cancelPendingResponseSpill(id); + pendingResponseSpillById.clear(); states.clear(); storedResponseBytes = 0; residentResponseBytes = 0; diff --git a/src/server/lifecycle.ts b/src/server/lifecycle.ts index b81d9bf46f..bf5d4ee292 100644 --- a/src/server/lifecycle.ts +++ b/src/server/lifecycle.ts @@ -458,8 +458,9 @@ export function trackStreamLifetime( export async function drainAndShutdown( server: ReturnType | undefined, timeoutMs: number, -): Promise { +): Promise { const s = server ?? _serverRef; + let shutdownSucceeded = true; // One absolute budget covers both a pre-existing scoped profile drain and // ordinary in-flight turns. A stuck scoped owner must not pin shutdown forever. const deadline = Date.now() + Math.max(0, timeoutMs); @@ -491,9 +492,11 @@ export async function drainAndShutdown( // shutdown is usually part of. const stateFlush = await Promise.allSettled([flushResponseState(), flushAntigravityReplay()]); if (stateFlush[0]?.status === "rejected") { + shutdownSucceeded = false; console.warn("[responses] state flush during shutdown failed"); } if (stateFlush[1]?.status === "rejected") { + shutdownSucceeded = false; console.warn("[antigravity] replay flush during shutdown failed"); } @@ -546,4 +549,5 @@ export async function drainAndShutdown( // never resume admission merely because shutdown cleanup returned. } } + return shutdownSucceeded; } diff --git a/src/server/management-api.ts b/src/server/management-api.ts index c850c3bbda..9f19831576 100644 --- a/src/server/management-api.ts +++ b/src/server/management-api.ts @@ -276,8 +276,13 @@ export async function handleManagementAPI( const { stripGrokConfig } = await import("../grok/inject"); const grok = stripGrokConfig(); setTimeout(async () => { - await drainAndShutdown(undefined, config.shutdownTimeoutMs ?? 5000); - process.exit(0); + let shutdownSucceeded = false; + try { + shutdownSucceeded = await drainAndShutdown(undefined, config.shutdownTimeoutMs ?? 5000); + } catch { + console.warn("[opencodex] shutdown drain failed"); + } + process.exit(shutdownSucceeded ? 0 : 1); }, 200); const grokNote = grok.ok ? "" : ` Grok config cleanup failed: ${grok.message}`; return jsonResponse(restore.success diff --git a/src/server/management/system-restart.ts b/src/server/management/system-restart.ts index 56d393800c..cfcc9ca416 100644 --- a/src/server/management/system-restart.ts +++ b/src/server/management/system-restart.ts @@ -78,11 +78,11 @@ let restartIo: SystemRestartIo = {}; /** Prevents double-scheduling in the 200ms window before drainAndShutdown sets draining. */ let restartAccepted = false; -type RestartDrainOutcome = "completed" | "rejected" | "deadline"; +type RestartDrainOutcome = "completed" | "failed" | "rejected" | "deadline"; type BoundedSettlementOutcome = "completed" | "rejected" | "deadline"; function waitForRestartDrain( - drainPromise: Promise, + drainPromise: Promise, deadlineMs: number, now: () => number, scheduleDeadline: NonNullable, @@ -105,7 +105,7 @@ function waitForRestartDrain( cancelDeadline = scheduleDeadline(() => finish("deadline"), remainingMs); if (settled) cancelDeadline(); void drainPromise.then( - () => finish("completed"), + succeeded => finish(succeeded === false ? "failed" : "completed"), () => finish("rejected"), ); }); @@ -382,7 +382,7 @@ export function acceptSystemRestart(io: SystemRestartIo = restartIo): { await completeDeadlineRestartHandoff(io, exitProcess, restartPort, scheduleDeadline); return; } - if (drainOutcome === "rejected") { + if (drainOutcome === "failed" || drainOutcome === "rejected") { // drainAndShutdown stops the listener in finally. Even if ancillary cleanup // rejects, an accepted restart must still reach replacement or terminal exit. console.warn("Drain-and-restart cleanup failed; continuing terminal restart handoff"); @@ -422,7 +422,7 @@ export function acceptSystemRestart(io: SystemRestartIo = restartIo): { return; } (io.markRecycling ?? markRecyclingForExit)(); - exitProcess(0); + exitProcess(drainOutcome === "failed" || drainOutcome === "rejected" ? 1 : 0); }, 200); } diff --git a/structure/02_config-and-codex-home.md b/structure/02_config-and-codex-home.md index 3685fb5bb4..8411884b5c 100644 --- a/structure/02_config-and-codex-home.md +++ b/structure/02_config-and-codex-home.md @@ -173,6 +173,50 @@ are consumed incrementally and at most 512 stale files are attempted per process - λ‹€λ₯Έ λŒ€μ•ˆ λŒ€μ‹  이 방식을 μ„ νƒν•œ 이유: It repairs known remnants without broad authority over unrelated temp files or active writers. - μž₯점, 단점 및 영ν–₯: Old dead-PID files are reclaimed automatically; locked or conservatively classified files remain for a later retry. +Windows runtime response spills never wait on `icacls` through `Bun.spawnSync`. Linux and macOS +retain the immediate synchronous publication path. On Windows, the resident continuation enters one +serialized publication queue and remains replayable while `hardenSecretDirAsync` and +`hardenSecretPathAsync` run. Publication installs a spill stub only when the map still contains the +same resident object; a superseded job deletes its newly published file instead of overwriting newer +state. Pending payloads are pinned and capped at 256 MiB, so an ACL outage cannot grow an unbounded +queue or be misreported as evictable memory. One caller-owned retry is allowed after a real +`ETIMEDOUT`; the first timeout does not install a `spill-failed` tombstone. Required ACL failures +remain fail-closed after that bounded recovery. Optional config-directory hardening uses a separate +per-directory async single-flight, while required config mutation writers retain their existing +awaited or synchronous fail-closed boundary. + +Each ordinary async spill write attempt owns one 30-second ACL budget shared across directory, temp, +and exclusive-copy destination hardening; the single timeout retry receives one fresh whole-attempt +budget. No harden step may reopen an independent 30-second window inside either attempt. +Both icacls and effective-principal subprocess waits are settlement-bounded: at deadline the child is +killed, unref'd, and abandoned without awaiting `proc.exited`. The caller-level deadline also bounds +injected/shared runners, so a child that ignores termination cannot pin the serialized spill queue. + +Graceful shutdown drains that serialized publication queue to a stable fixed point before snapshot +serialization. The drain has a wall-clock cap with a reserved synchronous fallback budget; expiry +supersedes the async writer, claims and removes any temp or destination it still owns, and only then +starts fallback publication. The writer rechecks supersession before no-replace publication, while +the fallback splits its reserve across the directory and file ACL hardens. This ordering is +load-bearing because resident entries over 2 MiB are deliberately excluded from +`responses-state.json`: serializing first could omit the resident before its durable spill stub +exists, losing the continuation on restart. Cleanup is attempted for every abandoned writer; any +failure is retained while fallback and snapshot persistence continue, then returned through the +shutdown status so process exit is non-zero without sacrificing unrelated replay state. +If the fallback reserve expires, every remaining resident candidate is terminalized as a bounded +`spill-failed` tombstone before pruning, so no payload remains eligible for shutdown requeue and the +snapshot flush always regains control. +The terminalization pass itself is hard-capped at `MAX_STORED_RESPONSES + 1`; exceeding that +structural bound records a bounded failure, fail-closes every remaining resident, and returns control +to snapshot persistence instead of relying on the progress argument alone. + +[Decision Log] +- λͺ©μ κ³Ό μ˜λ„: Keep `/healthz` and unrelated requests responsive during intermittent Windows ACL stalls without publishing an unhardened continuation. +- κΈ°μ‘΄ κ΅¬ν˜„ 및 μ œμ•½ 쑰건: Response demotion called the synchronous spill writer from request-time state mutations; `Bun.spawnSync(icacls)` could block the only Bun event loop for the full timeout and immediately replace replayable state with a tombstone. +- κ²€ν† ν•œ μ£Όμš” λŒ€μ•ˆ: Increase the ACL timeout, weaken required ACL checks, publish before hardening, move every platform to async state mutation, or isolate only the Windows ACL-dependent publication boundary. +- μ„ νƒν•œ 방식: Preserve non-Windows behavior; serialize Windows publications through async ACL APIs, retain the exact resident generation until compare-before-swap succeeds, cap pending bytes, and retry one proven timeout. +- λ‹€λ₯Έ λŒ€μ•ˆ λŒ€μ‹  이 방식을 μ„ νƒν•œ 이유: Longer waits worsen liveness, early publication weakens secret-file ACLs, and a cross-platform async rewrite would disturb mature immediate memory and crash-ordering contracts that do not cause this incident. +- μž₯점, 단점 및 영ν–₯: Windows health stays schedulable and transient ACL stalls retain continuation replay; pending payloads can temporarily exceed the 64 MiB resident target but are pinned under a 256 MiB local ceiling and remain inside the documented 512 MiB process-owned worst case. + ## Config surface ### OpenCodex home and live process state diff --git a/tests/config.test.ts b/tests/config.test.ts index 6f0416b35f..62d3f6d782 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -35,6 +35,7 @@ import * as windowsAcl from "../src/lib/windows-secret-acl"; import { setTrustedWindowsSystemDirectoryResolverForTests } from "../src/lib/windows-elevation"; import { AtomicWriteResidualTempError, atomicWriteFile, atomicWriteFileAsync, hardenConfigDir, hardenExistingSecret, renameAtomicFile, saveConfig } from "../src/config"; import { nextAtomicTempSequence } from "../src/config/atomic-write"; +import { flushConfigDirHardeningForTests } from "../src/config/paths"; import { providerManagementConfigError } from "../src/server/auth-cors"; let testDir = ""; @@ -2613,25 +2614,37 @@ describe("config.ts – Windows ACL hardening integration", () => { } }); - test("hardenConfigDir delegates to hardenSecretDir with required:false on win32", () => { + test("hardenConfigDir delegates to one async optional flight on win32", async () => { const origPlatform = process.platform; Object.defineProperty(process, "platform", { value: "win32", configurable: true }); try { - const spy = spyOn(windowsAcl, "hardenSecretDir").mockReturnValue({ ok: true }); + let release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + const spy = spyOn(windowsAcl, "hardenSecretDirAsync").mockImplementation(async () => { + await pending; + return { ok: true }; + }); mkdirSync(testDir, { recursive: true }); hardenConfigDir(); - expect(spy).toHaveBeenCalledWith(testDir, { required: false }); - spy.mockRestore(); + hardenConfigDir(); + try { + expect(spy).toHaveBeenCalledTimes(1); + expect(spy).toHaveBeenCalledWith(testDir, { required: false }); + } finally { + release(); + await flushConfigDirHardeningForTests(); + spy.mockRestore(); + } } finally { Object.defineProperty(process, "platform", { value: origPlatform, configurable: true }); } }); - test("hardenConfigDir does not call hardenSecretDir on non-Windows", () => { + test("hardenConfigDir does not call async ACL hardening on non-Windows", () => { const origPlatform = process.platform; Object.defineProperty(process, "platform", { value: "linux", configurable: true }); try { - const spy = spyOn(windowsAcl, "hardenSecretDir"); + const spy = spyOn(windowsAcl, "hardenSecretDirAsync"); mkdirSync(testDir, { recursive: true }); hardenConfigDir(); expect(spy).not.toHaveBeenCalled(); diff --git a/tests/grok-lifecycle.test.ts b/tests/grok-lifecycle.test.ts index 31374cae72..f245e917b5 100644 --- a/tests/grok-lifecycle.test.ts +++ b/tests/grok-lifecycle.test.ts @@ -170,7 +170,7 @@ describe("Grok fence lifecycle wiring", () => { expect(startFn).toContain("if (!restored.success)"); expect(startFn).toContain("cleanupSucceeded = false"); expect(startFn).toContain("Native Codex restore failed during shutdown"); - expect(startFn).toContain("process.exit(restored ? 0 : 1)"); + expect(startFn).toContain("process.exit(restored && shutdownSucceeded ? 0 : 1)"); }); }); @@ -222,6 +222,12 @@ describe("POST /api/stop teardown", () => { expect(handler).toContain("stripGrokConfig()"); }); + test("maps a failed shutdown drain to a nonzero process exit", () => { + const handler = sliceFn(MANAGEMENT_SOURCE, '"/api/stop"', "/api/codex-auth/"); + expect(handler).toContain("shutdownSucceeded = await drainAndShutdown"); + expect(handler).toContain("process.exit(shutdownSucceeded ? 0 : 1)"); + }); + test("a 409 does not escalate to a forced kill", () => { // Escalating would run the daemon's cleanup and strip shared config while the foreign // service keeps the proxy alive β€” the exact hole the ownership gate exists to close. diff --git a/tests/helpers/responses-state-never-settling-acl-child.ts b/tests/helpers/responses-state-never-settling-acl-child.ts new file mode 100644 index 0000000000..790b750fdd --- /dev/null +++ b/tests/helpers/responses-state-never-settling-acl-child.ts @@ -0,0 +1,59 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + clearResponseStateMemoryForTests, + awaitResponseSpillPublicationTailForTests, + pendingResponseSpillMetricsForTests, + rememberResponseState, + responseStateMetrics, + setResponseSpillAsyncAclAttemptBudgetForTests, + setResponseStateByteCapForTests, +} from "../../src/responses/state"; +import { + setAsyncIcaclsRunnerForTests, + setPlatformForTests, +} from "../../src/lib/windows-secret-acl"; +import { setAsyncWindowsPrincipalRunnerForTests } from "../../src/lib/windows-user-principal"; + +type Mode = "principal" | "icacls"; + +function rememberLarge(id: string): void { + const text = id.repeat(1_000); + rememberResponseState( + { model: "test/model", input: text, store: false }, + { id, output: [{ type: "message", role: "assistant", content: text }], status: "completed" }, + undefined, + { force: true }, + ); +} + +const mode = process.argv[2]; +if (mode !== "principal" && mode !== "icacls") { + throw new Error(`Unknown never-settling ACL mode: ${mode ?? ""}`); +} + +const home = mkdtempSync(join(tmpdir(), "ocx-never-settling-acl-child-")); +process.env.OPENCODEX_HOME = home; +clearResponseStateMemoryForTests(); +setPlatformForTests("win32"); +setResponseSpillAsyncAclAttemptBudgetForTests(100); +setResponseStateByteCapForTests(1_024); + +if (mode === "principal") { + setAsyncWindowsPrincipalRunnerForTests(() => new Promise(() => {})); + setAsyncIcaclsRunnerForTests(async () => ({ success: true, exitCode: 0, timedOut: false, stdout: "" })); +} else { + setAsyncIcaclsRunnerForTests(() => new Promise(() => {})); +} + +rememberLarge(`resp_never_settling_${mode}_first`); +rememberLarge(`resp_never_settling_${mode}_second`); +await awaitResponseSpillPublicationTailForTests(); + +console.log(JSON.stringify({ + settled: true, + pending: pendingResponseSpillMetricsForTests(), + metrics: responseStateMetrics(), +})); +rmSync(home, { recursive: true, force: true }); diff --git a/tests/helpers/responses-state-shutdown-budget-child.ts b/tests/helpers/responses-state-shutdown-budget-child.ts new file mode 100644 index 0000000000..a335d3fa5c --- /dev/null +++ b/tests/helpers/responses-state-shutdown-budget-child.ts @@ -0,0 +1,132 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + clearResponseStateForTests, + clearResponseStateMemoryForTests, + expandPreviousResponseInput, + flushResponseState, + pendingResponseSpillMetricsForTests, + rememberResponseState, + responseStateMetrics, + setResponseSpillShutdownBudgetForTests, + setResponseSpillShutdownTerminalizationPassLimitForTests, + setResponseStateByteCapForTests, +} from "../../src/responses/state"; +import { + resetHardenedStateForTests, + setAsyncIcaclsRunnerForTests, + setIcaclsRunnerForTests, + setPlatformForTests, +} from "../../src/lib/windows-secret-acl"; + +type Scenario = "exhaustion" | "guard"; + +function fixedResponse(id: string, output: unknown[]): { id: string; output: unknown[]; status: string } { + return { id, output, status: "completed" }; +} + +function rememberLarge(id: string, text: string): void { + rememberResponseState( + { model: "test/model", input: text, store: false }, + fixedResponse(id, [{ type: "message", role: "assistant", content: text }]), + undefined, + { force: true }, + ); +} + +function errorMessages(error: unknown): string[] { + if (!(error instanceof Error)) return []; + const nested = error instanceof AggregateError + ? error.errors.flatMap(errorMessages) + : []; + return [error.message, ...nested]; +} + +async function runScenario(scenario: Scenario): Promise> { + const home = mkdtempSync(join(tmpdir(), "ocx-shutdown-budget-child-")); + const priorHome = process.env.OPENCODEX_HOME; + process.env.OPENCODEX_HOME = home; + clearResponseStateMemoryForTests(); + try { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 60, fallbackReserveMs: 40 }); + setResponseSpillShutdownTerminalizationPassLimitForTests(scenario === "guard" ? 0 : null); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let announced = false; + setAsyncIcaclsRunnerForTests(async () => { + if (!announced) { + announced = true; + entered(); + await gate; + } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setIcaclsRunnerForTests((_args, timeoutMs) => { + Bun.sleepSync(timeoutMs + 50); + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_budget_exhausted_first", "a".repeat(2 * 1024 * 1024 + 4_096)); + await started; + rememberLarge("resp_budget_exhausted_final", "b".repeat(2 * 1024 * 1024 + 4_096)); + setResponseStateByteCapForTests(1_000_000_000); + rememberResponseState( + { model: "test/model", input: "budget-safe-input", store: false }, + fixedResponse("resp_budget_unrelated", [{ type: "message", role: "assistant", content: "budget-safe-output" }]), + undefined, + { force: true }, + ); + + let reported: unknown; + try { + const flushing = flushResponseState(); + setResponseStateByteCapForTests(scenario === "guard" ? 1 : 1_024); + await flushing; + } catch (error) { + reported = error; + } finally { + release(); + } + const pending = pendingResponseSpillMetricsForTests(); + const metrics = responseStateMetrics(); + const messages = errorMessages(reported); + + clearResponseStateMemoryForTests(); + setResponseStateByteCapForTests(1_024); + const replay = JSON.stringify(expandPreviousResponseInput({ + previous_response_id: "resp_budget_unrelated", + input: "next", + })); + return { + settled: true, + reported: reported instanceof Error, + pending, + metrics, + replayedUnrelated: replay.includes("budget-safe-input") && replay.includes("budget-safe-output"), + guardReported: messages.some(message => message.includes("terminalization pass limit")), + }; + } finally { + setAsyncIcaclsRunnerForTests(null); + setIcaclsRunnerForTests(null); + setPlatformForTests(null); + resetHardenedStateForTests(); + setResponseSpillShutdownBudgetForTests(null); + setResponseSpillShutdownTerminalizationPassLimitForTests(null); + setResponseStateByteCapForTests(null); + clearResponseStateForTests(); + rmSync(home, { recursive: true, force: true }); + if (priorHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = priorHome; + } +} + +const scenario = process.argv[2]; +if (scenario !== "exhaustion" && scenario !== "guard") { + throw new Error(`Unknown shutdown budget scenario: ${scenario ?? ""}`); +} + +console.log(JSON.stringify(await runScenario(scenario))); diff --git a/tests/responses-state.test.ts b/tests/responses-state.test.ts index aa05690e38..8b6690db4e 100644 --- a/tests/responses-state.test.ts +++ b/tests/responses-state.test.ts @@ -43,20 +43,26 @@ import { responseStatePersistPendingForTests, responseContinuationRetainedStoreSnapshot, runPendingResponseStatePersistForTests, + setResponseSpillAsyncAclAttemptBudgetForTests, setResponseStateByteCapForTests, setResponseStatePersistAttemptHookForTests, + setResponseSpillShutdownBudgetForTests, getStoredResponseBytesForTests, + flushPendingResponseSpillsForTests, + pendingResponseSpillMetricsForTests, } from "../src/responses/state"; import { readResponseSpill, deleteResponseSpill, recoverOrphanedResponseSpills, responseSpillDirectory, + setResponseSpillNowForTests, setResponseSpillPayloadCapForTests, setSpillIoForTest, writeResponseSpillDurably, } from "../src/responses/spill-store"; import { adapterNeedsForcedContinuation, injectDeveloperMessage } from "../src/server/responses"; +import { watchdogMs } from "./helpers/ci-watchdog"; /** * Windows without Developer Mode or admin cannot create a file symlink (EPERM). @@ -80,8 +86,11 @@ import { hardenSecretPath, hardenedSecretPathCountForTests, resetHardenedStateForTests, + setAsyncIcaclsRunnerForTests, setIcaclsRunnerForTests, + setNowForTests, setPlatformForTests, + setStatForTests, timedOutSecretPathCountForTests, } from "../src/lib/windows-secret-acl"; @@ -117,6 +126,100 @@ function spillFileNames(home: string): string[] { return existsSync(dir) ? readdirSync(dir).filter(name => name.endsWith(".spill.json")) : []; } +function spillTempNames(home: string): string[] { + const dir = responseSpillDirectory(home); + return existsSync(dir) ? readdirSync(dir).filter(name => name.endsWith(".tmp")) : []; +} + +interface ShutdownBudgetChildResult { + settled: boolean; + reported: boolean; + pending: { count: number; bytes: number }; + metrics: { residentCount: number; tombstoneCount: number }; + replayedUnrelated: boolean; + guardReported: boolean; +} + +interface NeverSettlingAclChildResult { + settled: boolean; + pending: { count: number; bytes: number }; + metrics: { tombstoneCount: number }; +} + +async function runShutdownBudgetChild( + scenario: "exhaustion" | "guard", +): Promise { + const timeoutMs = watchdogMs(3_000); + const child = Bun.spawn([ + process.execPath, + join(import.meta.dir, "helpers", "responses-state-shutdown-budget-child.ts"), + scenario, + ], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env }, + stdout: "pipe", + stderr: "pipe", + }); + const stdoutPromise = new Response(child.stdout).text(); + const stderrPromise = new Response(child.stderr).text(); + let timedOut = false; + let timer: ReturnType | undefined; + const timeout = new Promise(resolve => { + timer = setTimeout(() => { + timedOut = true; + try { child.kill("SIGKILL"); } catch { /* already exited */ } + void child.exited.then(resolve, () => resolve(-1)); + }, timeoutMs); + }); + const exitCode = await Promise.race([child.exited, timeout]); + if (timer !== undefined) clearTimeout(timer); + const [stdout, stderr] = await Promise.all([stdoutPromise, stderrPromise]); + if (timedOut) { + throw new Error(`response spill shutdown budget child timed out after ${timeoutMs}ms (${scenario})`); + } + if (exitCode !== 0) { + throw new Error(`response spill shutdown budget child exited ${exitCode} (${scenario}): ${stderr.trim()}`); + } + const line = stdout.trim().split(/\r?\n/).at(-1); + if (!line) throw new Error(`response spill shutdown budget child produced no result (${scenario})`); + return JSON.parse(line) as ShutdownBudgetChildResult; +} + +async function runNeverSettlingAclChild( + mode: "principal" | "icacls", +): Promise { + const timeoutMs = watchdogMs(1_500); + const child = Bun.spawn([ + process.execPath, + join(import.meta.dir, "helpers", "responses-state-never-settling-acl-child.ts"), + mode, + ], { + cwd: join(import.meta.dir, ".."), + env: { ...process.env }, + stdout: "pipe", + stderr: "pipe", + }); + const stdoutPromise = new Response(child.stdout).text(); + const stderrPromise = new Response(child.stderr).text(); + let timedOut = false; + let timer: ReturnType | undefined; + const timeout = new Promise(resolve => { + timer = setTimeout(() => { + timedOut = true; + try { child.kill("SIGKILL"); } catch { /* already exited */ } + void child.exited.then(resolve, () => resolve(-1)); + }, timeoutMs); + }); + const exitCode = await Promise.race([child.exited, timeout]); + if (timer !== undefined) clearTimeout(timer); + const [stdout, stderr] = await Promise.all([stdoutPromise, stderrPromise]); + if (timedOut) throw new Error(`never-settling ${mode} child timed out after ${timeoutMs}ms`); + if (exitCode !== 0) throw new Error(`never-settling ${mode} child exited ${exitCode}: ${stderr.trim()}`); + const line = stdout.trim().split(/\r?\n/).at(-1); + if (!line) throw new Error(`never-settling ${mode} child produced no result`); + return JSON.parse(line) as NeverSettlingAclChildResult; +} + function rememberLarge(id: string, text: string, providers?: Parameters[2]): void { rememberResponseState( { model: "test/model", input: text, store: false }, @@ -168,9 +271,16 @@ describe("Responses previous_response_id state", () => { afterEach(() => { setSpillIoForTest(null); + setResponseSpillNowForTests(null); + setAsyncIcaclsRunnerForTests(null); setIcaclsRunnerForTests(null); + setNowForTests(null); setPlatformForTests(null); + setStatForTests(null); resetHardenedStateForTests(); + delete process.env.OPENCODEX_ACL_TIMEOUT_MS; + setResponseSpillShutdownBudgetForTests(null); + setResponseSpillAsyncAclAttemptBudgetForTests(null); setResponseStateByteCapForTests(null); clearResponseStateForTests(); rmSync(home, { recursive: true, force: true }); @@ -710,6 +820,486 @@ describe("Responses previous_response_id state", () => { expect(events).toEqual(["write", "fsync", "close", "harden", "publish", "dir-fsync", "stub-swap"]); }); + test("Windows spill ACL hardening yields the event loop and swaps only after publication", async () => { + setPlatformForTests("win32"); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let announced = false; + setAsyncIcaclsRunnerForTests(async () => { + if (!announced) { + announced = true; + entered(); + } + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + + rememberLarge("resp_async_acl", "x".repeat(8_000)); + await started; + try { + let unrelatedTickRan = false; + setTimeout(() => { unrelatedTickRan = true; }, 0); + await new Promise(resolve => setTimeout(resolve, 0)); + + expect(unrelatedTickRan).toBe(true); + expect(pendingResponseSpillMetricsForTests()).toMatchObject({ count: 1 }); + expect(responseStateMetrics()).toMatchObject({ residentCount: 1, spillStubCount: 0, spillWriteFailures: 0 }); + expect(JSON.stringify(expandPreviousResponseInput({ previous_response_id: "resp_async_acl", input: "next" }))) + .toContain("xxxxxxxx"); + } finally { + release(); + } + await flushPendingResponseSpillsForTests(); + expect(pendingResponseSpillMetricsForTests()).toEqual({ count: 0, bytes: 0 }); + expect(responseStateMetrics()).toMatchObject({ residentCount: 0, spillStubCount: 1, spillWrites: 1, spillWriteFailures: 0 }); + }); + + test("Windows spill retries one transient ACL timeout without installing a tombstone", async () => { + setPlatformForTests("win32"); + process.env.OPENCODEX_ACL_TIMEOUT_MS = "1000"; + let clock = 0; + let grantCalls = 0; + setNowForTests(() => clock); + setAsyncIcaclsRunnerForTests(async args => { + if (args.includes("/grant:r")) { + grantCalls += 1; + if (grantCalls === 1) { + clock = 1_000; + return { success: false, exitCode: null, timedOut: true, stdout: "" }; + } + } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + + rememberLarge("resp_async_acl_retry", "r".repeat(8_000)); + await flushPendingResponseSpillsForTests(); + + expect(grantCalls).toBeGreaterThanOrEqual(2); + expect(responseStateMetrics()).toMatchObject({ + residentCount: 0, + spillStubCount: 1, + tombstoneCount: 0, + spillWrites: 1, + spillWriteFailures: 0, + }); + }); + + test("Windows async spill attempts share one bounded ACL budget across every harden", async () => { + setPlatformForTests("win32"); + let clock = 0; + let firstGrant = true; + const deadlines: number[] = []; + const grantDeadlines: number[] = []; + const hardenTargets: string[] = []; + setNowForTests(() => clock); + setResponseSpillNowForTests(() => clock); + setAsyncIcaclsRunnerForTests(async (args, timeoutMs) => { + deadlines.push(timeoutMs); + if (args.includes("/grant:r")) grantDeadlines.push(timeoutMs); + if (firstGrant && args.includes("/grant:r")) { + firstGrant = false; + clock += timeoutMs; + return { success: false, exitCode: null, timedOut: true, stdout: "" }; + } + const target = String(args[0]); + if (!hardenTargets.includes(target)) hardenTargets.push(target); + clock += [6_000, 2_000, 1_000][hardenTargets.indexOf(target)] ?? 1_000; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setSpillIoForTest({ + link: () => { throw Object.assign(new Error("injected link fallback"), { code: "EPERM" }); }, + }); + setResponseStateByteCapForTests(1_024); + + rememberLarge("resp_async_acl_attempt_budget", "q".repeat(8_000)); + await flushPendingResponseSpillsForTests(); + + expect(deadlines.length).toBeGreaterThanOrEqual(10); + expect(Math.max(...deadlines)).toBeLessThanOrEqual(15_000); + const retryAttemptGrantDeadlines = grantDeadlines.slice(-3); + expect(retryAttemptGrantDeadlines).toHaveLength(3); + expect(retryAttemptGrantDeadlines[1]!).toBeLessThan(retryAttemptGrantDeadlines[0]!); + expect(retryAttemptGrantDeadlines[2]!).toBeLessThan(retryAttemptGrantDeadlines[1]!); + expect(responseStateMetrics()).toMatchObject({ + residentCount: 0, + spillStubCount: 1, + tombstoneCount: 0, + spillWrites: 1, + spillWriteFailures: 0, + }); + }); + + test("Windows spill queue advances past never-settling principal and icacls runners", async () => { + for (const mode of ["principal", "icacls"] as const) { + const result = await runNeverSettlingAclChild(mode); + expect(result).toMatchObject({ + settled: true, + pending: { count: 0, bytes: 0 }, + metrics: { tombstoneCount: 2 }, + }); + } + }, { timeout: (2 * watchdogMs(1_500)) + 2_000 }); + + test("Windows pending spill publication cannot overwrite a newer same-id generation", async () => { + setPlatformForTests("win32"); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let announced = false; + setAsyncIcaclsRunnerForTests(async () => { + if (!announced) { + announced = true; + entered(); + } + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + + rememberLarge("resp_async_replace", `old-${"a".repeat(8_000)}`); + await started; + rememberLarge("resp_async_replace", `new-${"b".repeat(8_000)}`); + release(); + await flushPendingResponseSpillsForTests(); + + const replay = JSON.stringify(expandPreviousResponseInput({ + previous_response_id: "resp_async_replace", + input: "next", + })); + expect(replay).toContain("new-bbbbbbbb"); + expect(replay).not.toContain("old-aaaaaaaa"); + expect(spillFileNames(home)).toHaveLength(1); + expect(responseStateMetrics()).toMatchObject({ spillStubCount: 1, spillWriteFailures: 0 }); + }); + + test("shutdown flush stays pending while Windows spill ACL publication is gated", async () => { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 1_000, fallbackReserveMs: 500 }); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let announced = false; + setAsyncIcaclsRunnerForTests(async () => { + if (!announced) { + announced = true; + entered(); + } + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_shutdown_pending", "p".repeat(2 * 1024 * 1024 + 4_096)); + await started; + + let flushed = false; + const flushing = flushResponseState().then(() => { flushed = true; }); + try { + await new Promise(resolve => setTimeout(resolve, 25)); + expect(flushed).toBe(false); + } finally { + release(); + } + await flushing; + }); + + test("shutdown flush installs an oversized spill before snapshot and restart replay", async () => { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 1_000, fallbackReserveMs: 500 }); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let announced = false; + setAsyncIcaclsRunnerForTests(async () => { + if (!announced) { + announced = true; + entered(); + } + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + const payload = `restart-${"r".repeat(2 * 1024 * 1024 + 4_096)}`; + rememberLarge("resp_shutdown_restart", payload); + await started; + + let flushed = false; + const flushing = flushResponseState().then(() => { flushed = true; }); + await new Promise(resolve => setTimeout(resolve, 25)); + expect(flushed).toBe(false); + release(); + await flushing; + expect(responseStateMetrics()).toMatchObject({ residentCount: 0, spillStubCount: 1 }); + + clearResponseStateMemoryForTests(); + setResponseStateByteCapForTests(1_024); + const replay = JSON.stringify(expandPreviousResponseInput({ + previous_response_id: "resp_shutdown_restart", + input: "next", + })); + expect(replay).toContain("restart-rrrrrrrr"); + expect(responseStateMetrics().spillStubCount).toBe(1); + }); + + test("shutdown drain reaches a stable tail after a publication is appended mid-drain", async () => { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 1_000, fallbackReserveMs: 500 }); + let releaseFirst!: () => void; + let releaseSecond!: () => void; + let firstEntered!: () => void; + let secondEntered!: () => void; + const firstGate = new Promise(resolve => { releaseFirst = resolve; }); + const secondGate = new Promise(resolve => { releaseSecond = resolve; }); + const firstStarted = new Promise(resolve => { firstEntered = resolve; }); + const secondStarted = new Promise(resolve => { secondEntered = resolve; }); + let aclCalls = 0; + setAsyncIcaclsRunnerForTests(async () => { + aclCalls += 1; + if (aclCalls === 1) { + firstEntered(); + await firstGate; + } else if (aclCalls === 7) { + secondEntered(); + await secondGate; + } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_fixed_point_first", "a".repeat(8_000)); + await firstStarted; + + let flushed = false; + const flushing = flushResponseState().then(() => { flushed = true; }); + rememberLarge("resp_fixed_point_second", "b".repeat(8_000)); + releaseFirst(); + await secondStarted; + try { + await new Promise(resolve => setTimeout(resolve, 25)); + expect(flushed).toBe(false); + } finally { + releaseSecond(); + } + await flushing; + expect(responseStateMetrics()).toMatchObject({ residentCount: 0, spillStubCount: 2 }); + }); + + test("shutdown drain cap expiry enters the synchronous spill fallback", async () => { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 120, fallbackReserveMs: 80 }); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + setAsyncIcaclsRunnerForTests(async () => { + entered(); + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + let synchronousCalls = 0; + setIcaclsRunnerForTests(() => { + synchronousCalls += 1; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_shutdown_fallback", "f".repeat(2 * 1024 * 1024 + 4_096)); + await started; + + try { + await flushResponseState(); + expect(synchronousCalls).toBeGreaterThan(0); + expect(pendingResponseSpillMetricsForTests()).toEqual({ count: 0, bytes: 0 }); + expect(responseStateMetrics()).toMatchObject({ residentCount: 0, spillStubCount: 1 }); + } finally { + release(); + } + }); + + test("shutdown fallback spends only its reserved ACL budget", async () => { + setPlatformForTests("win32"); + const totalMs = 500; + const fallbackReserveMs = 300; + setResponseSpillShutdownBudgetForTests({ totalMs, fallbackReserveMs }); + let release!: () => void; + let entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + let aclClock = 0; + setNowForTests(() => aclClock); + setAsyncIcaclsRunnerForTests(async () => { + entered(); + await gate; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + const deadlines: number[] = []; + setIcaclsRunnerForTests((_args, timeoutMs) => { + deadlines.push(timeoutMs); + aclClock += 20; + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_shutdown_budget", "b".repeat(2 * 1024 * 1024 + 4_096)); + await started; + + try { + await flushResponseState(); + } finally { + release(); + } + const logicalElapsedMs = totalMs - fallbackReserveMs + aclClock; + expect(deadlines.length).toBeGreaterThanOrEqual(6); + expect(Math.max(...deadlines)).toBeLessThanOrEqual(Math.floor(fallbackReserveMs / 2)); + expect(logicalElapsedMs).toBeLessThanOrEqual(totalMs); + }); + + test("late async spill completion cannot overwrite the shutdown fallback", async () => { + setPlatformForTests("win32"); + setStatForTests(() => ({ dev: 1n, ino: 10n, ctimeNs: 100n })); + setResponseSpillShutdownBudgetForTests({ totalMs: 120, fallbackReserveMs: 80 }); + let release!: () => void; + let entered!: () => void; + let tempHardenFinished!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const hardened = new Promise(resolve => { tempHardenFinished = resolve; }); + let publishCount = 0; + setSpillIoForTest({ + record: event => { + if (event !== "publish") return; + publishCount += 1; + }, + }); + let tempHardenCalls = 0; + setAsyncIcaclsRunnerForTests(async args => { + if (String(args[0]).includes(".response-spill.")) { + tempHardenCalls += 1; + if (tempHardenCalls === 1) { + entered(); + await gate; + } + if (tempHardenCalls === 3) tempHardenFinished(); + } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setIcaclsRunnerForTests(() => ({ success: true, exitCode: 0, timedOut: false, stdout: "" })); + setResponseStateByteCapForTests(1_024); + const payload = `fallback-${"z".repeat(2 * 1024 * 1024 + 4_096)}`; + rememberLarge("resp_shutdown_late", payload); + await started; + + let fallbackFile: string | undefined; + let abandonedTempCount = -1; + try { + await flushResponseState(); + fallbackFile = spillFileNames(home)[0]; + expect(fallbackFile).toBeDefined(); + abandonedTempCount = spillTempNames(home).length; + } finally { + release(); + } + await hardened; + await new Promise(resolve => setTimeout(resolve, 0)); + + expect(spillFileNames(home)).toEqual([fallbackFile!]); + expect({ abandonedTempCount, publishCount }).toEqual({ abandonedTempCount: 0, publishCount: 1 }); + const replay = JSON.stringify(expandPreviousResponseInput({ + previous_response_id: "resp_shutdown_late", + input: "next", + })); + expect(replay).toContain("fallback-zzzzzzzz"); + expect(responseStateMetrics()).toMatchObject({ residentCount: 0, spillStubCount: 1, spillWrites: 1 }); + }); + + test("shutdown cleanup failure still persists unrelated response state and reports failure", async () => { + setPlatformForTests("win32"); + setResponseSpillShutdownBudgetForTests({ totalMs: 120, fallbackReserveMs: 80 }); + let release!: () => void; + let entered!: () => void; + let tempHardenFinished!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const hardened = new Promise(resolve => { tempHardenFinished = resolve; }); + let tempHardenCalls = 0; + setAsyncIcaclsRunnerForTests(async args => { + if (String(args[0]).includes(".response-spill.")) { + tempHardenCalls += 1; + if (tempHardenCalls === 1) { + entered(); + await gate; + } + if (tempHardenCalls === 3) tempHardenFinished(); + } + return { success: true, exitCode: 0, timedOut: false, stdout: "" }; + }); + setIcaclsRunnerForTests(() => ({ success: true, exitCode: 0, timedOut: false, stdout: "" })); + setResponseStateByteCapForTests(1_024); + rememberLarge("resp_cleanup_failure", "x".repeat(2 * 1024 * 1024 + 4_096)); + await started; + const abandonedTempPath = join(responseSpillDirectory(home), spillTempNames(home)[0]!); + setSpillIoForTest({ + unlink: path => { + if (path === abandonedTempPath) { + throw Object.assign(new Error("injected abandoned temp unlink failure"), { code: "EPERM" }); + } + unlinkSync(path); + }, + }); + rememberResponseState( + { model: "test/model", input: "safe-small-input", store: false }, + fixedResponse("resp_cleanup_unrelated", [{ type: "message", role: "assistant", content: "safe-small-output" }]), + undefined, + { force: true }, + ); + + let reported: unknown; + try { + await flushResponseState(); + } catch (error) { + reported = error; + } finally { + release(); + } + await hardened; + await new Promise(resolve => setTimeout(resolve, 0)); + expect(reported).toBeInstanceOf(Error); + + setSpillIoForTest(null); + clearResponseStateMemoryForTests(); + setResponseStateByteCapForTests(1_024); + const replay = JSON.stringify(expandPreviousResponseInput({ + previous_response_id: "resp_cleanup_unrelated", + input: "next", + })); + expect(replay).toContain("safe-small-input"); + expect(replay).toContain("safe-small-output"); + }); + + test("shutdown fallback budget exhaustion is contained by a child watchdog", async () => { + const result = await runShutdownBudgetChild("exhaustion"); + expect(result).toMatchObject({ + settled: true, + reported: true, + pending: { count: 0, bytes: 0 }, + metrics: { residentCount: 1, tombstoneCount: 2 }, + replayedUnrelated: true, + }); + }, { timeout: watchdogMs(3_000) + 2_000 }); + + test("shutdown terminalization pass guard reports a bounded failure", async () => { + const result = await runShutdownBudgetChild("guard"); + expect(result).toMatchObject({ + settled: true, + reported: true, + pending: { count: 0, bytes: 0 }, + guardReported: true, + }); + }, { timeout: watchdogMs(3_000) + 2_000 }); + test("directory fsync follows spill unlink", () => { const ref = writeResponseSpillDurably("resp_unlink_order", { createdAt: Date.now(), items: ["x"] }); const events: string[] = []; diff --git a/tests/system-restart.test.ts b/tests/system-restart.test.ts index 1b43ca2592..cd62c87dd3 100644 --- a/tests/system-restart.test.ts +++ b/tests/system-restart.test.ts @@ -426,6 +426,31 @@ describe("acceptSystemRestart", () => { ]); }); + test("a reported drain failure uses the uncertain-cleanup restart handoff", async () => { + const calls: string[] = []; + let scheduled: (() => void | Promise) | null = null; + + acceptSystemRestart({ + isDraining: () => false, + getActiveTurnCount: () => 0, + isSupervisedServiceChild: () => false, + listenPort: () => 10123, + schedule: fn => { scheduled = fn; }, + scheduleDeadline: () => () => {}, + setDraining: () => {}, + drainAndShutdown: async () => false, + stopListener: () => { calls.push("stop"); }, + spawnStart: (port, waitForHealth) => { + calls.push(`start:${port}:${waitForHealth ? "ready" : "deferred"}`); + }, + markRecycling: () => { calls.push("recycle"); }, + exitProcess: code => { calls.push(`exit:${code}`); }, + }); + + await scheduled!(); + expect(calls).toEqual(["stop", "start:10123:deferred", "recycle", "exit:1"]); + }); + test("late drain rejection after timeout is observed without a second terminal action", async () => { const calls: string[] = []; let scheduled: (() => void | Promise) | null = null; @@ -620,7 +645,7 @@ describe("acceptSystemRestart", () => { }); await scheduled!(); - expect(calls).toEqual(["latched", "drain", "stop", "start:10123", "recycle", "exit:0"]); + expect(calls).toEqual(["latched", "drain", "stop", "start:10123", "recycle", "exit:1"]); }); test("spawn failure clears OCX_SERVICE so exit cleanup can restore fences", async () => {