diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index 845980f5cb61..ba699a174b13 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -3585,6 +3585,12 @@ where if let Some(server) = self.server { // SAFETY: BACKREF let server = &*server; + // The "error" may be the worker's TerminationException raised + // inside on_request's fetch-handler call; entering the user + // error() handler with it pending trips assertNoException(). + if server.vm().script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { + return; + } let on_error = server.config().on_error; if !on_error.is_empty() && !self.flags.has_called_error_handler() { self.flags.set_has_called_error_handler(true); @@ -3595,6 +3601,9 @@ where &[value], ) .unwrap_or_else(|err| server.global_this().take_exception(err)); + if server.global_this().has_exception() { + return; + } let _keep = jsc::EnsureStillAlive(result); if !result.is_empty_or_undefined_or_null() { if let Some(err) = result.to_error() { diff --git a/src/runtime/server/ServerWebSocket.rs b/src/runtime/server/ServerWebSocket.rs index c6acf0d4a4cb..b292f9f57790 100644 --- a/src/runtime/server/ServerWebSocket.rs +++ b/src/runtime/server/ServerWebSocket.rs @@ -420,7 +420,7 @@ impl ServerWebSocket { let global_object = handler.global_object(); let on_open_handler = handler.on_open; let on_error = handler.on_error; - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { bun_output::scoped_log!(WebSocketServer, "onOpen called after script execution"); ws.close(); return; @@ -493,7 +493,7 @@ impl ServerWebSocket { let global_object = self.handler().global_object(); // This is the start of a task. let vm = self.handler().vm(); - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { bun_output::scoped_log!(WebSocketServer, "onMessage called after script execution"); ws.close(); return; @@ -501,19 +501,18 @@ impl ServerWebSocket { let _loop_guard = vm.enter_event_loop_scope(); + let payload = match opcode { + Opcode::Text => jsc::bun_string_jsc::create_utf8_for_js(global_object, message), + Opcode::Binary => self.binary_to_js(global_object, message), + _ => unreachable!(), + }; + let Ok(payload) = payload else { return }; let arguments = [ self.this_value .get() .try_get() .unwrap_or(JSValue::UNDEFINED), - match opcode { - Opcode::Text => jsc::bun_string_jsc::create_utf8_for_js(global_object, message) - .unwrap_or(JSValue::ZERO), // TODO: properly propagate exception upwards - Opcode::Binary => self - .binary_to_js(global_object, message) - .unwrap_or(JSValue::ZERO), // TODO: properly propagate exception upwards - _ => unreachable!(), - }, + payload, ]; let mut corker = Corker { @@ -562,7 +561,9 @@ impl ServerWebSocket { bun_output::scoped_log!(WebSocketServer, "onDrain"); let handler = self.handler(); let vm = handler.vm(); - if self.is_closed() || vm.is_shutting_down() { + if self.is_closed() + || vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running + { return; } @@ -610,7 +611,9 @@ impl ServerWebSocket { let cb = handler.on_ping; let on_error = handler.on_error; let vm = handler.vm(); - if cb.is_empty_or_undefined_or_null() || vm.is_shutting_down() { + if cb.is_empty_or_undefined_or_null() + || vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running + { return; } let global_this = handler.global_object(); @@ -618,13 +621,15 @@ impl ServerWebSocket { // This is the start of a task. let _loop_guard = vm.enter_event_loop_scope(); + let Ok(payload) = self.binary_to_js(global_this, data) else { + return; + }; let args = [ self.this_value .get() .try_get() .unwrap_or(JSValue::UNDEFINED), - self.binary_to_js(global_this, data) - .unwrap_or(JSValue::ZERO), // TODO: properly propagate exception upwards + payload, ]; if let Err(e) = cb.call(global_this, JSValue::UNDEFINED, &args) { let err = global_this.take_exception(e); @@ -646,20 +651,22 @@ impl ServerWebSocket { let global_this = handler.global_object(); let vm = handler.vm(); - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return; } // This is the start of a task. let _loop_guard = vm.enter_event_loop_scope(); + let Ok(payload) = self.binary_to_js(global_this, data) else { + return; + }; let args = [ self.this_value .get() .try_get() .unwrap_or(JSValue::UNDEFINED), - self.binary_to_js(global_this, data) - .unwrap_or(JSValue::ZERO), // TODO: properly propagate exception upwards + payload, ]; if let Err(e) = cb.call(global_this, JSValue::UNDEFINED, &args) { let err = global_this.take_exception(e); @@ -723,7 +730,7 @@ impl ServerWebSocket { }); let vm = handler.vm(); - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return; } diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index 9551413e57ed..18890ee174f5 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -554,7 +554,15 @@ impl NewServer { /// still live (its WriteBarrier slots still root the handlers); only /// `Finalized` means the slots are gone and the `config` shadows may point /// at freed cells. Dispatch trampolines answer 503+close on `None`. + /// + /// Also `None` once a worker's terminate() has armed the trap: a previous + /// request's handler in the same poll sweep may have left the sticky + /// TerminationException pending, and every request-family entry calls this + /// before its first JS entry. pub(crate) fn js_value_for_dispatch(&self) -> Option { + if self.vm().script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { + return None; + } self.js_value.try_get() } } @@ -672,6 +680,13 @@ impl NewServer { // S008: `Response` is a ZST opaque — safe `*mut → &mut` deref. let resp_ref = bun_opaque::opaque_deref_mut(resp); + // Belt-and-braces for `prepare_and_save_js_request_context` (bake), + // which reaches here without the `js_value_for_dispatch()` gate. + if server.vm().script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { + server_body::respond_stopped_503(resp_ref); + return None; + } + // We need to register the handler immediately since uSockets will not buffer. // // We first validate the self-reported request body length so that @@ -1756,9 +1771,9 @@ impl NewServer { } if self.pending_requests == 0 && !self.has_listener() && !self.has_active_web_sockets() { // Make the wrapper collectible. Dispatch still works while it is - // `Weak` (its WriteBarrier slots still root the handlers); the - // `js_value_for_dispatch` gate only trips once the wrapper is - // actually finalized. + // `Weak` (its WriteBarrier slots still root the handlers); see + // `js_value_for_dispatch` for the conditions under which the gate + // trips. self.js_value.downgrade(); if let Some(ws) = self.config.websocket.as_mut() { ws.handler.app = None; diff --git a/src/runtime/shell/interpreter.rs b/src/runtime/shell/interpreter.rs index 4e925f76cfb2..42789534f12e 100644 --- a/src/runtime/shell/interpreter.rs +++ b/src/runtime/shell/interpreter.rs @@ -1199,6 +1199,13 @@ impl Interpreter { let global_this = self .global_this_ref() .expect("global_this set on Js event-loop path"); + if global_this.bun_vm().script_execution_status() + != bun_jsc::ScriptExecutionStatus::Running + { + self.keep_alive.with_mut(|k| k.disable()); + self.deref_root_shell_and_io_if_needed(true); + return Yield::done(); + } let buffered_stdout = self.get_buffered_stdout(global_this); let buffered_stderr = self.get_buffered_stderr(global_this); self.keep_alive.with_mut(|k| k.disable()); diff --git a/src/runtime/socket/Handlers.rs b/src/runtime/socket/Handlers.rs index cde8a3463988..a0e9588ed555 100644 --- a/src/runtime/socket/Handlers.rs +++ b/src/runtime/socket/Handlers.rs @@ -215,7 +215,7 @@ impl Handlers { pub(crate) fn resolve_promise(&self, value: JSValue) -> JsResult<()> { let vm = self.vm; - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return Ok(()); } @@ -231,7 +231,7 @@ impl Handlers { pub(crate) fn reject_promise(&self, value: JSValue) -> JsResult { let vm = self.vm; - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return Ok(true); } @@ -277,7 +277,7 @@ impl Handlers { pub(crate) fn call_error_handler(&self, this_value: JSValue, args: &[JSValue; 2]) -> bool { let vm = self.vm; - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return false; } diff --git a/src/runtime/socket/Listener.rs b/src/runtime/socket/Listener.rs index 6b2dbf01aba5..68f87a0b260e 100644 --- a/src/runtime/socket/Listener.rs +++ b/src/runtime/socket/Listener.rs @@ -1688,7 +1688,7 @@ pub struct WindowsNamedPipeListeningContext { pub(crate) listener: Option>, pub global_this: GlobalRef, /// JSC_BORROW: process-lifetime singleton; `&'static` so call sites read - /// `self.vm.is_shutting_down()` without a raw-pointer deref. + /// `self.vm.()` without a raw-pointer deref. pub(crate) vm: &'static VirtualMachine, pub ctx: Option>, // server reuses the same ctx } @@ -1713,7 +1713,8 @@ impl WindowsNamedPipeListeningContext { // SAFETY: `this` is the `data` pointer libuv hands back; it was set to a // live heap `WindowsNamedPipeListeningContext` in `listen_named_pipe`. let this_ref = unsafe { &mut *this }; - let shutting_down = this_ref.vm.is_shutting_down(); + let shutting_down = + this_ref.vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running; if status != uv::ReturnCode::ZERO || shutting_down || this_ref.listener.is_none() { // connection dropped or vm is shutting down or we are deiniting/closing return; @@ -1915,7 +1916,7 @@ pub(crate) extern "C" fn us_dispatch_socket_server_name( return core::ptr::null_mut(); } let handlers = tls.get_handlers(); - if handlers.vm.is_shutting_down() { + if handlers.vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return core::ptr::null_mut(); } let callback = handlers.on_server_name(); @@ -2008,7 +2009,7 @@ extern "C" fn us_dispatch_server_name( // duration of this synchronous handshake dispatch. let listener = unsafe { bun_ptr::ThisPtr::new(listener_ptr) }; let handlers = &listener.handlers; - if handlers.vm.is_shutting_down() { + if handlers.vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return core::ptr::null_mut(); } let callback = handlers.on_server_name(); diff --git a/src/runtime/socket/UpgradedDuplex.rs b/src/runtime/socket/UpgradedDuplex.rs index 5b5a364a17af..aae797de204c 100644 --- a/src/runtime/socket/UpgradedDuplex.rs +++ b/src/runtime/socket/UpgradedDuplex.rs @@ -207,7 +207,7 @@ impl UpgradedDuplex { // `vm` is always set via `from()`; `None` only in the zeroed placeholder // state, which never reaches here. let Some(vm) = self.vm else { return }; - if vm.is_shutting_down() { + if vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { return; } let duplex = self.origin; @@ -229,6 +229,9 @@ impl UpgradedDuplex { // Best-effort probe: consume the exception and fall through. Err(err) => drop(global.take_exception(err)), } + if global.has_exception() { + return; + } } let name = if msg_more { "write" } else { "end" }; diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index ebdd1a879828..11a5a22e8575 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -109,7 +109,11 @@ extern "C" fn select_alpn_callback( { let handlers = this.get_handlers(); let callback = handlers.on_alpn_callback(); - if !callback.is_empty() && !handlers.vm.is_shutting_down() && !in_.is_null() && inlen > 0 { + if !callback.is_empty() + && handlers.vm.script_execution_status() == jsc::ScriptExecutionStatus::Running + && !in_.is_null() + && inlen > 0 + { let scope = handlers.enter(); let global = handlers.global_object; let this_value = this.get_this_value(&global); @@ -838,7 +842,7 @@ impl NewSocket { log!("handleError"); let handlers = self.get_handlers(); let vm = handlers.vm; - if vm.is_shutting_down() { + if vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { return; } // the handlers must be kept alive for the duration of the function call @@ -879,7 +883,7 @@ impl NewSocket { } let vm = handlers.vm; - if vm.is_shutting_down() { + if vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { return; } // Hold the socket alive for the rest of the dispatch: `internal_flush` @@ -971,7 +975,7 @@ impl NewSocket { if callback.is_empty() || this.flags.get().contains(Flags::FINALIZING) { return; } - if handlers.vm.is_shutting_down() { + if handlers.vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { return; } @@ -1615,7 +1619,9 @@ impl NewSocket { let callback = handlers.on_end(); let vm = handlers.vm; - if callback.is_empty() || vm.is_shutting_down() { + if callback.is_empty() + || vm.script_execution_status() != jsc::ScriptExecutionStatus::Running + { this.poll_ref.with_mut(|p| p.unref(js_loop_ctx())); // If you don't handle TCP fin, we assume you're done. @@ -1758,8 +1764,8 @@ impl NewSocket { let mut is_open = false; if handlers.vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { - // `on_close` is single-JS-entry, so the native close it routes - // through here just takes the trap and returns. + // `on_close` skips its JS dispatch under the same gate, so the + // native close is still safe here. if reject_unauthorized { this.reject_unauthorized_connection(); } @@ -2036,7 +2042,7 @@ impl NewSocket { return Ok(()); } - if vm.is_shutting_down() { + if vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { drop(cleanup); return Ok(()); } @@ -2112,7 +2118,7 @@ impl NewSocket { if callback.is_empty() || this.flags.get().contains(Flags::FINALIZING) { return; } - if handlers.vm.is_shutting_down() { + if handlers.vm.script_execution_status() != jsc::ScriptExecutionStatus::Running { return; } diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 6b9f7c1b3366..17108bf1dbc7 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -455,3 +455,70 @@ test.skipIf(!isDebug)( }, 120_000, ); + +// Regression: the remaining NewSocket dispatch entry points (on_end, +// on_writable, on_data, on_timeout, on_close, handle_error, ALPN) still gated +// on is_shutting_down(), so a worker terminate() raised inside one socket's +// callback left the TerminationException pending for the next socket's +// dispatch in the same uSockets tick. This shape fires on plain TCP via +// on_end: the worker accepts NCLI connections, the parent FINs them all in +// one burst (so multiple on_end dispatches land in one poll sweep), then +// terminates the worker mid-sweep. +test.skipIf(!isDebug)( + "terminate() while a worker's Bun.listen end handler is firing does not trip assertNoException()", + async () => { + const ROUNDS = 15; + const NCLI = 60; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const net = require("node:net"); + const src = + "const { parentPort, workerData: d } = require('node:worker_threads');" + + "let opens = 0;" + + "const srv = Bun.listen({ hostname: '127.0.0.1', port: 0, allowHalfOpen: true, socket: {" + + " open(s) { if (++opens === d.ncli) parentPort.postMessage('ready'); }," + + " data() {}, end(s) { s.end(); }, close() {}, error() {} } });" + + "parentPort.postMessage(srv.port);"; + for (let r = 0; r < ${ROUNDS}; r++) { + const w = new Worker(src, { eval: true, workerData: { ncli: ${NCLI} } }); + const msgs = []; + const port = await new Promise((res, rej) => { + w.once("error", rej); + w.on("message", (m) => { msgs.push(m); if (msgs.length === 1) res(m); }); + }); + const clients = []; + for (let i = 0; i < ${NCLI}; i++) { + const c = net.connect(port, "127.0.0.1"); + c.on("error", () => {}); c.on("data", () => {}); + clients.push(c); + } + await new Promise((res, rej) => { + w.once("error", rej); + if (msgs.includes("ready")) return res(); + w.on("message", (m) => { if (m === "ready") res(); }); + }); + w.on("error", () => {}); + await Bun.sleep(r % 4); + for (const c of clients) c.end(); + await w.terminate(); + for (const c of clients) c.destroy(); + } + console.log("PASS"); + `, + ], + env: { ...bunEnv, BUN_JSC_validateExceptionChecks: "1" }, + 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("PASS\n"); + expect(exitCode).toBe(0); + }, + 120_000, +);