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(); }