fix(worker-pool): clear parse timeouts so --full exits promptly (#146)

* fix(worker-pool): clear parse timeouts so --full exits promptly

Promise.race with uncancellable delay() left 120s timers on the event
loop after workers finished. Replace with explicit setTimeout +
clearTimeout on worker response and dispose.

* test(worker-pool): subprocess exit guard for timer leak

Review follow-up: in-process wall time did not catch orphaned event-loop
timers. Spawn a child that runs parseFilesParallel and assert it exits in
<5s. Export INLINE_PARSE_MAX for fixture gate; remove unused delay().

* test(worker-pool): fail fast when subprocess hangs on timer leak

Race child exit against 6s and kill on hang so regressions fail with a
clear assertion instead of burning the full test timeout.

* test(ci): tighten exit-delay guards

Drop redundant 5s elapsed assertion; hang race is the regression signal.
Wrap Node dist --full smoke in timeout 45 so a timer leak fails CI fast.

* test(worker-pool): address review nits before merge

Stop exporting INLINE_PARSE_MAX; mirror the threshold in tests. Add
node dist --full subprocess exit regression (skips when dist/ absent).

Author: SutuSebastian

.changeset/fix-worker-pool-exit-delay.md

@@ -0,0 +1,5 @@
+---
+"@stainless-code/codemap": patch
+---
+
+Fix `codemap --full` (and other worker-pool parses) appearing to hang ~120s after stats print — clear parse timeout timers when workers respond instead of leaving orphaned `setTimeout` handles on the event loop.

.github/workflows/ci.yml

@@ -131,9 +131,10 @@ jobs:
       # Full rebuild exercises multi-statement DDL (`runSql`), `dropAll`, workers path — Node only.
       - name: Node full index (fixtures/minimal)
         run: |
+          set -euo pipefail
           export CODEMAP_ROOT="$GITHUB_WORKSPACE/fixtures/minimal"
           rm -f "$CODEMAP_ROOT/.codemap.db"
-          node dist/index.mjs --full
+          timeout 45 node dist/index.mjs --full
           node dist/index.mjs query "SELECT COUNT(*) AS files FROM files"
 
   benchmark:

docs/architecture.md

@@ -771,7 +771,7 @@ The full rebuild (`--full`) applies several optimizations that are not safe for
 
 ### Worker thread parallelism
 
-File I/O and parsing dominate full rebuild time. The indexer spawns N worker threads (capped at CPU count, min 2, max 6; override **`CODEMAP_PARSE_WORKERS`**, max 32) via `parse-worker.ts`. Each worker receives a chunk of file paths, reads files from disk, and runs the appropriate parser (oxc-parser, lightningcss, or regex). Per-file parse budget: **`CODEMAP_PARSE_TIMEOUT_MS`** when set, else 10s + ~1ms per 50KB file size capped at 30s (`parse-timeout.ts`). Workers recycle after **`CODEMAP_WORKER_RECYCLE_EVERY`** files (default 250) to limit memory growth. Timeouts and other per-file failures append to `<state-dir>/errors.log` without aborting the run. Workers return structured `ParsedFile` results to the main thread, which handles import resolution and database inserts serially.
+File I/O and parsing dominate full rebuild time. The indexer spawns N worker threads (capped at CPU count, min 2, max 6; override **`CODEMAP_PARSE_WORKERS`**, max 32) via `parse-worker.ts`. Each worker receives a chunk of file paths, reads files from disk, and runs the appropriate parser (oxc-parser, lightningcss, or regex). Per-file parse budget: **`CODEMAP_PARSE_TIMEOUT_MS`** when set, else 10s + ~1ms per 50KB file size capped at 30s (`parse-timeout.ts`); multi-file worker messages cap the sum at 120s. `worker-pool.ts` clears the parse timeout when the worker responds (or on `dispose`) so the process does not wait on orphaned timers after indexing finishes. Workers recycle after **`CODEMAP_WORKER_RECYCLE_EVERY`** files (default 250) to limit memory growth. Timeouts and other per-file failures append to `<state-dir>/errors.log` without aborting the run. Workers return structured `ParsedFile` results to the main thread, which handles import resolution and database inserts serially.
 
 ### Deferred index creation
 

src/application/parse-timeout.ts

@@ -44,7 +44,3 @@ export class ParseTimeoutError extends Error {
     this.timeoutMs = timeoutMs;
   }
 }
-
-export function delay(ms: number): Promise<void> {
-  return new Promise((resolve) => setTimeout(resolve, ms));
-}

src/worker-pool.dist.test.ts

@@ -0,0 +1,56 @@
+/**
+ * Node + dist regression for orphaned parse timers on `--full`.
+ * Binary-level dist smoke also lives in `.github/workflows/ci.yml`.
+ */
+
+import { describe, expect, test } from "bun:test";
+import { existsSync } from "node:fs";
+import { join } from "node:path";
+
+import { globSync } from "./glob-sync";
+
+const repoRoot = join(import.meta.dir, "..");
+const minimalRoot = join(repoRoot, "fixtures", "minimal");
+const distEntry = join(repoRoot, "dist", "index.mjs");
+/** Mirrors `INLINE_PARSE_MAX` in `worker-pool.ts`. */
+const WORKER_POOL_INLINE_PARSE_MAX = 12;
+
+async function expectSubprocessExits(
+  spawn: () => ReturnType<typeof Bun.spawn>,
+): Promise<void> {
+  const proc = spawn();
+  const hangMs = 6_000;
+  const exitOrHang = await Promise.race([
+    proc.exited.then((code) => ({ kind: "exit" as const, code })),
+    Bun.sleep(hangMs).then(async () => {
+      proc.kill();
+      await proc.exited;
+      return { kind: "hang" as const };
+    }),
+  ]);
+
+  expect(exitOrHang.kind).toBe("exit");
+  if (exitOrHang.kind === "exit") {
+    expect(exitOrHang.code).toBe(0);
+  }
+}
+
+describe("node dist --full exit delay", () => {
+  test.skipIf(!existsSync(distEntry))(
+    "subprocess exits promptly after node dist --full (no orphaned timers)",
+    async () => {
+      const files = globSync(["**/*.ts", "**/*.tsx", "**/*.css"], minimalRoot);
+      expect(files.length).toBeGreaterThan(WORKER_POOL_INLINE_PARSE_MAX);
+
+      await expectSubprocessExits(() =>
+        Bun.spawn(["node", distEntry, "--full"], {
+          cwd: repoRoot,
+          env: { ...process.env, CODEMAP_ROOT: minimalRoot },
+          stdout: "ignore",
+          stderr: "pipe",
+        }),
+      );
+    },
+    8_000,
+  );
+});

src/worker-pool.test.ts

@@ -1,11 +1,20 @@
 import { describe, expect, test } from "bun:test";
+import { join } from "node:path";
 
+import { resolveCodemapConfig } from "./config";
+import { globSync } from "./glob-sync";
+import { initCodemap } from "./runtime";
 import {
   parseFilesParallel,
   parseParseWorkerCountOverride,
   parseWorkerRecycleEvery,
 } from "./worker-pool";
 
+const repoRoot = join(import.meta.dir, "..");
+const minimalRoot = join(repoRoot, "fixtures", "minimal");
+/** Mirrors `INLINE_PARSE_MAX` in `worker-pool.ts`. */
+const WORKER_POOL_INLINE_PARSE_MAX = 12;
+
 describe("parseParseWorkerCountOverride", () => {
   test("accepts valid decimal integers", () => {
     expect(parseParseWorkerCountOverride("2")).toBe(2);
@@ -41,4 +50,56 @@ describe("parseFilesParallel", () => {
   test("resolves immediately for an empty file list", async () => {
     await expect(parseFilesParallel([])).resolves.toEqual([]);
   });
+
+  test("returns parsed results via the worker pool path", async () => {
+    initCodemap(resolveCodemapConfig(minimalRoot, undefined));
+    const files = globSync(["**/*.ts", "**/*.tsx", "**/*.css"], minimalRoot);
+    expect(files.length).toBeGreaterThan(WORKER_POOL_INLINE_PARSE_MAX);
+
+    const results = await parseFilesParallel(files);
+    expect(results.length).toBe(files.length);
+  });
+
+  test("subprocess exits promptly after worker-pool parse (no orphaned timers)", async () => {
+    const files = globSync(["**/*.ts", "**/*.tsx", "**/*.css"], minimalRoot);
+    expect(files.length).toBeGreaterThan(WORKER_POOL_INLINE_PARSE_MAX);
+
+    const script = `
+import { resolveCodemapConfig } from "./src/config.ts";
+import { globSync } from "./src/glob-sync.ts";
+import { initCodemap } from "./src/runtime.ts";
+import { parseFilesParallel } from "./src/worker-pool.ts";
+
+const inlineParseMax = ${WORKER_POOL_INLINE_PARSE_MAX};
+const root = ${JSON.stringify(minimalRoot)};
+initCodemap(resolveCodemapConfig(root, undefined));
+const files = globSync(["**/*.ts", "**/*.tsx", "**/*.css"], root);
+if (files.length <= inlineParseMax) {
+  throw new Error("fixture too small for worker-pool path");
+}
+await parseFilesParallel(files);
+`;
+
+    const proc = Bun.spawn([process.execPath, "-e", script], {
+      cwd: repoRoot,
+      stdout: "ignore",
+      stderr: "pipe",
+    });
+    const hangMs = 6_000;
+    const exitOrHang = await Promise.race([
+      proc.exited.then((code) => ({ kind: "exit" as const, code })),
+      Bun.sleep(hangMs).then(async () => {
+        proc.kill();
+        await proc.exited;
+        return { kind: "hang" as const };
+      }),
+    ]);
+    const stderr = await new Response(proc.stderr).text();
+
+    expect(exitOrHang.kind).toBe("exit");
+    if (exitOrHang.kind === "exit") {
+      expect(exitOrHang.code).toBe(0);
+    }
+    expect(stderr).toBe("");
+  }, 8_000);
 });

src/worker-pool.ts

@@ -7,7 +7,6 @@ import { Worker as NodeWorker } from "node:worker_threads";
 import {
   ParseTimeoutError,
   computeParseTimeoutMs,
-  delay,
   parseParseTimeoutMsOverride,
 } from "./application/parse-timeout";
 import { CODEMAP_BUILD_OUTPUT_DIR } from "./build-output";
@@ -89,60 +88,106 @@ interface ParseWorkerSession {
   dispose(): void;
 }
 
+interface PendingParse {
+  gen: number;
+  resolve: (value: WorkerOutput) => void;
+  reject: (err: Error) => void;
+  clearTimer: () => void;
+}
+
 function createParseWorkerSession(): ParseWorkerSession {
   let generation = 0;
-  let pending:
-    | {
-        gen: number;
-        resolve: (value: WorkerOutput) => void;
-        reject: (err: Error) => void;
-      }
-    | undefined;
+  let pending: PendingParse | undefined;
+
+  const clearPending = (): void => {
+    if (pending === undefined) return;
+    pending.clearTimer();
+    pending = undefined;
+  };
+
+  const settleSuccess = (data: WorkerOutput): void => {
+    if (!pending || pending.gen !== generation) return;
+    const { resolve, clearTimer } = pending;
+    clearTimer();
+    pending = undefined;
+    resolve(data);
+  };
+
+  const settleError = (err: Error): void => {
+    if (!pending || pending.gen !== generation) return;
+    const { reject, clearTimer } = pending;
+    clearTimer();
+    pending = undefined;
+    reject(err);
+  };
+
+  const parseWithTimeout = (
+    timeoutMs: number,
+    postMessage: () => void,
+    recycleWorker: () => void,
+  ): Promise<WorkerOutput> => {
+    const gen = generation;
+    return new Promise<WorkerOutput>((resolve, reject) => {
+      let timer: ReturnType<typeof setTimeout> | undefined = setTimeout(() => {
+        timer = undefined;
+        if (!pending || pending.gen !== gen) return;
+        pending = undefined;
+        generation++;
+        recycleWorker();
+        reject(new ParseTimeoutError(timeoutMs));
+      }, timeoutMs);
+
+      pending = {
+        gen,
+        resolve,
+        reject,
+        clearTimer: () => {
+          if (timer !== undefined) {
+            clearTimeout(timer);
+            timer = undefined;
+          }
+        },
+      };
+      postMessage();
+    });
+  };
+
+  const disposeSession = (terminateWorker: () => void): void => {
+    clearPending();
+    generation++;
+    terminateWorker();
+  };
 
   if (IS_BUN) {
     let worker = new Worker(WORKER_URL_BUN);
     const attach = (): void => {
       worker.onmessage = (event: MessageEvent<WorkerOutput>) => {
-        if (!pending || pending.gen !== generation) return;
-        const { resolve } = pending;
-        pending = undefined;
-        resolve(event.data);
+        settleSuccess(event.data);
       };
       worker.onerror = (event: ErrorEvent) => {
-        if (!pending || pending.gen !== generation) return;
-        const { reject } = pending;
-        pending = undefined;
-        reject(event.error ?? new Error(event.message));
+        settleError(event.error ?? new Error(event.message));
       };
     };
     attach();
 
     return {
       parse(input, timeoutMs) {
-        const gen = generation;
-        return Promise.race([
-          new Promise<WorkerOutput>((resolve, reject) => {
-            pending = { gen, resolve, reject };
+        return parseWithTimeout(
+          timeoutMs,
+          () => {
             worker.postMessage(input);
-          }),
-          delay(timeoutMs).then(() => {
-            throw new ParseTimeoutError(timeoutMs);
-          }),
-        ]).catch((err) => {
-          if (err instanceof ParseTimeoutError) {
-            generation++;
-            pending = undefined;
+          },
+          () => {
             worker.terminate();
             worker = new Worker(WORKER_URL_BUN);
             attach();
-          }
-          throw err;
-        });
+          },
+        );
       },
       dispose() {
-        pending = undefined;
-        generation++;
-        worker.terminate();
+        disposeSession(() => {
+          worker.terminate();
+        });
       },
     };
   }
@@ -153,48 +198,34 @@ function createParseWorkerSession(): ParseWorkerSession {
 
   const attachNode = (): void => {
     worker.on("message", (data: WorkerOutput) => {
-      if (!pending || pending.gen !== generation) return;
-      const { resolve } = pending;
-      pending = undefined;
-      resolve(data);
+      settleSuccess(data);
     });
     worker.on("error", (err: Error) => {
-      if (!pending || pending.gen !== generation) return;
-      const { reject } = pending;
-      pending = undefined;
-      reject(err);
+      settleError(err);
     });
   };
   attachNode();
 
   return {
     parse(input, timeoutMs) {
-      const gen = generation;
-      return Promise.race([
-        new Promise<WorkerOutput>((resolve, reject) => {
-          pending = { gen, resolve, reject };
+      return parseWithTimeout(
+        timeoutMs,
+        () => {
           worker.postMessage(input);
-        }),
-        delay(timeoutMs).then(() => {
-          throw new ParseTimeoutError(timeoutMs);
-        }),
-      ]).catch((err) => {
-        if (err instanceof ParseTimeoutError) {
-          generation++;
-          pending = undefined;
+        },
+        () => {
           void worker.terminate();
           worker = new NodeWorker(NODE_WORKER_PATH, {
             type: "module",
           } as import("node:worker_threads").WorkerOptions);
           attachNode();
-        }
-        throw err;
-      });
+        },
+      );
     },
     dispose() {
-      pending = undefined;
-      generation++;
-      void worker.terminate();
+      disposeSession(() => {
+        void worker.terminate();
+      });
     },
   };
 }