diff --git a/src/jsc/VM.rs b/src/jsc/VM.rs index 41f780c67f5b..77e7f8e0bc76 100644 --- a/src/jsc/VM.rs +++ b/src/jsc/VM.rs @@ -118,10 +118,6 @@ impl VM { JSC__VM__notifyNeedTermination(self) } - pub(crate) fn clear_has_termination_request(&self) { - crate::cpp::JSC__VM__clearHasTerminationRequest(self) - } - #[track_caller] pub fn throw_error(&self, global_object: &JSGlobalObject, value: JSValue) -> JsError { crate::validation_scope!(scope, global_object); diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index 59947724bd44..aba1efd1ce2b 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/src/jsc/bindings/bindings.cpp @@ -4923,11 +4923,6 @@ bool JSC__VM__isTerminationException(JSC::VM* vm, JSC::Exception* exception) return vm->isTerminationException(exception); } -[[ZIG_EXPORT(nothrow)]] -void JSC__VM__clearHasTerminationRequest(JSC::VM* vm) -{ - vm->clearHasTerminationRequest(); -} [[ZIG_EXPORT(nothrow)]] bool JSC__VM__hasTerminationRequest(JSC::VM* vm) { diff --git a/src/jsc/web_worker.rs b/src/jsc/web_worker.rs index d4e1759d2f7a..ffd99b6ce6f8 100644 --- a/src/jsc/web_worker.rs +++ b/src/jsc/web_worker.rs @@ -1243,12 +1243,11 @@ impl WebWorker { // SAFETY: vm_ptr valid; unpublished above under vm_lock, so no // other thread can dereference it now — `&mut` is exclusive. let vm = unsafe { &mut *vm_ptr }; - // terminate() set the JSC termination flag to interrupt running JS; - // clear it so process.on('exit') handlers can run. teardownJSCVM - // re-sets it for the JSC VM teardown. - vm.jsc_vm().clear_has_termination_request(); vm.is_shutting_down = true; + // on_exit first: dispatchExitInternal skips 'exit' on terminate (node semantics). vm.on_exit(); + // Cleared here so close_all_socket_groups can re-enter JS; teardownJSCVM re-sets it. + vm.global().clear_termination_exception(); if let Some(hooks) = runtime_hooks() { (hooks.cron_clear_all_teardown)(vm); // Drain `TimeoutObject`s from this worker's timer heap before diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index 87bd776e2c12..5f5be1fa6f93 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -2129,6 +2129,14 @@ where stream_log!("returned a promise"); this.drain_microtasks(); + // terminate() inside the drain leaves a pending TerminationException; every branch below re-enters JS. + if global_this.has_exception() { + response_stream.sink.on_first_write = None; + response_stream.sink.ctx = None; + this.flags.set_has_marked_pending(true); + return; + } + // `MarkHandled` matters for the Rejected arm: the promise // settled before any reaction was attached, so without the // flag the VM would report it as an unhandled rejection even @@ -2679,6 +2687,11 @@ where response_value.ensure_still_alive(); ctx.drain_microtasks(); + // Same as do_render_stream: terminate() inside the drain leaves a pending TerminationException. + if this.global_this().has_exception() { + return; + } + if ctx.is_aborted_or_ended() { return; } diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index 61044580f1db..d630754e4021 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -1313,6 +1313,11 @@ impl NewServer { needs_to_drain = false; // SAFETY: `vm` is the process-static VirtualMachine. unsafe { (*vm).drain_microtasks() }; + // Same as RequestContext::on_response: terminate() inside the drain leaves a pending TerminationException. + if global.has_exception() { + strong_promise.deinit(); + break 'brk HttpResult::Success; + } status = promise.status(); } diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 374246033f82..02a9559aa5a1 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -11,6 +11,198 @@ const rounds = slow ? 4 : 8; const perRound = slow ? 12 : 32; const timeout = slow ? 60_000 : 20_000; +// Regression: Bun.serve() inside a worker streaming a type:"direct" body, +// then worker.terminate() mid-stream. drain_microtasks() drives the async +// pull loop; when the termination trap fires inside it, the outermost +// VMEntryScope's executeEntryScopeServicesOnExit clears hasTerminationRequest +// while the TerminationException stays pending. do_render_stream then kept +// going and (a) lazy-initialized NativePromiseContextStructure inside a +// DeferTerminationForAWhile scope, asserting hasTerminationRequest() in +// VMTraps::deferTerminationSlow, and (b) re-entered JS via then_with_value, +// asserting !exception() in executeCallImpl. Separately, WebWorker::shutdown +// cleared only the request flag before close_all_socket_groups, leaving the +// sticky exception in place for the same class of assert. Both asserts are +// compiled out outside debug/ASAN builds. +test.skipIf(!(isDebug || isASAN))( + 'terminate() while Bun.serve is streaming a type:"direct" body does not trip DeferTermination / assertNoException', + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const src = 'const server = Bun.serve({ port: 0, fetch() {' + + ' return new Response(new ReadableStream({ type: "direct", async pull(ctrl) {' + + ' for (let i = 0; i < 20000; i++) {' + + ' ctrl.write(new Uint8Array(8192).fill(i));' + + ' await ctrl.flush();' + + ' }' + + ' ctrl.close();' + + ' }, cancel() {} }));' + + '} });' + + 'require("node:worker_threads").parentPort.postMessage(server.port);'; + for (let i = 0; i < ${rounds}; i++) { + const w = new Worker(src, { eval: true }); + const port = await new Promise(r => w.once("message", r)); + const res = await fetch("http://127.0.0.1:" + port + "/"); + const rd = res.body.getReader(); + await rd.read(); + const done = new Promise(r => w.once("exit", r)); + w.terminate(); + await done; + await rd.cancel().catch(() => {}); + } + console.log("survived"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("survived\n"); + expect(exitCode).toBe(0); + }, + timeout, +); + +// Sibling of the above: on_response's drain_microtasks() driving an async +// fetch handler that never yields past `await 1`. Same DeferTermination / +// assertNoException pair, same fix shape. Also asserts Node's +// "process.on('exit') is skipped on terminate()" when terminate lands while +// JS is on the stack (sab[1] stays 0): in that state the NeedTermination trap +// has already been consumed, so only the sticky TerminationException guards +// dispatchExitInternal; clearing it before on_exit would let the handler run. +test.skipIf(!(isDebug || isASAN))( + "terminate() while Bun.serve's async fetch is looping in drain_microtasks does not trip DeferTermination", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const sab = new SharedArrayBuffer(8); + const flag = new Int32Array(sab); + const src = 'const { workerData } = require("node:worker_threads");' + + 'const flag = new Int32Array(workerData.sab);' + + 'process.on("exit", () => Atomics.add(flag, 1, 1));' + + 'const server = Bun.serve({ port: 0, async fetch() {' + + ' for (;;) { Atomics.add(flag, 0, 1); await 1; }' + + '} });' + + 'require("node:worker_threads").parentPort.postMessage(server.port);'; + for (let i = 0; i < ${rounds}; i++) { + Atomics.store(flag, 0, 0); + Atomics.store(flag, 1, 0); + const w = new Worker(src, { eval: true, workerData: { sab } }); + const port = await new Promise(r => w.once("message", r)); + const ac = new AbortController(); + fetch("http://127.0.0.1:" + port + "/", { signal: ac.signal }).catch(() => {}); + while (Atomics.load(flag, 0) < 100) await Bun.sleep(0); + const done = new Promise(r => w.once("exit", r)); + w.terminate(); + await done; + ac.abort(); + if (Atomics.load(flag, 1) !== 0) throw new Error("process.on('exit') fired on terminate"); + } + console.log("survived"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("survived\n"); + expect(exitCode).toBe(0); + }, + timeout, +); + +// Same again for the node:http-compat path (on_node_http_request_with_upgrade_ctx). +test.skipIf(!(isDebug || isASAN))( + "terminate() while a node:http server's async handler is looping in drain_microtasks does not trip DeferTermination", + async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const sab = new SharedArrayBuffer(4); + const flag = new Int32Array(sab); + const src = 'const { workerData } = require("node:worker_threads");' + + 'const flag = new Int32Array(workerData.sab);' + + 'const http = require("node:http");' + + 'const server = http.createServer(async (req, res) => {' + + ' for (;;) { Atomics.add(flag, 0, 1); await 1; }' + + '}).listen(0, () => require("node:worker_threads").parentPort.postMessage(server.address().port));'; + for (let i = 0; i < ${rounds}; i++) { + Atomics.store(flag, 0, 0); + const w = new Worker(src, { eval: true, workerData: { sab } }); + const port = await new Promise(r => w.once("message", r)); + const ac = new AbortController(); + fetch("http://127.0.0.1:" + port + "/", { signal: ac.signal }).catch(() => {}); + while (Atomics.load(flag, 0) < 100) await Bun.sleep(0); + const done = new Promise(r => w.once("exit", r)); + w.terminate(); + await done; + ac.abort(); + } + console.log("survived"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("survived\n"); + expect(exitCode).toBe(0); + }, + timeout, +); + +// Node skips the worker's process.on('exit') listeners on worker.terminate() +// when the worker is idle (the NeedTermination trap bit is still armed here, +// so this is the simpler half of the invariant; the mid-JS half is checked in +// the async-fetch test above). +test("process.on('exit') does not fire on worker.terminate() when idle", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const sab = new SharedArrayBuffer(4); + const counter = new Int32Array(sab); + const src = 'const { workerData, parentPort } = require("node:worker_threads");' + + 'const counter = new Int32Array(workerData.sab);' + + 'process.on("exit", () => Atomics.add(counter, 0, 1));' + + 'parentPort.postMessage("ready");' + + 'setInterval(() => {}, 100000);'; + const w = new Worker(src, { eval: true, workerData: { sab } }); + await new Promise(r => w.once("message", r)); + const done = new Promise(r => w.once("exit", r)); + w.terminate(); + await done; + console.log("exitCalls=" + Atomics.load(counter, 0)); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("exitCalls=0\n"); + expect(exitCode).toBe(0); +}); + // Regression: `new Worker(url, { ref: false })` was silently ignored — the // Zig-side `user_keep_alive` field was set from it but never read, and the // parent keep-alive was taken unconditionally in `create()`. `.unref()` after