From ef2d2697b84242d79e1f1080d82967e67c6aacb8 Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Tue, 21 Jul 2026 22:54:59 -0700 Subject: [PATCH] fix: reject peer connects arriving after shutdown instead of hanging the drain --- src/worker-pool.ts | 39 +++++++++++++------ test/fixtures/peer-exec.ts | 23 ++++++++++- test/fixtures/runtime-def.ts | 2 + .../fixtures/runtime-shutdown-connect-main.ts | 12 ++++++ test/runtime-lifecycle.test.ts | 9 +++++ 5 files changed, 73 insertions(+), 12 deletions(-) create mode 100644 test/fixtures/runtime-shutdown-connect-main.ts diff --git a/src/worker-pool.ts b/src/worker-pool.ts index 60f9989..9406b51 100644 --- a/src/worker-pool.ts +++ b/src/worker-pool.ts @@ -93,17 +93,15 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption // 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 { + // Ends of factory channels that can never reach a live peer (the peer was + // already dead at connect time, or the pool is shutting down): 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 adoptFailingEnd(port: MessagePort, error: Error): 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.postMessage({ callId: msg.callId, error }); } }); port.unref(); @@ -118,9 +116,28 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption if (!Number.isInteger(peer) || peer < 0 || peer >= size || peer === fromIdx) { throw new Error(`Worker ${fromIdx} requested invalid peer ${peer}`); } - if (disposed) return; // teardown in progress; workers are dying anyway + // Plumbed pairs dedupe even during teardown — the requester's end is + // already delivered or in flight (per-port FIFO), so its waiting + // resolver settles without our help. const key = `${Math.min(fromIdx, peer)}:${Math.max(fromIdx, peer)}`; if (plumbed.has(key)) return; + if (disposed) { + // Teardown in progress: dropping the frame would strand the + // requester's waiting resolver and hang the shutdown drain forever. + // Deliver a real end whose far side fails every call, so the + // requester's pending dispatch rejects and the drain completes. + // NOT added to plumbed: no end was pushed to the other side, so a + // connect from it must get its own failing channel, not a dedupe. + const requester = pool[fromIdx]; + if (requester === undefined) return; // pool already terminated + const { port1, port2 } = new MessageChannel(); + requester.postMessage({ __ctrl__: 'peer', peer, port: port1 }, [port1]); + adoptFailingEnd( + port2, + new Error('The runtime is shutting down; peer connections are no longer available'), + ); + return; + } plumbed.add(key); const { port1, port2 } = new MessageChannel(); pool[fromIdx].postMessage({ __ctrl__: 'peer', peer, port: port1 }, [port1]); @@ -128,7 +145,7 @@ export function workers(sizeOrOpts?: number | WorkerOptions, opts?: WorkerOption // 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); + adoptFailingEnd(port2, new Error(`Worker ${peer} has exited; the runtime does not restart workers`)); } else { pool[peer].postMessage({ __ctrl__: 'peer', peer: fromIdx, port: port2 }, [port2]); } diff --git a/test/fixtures/peer-exec.ts b/test/fixtures/peer-exec.ts index dda3a60..99ea9a6 100644 --- a/test/fixtures/peer-exec.ts +++ b/test/fixtures/peer-exec.ts @@ -1,4 +1,5 @@ -import { mo } from 'moroutine'; +import { workerData } from 'node:worker_threads'; +import { mo, runtime } from 'moroutine'; import { getPeerPort } from '../../src/peers.ts'; import { execute } from '../../src/execute.ts'; import { double } from './math.ts'; @@ -9,3 +10,23 @@ export const viaPeer = mo(import.meta, async (peer: number, n: number): Promise< const port = await getPeerPort(peer); return await execute(port, double.id, [n]); }); + +// Parks until the shutdown signal reaches this worker, THEN connects to a +// not-yet-plumbed peer. The connect ctrl frame arrives at main after +// disposal — it must produce a rejection, not a silently dropped frame that +// strands the waiting resolver and hangs main's shutdown drain forever. +export const peerAfterShutdown = mo(import.meta, async (): Promise => { + const wd = workerData as { index: number; size: number }; + const peer = (wd.index + 1) % wd.size; // a sibling — never plumbed in this fixture + await new Promise((resolve) => { + if (runtime.signal.aborted) return resolve(); + runtime.signal.addEventListener('abort', () => resolve(), { once: true }); + }); + try { + const port = await getPeerPort(peer); + await execute(port, double.id, [1]); + return 'no-reject'; + } catch (err) { + return 'rejected: ' + (err as Error).message; + } +}); diff --git a/test/fixtures/runtime-def.ts b/test/fixtures/runtime-def.ts index bfed5a9..c61e8c2 100644 --- a/test/fixtures/runtime-def.ts +++ b/test/fixtures/runtime-def.ts @@ -2,3 +2,5 @@ import { define } from 'moroutine'; import type { RuntimeDefinition } from 'moroutine'; export const tinyRuntime: RuntimeDefinition = define(import.meta, { size: 1 }); + +export const smallRuntime: RuntimeDefinition = define(import.meta, { size: 2 }); diff --git a/test/fixtures/runtime-shutdown-connect-main.ts b/test/fixtures/runtime-shutdown-connect-main.ts new file mode 100644 index 0000000..6c09e07 --- /dev/null +++ b/test/fixtures/runtime-shutdown-connect-main.ts @@ -0,0 +1,12 @@ +import { runtime, registerRuntime } from 'moroutine'; +import { smallRuntime } from './runtime-def.ts'; +import { peerAfterShutdown } from './peer-exec.ts'; + +registerRuntime(smallRuntime); +// Parks until the shutdown signal, then connects to an unplumbed peer — the +// connect ctrl frame reaches main only AFTER shutdown() marked the pool +// disposed. Dropping that frame would hang the drain (and this process). +const pending = runtime.run(peerAfterShutdown()); +const shutdown = runtime.shutdown(); +const [result] = await Promise.all([pending, shutdown]); +console.log('CONNECT-AFTER-SHUTDOWN ' + result); diff --git a/test/runtime-lifecycle.test.ts b/test/runtime-lifecycle.test.ts index dd87dcd..c45bb7e 100644 --- a/test/runtime-lifecycle.test.ts +++ b/test/runtime-lifecycle.test.ts @@ -112,6 +112,15 @@ describe('global runtime lifecycle', () => { assert.match(stdout, /CONNECT-AFTER-DEATH rejected: .*has exited/); }); + it('peer connect arriving after shutdown() rejects instead of hanging the drain', async () => { + const { stdout } = await exec( + process.execPath, + ['--no-warnings', join(fixturesDir, 'runtime-shutdown-connect-main.ts')], + { timeout: 15000 }, + ); + assert.match(stdout, /CONNECT-AFTER-SHUTDOWN rejected: .*shut(ting)? down/i); + }); + it('fire-and-forget worker dispatch keeps the process alive to completion', async () => { const { stdout } = await exec( process.execPath, -- 2.51.2