diff --git a/src/execute.ts b/src/execute.ts --- a/src/execute.ts +++ b/src/execute.ts @@ -19,9 +19,33 @@ let nextCallId = 0; const pending = new Map void; reject: (reason: any) => void }>(); const streamPortStack: MessagePort[][] = []; -// Keyed by endpoint (per Task 8: endpoint-death rejection). Reads are no-ops -// until that task populates it. +// Outstanding callIds per endpoint, so failEndpoint can reject every pending +// call when the endpoint's thread exits or its port closes. const endpointCalls = new WeakMap>(); + +function trackCall(endpoint: Endpoint, callId: number): void { + let set = endpointCalls.get(endpoint); + if (set === undefined) { + set = new Set(); + endpointCalls.set(endpoint, set); + } + set.add(callId); +} + +/** Rejects every pending call outstanding on `endpoint` — call when the + * endpoint's thread exits or its port closes. */ +export function failEndpoint(endpoint: Endpoint, reason: Error): void { + const set = endpointCalls.get(endpoint); + if (set === undefined) return; + for (const callId of set) { + const call = pending.get(callId); + if (call !== undefined) { + pending.delete(callId); + call.reject(reason); + } + } + set.clear(); +} /** Anything dispatch can talk over: a Worker from the main thread, or a * MessagePort between peers. Structurally satisfied by both. */ @@ -35,7 +59,7 @@ const call = pending.get(msg.callId); if (!call) return; pending.delete(msg.callId); - endpointCalls.get(endpoint)?.delete(msg.callId); // no-op until Task 8 adds the map + endpointCalls.get(endpoint)?.delete(msg.callId); if (msg.error !== undefined) { call.reject(msg.error); } else { @@ -116,6 +140,7 @@ const callId = nextCallId++; return new Promise((resolve, reject) => { pending.set(callId, { resolve, reject }); + trackCall(endpoint, callId); const extracted = extractTransferables(args); streamPortStack.push([]); const preparedArgs = extracted.args.map(prepareArg); diff --git a/src/worker-pool.ts b/src/worker-pool.ts --- a/src/worker-pool.ts +++ b/src/worker-pool.ts @@ -1,8 +1,9 @@ import { setTimeout } from 'node:timers/promises'; import { transferableAbortSignal } from 'node:util'; import { Worker, MessageChannel } from 'node:worker_threads'; +import type { MessagePort } from 'node:worker_threads'; import { availableParallelism } from 'node:os'; -import { setupWorker, execute, dispatchStream } from './execute.ts'; +import { setupWorker, execute, dispatchStream, failEndpoint } from './execute.ts'; import { AsyncIterableTask } from './stream-task.ts'; import { leastBusy } from './balancers.ts'; import { ActiveCounts } from './active-counts.ts'; @@ -74,6 +75,10 @@ const worker = new Worker(workerEntryUrl, { workerData: handshake, transferList: transferList as any[] }); const idx = i; setupWorker(worker, (frame) => handleCtrl(idx, frame)); + worker.on('exit', (code) => { + dead.add(idx); + failEndpoint(worker, new Error(`Worker ${idx} exited with code ${code} while calls were in flight`)); + }); pool.push(worker); } @@ -84,6 +89,25 @@ // Only main creates channels, so simultaneous first contact from both sides // cannot produce duplicates. const plumbed = new Set(); + // Workers that have exited. Checked at connect time; the runtime does not + // restart workers, so membership is permanent. + const dead = new Set(); + + // Ends of factory channels whose peer was already dead at connect time: + // fail every request that arrives (the requester dispatches into its end + // believing the peer is live; each call gets an error response instead of + // a hang). attachEndpointHandler is worker-side — main wires manually. + function adoptDeadEnd(deadIdx: number, port: MessagePort): void { + port.on('message', (msg: { callId?: number }) => { + if (msg.callId !== undefined) { + port.postMessage({ + callId: msg.callId, + error: new Error(`Worker ${deadIdx} has exited; the runtime does not restart workers`), + }); + } + }); + port.unref(); + } function handleCtrl(fromIdx: number, frame: CtrlFrame): void { switch (frame.__ctrl__) { @@ -100,7 +124,14 @@ plumbed.add(key); const { port1, port2 } = new MessageChannel(); pool[fromIdx].postMessage({ __ctrl__: 'peer', peer, port: port1 }, [port1]); - pool[peer].postMessage({ __ctrl__: 'peer', peer: fromIdx, port: port2 }, [port2]); + if (dead.has(peer)) { + // Deliver the requester's end as usual, but the far side is gone: + // hold port2 and fail every call that arrives on it. (Channels that + // were live when the peer died are a known v1 gap — see TODO.) + adoptDeadEnd(peer, port2); + } else { + pool[peer].postMessage({ __ctrl__: 'peer', peer: fromIdx, port: port2 }, [port2]); + } return; } default: diff --git a/test/async-dispose.test.ts b/test/async-dispose.test.ts --- a/test/async-dispose.test.ts +++ b/test/async-dispose.test.ts @@ -36,11 +36,13 @@ it('force-terminates after shutdownTimeout', async () => { const run = workers(1, { shutdownTimeout: 50 }); - run(slowTask(10_000)); // will not finish in time + const pending = run(slowTask(10_000)); // will not finish in time const start = performance.now(); await run[Symbol.asyncDispose](); const elapsed = performance.now() - start; assert.ok(elapsed < 500, `Expected fast teardown, took ${elapsed}ms`); + // Termination rejects the still-pending call instead of leaving it hanging. + await assert.rejects(pending, { message: /exited/ }); }); it('waits for streaming task to finish', async () => { diff --git a/test/runtime-lifecycle.test.ts b/test/runtime-lifecycle.test.ts --- a/test/runtime-lifecycle.test.ts +++ b/test/runtime-lifecycle.test.ts @@ -95,4 +95,20 @@ }); assert.match(stdout, /BARE-AWAIT .*not part of the global runtime/); }); + + it('worker death rejects its in-flight calls instead of hanging', async () => { + const { stdout } = await exec(process.execPath, ['--no-warnings', join(fixturesDir, 'peer-death-main.ts')], { + timeout: 15000, + }); + assert.match(stdout, /DEATH rejected: .*(exited|died|terminated)/i); + }); + + it('peer connect after worker death rejects instead of hanging', async () => { + const { stdout } = await exec( + process.execPath, + ['--no-warnings', join(fixturesDir, 'peer-death-connect-main.ts')], + { timeout: 15000 }, + ); + assert.match(stdout, /CONNECT-AFTER-DEATH rejected: .*has exited/); + }); }); diff --git a/test/fixtures/peer-death-connect-main.ts b/test/fixtures/peer-death-connect-main.ts new file mode 100644 --- /dev/null +++ b/test/fixtures/peer-death-connect-main.ts @@ -0,0 +1,13 @@ +import { workers, assign } from 'moroutine'; +import { die } from './peer-death.ts'; +import { viaPeer } from './peer-exec.ts'; + +const run = workers(2); +await run(assign(run.workers[1], die())).catch(() => {}); // worker 1 exits +try { + await run(assign(run.workers[0], viaPeer(1, 5))); + console.log('CONNECT-AFTER-DEATH no-reject'); +} catch (err) { + console.log('CONNECT-AFTER-DEATH rejected: ' + (err as Error).message); +} +run[Symbol.dispose](); diff --git a/test/fixtures/peer-death-main.ts b/test/fixtures/peer-death-main.ts new file mode 100644 --- /dev/null +++ b/test/fixtures/peer-death-main.ts @@ -0,0 +1,17 @@ +import { workers, assign } from 'moroutine'; +import { die, slowEcho } from './peer-death.ts'; + +const run = workers(2); +// Start a long call on worker 1, then kill worker 1. The pending call must +// REJECT (worker exit), not hang. +const pendingCall = run(assign(run.workers[1], slowEcho(7))); +setTimeout(() => { + void run(assign(run.workers[1], die())).catch(() => {}); +}, 100); +try { + await pendingCall; + console.log('DEATH no-reject'); +} catch (err) { + console.log('DEATH rejected: ' + (err as Error).message); +} +run[Symbol.dispose](); diff --git a/test/fixtures/peer-death.ts b/test/fixtures/peer-death.ts new file mode 100644 --- /dev/null +++ b/test/fixtures/peer-death.ts @@ -0,0 +1,10 @@ +import { mo } from 'moroutine'; + +export const die = mo(import.meta, (): never => { + process.exit(1); +}); + +export const slowEcho = mo(import.meta, async (v: number): Promise => { + await new Promise((r) => setTimeout(r, 60_000)); // never finishes in test time + return v; +});