diff --git a/src/storage/cleanup.ts b/src/storage/cleanup.ts index f6631948df..c39bbeedf1 100644 --- a/src/storage/cleanup.ts +++ b/src/storage/cleanup.ts @@ -2578,6 +2578,11 @@ export interface RestoreTestHooks { * reconcile, so cleanup can race an in-flight restore. */ holdAfterFileMovesMs?: number; + /** + * Test-only: publish a ready file after rollout moves, then wait until the + * release file exists. This makes cross-thread race tests phase-driven. + */ + pauseAfterFileMoves?: { readyPath: string; releasePath: string }; } /** @@ -2917,6 +2922,11 @@ export function restoreTrashEntry( return { ok: false, trashDir: id, ...partialCounts, error }; }; + if (hooks?.pauseAfterFileMoves) { + writeFileSync(hooks.pauseAfterFileMoves.readyPath, "ready\n"); + while (!existsSync(hooks.pauseAfterFileMoves.releasePath)) Bun.sleepSync(10); + } + if (hooks?.holdAfterFileMovesMs !== undefined) { const holdMs = Math.max(0, Math.floor(hooks.holdAfterFileMovesMs)); if (holdMs > 0) { diff --git a/src/storage/storage-mutation-coordinator.ts b/src/storage/storage-mutation-coordinator.ts index 695711dfca..c365420c6a 100644 --- a/src/storage/storage-mutation-coordinator.ts +++ b/src/storage/storage-mutation-coordinator.ts @@ -16,6 +16,12 @@ export type StorageMutationBusyError = "storage_mutation_busy"; export interface StorageMutationCoordinatorTestHooks { /** Block after acquiring the slot, before mutation work (race tests). */ blockMs?: number; + /** Test-only cross-thread handshake after one mutation kind acquires its slot. */ + pauseAfterAcquire?: { + kind: StorageMutationKind; + readyPath: string; + releasePath: string; + }; } interface ActiveSlot { @@ -93,7 +99,12 @@ export function endStorageMutation(codexHome?: string): void { slot.lease.release(); } -async function applyCoordinatorBlock(): Promise { +async function applyCoordinatorBlock(kind: StorageMutationKind): Promise { + const pause = testHooks?.pauseAfterAcquire; + if (pause?.kind === kind) { + await Bun.write(pause.readyPath, "ready\n"); + while (!Bun.file(pause.releasePath).size) await Bun.sleep(10); + } const blockMs = testHooks?.blockMs; if (typeof blockMs === "number" && Number.isFinite(blockMs) && blockMs > 0) { await Bun.sleep(Math.floor(blockMs)); @@ -114,7 +125,7 @@ export async function runPolicyStorageMutation( return { ok: false, error: "storage_mutation_busy" }; } try { - await applyCoordinatorBlock(); + await applyCoordinatorBlock("policy"); return await work(); } finally { gate.lease.release(); @@ -131,7 +142,7 @@ export async function withStorageMutationSlot( return { ok: false, error: "storage_mutation_busy" }; } try { - await applyCoordinatorBlock(); + await applyCoordinatorBlock(kind); return await work(); } finally { gate.lease.release(); diff --git a/tests/init-eof.test.ts b/tests/init-eof.test.ts index 297abe4679..53c4c857bf 100644 --- a/tests/init-eof.test.ts +++ b/tests/init-eof.test.ts @@ -3,6 +3,24 @@ import { existsSync, mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +async function waitForOutput( + stream: ReadableStream, + expected: string, +): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let output = ""; + try { + while (!output.includes(expected)) { + const { value, done } = await reader.read(); + if (done) throw new Error(`init exited before writing ${JSON.stringify(expected)}`); + output += decoder.decode(value, { stream: true }); + } + } finally { + reader.releaseLock(); + } +} + describe("ocx init piped stdin (#754)", () => { const dirs: string[] = []; afterEach(() => { @@ -20,14 +38,20 @@ describe("ocx init piped stdin (#754)", () => { stdout: "pipe", stderr: "pipe", }); - proc.stdin.end(); - const exit = await Promise.race([ - proc.exited, - new Promise((_, reject) => setTimeout(() => reject(new Error("init did not exit after stdin EOF")), 8_000)), - ]); - expect(exit).toBe(1); - const stderr = await new Response(proc.stderr).text(); - expect(stderr.toLowerCase()).toMatch(/stdin (closed|reached eof)/); - expect(existsSync(join(home, "config.json"))).toBe(false); - }); + const stderrPromise = new Response(proc.stderr).text(); + try { + // Synchronize on the behavior under test, not Windows process startup/import time. + // EOF now arrives while readline is waiting for the first answer. + await waitForOutput(proc.stdout, "Select default provider (number):"); + proc.stdin.end(); + + expect(await proc.exited).toBe(1); + const stderr = await stderrPromise; + expect(stderr.toLowerCase()).toMatch(/stdin (closed|reached eof)/); + expect(existsSync(join(home, "config.json"))).toBe(false); + } finally { + if (proc.exitCode === null) proc.kill(); + await proc.exited.catch(() => {}); + } + }, 30_000); }); diff --git a/tests/service.test.ts b/tests/service.test.ts index ed1bc1d944..3b48a76c43 100644 --- a/tests/service.test.ts +++ b/tests/service.test.ts @@ -3,6 +3,7 @@ import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, wri import { execFileSync } from "node:child_process"; import { tmpdir } from "node:os"; import { delimiter, isAbsolute, join, posix, win32 } from "node:path"; +import { pathToFileURL } from "node:url"; import * as serviceModule from "../src/service"; import { saveConfig } from "../src/config"; import { windowsEnvIndirectBatchValue } from "../src/lib/win-paths"; @@ -956,15 +957,24 @@ describe("launchd service plist", () => { const shimDir = join(root, "shims"); const v1 = join(root, "installs", "2.35.0 package's"); const v2 = join(root, "installs", "2.36.0 package's"); - const quoteForSh = (value: string): string => `'${value.replaceAll("'", "'\"'\"'")}'`; mkdirSync(shimDir, { recursive: true }); mkdirSync(v1, { recursive: true }); mkdirSync(v2, { recursive: true }); - writeFileSync(join(v1, "ocx"), "#!/bin/sh\necho V1 \"$@\"\n", { mode: 0o755 }); - writeFileSync(join(v2, "ocx"), "#!/bin/sh\necho V2 \"$@\"\n", { mode: 0o755 }); + const v1Entry = join(v1, "ocx"); + const v2Entry = join(v2, "ocx"); + writeFileSync(v1Entry, 'console.log("V1", Bun.argv.slice(2).join(" "));\n'); + writeFileSync(v2Entry, 'console.log("V2", Bun.argv.slice(2).join(" "));\n'); const shim = join(shimDir, "ocx"); - writeFileSync(shim, `#!/bin/sh\nexec ${quoteForSh(join(v1, "ocx"))} "\$@"\n`, { mode: 0o755 }); + const retargetShim = (target: string): void => { + writeFileSync(shim, `await import(${JSON.stringify(pathToFileURL(target).href)});\n`); + }; + const runShim = (): string => execFileSync( + process.execPath, + [shim, "start", "--port", "1"], + { encoding: "utf8" }, + ); + retargetShim(v1Entry); // stableLauncherEntry finds the shim lexically from PATH — not its versioned target. const found = buildUnit(resolvedProxyEnv({}), { launcher: shim }); @@ -977,13 +987,15 @@ describe("launchd service plist", () => { const windowsUnit = buildUnit(resolvedProxyEnv({}), { launcher: windowsShim }); expectTextToContainPath(windowsUnit, windowsShim); - expect(execFileSync(shim, ["start", "--port", "1"], { encoding: "utf8" })).toContain("V1"); + // Exercise the retarget through Bun on every host. Directly executing the old + // extensionless #!/bin/sh fixture was itself a POSIX-only assumption. + expect(runShim()).toContain("V1"); // The upgrade: shim retargeted, old version removed. - writeFileSync(shim, `#!/bin/sh\nexec ${quoteForSh(join(v2, "ocx"))} "\$@"\n`, { mode: 0o755 }); + retargetShim(v2Entry); rmSync(v1, { recursive: true, force: true }); - expect(existsSync(join(v1, "ocx"))).toBe(false); - expect(execFileSync(shim, ["start", "--port", "1"], { encoding: "utf8" })).toContain("V2"); + expect(existsSync(v1Entry)).toBe(false); + expect(runShim()).toContain("V2"); rmSync(root, { recursive: true, force: true }); }); diff --git a/tests/storage-mutation-race.test.ts b/tests/storage-mutation-race.test.ts index 846ae1fad5..3f7a95382f 100644 --- a/tests/storage-mutation-race.test.ts +++ b/tests/storage-mutation-race.test.ts @@ -144,6 +144,18 @@ async function waitForPolicyJob( throw new Error("policy job did not finish"); } +async function waitForCondition( + description: string, + condition: () => boolean, + timeoutMs = 8_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() >= deadline) throw new Error(`timed out waiting for ${description}`); + await Bun.sleep(20); + } +} + /** Windows can keep SQLite/job handles briefly after stop; retry only transient cleanup codes. */ function removeTree(path: string): void { let lastError: unknown; @@ -313,13 +325,20 @@ describe("storage mutation coordinator", () => { test("cleanup quarantine and permanent are rejected while restore holds slot after file moves", async () => { const home = isolatedCodexHome!.path; - const holdMs = 2500; + const movedReadyPath = join(testDir, "restore-files-moved.ready"); + const releaseRestorePath = join(testDir, "release-restore"); setRestoreTrashJobTestHooks({ - restoreTest: { holdAfterFileMovesMs: holdMs }, + restoreTest: { + pauseAfterFileMoves: { + readyPath: movedReadyPath, + releasePath: releaseRestorePath, + }, + }, }); seedArchivedPair(home); const server = startServer(0); + let restorePromise: Promise | null = null; try { const preview = await previewDigest(server.url, 50); const cleanupRes = await fetch(new URL("/api/storage/cleanup", server.url), { @@ -336,17 +355,17 @@ describe("storage mutation coordinator", () => { const remainingPreview = await previewDigest(server.url, 50); expect(remainingPreview.count).toBe(1); - const restorePromise = fetch(new URL("/api/storage/trash/restore", server.url), { + restorePromise = fetch(new URL("/api/storage/trash/restore", server.url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ id: trashId }), }); const restoredPath = join(home, "archived_sessions", "rollout-old.jsonl"); - const movedDeadline = Date.now() + 8000; - while (!existsSync(restoredPath) && Date.now() < movedDeadline) { - await Bun.sleep(20); - } + await waitForCondition( + "restore worker to finish file moves", + () => existsSync(movedReadyPath), + ); expect(existsSync(restoredPath)).toBe(true); expect(existsSync(join(trashStage, "rollout-old.jsonl"))).toBe(false); expect(existsSync(join(trashStage, "restore-pending.json"))).toBe(true); @@ -378,6 +397,7 @@ describe("storage mutation coordinator", () => { expect(permanentDuring.status).toBe(409); expect((await permanentDuring.json()).error).toBe("storage_mutation_busy"); + writeFileSync(releaseRestorePath, "release\n"); const restoreRes = await restorePromise; expect(restoreRes.status).toBe(200); const restored = await restoreRes.json(); @@ -386,26 +406,41 @@ describe("storage mutation coordinator", () => { expect(threadCount(home)).toBe(2); expect(readFileSync(restoredPath, "utf8")).toBe("o".repeat(100)); } finally { + // Never strand the Worker if an assertion above fails; also consume the + // request so its rejection cannot leak into the next isolated test. + writeFileSync(releaseRestorePath, "release\n"); + if (restorePromise) await restorePromise.catch(() => undefined); await stopRaceServer(server); } }, { timeout: 45_000 }); test("restore is rejected while cleanup holds the shared mutation slot", async () => { const home = isolatedCodexHome!.path; - const blockMs = 1200; - setArchivedCleanupJobTestHooks({ blockMs }); + const cleanupReadyPath = join(testDir, "cleanup-slot-acquired.ready"); + const releaseCleanupPath = join(testDir, "release-cleanup"); + setArchivedCleanupJobTestHooks({ + pauseAfterAcquire: { + kind: "cleanup", + readyPath: cleanupReadyPath, + releasePath: releaseCleanupPath, + }, + }); seedArchivedPair(home); const server = startServer(0); + let cleanupPromise: Promise | null = null; try { const preview = await previewDigest(server.url, 50); - const cleanupPromise = fetch(new URL("/api/storage/cleanup", server.url), { + cleanupPromise = fetch(new URL("/api/storage/cleanup", server.url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ percent: 50, mode: "quarantine", digest: preview.digest }), }); - await Bun.sleep(80); + await waitForCondition( + "manual cleanup to acquire the storage mutation slot", + () => existsSync(cleanupReadyPath), + ); const restoreAttempt = await fetch(new URL("/api/storage/trash/restore", server.url), { method: "POST", @@ -418,6 +453,7 @@ describe("storage mutation coordinator", () => { expect(existsSync(join(home, "archived_sessions", "rollout-new.jsonl"))).toBe(true); expect(trashStageCount(home)).toBe(0); + writeFileSync(releaseCleanupPath, "release\n"); const cleanupRes = await cleanupPromise; expect(cleanupRes.status).toBe(200); const cleanup = await cleanupRes.json(); @@ -427,6 +463,8 @@ describe("storage mutation coordinator", () => { expect(existsSync(join(home, "archived_sessions", "rollout-new.jsonl"))).toBe(true); expect(threadCount(home)).toBe(1); } finally { + writeFileSync(releaseCleanupPath, "release\n"); + if (cleanupPromise) await cleanupPromise.catch(() => undefined); await stopRaceServer(server); } }, { timeout: 30_000 });