diff --git a/src/runtime/server/RequestContext.rs b/src/runtime/server/RequestContext.rs index d44f8b4afa0f..8eaa58fe3848 100644 --- a/src/runtime/server/RequestContext.rs +++ b/src/runtime/server/RequestContext.rs @@ -924,21 +924,7 @@ where // incomplete message instead of a second header block spliced into // the chunked body. if !has_responded && ctx.flags.has_written_status() { - if !value.is_empty_or_undefined_or_null() - && let Some(server) = ctx.server - { - // SAFETY: BACKREF; see drain_microtasks() re: the const→mut cast. - unsafe { - (*std::ptr::from_ref::((*server).vm()).cast_mut()) - .run_error_handler(value, None); - } - } - let state = resp.state(); - if state.is_http_write_called() && state.is_response_pending() { - ctx.force_close(); - } else { - ctx.end_stream(ctx.should_close_connection()); - } + ctx.terminate_body_after_committed_status(value); return; } @@ -1229,6 +1215,34 @@ where } } + /// A body failure after the status line has been committed: `error()` + /// cannot replace the response, so report the failure and terminate the + /// body so the client observes an incomplete message (RFC 9112 section 7). + /// If body bytes have already been written the connection is force-closed + /// without the terminating chunk; otherwise the chunked body is ended + /// empty. The uncork flushes bytes written inside the `do_render_stream` + /// cork and is a no-op when not corked. + fn terminate_body_after_committed_status(&mut self, err: JSValue) { + if !err.is_empty_or_undefined_or_null() + && let Some(server) = self.server + { + // SAFETY: BACKREF; see drain_microtasks() re: the const→mut cast. + unsafe { + (*std::ptr::from_ref::((*server).vm()).cast_mut()) + .run_error_handler(err, None); + } + } + if let Some(resp) = self.resp { + let state = resp.state(); + if state.is_http_write_called() && state.is_response_pending() { + resp.uncork(); + self.force_close(); + return; + } + } + self.end_stream(self.should_close_connection()); + } + /// # Safety /// `this` must be the live `RequestContext` user-data pointer registered with uWS. pub(crate) fn on_writable_response_buffer( @@ -2048,11 +2062,12 @@ where // we use this memory address to disable signals being sent response_stream.sink.signal.clear(); debug_assert!(response_stream.sink.signal.is_dead()); - // we need to render metadata before assignToStream because the stream can call res.end - // and this would auto write an 200 status - if !this.flags.has_written_status() { - this.render_metadata(); - } + // The status line is written lazily by `on_first_write` (fired from the + // sink just before the first body byte reaches uWS, and from + // `end_from_js` when the stream ends with no body). Writing it here + // would commit the user's status before the stream produced anything, + // so a stream that errors before its first yield could no longer be + // routed to `error()`. // We are already corked! // `Option>` is layout-compatible with `*mut c_void` (niche). @@ -2141,11 +2156,13 @@ where // keeps `handle_reject` from ending the response out // from under the sink while the stream is in flight. this.flags.set_has_marked_pending(true); - if !this.flags.has_written_status() { - response_stream.sink.on_first_write = None; - response_stream.sink.ctx = None; - this.render_metadata(); - } + // Keep `on_first_write` armed: if the stream rejects + // before producing a byte, `handle_reject_stream` can + // still hand the error to `error()`. The `ref_()` below + // keeps this context (and so the `ctx` pointer the sink + // holds) alive until the pump promise settles, and the + // Locked body value protects the Response for the + // deferred `render_metadata()`. // The sink only tracks `has_backpressure`; the // on_writable registration lives here so it is armed @@ -2185,20 +2202,6 @@ where } jsc::PromiseResult::Rejected(err) => { stream_log!("promise Rejected"); - // Consuming the rejection here is what keeps it out of - // the unhandledRejection reporter, so surface it here. - // DEBUG_MODE already reports it in handle_reject_stream. - if !DEBUG_MODE - && let Some(server) = this.server - && !err.is_empty_or_undefined_or_null() - { - // SAFETY: BACKREF; see drain_microtasks() re: the - // const→mut cast. - unsafe { - (*std::ptr::from_ref::((*server).vm()).cast_mut()) - .run_error_handler(err, None); - } - } let mut readable_ref = core::mem::take(&mut this.response_body_readable_stream_ref); Self::handle_reject_stream(this, global_this, err); @@ -2983,76 +2986,6 @@ where stream_log!("onReject()"); - // `resp` must not be dereferenced once the sink has already ended the - // response (see `end_already_responded_stream`). - if !ended_response && !req.flags.has_written_status() { - req.render_metadata(); - } - - if DEBUG_MODE { - if let Some(server) = req.server { - if !err.is_empty_or_undefined_or_null() { - // SAFETY: BACKREF - let server = &*server; - let mut exception_list: jsc::ExceptionList = Vec::new(); - // SAFETY: see drain_microtasks() re: const→mut cast. - unsafe { - (*std::ptr::from_ref::(server.vm()).cast_mut()) - .run_error_handler(err, Some(&mut exception_list)); - } - let exception_list = jsc_exceptions_to_api(exception_list); - - // The fallback page below writes into `resp`, which must - // not be dereferenced once the sink has already ended the - // response (see `end_already_responded_stream`). - if !ended_response && server.dev_server().is_some() { - // Render the error fallback HTML page like renderDefaultError does - if !req.flags.has_written_status() { - req.flags.set_has_written_status(true); - if let Some(resp) = req.resp { - resp.write_status(b"500 Internal Server Error"); - resp.write_header( - b"content-type", - &bun_http_types::MimeType::HTML.value, - ); - } - } - - // Create error message for the stream rejection - let cwd = bun_resolver::fs::FileSystem::get().top_level_dir; - let fallback_container = Box::new(Api::FallbackMessageContainer { - message: Some( - b"Stream error during server-side rendering" - .to_vec() - .into_boxed_slice(), - ), - router: None, - reason: Some(Api::FallbackStep::fetch_event_handler), - cwd: Some(cwd.to_vec().into_boxed_slice()), - problems: Some(Api::Problems { - code: 500, - name: b"StreamError".to_vec().into_boxed_slice(), - exceptions: exception_list, - build: Api::Log::default(), - }), - }); - - let mut bb: Vec = Vec::new(); - - Fallback::render_backend(&fallback_container, &mut bb) - .expect("unreachable"); - - if let Some(resp) = req.resp { - // SAFETY: FFI handle - resp.write(&bb); - } - - req.end_stream(req.should_close_connection()); - return; - } - } - } - } // HTTP/1 only: the sink already fully ended the response, so `resp` // can no longer be dereferenced (see `end_already_responded_stream`). // H3 keeps the end_stream() path: its `resp` is still alive here and @@ -3061,17 +2994,40 @@ where req.end_already_responded_stream(); return; } - // Body bytes were already written: close without the terminating chunk - // (RFC 9112 section 7) so the client sees an incomplete message, not a - // truncated body that looks like a complete, successful response. - if let Some(resp) = req.resp { - let state = resp.state(); - if state.is_http_write_called() && state.is_response_pending() { - req.force_close(); - return; - } + + // This stream is itself `error()`'s Response body: `error()` already + // ran, so `handle_reject` below would fall through to the default + // 500. Keep `error()`'s own status instead and terminate. + if req.flags.has_called_error_handler() && !req.flags.has_written_status() { + req.render_metadata(); } - req.end_stream(req.should_close_connection()); + + // No status line committed: the stream failed before its first body + // byte reached uWS, so `error()` can still replace the response. + // Bytes still in the sink buffer never reached the client and are + // discarded (the sink was marked done above), so they do not count + // as committed. + // `handle_reject` runs the user's handler (or the default 500 page) + // and reports the failure itself. + if !req.flags.has_written_status() { + // The Pending branch set `has_marked_pending` for the in-flight + // pump; clear it so `handle_reject` falls through to + // `render_missing` when `error()` returns nothing to render. + req.flags.set_has_marked_pending(false); + Self::handle_reject( + req, + if err.is_empty() { + JSValue::UNDEFINED + } else { + err + }, + ); + return; + } + + // Status already on the wire: a second status from `error()` cannot be + // sent. + req.terminate_body_after_committed_status(err); } pub fn do_render_with_body( @@ -3486,6 +3442,16 @@ where ) .unwrap_or_else(|err| server.global_this().take_exception(err)); let _keep = jsc::EnsureStillAlive(result); + // `handle_reject_stream` reaches here with the original + // `fetch()` Response still protected. Every arm below either + // ends the response itself or installs a new one, so release + // the old root before branching: `process_on_error_promise` + // overwrites `response_jsvalue` (Fulfilled) or hands it to + // `handle_resolve` (Pending) which asserts the flag is clear. + if !self.response_jsvalue.is_empty() && self.flags.response_protected() { + self.response_jsvalue.unprotect(); + self.flags.set_response_protected(false); + } if !result.is_empty_or_undefined_or_null() { if let Some(err) = result.to_error() { self.finish_running_error_handler(err, status); @@ -3500,6 +3466,29 @@ where // falls through to the default error page below. // SAFETY: `response` is the live, rooted cell pointer. if HTTPStatusText::is_sendable(unsafe { (*response).status_code() }) { + // `render()` may suspend (stream/file body); root + // the Response like `process_on_error_promise` and + // the normal fetch() path do, so the deferred + // `render_metadata()` can still read it. + self.response_jsvalue = result; + self.flags.set_response_protected(false); + // SAFETY: as above; body_value borrows the rooted + // Response cell, disjoint from `self`. + let body_value = unsafe { (*response).get_body_value() }; + body_value.to_blob_if_possible(); + match body_value { + Body::Value::Blob(blob) => { + if shim::blob_needs_to_read_file(blob) { + result.protect(); + self.flags.set_response_protected(true); + } + } + Body::Value::Locked(_) => { + result.protect(); + self.flags.set_response_protected(true); + } + _ => {} + } // SAFETY: as above. unsafe { self.render(response) }; return; diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index d4dfdc47f1cf..43b453c079f3 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -1773,6 +1773,10 @@ impl HTTPServerWritable { return bun_sys::Result::Ok(value); } } else { + // The owner's status line is written on the first body byte; an + // empty body never fires that, so fire it here before `res.end()` + // writes uWS's default `200 OK`. + self.handle_first_write_if_necessary(); if let Some(res) = self.any_res() { res.end(b"", false); } @@ -1903,6 +1907,11 @@ impl HTTPServerWritable { let _ = self.flush_no_wait(); self.done = true; + // The owner's status line is written on the first body byte; an + // empty body never fires that, so fire it here before + // `res.end_stream()` writes uWS's default `200 OK`. Same hazard + // as `end_from_js`'s empty-body branch. + self.handle_first_write_if_necessary(); if let Some(res) = self.any_res() { // is actually fine to call this if the socket is closed because of flushNoWait, the free will be defered by usockets res.end_stream(false); diff --git a/test/js/bun/http/async-iterator-stream.test.ts b/test/js/bun/http/async-iterator-stream.test.ts index ea4988566180..e4b1f13e2778 100644 --- a/test/js/bun/http/async-iterator-stream.test.ts +++ b/test/js/bun/http/async-iterator-stream.test.ts @@ -114,11 +114,15 @@ describe.concurrent("Streaming body via", () => { expect(exitCode).toBe(0); }); - test("async generator function throws an error but continues to send the headers", async () => { + test("async generator function that throws before yielding is routed to error()", async () => { + let result: { status: number; xHey: string | null; body: string } | undefined; const onMessage = mock(async url => { - const response = await fetch(url); - expect(response.headers.get("X-Hey")).toBe("123"); - subprocess?.kill(); + try { + const response = await fetch(url); + result = { status: response.status, xHey: response.headers.get("X-Hey"), body: await response.text() }; + } finally { + subprocess.kill(); + } }); await using subprocess = spawn({ @@ -132,7 +136,9 @@ describe.concurrent("Streaming body via", () => { let [exitCode, stderr] = await Promise.all([subprocess.exited, subprocess.stderr.text()]); expect(exitCode).toBeInteger(); - expect(stderr).toContain("error: Oops"); + // No body byte reached the wire: error() replaced the response, so the + // original X-Hey header is not sent and the failure is not double-reported. + expect({ result, stderr }).toEqual({ result: { status: 555, xHey: null, body: "E:Oops" }, stderr: "" }); expect(onMessage).toHaveBeenCalledTimes(1); }); diff --git a/test/js/bun/http/async-iterator-throws.fixture.js b/test/js/bun/http/async-iterator-throws.fixture.js index 2227f9dccde9..5967f5036398 100644 --- a/test/js/bun/http/async-iterator-throws.fixture.js +++ b/test/js/bun/http/async-iterator-throws.fixture.js @@ -1,6 +1,9 @@ const server = Bun.serve({ port: 0, idleTimeout: 0, + error(e) { + return new Response("E:" + e.message, { status: 555 }); + }, async fetch(req) { return new Response( diff --git a/test/js/bun/http/serve-body-error-before-first-byte-fixture.ts b/test/js/bun/http/serve-body-error-before-first-byte-fixture.ts new file mode 100644 index 000000000000..eec4b7308c77 --- /dev/null +++ b/test/js/bun/http/serve-body-error-before-first-byte-fixture.ts @@ -0,0 +1,168 @@ +// argv[2]: route name (see below) +// argv[3]: "no-error-handler" to omit error(), "async-error-handler" for an +// async error(), "async-fetch" for an async fetch() that awaits a timer +// before returning (so the handler promise goes Pending past the initial +// drain and on_resolve runs with is_async()=true) +// +// Prints one JSON object on stdout. Raw keep-alive socket so the exact wire +// framing (status line, body bytes, presence of the terminating chunk) can be +// asserted, including the truncated-chunked cases fetch() would mask. +import net from "node:net"; + +const routeName = process.argv[2]; +const errorMode = process.argv[3]; + +let unhandled = 0; +process.on("unhandledRejection", () => unhandled++); + +let errorCalls: string[] = []; +const errorHandler = + errorMode === "no-error-handler" + ? {} + : errorMode === "async-error-handler" + ? { + async error(e: Error) { + errorCalls.push(e?.message ?? String(e)); + await Bun.sleep(1); + return new Response("E:" + e?.message, { status: 500, headers: { "x-err": "1" } }); + }, + } + : { + error(e: Error) { + errorCalls.push(e?.message ?? String(e)); + return new Response("E:" + e?.message, { status: 500, headers: { "x-err": "1" } }); + }, + }; +function route(p: string): Response { + // async iterable: throws before the first yield + if (p === "/iter-throw-first") { + async function* g() { + throw new Error("boom-first"); + } + return new Response(g() as any, { status: 201, headers: { "x-orig": "1" } }); + } + // async iterable: awaits a timer, then throws before the first yield + if (p === "/iter-throw-first-slow") { + async function* g() { + await Bun.sleep(5); + throw new Error("boom-slow"); + } + return new Response(g() as any, { status: 201, headers: { "x-orig": "1" } }); + } + // async iterable: yields synchronously, then throws + if (p === "/iter-yield-then-throw") { + async function* g() { + yield "AAAA"; + yield "BBBB"; + throw new Error("boom-fast"); + } + return new Response(g() as any); + } + // async iterable: yields with awaited gaps, then throws + if (p === "/iter-yield-slow-then-throw") { + let resolve!: () => void; + const gate = new Promise(r => (resolve = r)); + (globalThis as any).__gate = resolve; + async function* g() { + yield "AAAA"; + await gate; + throw new Error("boom-mid"); + } + return new Response(g() as any); + } + // default ReadableStream: pull rejects before any enqueue + if (p === "/rs-pull-throw") { + return new Response( + new ReadableStream({ + async pull() { + throw new Error("rs-boom"); + }, + }), + ); + } + // async iterable: successful empty body (custom status/headers must survive) + if (p === "/iter-empty-ok") { + async function* g() {} + return new Response(g() as any, { status: 202, headers: { "x-custom": "yes" } }); + } + // ReadableStream: close() in start, no chunks (user status/headers must survive) + if (p === "/rs-empty-ok") { + return new Response(new ReadableStream({ start: c => c.close() }), { status: 202, headers: { "x-custom": "yes" } }); + } + // direct stream: async pull resolves without write/end (handle_resolve_stream + // reaches finalize()'s !done branch with the user status still unwritten) + if (p === "/direct-empty-ok") { + return new Response( + new ReadableStream({ + type: "direct", + async pull() { + await Bun.sleep(1); + }, + } as any), + { status: 202, headers: { "x-custom": "yes" } }, + ); + } + return new Response("unknown: " + p, { status: 404 }); +} + +await using server = Bun.serve({ + port: 0, + development: false, + ...errorHandler, + ...(errorMode === "async-fetch" + ? { + async fetch(req) { + await Bun.sleep(1); + return route(new URL(req.url).pathname); + }, + } + : { + fetch(req) { + return route(new URL(req.url).pathname); + }, + }), +}); + +// Keep-alive request: read exactly one response, then end() so the server +// frees the socket. The truncated-chunk variants force-close the connection, +// which may surface as ECONNRESET; that is an asserted outcome, not a failure. +function rawRequest(path: string): Promise<{ raw: string; resetAfterBytes: boolean }> { + return new Promise(resolve => { + let buf = ""; + let reset = false; + const sock = net.connect(server.port, "127.0.0.1", () => { + sock.write(`GET ${path} HTTP/1.1\r\nHost: x\r\n\r\n`); + }); + sock.on("data", d => { + buf += d.toString("latin1"); + if (buf.includes("AAAA")) (globalThis as any).__gate?.(); + // Once a complete Content-Length body or the terminating chunk has + // arrived, end() releases the keep-alive connection. + if (buf.endsWith("0\r\n\r\n")) sock.end(); + const clen = buf.match(/Content-Length: (\d+)\r\n/i); + const sep = buf.indexOf("\r\n\r\n"); + if (clen && sep >= 0 && buf.length - sep - 4 >= Number(clen[1])) sock.end(); + }); + sock.on("error", () => (reset = true)); + sock.on("close", () => resolve({ raw: buf, resetAfterBytes: reset })); + }); +} + +const { raw, resetAfterBytes } = await rawRequest("/" + routeName); + +for (let i = 0; i < 10; i++) await Bun.sleep(0); + +const sep = raw.indexOf("\r\n\r\n"); +const headers = sep >= 0 ? raw.slice(0, sep) : raw; +console.log( + JSON.stringify({ + statusLine: headers.split("\r\n")[0] ?? "", + xErr: /\bx-err: 1\b/i.test(headers), + xOrig: /\bx-orig: 1\b/i.test(headers), + xCustom: /\bx-custom: yes\b/i.test(headers), + body: sep >= 0 ? raw.slice(sep + 4) : "", + resetAfterBytes, + errorCalls, + unhandled, + }), +); diff --git a/test/js/bun/http/serve-direct-readable-stream.test.ts b/test/js/bun/http/serve-direct-readable-stream.test.ts index 13df3ec832b5..c61bf6fe9695 100644 --- a/test/js/bun/http/serve-direct-readable-stream.test.ts +++ b/test/js/bun/http/serve-direct-readable-stream.test.ts @@ -367,14 +367,11 @@ test.skipIf(!isASAN)( 60_000, ); -// A direct stream's pull() that throws synchronously reaches handle_reject -// AFTER do_render_stream already wrote the 200 status+headers. handle_reject -// gated only on has_responded() (response ended), not has_written_status(), -// so the server's error() handler was asked to produce a second Response and -// render_metadata wrote its status/headers into the in-flight body. Debug -// builds hit the !has_written_status assert in do_write_status and aborted; -// release builds spliced the error() header block into the chunked body. -describe("sync pull() throw after status is written does not re-render error()", () => { +// A direct stream's pull() throws synchronously. The boundary is whether a +// body byte has reached uWS by then: if one has, the status line is committed +// and error() must NOT be re-rendered (its header block would be spliced into +// the chunked body); if none has, error() can replace the response entirely. +describe("sync pull() throw: error() boundary is the first body byte", () => { function fixture(pullBody: string) { return ` const net = require("node:net"); @@ -428,7 +425,7 @@ describe("sync pull() throw after status is written does not re-render error()", expect(exitCode).toBe(0); }); - test("no body bytes flushed: stream is ended without splicing error() headers", async () => { + test("no body bytes flushed: error() replaces the response", async () => { await using proc = Bun.spawn({ cmd: [bunExe(), "-e", fixture(`throw new Error("boom");`)], env: bunEnv, @@ -436,15 +433,24 @@ describe("sync pull() throw after status is written does not re-render error()", stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toContain("error: boom"); const { wire, errorHandlerCalls } = JSON.parse(stdout); - // Status 200 was already written; the stream is ended empty. The error() - // response's status (500), headers, and body must not appear on the wire. - expect(wire).not.toContain("x-err"); - expect(wire).not.toContain("FROM-ERROR-HANDLER"); - expect(wire.startsWith("HTTP/1.1 200 OK\r\n")).toBe(true); - expect(errorHandlerCalls).toBe(0); - expect(exitCode).toBe(0); + // pull() threw before any body byte reached uWS, so no status line is + // committed yet and error() can replace the response in full. + expect({ + wire: wire.split("\r\n")[0], + hasXErr: wire.includes("x-err: 1"), + body: wire.split("\r\n\r\n")[1], + errorHandlerCalls, + stderr, + exitCode, + }).toEqual({ + wire: "HTTP/1.1 500 Internal Server Error", + hasXErr: true, + body: "FROM-ERROR-HANDLER", + errorHandlerCalls: 1, + stderr: "", + exitCode: 0, + }); }); }); diff --git a/test/js/bun/http/serve-stream-body-error.test.ts b/test/js/bun/http/serve-stream-body-error.test.ts index c0e5635ea7bd..1395ae420f7e 100644 --- a/test/js/bun/http/serve-stream-body-error.test.ts +++ b/test/js/bun/http/serve-stream-body-error.test.ts @@ -8,7 +8,7 @@ // 3. A body that errored after chunks were already sent was still terminated // with a clean `0\r\n\r\n`, so the client could not tell the truncated // body apart from a complete one. -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, isASAN } from "harness"; import { join } from "node:path"; @@ -25,66 +25,57 @@ async function runFixture(variant: string, ...extra: string[]) { return { stdout, stderr, exitCode }; } -// The wire-level contract for a stream source that errors before producing any -// body bytes is unchanged: the Response's own status and headers go out with -// an empty chunked body and the server `error()` callback is NOT invoked (see -// "throw on pull renders headers, does not call error handler" in -// serve.test.ts). What changes is that the rejection no longer reaches the -// unhandledRejection reporter; it is reported directly instead. +// A stream source that errors before producing any body bytes is routed to +// `error()`: no status line has been committed yet, so the handler's Response +// can replace the original one in full. The rejection must still not reach the +// unhandledRejection reporter. test.concurrent.each([ "pull-throw", "pull-async-reject", "controller-error", "start-async-reject", "deferred-pull-throw", -])("%s: the rejection is handled and the error is still reported", async variant => { +])("%s: error() is invoked and its Response is sent", async variant => { const { stdout, stderr, exitCode } = await runFixture(variant); - expect({ result: JSON.parse(stdout), exitCode }).toEqual({ + expect({ result: JSON.parse(stdout), stderr, exitCode }).toEqual({ result: { - statusLine: "HTTP/1.1 200 OK", - cleanChunkedTerminator: true, - body: "0\r\n\r\n", - errorCb: 0, + statusLine: "HTTP/1.1 500 Internal Server Error", + cleanChunkedTerminator: false, + body: "err-body", + errorCb: 2, unhandled: 0, - secondStatusLine: "HTTP/1.1 200 OK", + secondStatusLine: "HTTP/1.1 500 Internal Server Error", }, + stderr: "", exitCode: 0, }); - // With `development: false` the unhandledRejection report used to be the - // only place the error surfaced; it must still reach stderr without it. - expect(stderr).toContain("boom"); }); // Same under `development: true` (the DEBUG RequestContext monomorphization). -test.concurrent("pull-throw in development mode: the rejection is handled", async () => { +test.concurrent("pull-throw in development mode: error() is invoked", async () => { const { stdout, stderr, exitCode } = await runFixture("pull-throw", "development"); - expect({ result: JSON.parse(stdout), exitCode }).toEqual({ + expect({ result: JSON.parse(stdout), stderr, exitCode }).toEqual({ result: { - statusLine: "HTTP/1.1 200 OK", - cleanChunkedTerminator: true, - body: "0\r\n\r\n", - errorCb: 0, + statusLine: "HTTP/1.1 500 Internal Server Error", + cleanChunkedTerminator: false, + body: "err-body", + errorCb: 2, unhandled: 0, - secondStatusLine: "HTTP/1.1 200 OK", + secondStatusLine: "HTTP/1.1 500 Internal Server Error", }, + stderr: "", exitCode: 0, }); - expect(stderr).toContain("boom"); }); // The body errors after a chunk has already been flushed to the client. The // 200 is irrevocable at that point, but the connection must be closed without // the terminating `0\r\n\r\n` chunk (RFC 9112 section 7) so the client can -// tell the body is incomplete. -// -// No stderr assertion: a rejection on this path already had a reaction -// attached (it never became an unhandledRejection), so there is no lost -// report for this change to restore. `development: false` intentionally -// keeps handle_reject_stream quiet here, which the existing -// serve-direct-readable-stream.test.ts and serve-stream-reject-flush-leak -// tests rely on. The development-mode variant below asserts the report. +// tell the body is incomplete. `error()` is not invoked (a second status line +// cannot be sent) but the failure is reported to stderr. test.concurrent("mid-stream error: the chunked body is not terminated as complete", async () => { - const { stdout, exitCode } = await runFixture("mid-stream-reject"); + const { stdout, stderr, exitCode } = await runFixture("mid-stream-reject"); + expect(stderr).toContain("boom"); expect({ result: JSON.parse(stdout), exitCode }).toEqual({ result: { statusLine: "HTTP/1.1 200 OK", @@ -209,7 +200,9 @@ test.concurrent("a throwing cancel() on client abort does not kill the server pr // The whole point: with Bun's default unhandledRejection policy (no handler // installed), a single request whose Response body errors must not exit the -// server process. +// server process. With no error() handler the failure is reported like a +// fetch() throw (500 + stderr report + process.exitCode set), but the server +// keeps serving. test.concurrent("a stream body error does not kill the server process", async () => { await using proc = Bun.spawn({ cmd: [ @@ -218,7 +211,8 @@ test.concurrent("a stream body error does not kill the server process", async () `const server = Bun.serve({ port: 0, development: false, - fetch() { + fetch(req) { + if (new URL(req.url).pathname === "/ok") return new Response("ok"); return new Response(new ReadableStream({ pull() { throw new Error("boom"); } })); }, }); @@ -227,7 +221,8 @@ test.concurrent("a stream body error does not kill the server process", async () // Tick the event loop past the unhandledRejection checkpoint that used // to exit the process before this line was reached. for (let i = 0; i < 10; i++) await Bun.sleep(0); - console.log("alive", res.status); + const ok = await fetch(new URL("/ok", server.url)); + console.log("alive", res.status, ok.status, await ok.text()); server.stop(true);`, ], env: bunEnv, @@ -235,8 +230,9 @@ test.concurrent("a stream body error does not kill the server process", async () stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect({ stdout, exitCode }).toEqual({ stdout: "alive 200\n", exitCode: 0 }); - // The error must still be surfaced to the operator. + // exitCode is 1: the default error reporter sets it, exactly as it does when + // fetch() itself throws with no error() handler. + expect({ stdout, exitCode }).toEqual({ stdout: "alive 500 200 ok\n", exitCode: 1 }); expect(stderr).toContain("boom"); }); @@ -313,13 +309,186 @@ test.skipIf(!isASAN)( const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); // The stream errors after "EB" is on the wire, so the body is force-closed - // without the terminating 0\r\n\r\n chunk (RFC 9112 section 7). + // without the terminating 0\r\n\r\n chunk (RFC 9112 section 7). The + // nested failure is reported to stderr (error() already ran for the outer + // rejection and is not re-invoked). const expected = Array(6).fill({ status: "HTTP/1.1 597 HM", terminated: false }); - expect({ stderr, results: stdout.trim() ? JSON.parse(stdout) : stdout, exitCode }).toEqual({ - stderr: "", + expect({ results: stdout.trim() ? JSON.parse(stdout) : stdout, exitCode }).toEqual({ results: expected, exitCode: 0, }); + expect(stderr).toContain("nested"); }, 30_000, ); + +// A Response body driven by an async iterable (async generator or +// `[Symbol.asyncIterator]` object). Before the fix: +// * throw before the first yield -> a complete `200 OK` with an empty +// chunked body (cacheable as a successful response); +// * synchronous yields then throw -> connection reset with zero bytes +// (the already-yielded chunks discarded along with the status line); +// * awaited yields then throw -> chunked body truncated (the only case a +// client could detect); +// and error() was never invoked in any of them. +describe("Response body errors reach error() until the first body byte is written", () => { + const fixture = join(import.meta.dir, "serve-body-error-before-first-byte-fixture.ts"); + + async function run(route: string, ...extra: string[]) { + await using proc = Bun.spawn({ + cmd: [bunExe(), fixture, route, ...extra], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { result: JSON.parse(stdout), stderr, exitCode }; + } + + test.concurrent.each([ + ["iter-throw-first", "boom-first"], + ["iter-throw-first-slow", "boom-slow"], + ["rs-pull-throw", "rs-boom"], + ])("%s: error() is invoked and the original headers are not sent", async (route, message) => { + // No body byte reached uWS: error() replaces the response entirely. + expect(await run(route)).toEqual({ + result: { + statusLine: "HTTP/1.1 500 Internal Server Error", + xErr: true, + xOrig: false, + xCustom: false, + body: "E:" + message, + resetAfterBytes: false, + errorCalls: [message], + unhandled: 0, + }, + stderr: "", + exitCode: 0, + }); + }); + + // An async fetch() that goes Pending past the initial drain reaches + // do_render_stream with is_async()=true, so its own drain_microtasks is a + // no-op: small yields stay in the sink buffer, and on_reject_stream can + // fire in the same microtask drain as the yields, before the deferred + // auto-flusher runs. Those bytes never reached the client; error() must + // still replace the response. + test.concurrent.each([ + ["iter-throw-first", "boom-first"], + ["iter-yield-then-throw", "boom-fast"], + ])("async fetch %s: error() is invoked", async (r, message) => { + expect(await run(r, "async-fetch")).toEqual({ + result: { + statusLine: "HTTP/1.1 500 Internal Server Error", + xErr: true, + xOrig: false, + xCustom: false, + body: "E:" + message, + resetAfterBytes: false, + errorCalls: [message], + unhandled: 0, + }, + stderr: "", + exitCode: 0, + }); + }); + + // An async error() handler goes through process_on_error_promise; the + // original (protected) fetch() Response must be released first or it leaks + // one GC root per request (and trips handle_resolve's debug_assert). + test.concurrent("iter-throw-first with an async error(): the handler's Response is sent", async () => { + expect(await run("iter-throw-first", "async-error-handler")).toEqual({ + result: { + statusLine: "HTTP/1.1 500 Internal Server Error", + xErr: true, + xOrig: false, + xCustom: false, + body: "E:boom-first", + resetAfterBytes: false, + errorCalls: ["boom-first"], + unhandled: 0, + }, + stderr: "", + exitCode: 0, + }); + }); + + test.concurrent("iter-throw-first without error(): the default 500 is sent and the failure is reported", async () => { + const { result, stderr, exitCode } = await run("iter-throw-first", "no-error-handler"); + expect({ result, exitCode }).toEqual({ + result: { + statusLine: "HTTP/1.1 500 Internal Server Error", + xErr: false, + xOrig: false, + xCustom: false, + body: "Something went wrong!", + resetAfterBytes: false, + errorCalls: [], + unhandled: 0, + }, + exitCode: 1, + }); + expect(stderr).toContain("boom-first"); + }); + + // Yields then throws: by the time the error is observed the status line + // and the yielded chunks have been written to uWS, so error() cannot + // replace the response. The chunked body must be left unterminated; the + // failure is reported to stderr. + // + // iter-yield-then-throw reaches the force-close inside do_render_stream's + // cork; the uncork lets those bytes drain on POSIX, but Windows' + // SO_LINGER{1,0} reset may discard them, so only the invariants that hold + // on every platform are asserted there. + test.concurrent.each([ + ["iter-yield-then-throw", ["", "8\r\nAAAABBBB\r\n"], "boom-fast"], + ["iter-yield-slow-then-throw", ["4\r\nAAAA\r\n"], "boom-mid"], + ] as const)("%s: error() is not invoked and the body is not terminated", async (route, bodies, message) => { + const { result, stderr, exitCode } = await run(route); + expect({ + errorCalls: result.errorCalls, + unhandled: result.unhandled, + xErr: result.xErr, + terminated: result.body.endsWith("0\r\n\r\n"), + bodyIsExpected: bodies.includes(result.body), + body: result.body, + exitCode, + }).toEqual({ + errorCalls: [], + unhandled: 0, + xErr: false, + terminated: false, + bodyIsExpected: true, + body: result.body, + exitCode: 0, + }); + expect(stderr).toContain(message); + }); + + // Deferring the status write must not lose a successful empty body's own + // status/headers: on_first_write fires from the sink's empty-end paths + // (end_from_js and finalize()'s !done branch). + test.concurrent.each(["iter-empty-ok", "rs-empty-ok", "direct-empty-ok"])( + "%s: an empty body keeps the Response's own status and headers", + async route => { + const { result, stderr, exitCode } = await run(route); + expect({ + statusLine: result.statusLine, + xCustom: result.xCustom, + xErr: result.xErr, + errorCalls: result.errorCalls, + unhandled: result.unhandled, + stderr, + exitCode, + }).toEqual({ + statusLine: "HTTP/1.1 202 Accepted", + xCustom: true, + xErr: false, + errorCalls: [], + unhandled: 0, + stderr: "", + exitCode: 0, + }); + }, + ); +}); diff --git a/test/js/bun/http/serve-stream-reject-flush-leak.test.ts b/test/js/bun/http/serve-stream-reject-flush-leak.test.ts index 3d1a615b44d1..b918c67848d4 100644 --- a/test/js/bun/http/serve-stream-reject-flush-leak.test.ts +++ b/test/js/bun/http/serve-stream-reject-flush-leak.test.ts @@ -23,7 +23,10 @@ test.skipIf(isWindows)( const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); + // The fixture's stream rejects after a body chunk reached uWS, so the + // failure is reported to stderr; only unexpected output is a failure. + const unexpected = stderr.split("\n").filter(l => l && !l.includes("boom") && !l.match(/^(\s|at |[0-9]+ \|)/)); + expect(unexpected).toEqual([]); const result = JSON.parse(stdout.trim()); // Sanity: we actually hit the backpressure → pending_flush path. expect(result.flushPending).toBeGreaterThanOrEqual(result.iterations / 2); diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index 6aa45fb3ac49..f944201dd503 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -616,12 +616,12 @@ it.each([ describe("streaming", () => { describe("error handler", () => { - it("throw on pull renders headers, does not call error handler", async () => { + it("throw on pull before the first body byte calls the error handler", async () => { const onMessage = mock(async url => { const response = await fetch(url); - expect(response.status).toBe(402); - expect(response.headers.get("X-Hey")).toBe("123"); - expect(response.text()).resolves.toBe(""); + expect(response.status).toBe(555); + expect(response.headers.get("X-Hey")).toBe(null); + expect(await response.text()).toBe("Failed"); subprocess.kill(); }); @@ -636,17 +636,21 @@ describe("streaming", () => { let [exitCode, stderr] = await Promise.all([subprocess.exited, subprocess.stderr.text()]); expect(exitCode).toBeInteger(); - expect(stderr).toContain("error: Oops"); + expect(stderr).toBe(""); expect(onMessage).toHaveBeenCalled(); }); - it("throw on pull after writing should not call the error handler", async () => { + it("throw on pull after close() calls the error handler", async () => { const onMessage = mock(async href => { const url = new URL("write", href); const response = await fetch(url); - expect(response.status).toBe(402); - expect(response.headers.get("X-Hey")).toBe("123"); - expect(response.text()).resolves.toBe(""); + // pull() enqueues, closes, then throws: the throw errors the stream, + // and the queued chunks are discarded when the server's reader + // observes the errored state. No body byte reaches the wire, so + // error() can still replace the response. + expect(response.status).toBe(555); + expect(response.headers.get("X-Hey")).toBe(null); + expect(await response.text()).toBe("Failed"); subprocess.kill(); }); @@ -661,7 +665,7 @@ describe("streaming", () => { let [exitCode, stderr] = await Promise.all([subprocess.exited, subprocess.stderr.text()]); expect(exitCode).toBeInteger(); - expect(stderr).toContain("error: Oops"); + expect(stderr).toBe(""); expect(onMessage).toHaveBeenCalled(); });