Bun.serve: route Response body stream errors to error() until the first byte is written

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::<VirtualMachine>((*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::<VirtualMachine>((*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<NonNull<c_void>>` 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::<VirtualMachine>((*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::<VirtualMachine>(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<u8> = 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;

src/runtime/webcore/streams.rs

@@ -1773,6 +1773,10 @@ impl<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
                 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<const SSL: bool, const HTTP3: bool> HTTPServerWritable<SSL, HTTP3> {
             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);

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);
   });
 

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(