diff --git a/src/runtime/server/ServerWebSocket.rs b/src/runtime/server/ServerWebSocket.rs index 5225c12379fb..915ee1894870 100644 --- a/src/runtime/server/ServerWebSocket.rs +++ b/src/runtime/server/ServerWebSocket.rs @@ -708,6 +708,13 @@ impl ServerWebSocket { return; } + // on_open's error branch closes the socket, landing here with the + // termination from its handler still pending. Both branches below + // enter JS, which trips assertNoException(). + if handler.global_object().has_exception() { + return; + } + // Copy to a stack local before `sig.signal()` re-enters JS: a GC // between the test and the `.call(...)` could otherwise collect it. let on_close_handler = handler.on_close; diff --git a/src/runtime/server/WebSocketServerContext.rs b/src/runtime/server/WebSocketServerContext.rs index 41d8be5ac37d..ce65d055ae8a 100644 --- a/src/runtime/server/WebSocketServerContext.rs +++ b/src/runtime/server/WebSocketServerContext.rs @@ -85,6 +85,11 @@ impl Handler { global_object: &JSGlobalObject, error_value: JSValue, ) { + // Termination raised inside the preceding callback.call() cannot be + // cleared; entering JS again trips executeCallImpl's assertNoException. + if global_object.has_exception() { + return; + } if !on_error.is_empty_or_undefined_or_null() { let _ = on_error .call(global_object, JSValue::UNDEFINED, &[error_value]) diff --git a/src/runtime/socket/Handlers.rs b/src/runtime/socket/Handlers.rs index a83944b4153c..8819de1760ed 100644 --- a/src/runtime/socket/Handlers.rs +++ b/src/runtime/socket/Handlers.rs @@ -282,6 +282,11 @@ impl Handlers { } let global_object = self.global_object; + // Termination raised inside the preceding callback.call() cannot be + // cleared; entering JS again trips executeCallImpl's assertNoException. + if global_object.has_exception() { + return false; + } let on_error = self.on_error(); if on_error.is_empty() { diff --git a/src/runtime/socket/socket_body.rs b/src/runtime/socket/socket_body.rs index f0967f4f7a99..0ed1b63dc8c6 100644 --- a/src/runtime/socket/socket_body.rs +++ b/src/runtime/socket/socket_body.rs @@ -2032,6 +2032,14 @@ impl NewSocket { return Ok(()); } + // An earlier callback in this dispatch may have left a termination + // pending — on_open's error branch closes the socket from + // mark_inactive(), landing here. Entering JS trips assertNoException(). + if handlers.global_object.has_exception() { + drop(cleanup); + return Ok(()); + } + // the handlers must be kept alive for the duration of the function call // that way if we need to call the error handler, we can let scope = handlers.enter(); diff --git a/test/js/bun/net/socket-handler-worker-terminate.test.ts b/test/js/bun/net/socket-handler-worker-terminate.test.ts new file mode 100644 index 000000000000..c2d44669de8a --- /dev/null +++ b/test/js/bun/net/socket-handler-worker-terminate.test.ts @@ -0,0 +1,223 @@ +// worker.terminate() firing while a Bun.listen socket handler is mid-call must +// not re-enter JS with the termination exception still pending. The socket +// dispatch path calls the error handler when the primary handler throws, and +// termination cannot be cleared: entering JS again trips Interpreter:: +// executeCallImpl's `assertNoException`. Repro for the +// test/js/node/test/parallel/test-http2-reset-flood.js SIGABRT. +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isWindows } from "harness"; + +// The handler body shared by both the Bun.listen close handler and the +// Bun.serve websocket message handler: a two-phase Atomics handshake so the +// safepoint spin starts only once the parent is already calling terminate(), +// making the window build-speed-independent. +// shared[0]: worker -> parent ("I am inside the handler"; 2 = spin ran out) +// shared[1]: parent -> worker ("terminate() is in flight") +const handlerBody = ` + Atomics.store(shared, 0, 1); + Atomics.notify(shared, 0); + Atomics.wait(shared, 1, 0, 5000); + // terminate() is now in flight; spin on safepoints so the termination + // exception is raised inside this handler frame. Bounded so a missed + // terminate cannot hang. + let sink = 0; + for (let i = 0; i < 10_000_000; i++) sink += Atomics.load(shared, 1); + Atomics.store(shared, 0, 2); +`; + +const workerSource = ` +const { parentPort } = require("worker_threads"); +const shared = new Int32Array(require("worker_threads").workerData); + +const server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { socket.write("hello"); }, + data() {}, + // socket_body.rs:on_close: a termination raised here returns Err to the + // native caller, which then invokes Handlers::call_error_handler. + close() {${handlerBody}}, + error() {}, + }, +}); +parentPort.postMessage(server.port); +`; + +const wsWorkerSource = ` +const { parentPort } = require("worker_threads"); +const shared = new Int32Array(require("worker_threads").workerData); + +const server = Bun.serve({ + port: 0, + fetch(req, server) { + if (server.upgrade(req)) return; + return new Response("no upgrade", { status: 400 }); + }, + websocket: { + open(ws) { ws.send("hello"); }, + // ServerWebSocket.rs:on_message: a termination raised here returns Err to + // the native caller, which then invokes WebSocketServerContext:: + // run_error_callback. + message() {${handlerBody}}, + close() {}, + error() {}, + }, +}); +parentPort.postMessage(server.port); +`; + +function parentSource(workerSrc: string, driveHandler: string, cleanup: string) { + return ` +const { Worker } = require("worker_threads"); +const net = require("net"); + +(async () => { + // Each iteration is an independent opportunity for terminate() to land inside + // the handler. The two-phase handshake makes it land on the first try; a few + // repeats leave headroom for CI scheduling jitter. + let hits = 0; + for (let i = 0; i < 6; i++) { + const sab = new SharedArrayBuffer(8); + const shared = new Int32Array(sab); + const worker = new Worker(${JSON.stringify(workerSrc)}, { eval: true, workerData: sab }); + const port = await new Promise(resolve => worker.once("message", resolve)); + ${driveHandler} + // Wait until the worker is inside the handler, then release it from its + // own wait and terminate so the termination exception lands mid-handler. + if (Atomics.wait(shared, 0, 0, 5000) === "timed-out") throw new Error("handler never fired"); + Atomics.store(shared, 1, 1); + Atomics.notify(shared, 1); + await worker.terminate(); + ${cleanup} + // shared[0] === 1 means termination landed mid-handler (the spin loop was + // interrupted); 2 means the handler returned normally first. + if (Atomics.load(shared, 0) === 1) hits++; + } + if (hits === 0) throw new Error("terminate() never landed inside the handler (0/6)"); + console.log("ok", hits); +})(); +`; +} + +// A terminate inside the *open* handler takes a different route back into JS: +// on_open's error branch runs mark_inactive() -> close_and_detach() -> +// us_socket_close, which synchronously dispatches on_close. +const openWorkerSource = ` +const { parentPort } = require("worker_threads"); +const shared = new Int32Array(require("worker_threads").workerData); +const net = require("net"); + +const server = net.createServer(socket => { + socket.write(Buffer.alloc(1 << 16, "x").toString()); + ${handlerBody} +}); +server.listen(0, "127.0.0.1", () => parentPort.postMessage(server.address().port)); +`; + +// Same shape for Bun.serve websockets: ServerWebSocket::on_open's error branch +// calls websocket().close(), which re-enters ServerWebSocket::on_close. +const wsOpenWorkerSource = ` +const { parentPort } = require("worker_threads"); +const shared = new Int32Array(require("worker_threads").workerData); + +const server = Bun.serve({ + port: 0, + fetch(req, server) { + if (server.upgrade(req)) return; + return new Response("no upgrade", { status: 400 }); + }, + websocket: { + open() {${handlerBody}}, + message() {}, + close() {}, + error() {}, + }, +}); +parentPort.postMessage(server.port); +`; + +const listenDriver = ` + const conn = net.connect({ port, host: "127.0.0.1" }); + await new Promise(resolve => conn.once("data", resolve)); + // Closing the client drives on_close on the server's per-connection socket. + conn.destroy(); +`; + +// The parent must not block in Atomics.wait before the bytes have actually left +// the socket, or the worker's handler never fires. +const openDriver = ` + const conn = net.connect({ port, host: "127.0.0.1" }); + conn.on("error", () => {}); + await new Promise(resolve => conn.once("connect", resolve)); +`; + +// Raw handshake rather than \`new WebSocket\`: the upgrade has to be flushed +// before the parent blocks, and the server's open() never lets the client's +// handshake complete. +const wsOpenDriver = ` + const conn = net.connect({ port, host: "127.0.0.1" }); + conn.on("error", () => {}); + conn.on("data", () => {}); + await new Promise(resolve => conn.once("connect", resolve)); + await new Promise(resolve => conn.write( + "GET / HTTP/1.1\\r\\nHost: 127.0.0.1\\r\\nUpgrade: websocket\\r\\nConnection: Upgrade\\r\\n" + + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\\r\\nSec-WebSocket-Version: 13\\r\\n\\r\\n", + resolve, + )); +`; + +const wsDriver = ` + const ws = new WebSocket("ws://127.0.0.1:" + port); + await new Promise((resolve, reject) => { ws.onopen = resolve; ws.onerror = reject; }); + // Terminating the worker ends the connection under the client; swallow the + // resulting ErrorEvent so it does not surface as an uncaught error. + ws.onerror = () => {}; + ws.send("go"); +`; + +// On Windows the usockets close path and Atomics.wait scheduling differ enough +// that the window does not open; the bug is platform-agnostic and is exercised +// on the POSIX lanes. +describe.skipIf(isWindows)( + "worker.terminate() mid-handler does not re-enter JS with a pending termination exception", + () => { + for (const [name, src] of [ + ["Bun.listen close handler (socket Handlers::call_error_handler)", parentSource(workerSource, listenDriver, "")], + [ + "Bun.serve websocket message handler (WebSocketServerContext::run_error_callback)", + parentSource(wsWorkerSource, wsDriver, "ws.close();"), + ], + [ + "node:net connection handler (on_open -> mark_inactive -> on_close)", + parentSource(openWorkerSource, openDriver, "conn.destroy();"), + ], + [ + "Bun.serve websocket open handler (ServerWebSocket::on_close)", + parentSource(wsOpenWorkerSource, wsOpenDriver, "conn.destroy();"), + ], + ] as const) { + test.concurrent( + name, + async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // The unpatched build aborts inside an iteration (exit 134, + // "ASSERTION FAILED: !exception()" on stderr, no stdout). Assert on + // stdout/exitCode so benign debug-build stderr noise cannot cause a + // false positive; the crash's stderr is in the diff either way. + expect({ stdout: stdout.trim(), stderr, exitCode }).toMatchObject({ + stdout: expect.stringMatching(/^ok [1-6]$/), + exitCode: 0, + }); + }, + 60_000, + ); + } + }, +);