From 3e815939165d8d5ff95585a378f0650db90c9886 Mon Sep 17 00:00:00 2001 From: Joseph Dale Banares Date: Sat, 24 Oct 2026 00:39:57 +0800 Subject: [PATCH] fix(layout): fail the layout when ELK's thread or child process dies --- packages/layout/src/elk-io.ts | 75 ++++++++++++++++++++ packages/layout/src/index.ts | 49 ++++--------- packages/layout/test/elk-io.test.ts | 103 ++++++++++++++++++++++++++++ 3 files changed, 193 insertions(+), 34 deletions(-) create mode 100644 packages/layout/src/elk-io.ts create mode 100644 packages/layout/test/elk-io.test.ts diff --git a/packages/layout/src/elk-io.ts b/packages/layout/src/elk-io.ts new file mode 100644 index 0000000..3576ffe --- /dev/null +++ b/packages/layout/src/elk-io.ts @@ -0,0 +1,75 @@ +import type { ChildProcessWithoutNullStreams } from 'node:child_process'; +import type { EventEmitter } from 'node:events'; + +/** What threadElk needs from a worker thread: enough to fake in a test. */ +export interface ThreadLike extends Pick { + postMessage(msg: unknown): void; + terminate(): unknown; +} + +/** elkjs's worker port over a thread; a thread that dies or exits fails what it was asked instead of hanging. */ +export function threadPort(thread: ThreadLike) { + const pending = new Set(); + const port = { + onmessage: null as ((answer: { data: unknown }) => void) | null, + postMessage(msg: { id: number }) { + pending.add(msg.id); + thread.postMessage(msg); + }, + terminate: () => void thread.terminate(), + }; + thread.on('message', (data: { id: number; error?: { message: string } }) => { + pending.delete(data.id); + port.onmessage?.({ data: data.error ? { ...data, error: new Error(data.error.message) } : data }); + }); + // A thread that dies (out of memory, say) answers nothing: fail what it was asked instead of hanging. + thread.on('error', (error) => { + for (const id of pending) port.onmessage?.({ data: { id, error } }); + pending.clear(); + }); + // An exit (terminate, process.exit, some resource limits) can come without an 'error': fail what's still waiting. + thread.on('exit', (code: number) => { + const error = new Error(`ELK thread stopped (exit code ${code})`); + for (const id of pending) port.onmessage?.({ data: { id, error } }); + pending.clear(); + }); + return port; +} + +/** Sends `input` to a child ELK and resolves its JSON answer; every way the child can fail rejects, once. */ +export function runChild(child: ChildProcessWithoutNullStreams, input: string): Promise { + return new Promise((resolve, reject) => { + let out = ''; + let err = ''; + let settled = false; + const succeed = (value: T) => { + if (!settled) { + settled = true; + resolve(value); + } + }; + const fail = (error: Error) => { + if (!settled) { + settled = true; + reject(error); + } + }; + child.stdout.on('data', (c: Buffer) => (out += c)); + child.stderr.on('data', (c: Buffer) => (err += c)); + child.on('error', (e) => fail(e)); + // EPIPE when the child dies before reading its input: 'close' reports why. + child.stdin.on('error', () => {}); + child.on('close', (code, signal) => { + if (code === 0) { + try { + succeed(JSON.parse(out) as T); + } catch (e) { + fail(new Error(`ELK answered with invalid JSON: ${(e as Error).message}`)); + } + } else { + fail(new Error(err.trim() || (signal ? `ELK was killed by ${signal}` : `ELK exited with code ${code}`))); + } + }); + child.stdin.end(input); + }); +} diff --git a/packages/layout/src/index.ts b/packages/layout/src/index.ts index 674a0a6..7d9d352 100644 --- a/packages/layout/src/index.ts +++ b/packages/layout/src/index.ts @@ -1,4 +1,4 @@ -import { spawn } from 'node:child_process'; +import { spawn, type ChildProcess } from 'node:child_process'; import { readFileSync } from 'node:fs'; import { createRequire } from 'node:module'; import { Worker as Thread } from 'node:worker_threads'; @@ -6,6 +6,7 @@ import ElkApi from 'elkjs/lib/elk-api.js'; import ElkBundled from 'elkjs/lib/elk.bundled.js'; import type { ElkExtendedEdge, ElkNode } from 'elkjs/lib/elk-api'; import { cardSize, isLaneKind, usesCompactCards, type DiagramDraft, type Direction, type LaidOutDiagram, type Point, type Rect } from '@stackmap/core'; +import { runChild, threadPort } from './elk-io'; import { placeLabels } from './labels'; import { layoutLanes } from './lanes'; import { layoutSequence } from './sequence'; @@ -53,31 +54,14 @@ parentPort.on('message', (msg) => elk.postMessage(msg)); function threadElk(workerUrl: string): { elk: InstanceType; dispose: () => void } { const thread = new Thread(NODE_THREAD, { eval: true, workerData: workerUrl, resourceLimits: { stackSizeMb: ELK_STACK_MB } }); - const pending = new Set(); - const port = { - onmessage: null as ((answer: { data: unknown }) => void) | null, - postMessage(msg: { id: number }) { - pending.add(msg.id); - thread.postMessage(msg); - }, - terminate: () => void thread.terminate(), - }; - thread.on('message', (data: { id: number; error?: { message: string } }) => { - pending.delete(data.id); - port.onmessage?.({ data: data.error ? { ...data, error: new Error(data.error.message) } : data }); - }); - // A thread that dies (out of memory, say) answers nothing: fail what it was asked instead of hanging. - thread.on('error', (error) => { - for (const id of pending) port.onmessage?.({ data: { id, error } }); - pending.clear(); - }); + const port = threadPort(thread); const elk = new ElkApi({ workerFactory: () => port as never }); return { elk, dispose: () => elk.terminateWorker() }; } /** * ELK in-process where a global `self` exists: its worker script, run with `self` hidden, exports the in-process worker - * instead of taking over `self.onmessage`. Bun's fallback for a large graph where a child process can't be started. + * instead of taking over `self.onmessage`. Bun's large-graph host on Windows, where the child process (a POSIX shell with ulimit) isn't available. */ function mainThreadElk(workerUrl: string): InstanceType { const mod: { exports: { Worker?: new () => unknown } } = { exports: {} }; @@ -106,22 +90,19 @@ process.stdin.on('data', (c) => (input += c)).on('end', () => `; /** For Bun's large graphs: ELK in a child Bun started with room for its recursion (the densest need ~100 MB). */ -function childElk(workerUrl: string): InstanceType { +function childElk(workerUrl: string): { elk: InstanceType; dispose: () => void } { const apiUrl = createRequire(import.meta.url).resolve('elkjs/lib/elk-api.js'); // The option only when the limit rose: telling JavaScriptCore the stack is bigger than it is crashes instead of throwing. const script = `if ulimit -s ${BUN_STACK_KB} 2>/dev/null; then export BUN_JSC_maxPerThreadStackUsage=${BUN_STACK_KB * 1024}; fi; exec "$0" -e "$1" "$2" "$3"`; - const layout = (graph: ElkNode) => - new Promise((resolve, reject) => { - const child = spawn('/bin/sh', ['-c', script, process.execPath, BUN_CHILD, workerUrl, apiUrl]); - let out = ''; - let err = ''; - child.stdout.on('data', (c: Buffer) => (out += c)); - child.stderr.on('data', (c: Buffer) => (err += c)); - child.on('error', reject); - child.on('close', (code) => (code === 0 ? resolve(JSON.parse(out) as ElkNode) : reject(new Error(err.trim() || `ELK exited with code ${code}`)))); - child.stdin.end(JSON.stringify(graph)); - }); - return { layout } as never; + const live = new Set(); + const layout = (graph: ElkNode) => { + const child = spawn('/bin/sh', ['-c', script, process.execPath, BUN_CHILD, workerUrl, apiUrl]); + live.add(child); + child.once('close', () => live.delete(child)); + return runChild(child, JSON.stringify(graph)); + }; + // A layout that threw elsewhere leaves its child running: kill it with the rest. + return { elk: { layout } as never, dispose: () => live.forEach((c) => c.kill()) }; } // elkjs's bundled build treats any runtime with a global `self` and no `document` as a web worker @@ -134,7 +115,7 @@ function createElk(large: boolean): { elk: InstanceType; disp if (node ? !large : g.document !== undefined || g.Worker === undefined) return { elk: new ElkBundled(), dispose: () => {} }; const workerUrl = createRequire(import.meta.url).resolve('elkjs/lib/elk-worker.min.js'); if (node) return threadElk(workerUrl); - if (large) return { elk: process.platform === 'win32' ? mainThreadElk(workerUrl) : childElk(workerUrl), dispose: () => {} }; + if (large) return process.platform === 'win32' ? { elk: mainThreadElk(workerUrl), dispose: () => {} } : childElk(workerUrl); const elk = new ElkApi({ workerUrl }); // A real Worker keeps the process alive until terminated; the in-process one has no terminate(). return { elk, dispose: () => elk.terminateWorker() }; diff --git a/packages/layout/test/elk-io.test.ts b/packages/layout/test/elk-io.test.ts new file mode 100644 index 0000000..1237a61 --- /dev/null +++ b/packages/layout/test/elk-io.test.ts @@ -0,0 +1,103 @@ +import type { ChildProcessWithoutNullStreams } from 'node:child_process'; +import { EventEmitter } from 'node:events'; +import { PassThrough } from 'node:stream'; +import { describe, expect, it } from 'vitest'; +import { runChild, threadPort } from '../src/elk-io'; + +function fakeThread() { + const emitter = new EventEmitter(); + const thread = Object.assign(emitter, { + posted: [] as unknown[], + postMessage(msg: unknown) { + thread.posted.push(msg); + }, + terminate() { + emitter.emit('exit', 1); + }, + }); + return thread; +} + +function fakeChild() { + const child = Object.assign(new EventEmitter(), { stdout: new PassThrough(), stderr: new PassThrough(), stdin: new PassThrough() }); + return child as unknown as ChildProcessWithoutNullStreams & { stdout: PassThrough; stderr: PassThrough; stdin: PassThrough }; +} + +describe('threadPort', () => { + it('fails a pending request when the thread exits', () => { + const thread = fakeThread(); + const port = threadPort(thread); + const answers: { data: { id: number; error?: Error } }[] = []; + port.onmessage = (a) => answers.push(a as never); + port.postMessage({ id: 1 }); + thread.emit('exit', 1); + expect(answers).toHaveLength(1); + expect(answers[0]!.data.id).toBe(1); + expect(answers[0]!.data.error?.message).toMatch(/exit code 1/); + }); + + it('does not fail an answered request when the thread later exits', () => { + const thread = fakeThread(); + const port = threadPort(thread); + const answers: { data: { id: number; error?: Error } }[] = []; + port.onmessage = (a) => answers.push(a as never); + port.postMessage({ id: 1 }); + thread.emit('message', { id: 1 }); + thread.emit('exit', 0); + expect(answers).toHaveLength(1); + expect(answers[0]!.data.error).toBeUndefined(); + }); + + it('fails a pending request when the thread errors', () => { + const thread = fakeThread(); + const port = threadPort(thread); + const answers: { data: { id: number; error?: Error } }[] = []; + port.onmessage = (a) => answers.push(a as never); + port.postMessage({ id: 1 }); + thread.emit('error', new Error('out of memory')); + expect(answers).toHaveLength(1); + expect(answers[0]!.data.error?.message).toBe('out of memory'); + }); +}); + +describe('runChild', () => { + it('resolves the JSON the child prints', async () => { + const child = fakeChild(); + const p = runChild(child, '{}'); + child.stdout.write('{"id":"root"}'); + child.emit('close', 0, null); + await expect(p).resolves.toEqual({ id: 'root' }); + }); + + it('rejects, rather than throwing outside the promise, on truncated output', async () => { + const child = fakeChild(); + const p = runChild(child, '{}'); + child.stdout.write('{"id":'); + child.emit('close', 0, null); + await expect(p).rejects.toThrow(/invalid JSON/); + }); + + it('rejects with the signal when the child is killed silently', async () => { + const child = fakeChild(); + const p = runChild(child, '{}'); + child.emit('close', null, 'SIGKILL'); + await expect(p).rejects.toThrow(/killed by SIGKILL/); + }); + + it('survives EPIPE on stdin and reports the child stderr', async () => { + const child = fakeChild(); + const p = runChild(child, '{}'); + child.stdin.emit('error', Object.assign(new Error('write EPIPE'), { code: 'EPIPE' })); + child.stderr.write('boom'); + child.emit('close', 1, null); + await expect(p).rejects.toThrow(/boom/); + }); + + it('settles with the first failure only', async () => { + const child = fakeChild(); + const p = runChild(child, '{}'); + child.emit('error', new Error('spawn failed')); + child.emit('close', 1, null); + await expect(p).rejects.toThrow('spawn failed'); + }); +});