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"]); + }); +});