From fdc88c6a376aa2a265ff33103fc4ed8d45a3c64c Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 20 Aug 2026 23:38:37 +0000 Subject: [PATCH] Bun.serve, node:http: run request handlers with the event loop entered The request dispatch paths called the user's handler with entered_event_loop_count at 0 and drained microtasks explicitly afterwards. A native call made from inside the handler that dispatches another callback through enter()/exit() (server.upgrade() running open(), ws.close() running close()) was therefore the outermost pair, and its exit() drained the handler's own nextTick and promise callbacks before the handler's next statement. The same happened to the continuation of an async handler, which runs from the explicit drain, and to a request's abort listeners. Add EventLoop::enter_scope_without_checkpoint, an enter() whose exit only balances the count, and hold it across the handler call and the explicit drains in every dispatch path: fetch(), routes, both WebSocket upgrade paths, node:http, the HTTP/3 paths, the bake framework path, and RequestContext::on_abort. The drain points do not change, so a request still performs the same number of checkpoints. --- src/jsc/VirtualMachine.rs | 11 + src/jsc/event_loop.rs | 61 ++++++ src/runtime/server/RequestContext.rs | 9 + src/runtime/server/mod.rs | 4 + src/runtime/server/server_body.rs | 4 + test/js/bun/http/serve-http3.test.ts | 66 ++++++ .../js/bun/websocket/websocket-server.test.ts | 189 ++++++++++++++++++ test/js/node/http/node-http-with-ws.test.ts | 63 +++++- 8 files changed, 405 insertions(+), 2 deletions(-) diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index 3a0e43b786b6..793c5481290f 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -974,6 +974,17 @@ impl VirtualMachine { unsafe { EventLoop::enter_scope(self.event_loop) } } + /// `event_loop().enter()` now, `.exit_without_checkpoint()` on drop, for a + /// dispatcher that drains microtasks itself: see + /// [`EventLoop::enter_scope_without_checkpoint`]. + #[inline] + pub fn enter_event_loop_scope_without_checkpoint( + &self, + ) -> crate::event_loop::EventLoopEnterNoCheckpointGuard { + // SAFETY: as `enter_event_loop_scope`. + unsafe { EventLoop::enter_scope_without_checkpoint(self.event_loop) } + } + /// Safe shared-reference accessor for the process-lifetime dotenv loader /// (`vm.transpiler.env`). The loader is allocated once during VM init and /// never freed; callers previously open-coded `unsafe { &*vm.transpiler.env }`. diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index be2b578fd645..31020cc33854 100644 --- a/src/jsc/event_loop.rs +++ b/src/jsc/event_loop.rs @@ -271,6 +271,25 @@ impl Drop for EventLoopEnterGuard { } } +/// RAII pairing for [`EventLoop::enter`] / [`EventLoop::exit_without_checkpoint`]. +/// +/// Holds the raw pointer for the same reason as [`EventLoopEnterGuard`]. +/// Construct via [`EventLoop::enter_scope_without_checkpoint`]. +#[must_use = "dropping immediately exits the event loop scope"] +pub struct EventLoopEnterNoCheckpointGuard { + loop_: *mut EventLoop, +} + +impl Drop for EventLoopEnterNoCheckpointGuard { + #[inline] + fn drop(&mut self) { + // SAFETY: as `EventLoopEnterGuard`: `loop_` was live at + // `enter_scope_without_checkpoint` and the VM owns it for the process + // lifetime; short-lived `&mut` only. + unsafe { (*self.loop_).exit_without_checkpoint() }; + } +} + impl EventLoop { /// Before your code enters JavaScript at the top of the event loop, call /// `loop.enter()`. If running a single callback, prefer `runCallback` instead. @@ -315,6 +334,48 @@ impl EventLoop { EventLoopEnterGuard { loop_ } } + /// Balance an [`enter`](Self::enter) without the checkpoint [`exit`](Self::exit) + /// runs at the outermost level. See [`Self::enter_scope_without_checkpoint`]. + #[inline] + pub fn exit_without_checkpoint(&mut self) { + bun_core::scoped_log!( + EventLoop, + "exit_without_checkpoint() = {}", + self.entered_event_loop_count - 1 + ); + self.entered_event_loop_count -= 1; + } + + /// `enter()` now, [`exit_without_checkpoint`](Self::exit_without_checkpoint) + /// on drop. + /// + /// For a dispatcher that runs the checkpoint itself once the callback has + /// returned, at points of its own choosing: the HTTP request paths drain + /// explicitly so that they can look at a returned promise that the drain + /// settled (`RequestContext::on_response`, the node:http dispatch), and a + /// checkpoint on exit would add an empty one per request. + /// + /// What the scope is for is the count. Only while it is above zero is the + /// callback's frame safe from a checkpoint in the middle of it: a native + /// call made from inside the callback that dispatches another callback + /// through `enter()`/`exit()` (`server.upgrade()` running `open()`, + /// `ws.close()` running `close()`) is then a nested pair, not the outermost + /// one, so its exit does not run the nextTicks and promise reactions the + /// callback queued before its next statement. The dispatcher's explicit + /// drains are unconditional, so the held count does not skip them, and the + /// continuations they run are covered by it as well. + /// + /// # Safety + /// As [`Self::enter_scope`]. + #[inline] + pub unsafe fn enter_scope_without_checkpoint( + loop_: *mut EventLoop, + ) -> EventLoopEnterNoCheckpointGuard { + // SAFETY: caller contract — `loop_` is live; short-lived `&mut` only. + unsafe { (*loop_).enter() }; + EventLoopEnterNoCheckpointGuard { loop_ } + } + pub fn exit_maybe_drain_microtasks( &mut self, allow_drain_microtask: bool, diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index 5cd7650acfd7..9a1e1f6c17f5 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -1352,6 +1352,9 @@ where let server = this.server(); let vm = server.vm(); let global_this = server.global_this(); + // Entered for the abort listeners below, and (dropped last) for the + // drains below and in the release of `_ref`. + let _entered = vm.enter_event_loop_scope_without_checkpoint(); let _ref = RequestContextRef::adopt(this.as_ctx_ptr()); // This is a task in the event loop. // If we called into JavaScript, we must drain the microtask queue. @@ -2576,6 +2579,12 @@ where // // - If you return a Promise, we drain the microtask queue once // - If you return a streaming Response, we drain the microtask queue (possibly the 2nd time this task!) + // + // Like a task, the handler and these drains run with the event loop entered + // (the dispatchers hold `enter_event_loop_scope_without_checkpoint`), so a + // callback the handler dispatches synchronously through `enter()`/`exit()` + // (`server.upgrade()` -> `open()`, `ws.close()` -> `close()`) does not + // drain in the middle of the handler. pub(crate) fn on_response( &self, this: &ThisServer, diff --git a/src/runtime/server/mod.rs b/src/runtime/server/mod.rs index a083f78e040b..b398f044c8fb 100644 --- a/src/runtime/server/mod.rs +++ b/src/runtime/server/mod.rs @@ -997,6 +997,7 @@ impl NewServer { // SAFETY: `this` is the live server backref for this request. let server = unsafe { &*this }; + let _entered = server.vm().enter_event_loop_scope_without_checkpoint(); let global = server.global_this(); let response_value = match callback.call(global, server_js, &args) { Ok(v) => v, @@ -1143,6 +1144,7 @@ impl NewServer { // SAFETY: `this` is the live server backref for this request. let server = unsafe { &*this }; + let _entered = server.vm().enter_event_loop_scope_without_checkpoint(); let on_request = server.config.on_request; debug_assert!(!on_request.is_empty()); @@ -1194,6 +1196,7 @@ impl NewServer { // SAFETY: `server` is the live backref stored in `user_route`. let server_ref = unsafe { &*server }; + let _entered = server_ref.vm().enter_event_loop_scope_without_checkpoint(); let global = server_ref.global_this(); let server_request_list = Self::js_route_list_get_cached(server_js).expect("routeList cached value missing"); @@ -1275,6 +1278,7 @@ impl NewServer { core::ptr::NonNull::new(this).expect("on_node_http_request: this non-null"), ); let vm = this_ref.vm_mut(); + let _entered = this_ref.vm().enter_event_loop_scope_without_checkpoint(); req.set_yield(false); resp.timeout(this_ref.config.idle_timeout); diff --git a/src/runtime/server/server_body.rs b/src/runtime/server/server_body.rs index 5b5f40098dd9..6c3af22870a9 100644 --- a/src/runtime/server/server_body.rs +++ b/src/runtime/server/server_body.rs @@ -2947,6 +2947,7 @@ where return; }; + let _entered = server_ref.vm().enter_event_loop_scope_without_checkpoint(); let server_request_list = Self::js_route_list_get_cached(server_js).unwrap(); let call_route = if Ctx::IS_H3 { Bun__ServerRouteList__callRouteH3 @@ -3045,6 +3046,7 @@ where // SAFETY: `self_ptr` is `self`, live for this frame. Shared — the // handler call below re-enters JS, so no `&mut` may span it. let server = unsafe { &*self_ptr }; + let _entered = server.vm().enter_event_loop_scope_without_checkpoint(); let on_request_fn = server.config.on_request; debug_assert!(!on_request_fn.is_empty()); @@ -3359,6 +3361,7 @@ where .upgrade_context .set(UpgradeState::Pending(NonNull::from(upgrade_ctx))) }; + let _entered = server_ref.vm().enter_event_loop_scope_without_checkpoint(); let server_request_list = Self::js_route_list_get_cached(server_js).unwrap(); // S008: `JSGlobalObject` is an `opaque_ffi!` ZST — safe deref. let global = bun_opaque::opaque_deref(server_ref.global_this); @@ -3441,6 +3444,7 @@ where resp.end_without_body(true); return; } + let _entered = this.vm().enter_event_loop_scope_without_checkpoint(); this.on_pending_request(); req.set_yield(false); // SAFETY: `request_pool` is non-null while the server is alive; `claim()` diff --git a/test/js/bun/http/serve-http3.test.ts b/test/js/bun/http/serve-http3.test.ts index 78e37448cbbb..d0ce2e64e30c 100644 --- a/test/js/bun/http/serve-http3.test.ts +++ b/test/js/bun/http/serve-http3.test.ts @@ -1,3 +1,4 @@ +import type { ServerWebSocket } from "bun"; import { describe, expect, test } from "bun:test"; import { createHash, createPrivateKey, randomBytes } from "crypto"; import { readFileSync } from "fs"; @@ -1449,3 +1450,68 @@ describe("Bun.serve HTTP/3 request validation", () => { expect({ selfSigned, chained }).toEqual({ selfSigned: "closed", chained: "200 1" }); }); }); + +// The HTTP/3 twin of the HTTP/1 cases in websocket-server.test.ts: ws.close() +// runs close() before it returns, and a request handler that calls it must still +// run to completion before the nextTick and promise callbacks it queued. The +// socket being closed lives on a plain HTTP/1 server, since HTTP/3 carries no +// WebSockets; any handler can close it. +describe("Bun.serve HTTP/3 request handlers run to completion before the callbacks they queued", () => { + async function openHeldSocket() { + const order: string[] = []; + const opened = Promise.withResolvers>(); + const closed = Promise.withResolvers(); + const wsServer = Bun.serve({ + port: 0, + fetch: (req, srv) => (srv.upgrade(req) ? undefined : new Response("upgrade() failed", { status: 500 })), + websocket: { + open: ws => opened.resolve(ws), + message() {}, + close() { + order.push("close()"); + }, + }, + }); + const client = new WebSocket(wsServer.url.href.replace(/^http/, "ws")); + client.onerror = () => closed.resolve(); + client.onclose = () => closed.resolve(); + const held = await opened.promise; + return { + order, + closed: closed.promise, + handler() { + process.nextTick(() => order.push("nextTick")); + Promise.resolve().then(() => order.push("microtask")); + held.close(); + order.push("rest of handler"); + return new Response("ok"); + }, + [Symbol.dispose]: () => wsServer.stop(true), + }; + } + + test("fetch() and a route handler closing an open ServerWebSocket", async () => { + using viaFetch = await openHeldSocket(); + using viaRoute = await openHeldSocket(); + await using server = Bun.serve({ + port: 0, + tls, + http3: true, + routes: { "/route": viaRoute.handler }, + fetch: viaFetch.handler, + }); + + const responses = { + fetch: await h3Exchange(server.port, requestHeaders("/")), + route: await h3Exchange(server.port, requestHeaders("/route")), + }; + await Promise.all([viaFetch.closed, viaRoute.closed]); + + const expectedOrder = ["close()", "rest of handler", "nextTick", "microtask"]; + expect({ responses, fetch: viaFetch.order, route: viaRoute.order }).toEqual({ + responses: { fetch: "200 ok", route: "200 ok" }, + fetch: expectedOrder, + route: expectedOrder, + }); + }); +}); diff --git a/test/js/bun/websocket/websocket-server.test.ts b/test/js/bun/websocket/websocket-server.test.ts index 9f1651cf855d..6585d97d866b 100644 --- a/test/js/bun/websocket/websocket-server.test.ts +++ b/test/js/bun/websocket/websocket-server.test.ts @@ -1998,3 +1998,192 @@ describe("server.upgrade() validates the opening handshake", () => { expect(upgradeResult).toBe(false); }); }); + +// server.upgrade() runs open() before it returns, and ws.close() runs close() +// before it returns. A request handler (or a request's abort listener) that +// calls one of them must still run to completion first: the nextTick and +// promise callbacks it queued run once it has returned, as they do for a timer +// or socket callback that does the same thing. They used to run inside the +// upgrade()/close() call. +describe.concurrent("request handlers run to completion before the callbacks they queued", () => { + function queueThen(order: string[], nativeCall: () => void) { + process.nextTick(() => order.push("nextTick")); + Promise.resolve().then(() => order.push("microtask")); + nativeCall(); + order.push("rest of handler"); + } + + function wsUrl(server: Server, pathname: string) { + return new URL(pathname, server.url.href.replace(/^http/, "ws")); + } + + // Resolves once the server has closed the socket (or the handshake failed). + function connectUntilClosed(server: Server, pathname: string) { + const { promise, resolve } = Promise.withResolvers(); + const ws = new WebSocket(wsUrl(server, pathname)); + ws.onerror = () => resolve(); + ws.onclose = () => resolve(); + return promise; + } + + const upgradeHandler = (order: string[]) => (req: Request, srv: Server) => { + queueThen(order, () => { + if (!srv.upgrade(req)) order.push("upgrade() failed"); + }); + }; + + const websocket = (order: string[], onOpen: (ws: ServerWebSocket) => void) => + ({ + open(ws) { + onOpen(ws); + }, + message() {}, + close() { + order.push("close()"); + }, + }) satisfies WebSocketHandler; + + it("fetch() calling server.upgrade()", async () => { + const order: string[] = []; + using server = serve({ + port: 0, + fetch: upgradeHandler(order), + websocket: websocket(order, ws => { + order.push("open()"); + ws.close(); + }), + }); + + await connectUntilClosed(server, "/"); + expect(order).toEqual(["open()", "close()", "rest of handler", "nextTick", "microtask"]); + }); + + it("a route handler calling server.upgrade()", async () => { + const order: string[] = []; + using server = serve({ + port: 0, + routes: { "/ws": upgradeHandler(order) }, + websocket: websocket(order, ws => { + order.push("open()"); + ws.close(); + }), + }); + + await connectUntilClosed(server, "/ws"); + expect(order).toEqual(["open()", "close()", "rest of handler", "nextTick", "microtask"]); + }); + + // Opens a websocket on `/ws` and hands back the server side of it. + async function openHeldSocket(server: Server, opened: Promise>) { + const closed = connectUntilClosed(server, "/ws"); + const held = await opened; + return { held, closed }; + } + + it("fetch() closing an open ServerWebSocket", async () => { + const order: string[] = []; + const opened = Promise.withResolvers>(); + let held: ServerWebSocket; + using server = serve({ + port: 0, + fetch(req, srv) { + if (new URL(req.url).pathname === "/ws") { + return srv.upgrade(req) ? undefined : new Response("upgrade() failed", { status: 500 }); + } + queueThen(order, () => held.close()); + return new Response("ok"); + }, + websocket: websocket(order, opened.resolve), + }); + + const sockets = await openHeldSocket(server, opened.promise); + held = sockets.held; + expect(await fetch(new URL("/close-it", server.url)).then(res => res.text())).toBe("ok"); + await sockets.closed; + expect(order).toEqual(["close()", "rest of handler", "nextTick", "microtask"]); + }); + + it("a route handler closing an open ServerWebSocket", async () => { + const order: string[] = []; + const opened = Promise.withResolvers>(); + let held: ServerWebSocket; + using server = serve({ + port: 0, + routes: { + "/ws": (req, srv) => (srv.upgrade(req) ? undefined : new Response("upgrade() failed", { status: 500 })), + "/close-it": () => { + queueThen(order, () => held.close()); + return new Response("ok"); + }, + }, + websocket: websocket(order, opened.resolve), + }); + + const sockets = await openHeldSocket(server, opened.promise); + held = sockets.held; + expect(await fetch(new URL("/close-it", server.url)).then(res => res.text())).toBe("ok"); + await sockets.closed; + expect(order).toEqual(["close()", "rest of handler", "nextTick", "microtask"]); + }); + + it("a request's abort listener closing an open ServerWebSocket", async () => { + const order: string[] = []; + const opened = Promise.withResolvers>(); + const reachedHandler = Promise.withResolvers(); + let held: ServerWebSocket; + using server = serve({ + port: 0, + fetch(req, srv) { + if (new URL(req.url).pathname === "/ws") { + return srv.upgrade(req) ? undefined : new Response("upgrade() failed", { status: 500 }); + } + req.signal.addEventListener("abort", () => queueThen(order, () => held.close())); + reachedHandler.resolve(); + // Never responds: the client aborts the request instead. + return new Promise(() => {}); + }, + websocket: websocket(order, opened.resolve), + }); + + const sockets = await openHeldSocket(server, opened.promise); + held = sockets.held; + const controller = new AbortController(); + const aborted = fetch(new URL("/abort-me", server.url), { signal: controller.signal }); + await reachedHandler.promise; + controller.abort(); + await expect(aborted).rejects.toThrow(); + await sockets.closed; + expect(order).toEqual(["close()", "rest of handler", "nextTick", "microtask"]); + }); + + // After an await, the rest of an async handler runs from the microtask + // checkpoint the server performs as soon as the handler returns its promise. + // Only a promise callback is queued here: a nextTick queued from inside a + // microtask is ordered differently from one queued by synchronous code, and + // that ordering is not what this test is about. + it("the continuation of an async fetch() closing an open ServerWebSocket", async () => { + const order: string[] = []; + const opened = Promise.withResolvers>(); + let held: ServerWebSocket; + using server = serve({ + port: 0, + async fetch(req, srv) { + if (new URL(req.url).pathname === "/ws") { + return srv.upgrade(req) ? undefined : new Response("upgrade() failed", { status: 500 }); + } + await Promise.resolve(); + Promise.resolve().then(() => order.push("microtask")); + held.close(); + order.push("rest of handler"); + return new Response("ok"); + }, + websocket: websocket(order, opened.resolve), + }); + + const sockets = await openHeldSocket(server, opened.promise); + held = sockets.held; + expect(await fetch(new URL("/close-it", server.url)).then(res => res.text())).toBe("ok"); + await sockets.closed; + expect(order).toEqual(["close()", "rest of handler", "microtask"]); + }); +}); diff --git a/test/js/node/http/node-http-with-ws.test.ts b/test/js/node/http/node-http-with-ws.test.ts index a3ef8cac6a29..0ab97a4b1d26 100644 --- a/test/js/node/http/node-http-with-ws.test.ts +++ b/test/js/node/http/node-http-with-ws.test.ts @@ -1,9 +1,11 @@ -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tls as options } from "harness"; +import http from "http"; import https from "https"; +import { once } from "node:events"; import type { AddressInfo } from "node:net"; import tls from "tls"; -import { WebSocketServer } from "ws"; +import { WebSocketServer, type WebSocket as WsWebSocket } from "ws"; test.concurrent("WebSocket upgrade should unref poll_ref from response", async () => { // Regression test for bug where poll_ref was not unref'd on WebSocket upgrade @@ -103,3 +105,60 @@ test.concurrent("should not crash when closing sockets after upgrade", async () await promise; expect().pass(); }); + +// ws.close() on a server-side socket runs the native close callback before it +// returns. A node:http 'request' or 'upgrade' handler that calls it must still +// run to completion first: the nextTick and promise callbacks the handler +// queued run once it has returned, as in Node.js. They used to run inside the +// close() call. +describe.concurrent("request handlers run to completion before the callbacks they queued", () => { + function queueThen(order: string[], nativeCall: () => void) { + process.nextTick(() => order.push("nextTick")); + Promise.resolve().then(() => order.push("microtask")); + nativeCall(); + order.push("rest of handler"); + } + + async function listen(server: http.Server) { + await once(server.listen(0, "127.0.0.1"), "listening"); + return (server.address() as AddressInfo).port; + } + + // Resolves once the server has closed the socket (or the handshake failed). + function connectUntilClosed(port: number) { + const { promise, resolve } = Promise.withResolvers(); + const ws = new WebSocket(`ws://127.0.0.1:${port}/`); + ws.onerror = () => resolve(); + ws.onclose = () => resolve(); + return promise; + } + + test("the 'connection' handler of a WebSocketServer, run from the upgrade request", async () => { + const order: string[] = []; + await using server = http.createServer(); + const wss = new WebSocketServer({ server }); + wss.on("connection", ws => queueThen(order, () => ws.close())); + + await connectUntilClosed(await listen(server)); + expect(order).toEqual(["rest of handler", "nextTick", "microtask"]); + }); + + test("a 'request' handler closing an open WebSocketServer socket", async () => { + const order: string[] = []; + const connected = Promise.withResolvers(); + let held: WsWebSocket; + await using server = http.createServer((req, res) => { + queueThen(order, () => held.close()); + res.end("ok"); + }); + const wss = new WebSocketServer({ server }); + wss.on("connection", connected.resolve); + + const port = await listen(server); + const closed = connectUntilClosed(port); + held = await connected.promise; + expect(await fetch(`http://127.0.0.1:${port}/`).then(res => res.text())).toBe("ok"); + await closed; + expect(order).toEqual(["rest of handler", "nextTick", "microtask"]); + }); +});