Merge pull request #813 from Wibias/fix/ci-windows-no-isolate

fix(storage): await Worker exit before isolate reclaim on Windows

Author: Wibias

src/config.ts

@@ -295,7 +295,10 @@ export function backupConfigBeforeOpenAiTierMigration(
     write: (target, bytes) => writeFileSync(target, bytes),
     harden: target => {
       try { chmodSync(target, 0o600); } catch { /* platform may ignore chmod */ }
-      if (process.platform === "win32") hardenSecretPath(target, { required: true });
+      // Soft-fail: a wedged/failed icacls on CI temp volumes must not abort
+      // startServer mid-suite (timeout + EBUSY cascade on shared TEST_DIR).
+      // chmod above still applies; live credential writes keep required:true.
+      if (process.platform === "win32") hardenSecretPath(target, { required: false });
     },
     publishNoReplace: (temp, backup) => linkSync(temp, backup),
     truncate: target => truncateSync(target, 0),

src/server/lifecycle.ts

@@ -1,10 +1,15 @@
 import { flushResponseState } from "../responses/state";
 import { setStorageCleanupPolicyLiveSink } from "../storage/policy";
 import {
-  abortStorageCleanupPolicyJob,
+  abortStorageCleanupPolicyJobAsync,
   setStorageCleanupPolicyJobLiveApply,
 } from "../storage/policy-job";
+import { abortRestoreTrashJobAsync } from "../storage/restore-job";
 import { stopStorageCleanupScheduler } from "../storage/policy-scheduler";
+import {
+  cancelQueuedStorageWorkerSpawns,
+  drainStorageWorkers,
+} from "../storage/worker-lifecycle";
 
 // ---------------------------------------------------------------------------
 // Active turn tracking + graceful shutdown drain
@@ -88,8 +93,32 @@ export async function drainAndShutdown(
   // previous_response_id chain survives the restart this shutdown is usually part of.
   await flushResponseState();
   // Tear down opt-in storage policy timers / worker / live-config sink so they cannot fire after stop.
+  // Await worker thread exit: on Windows, a still-exiting Bun Worker under
+  // `bun test --isolate` panics the whole process at the next realm reclaim.
+  // Abort each job independently so one wedged join cannot skip the other,
+  // then drain leftovers; failures must not prevent `server.stop`.
   stopStorageCleanupScheduler();
-  abortStorageCleanupPolicyJob();
+  cancelQueuedStorageWorkerSpawns();
+  const shutdownJoins = await Promise.allSettled([
+    abortStorageCleanupPolicyJobAsync(),
+    abortRestoreTrashJobAsync(),
+  ]);
+  for (const result of shutdownJoins) {
+    if (result.status === "rejected") {
+      console.warn(
+        "[storage] worker abort during shutdown failed:",
+        result.reason instanceof Error ? result.reason.message : result.reason,
+      );
+    }
+  }
+  try {
+    await drainStorageWorkers();
+  } catch (err) {
+    console.warn(
+      "[storage] worker drain during shutdown failed:",
+      err instanceof Error ? err.message : err,
+    );
+  }
   setStorageCleanupPolicyLiveSink(null);
   setStorageCleanupPolicyJobLiveApply(null);
   s?.stop(true);

src/server/relay.ts

@@ -854,8 +854,10 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void {
       stopDrain();
       return;
     }
+    // Do not unref: on Bun/Windows a pending `reader.read()` can be the only
+    // wake source; an unref'd timer may never run, so a silent post-cancel
+    // drain (time bound, no bytes) hangs the suite until the job timeout.
     drainTimer = setTimeout(stopDrain, drainMs);
-    (drainTimer as { unref?: () => void }).unref?.();
   };
   // Ends the bounded drain by cancelling the reader: the pending read settles
   // and the pump loop observes `drainStopped`. Deliberately NOT a shared

src/storage/policy-job.ts

@@ -19,9 +19,11 @@ import {
   type PolicySkipReason,
 } from "./policy";
 import {
+  cancelQueuedStorageWorkerSpawns,
   drainStorageWorkers,
   registerStorageWorker,
   terminateStorageWorker,
+  withStorageWorkerSpawnGate,
 } from "./worker-lifecycle";
 
 export type PolicyJobStatus = "idle" | "running";
@@ -148,10 +150,27 @@ export function resetStorageCleanupPolicyJobForTests(): void {
  * worker still exiting at that moment trips a Bun-internal assertion on Windows
  * and takes the whole run down, so a suite that spawns workers must be able to
  * wait for them rather than fire-and-forget.
+ *
+ * Do not route through the sync reset's `void terminate(...)`: that used to
+ * deregister the worker before the thread exited, so a follow-up drain saw an
+ * empty registry and returned while Windows was still reclaiming the thread.
  */
 export async function resetStorageCleanupPolicyJobForTestsAsync(): Promise<void> {
-  resetStorageCleanupPolicyJobForTests();
-  await drainStorageWorkers();
+  disownActiveRun();
+  cancelQueuedStorageWorkerSpawns();
+  const worker = activeWorker;
+  activeWorker = null;
+  inflight = null;
+  testHooks = null;
+  state = { status: "idle" };
+  // Join the worker before releasing the mutation slot so a concurrent run
+  // cannot acquire CODEX_HOME while the aborted thread is still mutating.
+  try {
+    if (worker) await terminateStorageWorker(worker);
+    await drainStorageWorkers();
+  } finally {
+    releaseHeldMutationSlot();
+  }
 }
 
 /** Terminate an in-flight worker during process shutdown. */
@@ -161,6 +180,8 @@ export function abortStorageCleanupPolicyJob(): void {
     void terminateStorageWorker(activeWorker);
     activeWorker = null;
   }
+  // Sync path cannot join; prefer abortStorageCleanupPolicyJobAsync when the
+  // caller can await so the mutation slot stays held until the thread exits.
   releaseHeldMutationSlot();
   if (state.status === "running") {
     state = {
@@ -177,6 +198,37 @@ export function abortStorageCleanupPolicyJob(): void {
   inflight = null;
 }
 
+/**
+ * Shutdown path that joins worker threads. Prefer this over the sync abort when
+ * the caller can await (server drain, test teardown) so Windows isolate reclaim
+ * does not race a still-exiting storage worker.
+ */
+export async function abortStorageCleanupPolicyJobAsync(): Promise<void> {
+  disownActiveRun();
+  cancelQueuedStorageWorkerSpawns();
+  const worker = activeWorker;
+  activeWorker = null;
+  if (state.status === "running") {
+    state = {
+      ...state,
+      status: "idle",
+      finishedAt: Date.now(),
+      lastError: "aborted",
+      lastOutcome: {
+        ok: false,
+        error: "evaluation_failed",
+      },
+    };
+  }
+  inflight = null;
+  try {
+    if (worker) await terminateStorageWorker(worker);
+    await drainStorageWorkers();
+  } finally {
+    releaseHeldMutationSlot();
+  }
+}
+
 function outcomeFromResult(result: PolicyRunResult): PolicyJobOutcome {
   return {
     ok: result.ok,
@@ -232,22 +284,13 @@ function applyMutationBusy(): void {
 }
 
 function runInWorker(opts: RequestPolicyRunOptions & { blockMs?: number }): Promise<PolicyRunResult> {
-  return new Promise((resolve, reject) => {
+  return withStorageWorkerSpawnGate(() => new Promise<PolicyRunResult>((resolve, reject) => {
     const requestId = crypto.randomUUID();
     let settled = false;
     const worker = new Worker(new URL("./policy-worker.ts", import.meta.url).href);
     registerStorageWorker(worker);
     activeWorker = worker;
 
-    const timer = setTimeout(() => {
-      if (settled) return;
-      settled = true;
-      cancelActiveRun = null;
-      void terminateStorageWorker(worker);
-      if (activeWorker === worker) activeWorker = null;
-      reject(new Error("storage_cleanup_worker_timeout"));
-    }, WORKER_TIMEOUT_MS);
-
     const finish = (fn: () => void) => {
       if (settled) return;
       settled = true;
@@ -256,10 +299,15 @@ function runInWorker(opts: RequestPolicyRunOptions & { blockMs?: number }): Prom
       if (activeWorker === worker) activeWorker = null;
       // Settle the caller only after the thread is actually gone, so a suite
       // that awaits its request cannot reach the next test file with a worker
-      // still exiting behind it.
+      // still exiting behind it. Also keeps executeJob's finally from releasing
+      // the mutation slot while the thread may still mutate CODEX_HOME.
       void terminateStorageWorker(worker).then(fn, fn);
     };
 
+    const timer = setTimeout(() => {
+      finish(() => reject(new Error("storage_cleanup_worker_timeout")));
+    }, WORKER_TIMEOUT_MS);
+
     cancelActiveRun = () => {
       finish(() => reject(new Error("aborted")));
     };
@@ -296,7 +344,7 @@ function runInWorker(opts: RequestPolicyRunOptions & { blockMs?: number }): Prom
         ...(process.env.OPENCODEX_HOME ? { OPENCODEX_HOME: process.env.OPENCODEX_HOME } : {}),
       },
     });
-  });
+  }));
 }
 
 async function executeJob(opts: RequestPolicyRunOptions): Promise<void> {

src/storage/policy-worker.ts

@@ -49,5 +49,11 @@ self.onmessage = (event: MessageEvent<unknown>) => {
       requestId,
       message: err instanceof Error ? err.message : "worker_failed",
     });
+  } finally {
+    // Close from inside the worker so the thread begins exiting before the
+    // parent’s terminate() races isolate realm reclaim on Windows.
+    try {
+      (self as unknown as { close?: () => void }).close?.();
+    } catch { /* already closing */ }
   }
 };

src/storage/restore-job.ts

@@ -17,9 +17,11 @@ import {
   type StorageMutationCoordinatorTestHooks,
 } from "./storage-mutation-coordinator";
 import {
+  cancelQueuedStorageWorkerSpawns,
   drainStorageWorkers,
   registerStorageWorker,
   terminateStorageWorker,
+  withStorageWorkerSpawnGate,
 } from "./worker-lifecycle";
 
 export interface RestoreJobTestHooks extends StorageMutationCoordinatorTestHooks {
@@ -83,9 +85,20 @@ export function resetRestoreTrashJobForTests(): void {
   resetStorageMutationCoordinatorForTests();
 }
 
-/** Await-able reset for test teardown; see policy-job's equivalent for why. */
+/**
+ * Await-able reset for test teardown; see policy-job's equivalent for why.
+ * Awaits the active worker directly so drain cannot race a fire-and-forget
+ * deregister from the sync reset.
+ */
 export async function resetRestoreTrashJobForTestsAsync(): Promise<void> {
-  resetRestoreTrashJobForTests();
+  cancelQueuedStorageWorkerSpawns();
+  const worker = activeWorker;
+  activeWorker = null;
+  cancelActiveRun?.();
+  cancelActiveRun = null;
+  testHooks = null;
+  resetStorageMutationCoordinatorForTests();
+  if (worker) await terminateStorageWorker(worker);
   await drainStorageWorkers();
 }
 
@@ -99,6 +112,17 @@ export function abortRestoreTrashJob(): void {
   cancelActiveRun = null;
 }
 
+/** Await-able shutdown sibling; joins the restore worker thread before returning. */
+export async function abortRestoreTrashJobAsync(): Promise<void> {
+  cancelQueuedStorageWorkerSpawns();
+  const worker = activeWorker;
+  activeWorker = null;
+  cancelActiveRun?.();
+  cancelActiveRun = null;
+  if (worker) await terminateStorageWorker(worker);
+  await drainStorageWorkers();
+}
+
 /** Test-only SSE/text stream served from the proxy while a worker is blocked. */
 export function getRestoreTrashTestStreamResponse(): Response | null {
   if (!testHooks?.enableTestStream) return null;
@@ -124,31 +148,28 @@ function runInWorker(opts: {
   blockMs?: number;
   restoreTest?: RestoreTestHooks;
 }): Promise<RestoreResult> {
-  return new Promise((resolve, reject) => {
+  return withStorageWorkerSpawnGate(() => new Promise<RestoreResult>((resolve, reject) => {
     const requestId = crypto.randomUUID();
     let settled = false;
     const worker = new Worker(new URL("./restore-worker.ts", import.meta.url).href);
     registerStorageWorker(worker);
     activeWorker = worker;
 
-    const timer = setTimeout(() => {
-      if (settled) return;
-      settled = true;
-      cancelActiveRun = null;
-      void terminateStorageWorker(worker);
-      if (activeWorker === worker) activeWorker = null;
-      reject(new Error("restore_worker_timeout"));
-    }, WORKER_TIMEOUT_MS);
-
     const finish = (fn: () => void) => {
       if (settled) return;
       settled = true;
       cancelActiveRun = null;
       clearTimeout(timer);
       if (activeWorker === worker) activeWorker = null;
+      // Join before settle so mutation coordination inside the restore worker
+      // cannot overlap a follow-on run after a watchdog timeout.
       void terminateStorageWorker(worker).then(fn, fn);
     };
 
+    const timer = setTimeout(() => {
+      finish(() => reject(new Error("restore_worker_timeout")));
+    }, WORKER_TIMEOUT_MS);
+
     cancelActiveRun = () => {
       finish(() => reject(new Error("aborted")));
     };
@@ -185,7 +206,7 @@ function runInWorker(opts: {
         ...(process.env.OPENCODEX_HOME ? { OPENCODEX_HOME: process.env.OPENCODEX_HOME } : {}),
       },
     });
-  });
+  }));
 }
 
 async function executeRestore(opts: {

src/storage/restore-worker.ts

@@ -46,6 +46,12 @@ self.onmessage = async (event: MessageEvent<unknown>) => {
       requestId,
       message: err instanceof Error ? err.message : "worker_failed",
     });
+  } finally {
+    // Close from inside the worker so the thread begins exiting before the
+    // parent’s terminate() races isolate realm reclaim on Windows.
+    try {
+      (self as unknown as { close?: () => void }).close?.();
+    } catch { /* already closing */ }
   }
 };