From cf2f061e50e1654e6bcc6a83d91e7d943faf717f Mon Sep 17 00:00:00 2001 From: Will Madden Date: Tue, 9 Jun 2026 17:02:40 +0200 Subject: [PATCH] fix: await worker-thread termination on stop/close The touch and segmenter worker pools called worker.terminate() without awaiting it, so stop() (and the local server close path that calls it) resolved while the worker threads were still tearing down. When @prisma/streams-local runs in-process inside @prisma/dev, those lingering worker threads outlive the server close and race the host process freeing PGlite (WebAssembly) JIT pages. That trips V8 process-global JIT bookkeeping (PKU/ThreadIsolation) and aborts the process on Linux: # Check failed: jit_page_->allocations_.erase(addr) == 1 Make the pool stop() await every terminate(), and thread that await up through TouchProcessorManager.stop, the app_core close, and the local server close so a resolved close guarantees the worker threads are gone. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Will Madden --- src/app_core.ts | 11 ++++++++--- src/local/server.ts | 4 ++-- src/segment/segmenter_workers.ts | 24 +++++++++++++++--------- src/server.ts | 4 ++-- src/touch/manager.ts | 6 +++--- src/touch/worker_pool.ts | 30 +++++++++++++++++++----------- 6 files changed, 49 insertions(+), 30 deletions(-) diff --git a/src/app_core.ts b/src/app_core.ts index 36c1604..a92d134 100644 --- a/src/app_core.ts +++ b/src/app_core.ts @@ -2858,12 +2858,17 @@ export function createAppCore(cfg: Config, opts: CreateAppCoreOptions): App { } }; - const close = () => { + const close = async () => { closing = true; - touch.stop(); + // Await the worker-thread pools so their threads are fully gone before we + // return. The host process (e.g. @prisma/dev) frees other native resources + // -- PGlite's WebAssembly JIT pages -- right after this resolves; a worker + // thread still tearing down at that moment races V8's process-global JIT + // bookkeeping and can abort the process on Linux. + await touch.stop(); + await segmenter.stop(true); uploader.stop(true); indexer?.stop(); - segmenter.stop(true); metricsEmitter.stop(); expirySweeper.stop(); streamSizeReconciler.stop(); diff --git a/src/local/server.ts b/src/local/server.ts index db5dc1b..7210ebf 100644 --- a/src/local/server.ts +++ b/src/local/server.ts @@ -65,7 +65,7 @@ export async function startLocalDurableStreamsServer(opts: { if (http) await http.close(); } finally { try { - app?.close(); + await app?.close(); } finally { await releaseLock(); } @@ -79,7 +79,7 @@ export async function startLocalDurableStreamsServer(opts: { // ignore } try { - app?.close(); + await app?.close(); } catch { // ignore } diff --git a/src/segment/segmenter_workers.ts b/src/segment/segmenter_workers.ts index d3d272f..0cf3df3 100644 --- a/src/segment/segmenter_workers.ts +++ b/src/segment/segmenter_workers.ts @@ -39,19 +39,25 @@ export class SegmenterWorkerPool implements SegmenterController { } } - stop(_hard?: boolean): void { + async stop(_hard?: boolean): Promise { if (!this.started) return; this.started = false; - for (const w of this.workers) { - try { - w.postMessage({ type: "stop" }); - } catch { - // ignore - } - void w.terminate(); - } + const workers = this.workers.slice(); this.workers.length = 0; this.workerMemory.clear(); + // Await termination so the worker threads are gone before stop() resolves; + // see the note in TouchProcessorWorkerPool.stop -- a lingering worker thread + // racing the host process's WASM teardown can abort the process on Linux. + await Promise.all( + workers.map((w) => { + try { + w.postMessage({ type: "stop" }); + } catch { + // ignore + } + return w.terminate(); + }), + ); } getMemoryStats(): SegmenterMemoryStats { diff --git a/src/server.ts b/src/server.ts index 78f2b54..1cb07ae 100644 --- a/src/server.ts +++ b/src/server.ts @@ -193,7 +193,7 @@ statsReporter?.start(); histReporter?.start(); let shuttingDown = false; -const shutdown = (signal: NodeJS.Signals) => { +const shutdown = async (signal: NodeJS.Signals) => { if (shuttingDown) return; shuttingDown = true; // eslint-disable-next-line no-console @@ -207,7 +207,7 @@ const shutdown = (signal: NodeJS.Signals) => { console.error("failed to stop HTTP server cleanly", err); } try { - app.close(); + await app.close(); } catch (err) { // eslint-disable-next-line no-console console.error("failed to close application cleanly", err); diff --git a/src/touch/manager.ts b/src/touch/manager.ts index ee083e6..8906b00 100644 --- a/src/touch/manager.ts +++ b/src/touch/manager.ts @@ -198,11 +198,11 @@ export class TouchProcessorManager { } } - stop(): void { + async stop(): Promise { this.stopping = true; if (this.timer) clearInterval(this.timer); this.timer = null; - this.pool.stop(); + await this.pool.stop(); this.liveMetrics.stop(); for (const j of this.journals.values()) j.stop(); this.journals.clear(); @@ -305,7 +305,7 @@ export class TouchProcessorManager { if (this.restartWorkerPoolRequested) { this.restartWorkerPoolRequested = false; try { - this.pool.restart(); + await this.pool.restart(); this.lastWorkerPoolRestartAtMs = Date.now(); } catch (e) { // eslint-disable-next-line no-console diff --git a/src/touch/worker_pool.ts b/src/touch/worker_pool.ts index fccc7e8..78373d4 100644 --- a/src/touch/worker_pool.ts +++ b/src/touch/worker_pool.ts @@ -38,28 +38,36 @@ export class TouchProcessorWorkerPool { for (let i = 0; i < this.workerCount; i++) this.spawnWorker(i, generation); } - stop(): void { + async stop(): Promise { if (!this.started) return; this.started = false; this.generation += 1; - for (const w of this.workers) { - try { - w.worker.postMessage({ type: "stop" }); - } catch { - // ignore - } - void w.worker.terminate(); - } + const workers = this.workers.slice(); this.workers.length = 0; this.queue.length = 0; for (const [id, p] of this.pending.entries()) { p.resolve(Result.err({ kind: "worker_pool_failure", message: "worker pool stopped" })); this.pending.delete(id); } + // Await termination so the worker threads are actually gone before stop() + // resolves. Callers (the local server's close path) rely on this: a worker + // thread still tearing down while the host process frees other native + // resources -- e.g. PGlite's WebAssembly JIT pages in @prisma/dev -- races + // V8's process-global JIT bookkeeping and can abort the process on Linux. + await Promise.all( + workers.map((w) => { + try { + w.worker.postMessage({ type: "stop" }); + } catch { + // ignore + } + return w.worker.terminate(); + }), + ); } - restart(): void { - this.stop(); + async restart(): Promise { + await this.stop(); this.start(); }