fix: propagate routing cancellation and track concurrent phase progress

lib/shared/generate-circuit-json.tsx

@@ -170,24 +170,30 @@ export async function generateCircuitJson({
 
   runner.add(<Component {...(injectedProps ?? {})} />)
 
-  runner.render()
-
   const loggedAsyncEffectNames = new Set<string>()
 
-  while (!runner.isDoneRendering()) {
-    for (const asyncEffect of runner.getRunningAsyncEffects()) {
-      const asyncEffectName = asyncEffect.effectName
-      if (!asyncEffectName || loggedAsyncEffectNames.has(asyncEffectName)) {
-        continue
+  try {
+    runner.render()
+
+    while (!runner.isDoneRendering()) {
+      for (const asyncEffect of runner.getRunningAsyncEffects()) {
+        const asyncEffectName = asyncEffect.effectName
+        if (!asyncEffectName || loggedAsyncEffectNames.has(asyncEffectName)) {
+          continue
+        }
+
+        loggedAsyncEffectNames.add(asyncEffectName)
+        onAsyncEffectStatus?.(asyncEffectName)
       }
 
-      loggedAsyncEffectNames.add(asyncEffectName)
-      onAsyncEffectStatus?.(asyncEffectName)
+      autorouterDiagnostics.checkTimeout()
+      await new Promise((resolve) => setTimeout(resolve, 100))
+      runner.render()
     }
-
-    autorouterDiagnostics.checkTimeout()
-    await new Promise((resolve) => setTimeout(resolve, 100))
-    runner.render()
+  } catch (error) {
+    // User projects can use older tscircuit versions without cancellation.
+    runner.cancelRendering?.(error)
+    throw error
   }
 
   runner.emit("renderComplete")

lib/shared/thread-worker-pool.ts

@@ -49,6 +49,7 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
   private stopped = false
   private stopReason: Error | null = null
   private heartbeatIntervalId: NodeJS.Timeout | null = null
+  private terminationPromise: Promise<void> | null = null
 
   constructor(
     options: ThreadWorkerPoolOptions<
@@ -202,6 +203,8 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
   }
 
   private replaceWorker(threadWorker: ThreadWorker<TJob, TResult>): void {
+    if (this.stopped) return
+
     this.clearWorkerTimeout(threadWorker)
     void threadWorker.worker.terminate().catch(() => undefined)
 
@@ -229,7 +232,6 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
     threadWorker.currentStatus = null
     threadWorker.busy = false
     action(job)
-    this.processQueue()
   }
 
   private attachWorkerHandlers(
@@ -238,7 +240,7 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
     const worker = threadWorker.worker
 
     worker.on("message", (message: TWorkerOutput) => {
-      if (threadWorker.worker !== worker) {
+      if (this.stopped || threadWorker.worker !== worker) {
         return
       }
 
@@ -263,10 +265,11 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
 
         job.resolve(this.options.getResult(message))
       })
+      this.processQueue()
     })
 
     worker.on("error", (error) => {
-      if (threadWorker.worker !== worker) {
+      if (this.stopped || threadWorker.worker !== worker) {
         return
       }
 
@@ -283,18 +286,16 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
     })
 
     worker.on("exit", (code) => {
-      if (threadWorker.worker !== worker) {
+      if (this.stopped || threadWorker.worker !== worker) {
         return
       }
 
-      if (code !== 0) {
-        this.finishJob(threadWorker, (job) => {
-          job.reject(new Error(`Worker exited with code ${code}`))
-        })
+      this.finishJob(threadWorker, (job) => {
+        job.reject(new Error(`Worker exited with code ${code}`))
+      })
 
-        this.replaceWorker(threadWorker)
-        this.processQueue()
-      }
+      this.replaceWorker(threadWorker)
+      this.processQueue()
     })
   }
 
@@ -330,14 +331,21 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
 
     await this.initWorkers()
 
+    if (this.stopped) {
+      throw this.stopReason ?? new Error("Worker pool stopped")
+    }
+
     return new Promise((resolve, reject) => {
       this.jobQueue.push({ job, resolve, reject })
       this.processQueue()
     })
   }
 
   async stop(reason: Error): Promise<void> {
-    if (this.stopped) return
+    if (this.stopped) {
+      await this.terminationPromise
+      return
+    }
 
     this.stopped = true
     this.stopHeartbeat()
@@ -346,17 +354,25 @@ export class ThreadWorkerPool<TJob, TWorkerInput, TWorkerOutput, TResult> {
       queuedJob.reject(reason)
     }
     this.jobQueue = []
-  }
+    const workers = this.workers
+    this.workers = []
+    this.initialized = false
 
-  async terminate(): Promise<void> {
-    this.stopHeartbeat()
-    await Promise.all(
-      this.workers.map((worker) => {
+    this.terminationPromise = Promise.all(
+      workers.map((worker) => {
         this.clearWorkerTimeout(worker)
+        worker.currentJob?.reject(reason)
+        worker.currentJob = null
+        worker.currentJobStartedAt = null
+        worker.currentStatus = null
+        worker.busy = false
         return worker.worker.terminate()
       }),
-    )
-    this.workers = []
-    this.initialized = false
+    ).then(() => undefined)
+    await this.terminationPromise
+  }
+
+  async terminate(): Promise<void> {
+    await this.stop(new Error("Worker pool terminated"))
   }
 }

tests/fixtures/get-thread-worker-pool-test-fixture.ts

@@ -0,0 +1,74 @@
+import fs from "node:fs"
+import os from "node:os"
+import path from "node:path"
+import { ThreadWorkerPool } from "lib/shared/thread-worker-pool"
+
+type TestJob = {
+  id: string
+  outcome: "hang" | "fatal" | "success" | "error" | "exit" | "clean_exit"
+}
+type TestMessage =
+  | { type: "progress"; id: string }
+  | { type: "complete"; id: string; fatal: boolean }
+
+export const getThreadWorkerPoolTestFixture = (concurrency = 1) => {
+  const directory = fs.mkdtempSync(path.join(os.tmpdir(), "worker-pool-"))
+  const workerPath = path.join(directory, "worker.cjs")
+  fs.writeFileSync(
+    workerPath,
+    `const { parentPort } = require("node:worker_threads")
+parentPort.on("message", (job) => {
+  parentPort.postMessage({ type: "progress", id: job.id })
+  if (job.outcome === "error") throw new Error("Worker routing failed")
+  if (job.outcome === "exit") return process.exit(1)
+  if (job.outcome === "clean_exit") return process.exit(0)
+  if (job.outcome === "hang") {
+    setInterval(() => {
+      parentPort.postMessage({ type: "progress", id: job.id })
+    }, 1000)
+    return
+  }
+  parentPort.postMessage({
+    type: "complete",
+    id: job.id,
+    fatal: job.outcome === "fatal",
+  })
+})`,
+  )
+
+  let resolveStarted: () => void = () => {}
+  const started = new Promise<void>((resolve) => {
+    resolveStarted = resolve
+  })
+  const logs: string[] = []
+  const cancellationError = new Error("Cancelled after fatal result")
+  const pool = new ThreadWorkerPool<TestJob, TestJob, TestMessage, string>({
+    concurrency,
+    workerEntrypointPath: workerPath,
+    createMessage: (job) => job,
+    isLogMessage: (message) => message.type === "progress",
+    getLogLines: (message) => [message.id],
+    isCompletionMessage: (message) => message.type === "complete",
+    getResult: (message) => message.id,
+    shouldStopOnMessage: (message) =>
+      message.type === "complete" && message.fatal,
+    cancellationError,
+    jobTimeoutMs: 0,
+    heartbeatIntervalMs: 0,
+    onLog: (lines) => {
+      logs.push(...lines)
+      resolveStarted()
+    },
+  })
+
+  return {
+    pool,
+    started,
+    logs,
+    cancellationError,
+    cleanup: async () => {
+      await pool.terminate()
+      fs.rmSync(directory, { recursive: true, force: true })
+    },
+  }
+}

tests/shared/autorouter-diagnostics-concurrent-isolated-phases.test.ts

@@ -0,0 +1,104 @@
+import { expect, test } from "bun:test"
+import { EventEmitter } from "node:events"
+import fs from "node:fs"
+import os from "node:os"
+import path from "node:path"
+import {
+  AutorouterDiagnostics,
+  AutorouterPhaseTimeoutError,
+} from "lib/shared/autorouter-diagnostics"
+
+test("completing one isolated sibling preserves the other sibling's progress and timeout", async () => {
+  const debugDir = fs.mkdtempSync(path.join(os.tmpdir(), "routing-scopes-"))
+  let parentDbReads = 0
+  const root = Object.assign(new EventEmitter(), {
+    db: {
+      toArray: () => {
+        parentDbReads++
+        return []
+      },
+    },
+  })
+  const diagnostics = new AutorouterDiagnostics({
+    timeoutMs: 1,
+    dumpSrj: "all",
+    debugDir,
+    log: () => {},
+  })
+  diagnostics.attachToRootCircuit(root)
+  const shared = { subcircuit_id: "subcircuit_0", routingPhaseIndex: 0 }
+  const first = { ...shared, isolatedSubcircuitPath: ["outer", "first"] }
+  const second = { ...shared, isolatedSubcircuitPath: ["outer", "second"] }
+
+  try {
+    root.emit("autorouting:start", {
+      ...first,
+      simpleRouteJson: { connections: [{ name: "first" }] },
+    })
+    root.emit("autorouting:start", {
+      ...second,
+      simpleRouteJson: { connections: [{ name: "second" }] },
+    })
+    root.emit("autorouting:progress", { ...first, progress: 0.25, steps: 12 })
+    root.emit("autorouting:progress", { ...second, progress: 0.75, steps: 99 })
+    root.emit("autorouting:end", {
+      ...second,
+      simpleRouteJson: {
+        traces: [{ type: "pcb_trace", pcb_trace_id: "pcb_trace_0", route: [] }],
+      },
+    })
+    root.emit("autorouting:progress", { ...second, progress: 1, steps: 100 })
+
+    await new Promise((resolve) => setTimeout(resolve, 5))
+    let timeout: unknown
+    try {
+      diagnostics.checkTimeout()
+    } catch (error) {
+      timeout = error
+    }
+    expect(timeout).toBeInstanceOf(AutorouterPhaseTimeoutError)
+    const artifactPath = (timeout as AutorouterPhaseTimeoutError)
+      .debugArtifactPath!
+    const artifact = JSON.parse(fs.readFileSync(artifactPath, "utf8"))
+    expect(artifact.isolatedSubcircuitPath).toEqual(
+      first.isolatedSubcircuitPath,
+    )
+    expect(artifact.phaseOrdinal).toBe(1)
+    expect(artifact.lastProgress).toMatchObject({ progress: 0.25, steps: 12 })
+    expect(artifact.files.boardCircuitJson).toBeUndefined()
+    expect(
+      JSON.parse(
+        fs.readFileSync(
+          path.join(debugDir, artifact.files.previousOutputTraces),
+          "utf8",
+        ),
+      ),
+    ).toEqual([])
+
+    diagnostics.finalize([])
+    const summary = JSON.parse(
+      fs.readFileSync(path.join(debugDir, "board.meta.json"), "utf8"),
+    )
+    expect(summary.phases).toHaveLength(1)
+    expect(summary.phases[0]).toMatchObject({
+      isolatedSubcircuitPath: second.isolatedSubcircuitPath,
+      phaseOrdinal: 1,
+      outputTraceCount: 1,
+    })
+    expect(
+      fs.existsSync(
+        path.join(
+          debugDir,
+          "isolated",
+          "outer",
+          "second",
+          "subcircuit_0",
+          "phase-0.input.simple-route.json",
+        ),
+      ),
+    ).toBe(true)
+    expect(parentDbReads).toBe(0)
+  } finally {
+    fs.rmSync(debugDir, { recursive: true, force: true })
+  }
+})

tests/shared/autorouter-diagnostics-concurrent-progress.test.ts

@@ -0,0 +1,36 @@
+import { expect, test } from "bun:test"
+import { EventEmitter } from "node:events"
+import { AutorouterDiagnostics } from "lib/shared/autorouter-diagnostics"
+
+test("long-running diagnostics report each active scope's own progress", async () => {
+  const root = new EventEmitter()
+  const logs: string[] = []
+  const diagnostics = new AutorouterDiagnostics({
+    longRunningLogThresholdMs: 1,
+    log: (message) => logs.push(message),
+  })
+  diagnostics.attachToRootCircuit(root)
+  const first = { subcircuit_id: "board", isolatedSubcircuitPath: ["first"] }
+  const second = { subcircuit_id: "board", isolatedSubcircuitPath: ["second"] }
+  root.emit("autorouting:start", first)
+  root.emit("autorouting:start", second)
+  root.emit("autorouting:progress", { ...first, progress: 0.25 })
+  root.emit("autorouting:progress", { ...second, progress: 0.75 })
+
+  await new Promise((resolve) => setTimeout(resolve, 5))
+  diagnostics.checkTimeout()
+
+  expect(
+    logs.some((line) =>
+      line.includes("[isolated first] progress: progress=25%"),
+    ),
+  ).toBe(true)
+  expect(
+    logs.some((line) =>
+      line.includes("[isolated second] progress: progress=75%"),
+    ),
+  ).toBe(true)
+  expect(logs.filter((line) => line.includes("has been running"))).toHaveLength(
+    2,
+  )
+})