Bun.serve: never terminate a failed body stream as a complete message

src/runtime/server/RequestContext.rs

@@ -1000,24 +1000,15 @@ where
         // SAFETY: FFI handle, just checked Some
         let has_responded = resp.has_responded();
 
-        // The status line is already committed (a direct ReadableStream's
-        // pull() threw synchronously after do_render_stream wrote headers).
-        // error() cannot replace a response whose status is on the wire;
-        // report the failure and terminate the body so the client observes an
-        // incomplete message instead of a second header block spliced into
-        // the chunked body.
+        // The status line is already committed (a direct stream's pull() threw
+        // synchronously after the headers were written): report and close.
         if !has_responded && self.flags.has_written_status() {
             if !value.is_empty_or_undefined_or_null()
                 && let Some(server) = self.server.get()
             {
                 server.vm().as_mut().run_error_handler(value, None);
             }
-            let state = resp.state();
-            if state.is_http_write_called() && state.is_response_pending() {
-                self.force_close();
-            } else {
-                self.end_stream(self.should_close_connection());
-            }
+            self.close_incomplete_stream();
             return;
         }
 
@@ -1290,6 +1281,18 @@ where
         }
     }
 
+    /// Closes a response whose body failed after the status line was committed.
+    /// Never the terminating chunk: it would make a truncated body look complete.
+    pub(crate) fn close_incomplete_stream(&self) {
+        if let Some(resp) = self.resp.get() {
+            if resp.state().is_response_pending() {
+                self.force_close();
+                return;
+            }
+        }
+        self.end_stream(self.should_close_connection());
+    }
+
     fn on_writable_complete_response_buffer(
         this: *mut Self,
         write_offset: u64,
@@ -3001,50 +3004,9 @@ where
             self.render_metadata();
         }
 
-        if DEBUG_MODE {
-            if let Some(server) = self.server.get() {
-                if !err.is_empty_or_undefined_or_null() {
-                    let server = &*server;
-                    let mut exception_list: jsc::ExceptionList = Vec::new();
-                    server
-                        .vm()
-                        .as_mut()
-                        .run_error_handler(err, Some(&mut 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 !self.flags.has_written_status() {
-                            self.flags.set_has_written_status(true);
-                            if let Some(resp) = self.resp.get() {
-                                resp.write_status(b"500 Internal Server Error");
-                                resp.write_header(
-                                    b"content-type",
-                                    &bun_http_types::MimeType::HTML.value,
-                                );
-                            }
-                        }
-
-                        let bb = DevErrorPage {
-                            message: b"Stream error during server-side rendering",
-                            cwd: bun_resolver::fs::FileSystem::get().top_level_dir,
-                            exceptions: &exception_list,
-                            log: None,
-                        }
-                        .render();
-
-                        if let Some(resp) = self.resp.get() {
-                            // SAFETY: FFI handle
-                            resp.write(&bb);
-                        }
-
-                        self.end_stream(self.should_close_connection());
-                        return;
-                    }
-                }
-            }
+        // Production mode keeps this asynchronous JS path quiet.
+        if DEBUG_MODE && self.report_committed_body_error(err, !ended_response) {
+            return;
         }
         // HTTP/1 only: the sink already fully ended the response, so `resp`
         // can no longer be dereferenced (see `end_already_responded_stream`).
@@ -3054,17 +3016,43 @@ where
             self.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.
+        self.close_incomplete_stream();
+    }
+
+    /// Reports a body failure that arrived after the status line was
+    /// committed. Under a bake dev server the error page ends the response:
+    /// returns `true`, and `resp` must not be touched again.
+    fn report_committed_body_error(&self, err: JSValue, resp_writable: bool) -> bool {
+        if err.is_empty_or_undefined_or_null() {
+            return false;
+        }
+        let server = self.server();
+        if !DEBUG_MODE || !resp_writable || server.dev_server().is_none() {
+            server.vm().as_mut().run_error_handler(err, None);
+            return false;
+        }
+
+        let mut exception_list: jsc::ExceptionList = Vec::new();
+        server
+            .vm()
+            .as_mut()
+            .run_error_handler(err, Some(&mut exception_list));
+
+        let bb = DevErrorPage {
+            message: b"Stream error during server-side rendering",
+            cwd: bun_resolver::fs::FileSystem::get().top_level_dir,
+            exceptions: &exception_list,
+            log: None,
+        }
+        .render();
+
         if let Some(resp) = self.resp.get() {
-            let state = resp.state();
-            if state.is_http_write_called() && state.is_response_pending() {
-                self.force_close();
-                return;
-            }
+            // SAFETY: FFI handle
+            resp.write(&bb);
         }
+
         self.end_stream(self.should_close_connection());
+        true
     }
 
     pub(crate) fn do_render_with_body(
@@ -3353,13 +3341,13 @@ where
                 this.run_error_handler(js_err);
                 return;
             }
-            if let Some(resp) = this.resp.get() {
-                let state = resp.state();
-                if state.is_http_write_called() && state.is_response_pending() {
-                    this.force_close();
-                    return;
-                }
+            // Committed status: report in both modes, then close.
+            let global_this = this.server().global_this();
+            if this.report_committed_body_error(err.to_js(global_this), true) {
+                return;
             }
+            this.close_incomplete_stream();
+            return;
         } else if !this.flags.has_written_status() {
             // Upstream ended cleanly before any chunk: flush the deferred
             // status/headers so the client sees them before the terminator.

test/js/bun/http/async-iterator-stream.test.ts

@@ -114,10 +114,17 @@ describe.concurrent("Streaming body via", () => {
     expect(exitCode).toBe(0);
   });
 
-  test("async generator function throws an error but continues to send the headers", async () => {
+  // The generator fails before producing any chunk. The status and headers
+  // were already committed to uWS, so the server closes the connection without
+  // a complete response instead of sending them with an empty body and a clean
+  // chunked terminator.
+  test("async generator function throws an error before its first chunk closes the connection", async () => {
+    let outcome: string | undefined;
     const onMessage = mock(async url => {
-      const response = await fetch(url);
-      expect(response.headers.get("X-Hey")).toBe("123");
+      outcome = await fetch(url).then(
+        response => `resolved ${response.status}`,
+        (err: any) => `rejected ${err.code}`,
+      );
       subprocess?.kill();
     });
 
@@ -132,6 +139,7 @@ describe.concurrent("Streaming body via", () => {
 
     let [exitCode, stderr] = await Promise.all([subprocess.exited, subprocess.stderr.text()]);
     expect(exitCode).toBeInteger();
+    expect(outcome).toBe("rejected ECONNRESET");
     expect(stderr).toContain("error: Oops");
     expect(onMessage).toHaveBeenCalledTimes(1);
   });

test/js/bun/http/readable-stream-throws.fixture.js

@@ -2,6 +2,7 @@ const server = Bun.serve({
   port: 0,
   idleTimeout: 0,
   error(err) {
+    console.error("error handler called");
     return new Response("Failed", { status: 555 });
   },
 

test/js/bun/http/serve-direct-readable-stream.test.ts

@@ -793,7 +793,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: connection is force-closed without splicing error() headers", async () => {
     await using proc = Bun.spawn({
       cmd: [bunExe(), "-e", fixture(`throw new Error("boom");`)],
       env: bunEnv,
@@ -803,11 +803,12 @@ describe("sync pull() throw after status is written does not re-render error()",
     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);
+    // Status 200 was already written to the corked response, so error() cannot
+    // replace it. Ending the stream here would send that 200 with an empty
+    // chunked body and a clean terminator: a complete-looking response for a
+    // body that failed. The connection is closed instead, and since the status
+    // never left the cork buffer the client sees an empty reply.
+    expect(wire).toBe("");
     expect(errorHandlerCalls).toBe(0);
     expect(exitCode).toBe(0);
   });