fix: fail fast when benchmark workers exit

Reject worker readiness when startup exits and terminate sibling workers so benchmark failures cannot hang. Validate recovery event records against concrete schemas before assertions.

See #2

Author: cardmagic

benchmarks/processes.ts

@@ -0,0 +1,59 @@
+import type { ChildProcess } from "node:child_process"
+
+export function waitForWorkerReady(worker: ChildProcess): Promise<void> {
+  const priorExit = workerExitError(worker, "before ready")
+  if (priorExit) return Promise.reject(priorExit)
+
+  return new Promise((resolvePromise, reject) => {
+    const cleanup = () => {
+      worker.off("error", onError)
+      worker.off("exit", onExit)
+      worker.off("message", onMessage)
+    }
+    const onError = (error: Error) => {
+      cleanup()
+      reject(error)
+    }
+    const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
+      cleanup()
+      reject(workerExitError({ exitCode: code, signalCode: signal }, "before ready"))
+    }
+    const onMessage = (message: string) => {
+      if (message !== "ready") return
+      cleanup()
+      resolvePromise()
+    }
+    worker.once("error", onError)
+    worker.once("exit", onExit)
+    worker.on("message", onMessage)
+  })
+}
+
+export function waitForWorkerExit(worker: ChildProcess): Promise<void> {
+  const priorExit = workerExitError(worker)
+  if (priorExit) {
+    return worker.exitCode === 0 ? Promise.resolve() : Promise.reject(priorExit)
+  }
+
+  return new Promise((resolvePromise, reject) => {
+    worker.once("error", reject)
+    worker.once("exit", (code, signal) => {
+      if (code === 0) {
+        resolvePromise()
+        return
+      }
+      reject(workerExitError({ exitCode: code, signalCode: signal }))
+    })
+  })
+}
+
+function workerExitError(
+  worker: Pick<ChildProcess, "exitCode" | "signalCode">,
+  phase?: string,
+): Error | undefined {
+  if (worker.exitCode === null && worker.signalCode === null) return undefined
+  const suffix = phase ? ` ${phase}` : ""
+  return new Error(
+    `benchmark worker exited${suffix} with code ${worker.exitCode} and signal ${worker.signalCode}`,
+  )
+}

benchmarks/run.ts

@@ -6,6 +6,7 @@ import { fileURLToPath } from "node:url"
 import { fork, type ChildProcess } from "node:child_process"
 import { performance } from "node:perf_hooks"
 import type { MessageReference } from "solid-objects"
+import { waitForWorkerExit, waitForWorkerReady } from "./processes.ts"
 import { BenchmarkCounter, benchmarkRuntime, type BenchmarkDatabase } from "./shared.ts"
 
 type Shape = "warm-hot" | "warm-many" | "cold-many"
@@ -82,7 +83,7 @@ try {
     } finally {
       shutdown.abort()
       await running
-      const workerExits = workers.map(waitForExit)
+      const workerExits = workers.map(waitForWorkerExit)
       for (const worker of workers) worker.send("stop")
       await Promise.all(workerExits)
       await runtime.testing.reset()
@@ -232,18 +233,17 @@ async function spawnWorkers(options: {
       { stdio: ["ignore", "ignore", "inherit", "ipc"] },
     ),
   )
-  await Promise.all(
-    workers.map(
-      (worker) =>
-        new Promise<void>((resolvePromise, reject) => {
-          worker.once("error", reject)
-          worker.on("message", (message) => {
-            if (message === "ready") resolvePromise()
-          })
-        }),
-    ),
-  )
-  return workers
+  try {
+    await Promise.all(workers.map(waitForWorkerReady))
+    return workers
+  } catch (error) {
+    const workerExits = workers.map(waitForWorkerExit)
+    for (const worker of workers) {
+      if (worker.exitCode === null && worker.signalCode === null) worker.kill()
+    }
+    await Promise.allSettled(workerExits)
+    throw error
+  }
 }
 
 async function readDatabaseVersion(runtime: ReturnType<typeof benchmarkRuntime>): Promise<string> {
@@ -263,24 +263,6 @@ async function readDatabaseVersion(runtime: ReturnType<typeof benchmarkRuntime>)
   })
 }
 
-function waitForExit(child: ChildProcess): Promise<void> {
-  if (child.exitCode !== null) {
-    return child.exitCode === 0
-      ? Promise.resolve()
-      : Promise.reject(new Error(`benchmark worker exited ${child.exitCode}`))
-  }
-  if (child.signalCode !== null) {
-    return Promise.reject(new Error(`benchmark worker exited with ${child.signalCode}`))
-  }
-  return new Promise((resolvePromise, reject) => {
-    child.once("error", reject)
-    child.once("exit", (code) => {
-      if (code === 0) resolvePromise()
-      else reject(new Error(`benchmark worker exited ${code}`))
-    })
-  })
-}
-
 function percentile(values: number[], percent: number): number {
   const ordered = [...values].sort((left, right) => left - right)
   const index = Math.max(0, Math.ceil((percent / 100) * ordered.length) - 1)

examples/failure-recovery/demo.ts

@@ -15,6 +15,18 @@ interface WorkerMessage {
   processed?: number
 }
 
+interface SerializationEvent {
+  event: "start" | "finish"
+  messageId: string
+  at: number
+}
+
+interface ExternalEffectEvent {
+  messageId: string
+  attempt: number
+  processId: number
+}
+
 const directory = await mkdtemp(join(tmpdir(), "solid-objects-recovery-"))
 const databasePath = join(directory, "state.sqlite3")
 const runtime = createRuntime({
@@ -58,7 +70,10 @@ async function proveSerialization(): Promise<{ finalState: number; overlap: fals
   const workers = [spawnWorker(), spawnWorker()]
   await Promise.all(workers.map(({ finished }) => finished))
   await Promise.all(messages.map((message) => message.result()))
-  const events = await jsonLines(join(controlDirectory, "serialization.jsonl"))
+  const events = await readJsonLines(
+    join(controlDirectory, "serialization.jsonl"),
+    parseSerializationEvent,
+  )
   assert.equal(events.length, 4)
   const starts = events.filter((event) => event.event === "start")
   const finishes = events.filter((event) => event.event === "finish")
@@ -121,7 +136,10 @@ async function recoveryResult(options: {
   const stored = await runtime.repository.findMessage(options.message.id)
   const attempts = Number(stored?.attempt_count)
   const snapshot = await options.reference.snapshot()
-  const effects = await jsonLines(join(options.controlDirectory, "external-effects.jsonl"))
+  const effects = await readJsonLines(
+    join(options.controlDirectory, "external-effects.jsonl"),
+    parseExternalEffectEvent,
+  )
   assert.equal(attempts, 2)
   assert.equal(snapshot.count, 1)
   assert.equal(effects.length, 2)
@@ -175,12 +193,39 @@ function spawnWorker(): {
   }
 }
 
-async function jsonLines(path: string): Promise<Array<Record<string, unknown>>> {
-  return (await readFile(path, "utf8"))
-    .trim()
-    .split("\n")
-    .filter(Boolean)
-    .map((line) => JSON.parse(line) as Record<string, unknown>)
+async function readJsonLines<Value>(
+  path: string,
+  parse: (line: string) => Value,
+): Promise<Value[]> {
+  return (await readFile(path, "utf8")).trim().split("\n").filter(Boolean).map(parse)
+}
+
+function parseSerializationEvent(line: string): SerializationEvent {
+  const event = JSON.parse(line) as Partial<SerializationEvent>
+  if (
+    (event.event !== "start" && event.event !== "finish") ||
+    typeof event.messageId !== "string" ||
+    typeof event.at !== "number"
+  ) {
+    throw new TypeError("invalid serialization event")
+  }
+  return { event: event.event, messageId: event.messageId, at: event.at }
+}
+
+function parseExternalEffectEvent(line: string): ExternalEffectEvent {
+  const event = JSON.parse(line) as Partial<ExternalEffectEvent>
+  if (
+    typeof event.messageId !== "string" ||
+    typeof event.attempt !== "number" ||
+    typeof event.processId !== "number"
+  ) {
+    throw new TypeError("invalid external effect event")
+  }
+  return {
+    messageId: event.messageId,
+    attempt: event.attempt,
+    processId: event.processId,
+  }
 }
 
 async function wait(milliseconds: number): Promise<void> {

test/benchmark-processes.test.ts

@@ -0,0 +1,17 @@
+import { fork } from "node:child_process"
+import { fileURLToPath } from "node:url"
+import { describe, expect, it } from "vitest"
+import { waitForWorkerReady } from "../benchmarks/processes.js"
+
+describe("benchmark worker processes", () => {
+  it("rejects when a worker exits before signaling readiness", async () => {
+    const worker = fork(
+      fileURLToPath(new URL("./fixtures/benchmark-worker-exit.mjs", import.meta.url)),
+      { stdio: ["ignore", "ignore", "ignore", "ipc"] },
+    )
+
+    await expect(waitForWorkerReady(worker)).rejects.toThrow(
+      "benchmark worker exited before ready with code 7 and signal null",
+    )
+  })
+})

test/fixtures/benchmark-worker-exit.mjs

@@ -0,0 +1 @@
+process.exitCode = 7