fix(layout): fail the layout when ELK's thread or child process dies

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<EventEmitter, 'on'> {
+  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<number>();
+  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<T>(child: ChildProcessWithoutNullStreams, input: string): Promise<T> {
+  return new Promise<T>((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);
+  });
+}

packages/layout/src/index.ts

@@ -1,11 +1,12 @@
-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';
 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<typeof ElkBundled>; dispose: () => void } {
   const thread = new Thread(NODE_THREAD, { eval: true, workerData: workerUrl, resourceLimits: { stackSizeMb: ELK_STACK_MB } });
-  const pending = new Set<number>();
-  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<typeof ElkBundled> {
   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<typeof ElkBundled> {
+function childElk(workerUrl: string): { elk: InstanceType<typeof ElkBundled>; 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<ElkNode>((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<ChildProcess>();
+  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<ElkNode>(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<typeof ElkBundled>; 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() };

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');
+  });
+});