Bun.serve(http3): reset the stream when a response body fails mid-body

packages/bun-usockets/src/quic.c

@@ -1244,21 +1244,29 @@ void us_quic_stream_close(us_quic_stream_t *s) {
 /* From lsquic_stream.h (not in the public header). */
 void lsquic_stream_maybe_reset(struct lsquic_stream *, uint64_t error_code, int);
 
-/* Abort the send half with RESET_STREAM(H3_REQUEST_CANCELLED) instead of
+/* Abort the send half with RESET_STREAM(error_code) instead of
  * FIN. lsquic_stream_close/shutdown queue FIN after the buffered tail,
  * which is a protocol error if a content-length was advertised and the
  * client is abandoning the upload short — the server's lsquic will
  * CONNECTION_CLOSE on the mismatch (RFC 9114 §4.1.2). RESET_STREAM is
  * the wire-level "I'm cancelling this send" and lets the server treat it
  * as a stream-level cancellation rather than a malformed message.
  * Sends nothing once lsquic_stream_close/shutdown has run, so call it first. */
-void us_quic_stream_reset(us_quic_stream_t *s) {
+void us_quic_stream_reset(us_quic_stream_t *s, uint64_t error_code) {
     if (!s->stream) return;
     /* do_close=0: with no reset due, maybe_reset's own close shuts only the read half. */
-    lsquic_stream_maybe_reset(s->stream, 0x10C, 0);
+    lsquic_stream_maybe_reset(s->stream, error_code, 0);
     lsquic_stream_close(s->stream);
 }
 
+/* Reads lsquic's own state, so the stream carries no copy. A STOP_SENDING
+ * that arrives after the RESET_STREAM replaces the code. */
+int us_quic_stream_peer_reset(us_quic_stream_t *s, uint64_t *code) {
+    if (!s->stream || !lsquic_stream_reset_received(s->stream)) return 0;
+    *code = lsquic_stream_get_error_code(s->stream);
+    return 1;
+}
+
 int us_quic_stream_has_unacked(us_quic_stream_t *s) {
     return s->stream ? lsquic_stream_has_unacked_data(s->stream) : 0;
 }

packages/bun-usockets/src/quic.h

@@ -143,7 +143,14 @@ void us_quic_stream_shutdown(us_quic_stream_t *s);
 void us_quic_stream_flush(us_quic_stream_t *s);
 void us_quic_stream_shutdown_read(us_quic_stream_t *s);
 void us_quic_stream_close(us_quic_stream_t *s);
-void us_quic_stream_reset(us_quic_stream_t *s);
+
+/* RESET_STREAM(error_code) instead of FIN: in HTTP/3 a FIN marks the bytes
+ * sent so far as a complete message. */
+void us_quic_stream_reset(us_quic_stream_t *s, uint64_t error_code);
+/* Non-zero, with the code in *code, once the peer has sent RESET_STREAM for
+ * the half we read from. Answers only while the stream is live, up to and
+ * including on_stream_close. */
+int us_quic_stream_peer_reset(us_quic_stream_t *s, uint64_t *code);
 int us_quic_stream_has_unacked(us_quic_stream_t *s);
 
 void *us_quic_stream_ext(us_quic_stream_t *s);

src/http/h3_client/ClientSession.rs

@@ -226,7 +226,7 @@ impl ClientSession {
             // is the correct "I'm abandoning this send half" so lsquic reaps
             // the stream instead of leaking it on the pooled session.
             if abort || !request_body_done {
-                qs.reset();
+                qs.reset(quic::H3ErrorCode::RequestCancelled);
             }
         }
         st.qstream = None;
@@ -324,6 +324,32 @@ impl ClientSession {
         false
     }
 
+    /// The lsquic stream closed without a FIN (a delivered FIN detaches the
+    /// stream inside `deliver`). `peer_reset` is the peer's RESET_STREAM code.
+    /// Before the response headers `deliver` re-sends the request once, but
+    /// only when no reset arrived (stale session) or the code is
+    /// H3_REQUEST_REJECTED, the one code that promises the server did no
+    /// application processing (RFC 9114 section 8.1; h2 retries only
+    /// REFUSED_STREAM). Otherwise the request fails.
+    pub(crate) fn on_stream_closed(&mut self, stream: *mut Stream, peer_reset: Option<u64>) {
+        let st = stream_mut(stream);
+        let Some(client_ptr) = st.client else {
+            return self.detach(stream);
+        };
+        let retryable = peer_reset.is_none_or(|c| c == quic::H3ErrorCode::RequestRejected as u64);
+        if !st.headers_delivered && retryable {
+            return self.deliver(stream, true);
+        }
+        let err = if client_mut(client_ptr).signals.get(Signal::Aborted) {
+            crate::Error::Aborted
+        } else if peer_reset.is_some() {
+            crate::Error::HTTP3StreamReset
+        } else {
+            crate::Error::ConnectionClosed
+        };
+        self.fail(stream, err);
+    }
+
     /// Runs from inside lsquic's process_conns via on_stream_{headers,data,close}.
     /// `done` = the lsquic stream is gone; deliver whatever is buffered then
     /// detach. Mirrors H2's `ClientSession.deliverStream` so the HTTPClient state

src/http/h3_client/callbacks.rs

@@ -308,13 +308,15 @@ extern "C" fn on_stream_writable(s: *mut quic::Stream) {
 extern "C" fn on_stream_close(s: *mut quic::Stream) {
     let s = qstream_arg(s);
     let Some(stream) = stream_of(s) else { return };
+    let peer_reset = s.peer_reset();
     *s.ext::<Stream>() = None;
     stream.qstream = None;
     bun_core::scoped_log!(
         h3_client,
-        "stream_close status={} delivered={}",
+        "stream_close status={} delivered={} peer_reset={:?}",
         stream.status_code,
         stream.headers_delivered,
+        peer_reset,
     );
-    stream.session_mut().deliver(stream, true);
+    stream.session_mut().on_stream_closed(stream, peer_reset);
 }

src/uws_sys/h3.rs

@@ -203,8 +203,10 @@ impl Response {
         let ip = unsafe { bun_core::ffi::slice(ip_ptr, len) };
         Some(SocketAddress::new(ip, port, is_ipv6))
     }
+    /// The body failed after the response started: `RESET_STREAM`, the
+    /// HTTP/3 form of HTTP/1's close without the terminating chunk.
     pub(crate) fn force_close(&mut self) {
-        c::uws_h3_res_force_close(self)
+        c::uws_h3_res_force_close(self, crate::quic::H3ErrorCode::InternalError as u64)
     }
 
     pub(crate) fn on_writable<UD, H>(&mut self, _handler: H, ud: *mut UD)
@@ -691,7 +693,7 @@ mod c {
         pub(super) safe fn uws_h3_res_state(res: &mut Response) -> State;
         pub(super) fn uws_h3_res_end(res: *mut Response, p: *const u8, n: usize, close: bool);
         pub(super) safe fn uws_h3_res_end_stream(res: &mut Response, close: bool);
-        pub(super) safe fn uws_h3_res_force_close(res: &mut Response);
+        pub(super) safe fn uws_h3_res_force_close(res: &mut Response, error_code: u64);
         pub(super) fn uws_h3_res_try_end(
             res: *mut Response,
             p: *const u8,

src/uws_sys/libuwsockets_h3.cpp

@@ -116,10 +116,12 @@ void uws_h3_res_end_stream(uws_h3_res_t* res, bool close_connection)
     r->sendTerminatingChunk(close_connection);
 }
 
-void uws_h3_res_force_close(uws_h3_res_t* res)
+/* RESET_STREAM(error_code), not FIN: in HTTP/3 a FIN would deliver the
+ * bytes sent so far as a complete message. */
+void uws_h3_res_force_close(uws_h3_res_t* res, uint64_t error_code)
 {
     ((Http3Response*)res)->clearOnWritableAndAborted();
-    us_quic_stream_close((us_quic_stream_t*)res);
+    us_quic_stream_reset((us_quic_stream_t*)res, error_code);
 }
 
 bool uws_h3_res_try_end(uws_h3_res_t* res, const char* bytes, size_t len, size_t total_len, bool close)

src/uws_sys/quic.rs

@@ -22,6 +22,7 @@ pub mod stream;
 pub use self::context::Context;
 pub use self::pending_connect::PendingConnect;
 pub use self::socket::Socket;
+pub use self::stream::H3ErrorCode;
 pub use self::stream::Stream;
 
 pub use self::header::Header;

src/uws_sys/quic/Stream.rs

@@ -11,6 +11,16 @@ bun_opaque::opaque_ffi! {
     pub struct Stream;
 }
 
+/// HTTP/3 error codes (RFC 9114 §8.1) carried by `RESET_STREAM`.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[repr(u16)]
+pub enum H3ErrorCode {
+    InternalError = 0x102,
+    /// The server did no application processing; the request can be re-sent.
+    RequestRejected = 0x10B,
+    RequestCancelled = 0x10C,
+}
+
 // `Stream` is an `opaque_ffi!` ZST (`UnsafeCell<[u8; 0]>`), so `&mut Stream` is
 // ABI-identical to a non-null `*mut Stream` with no `noalias`/`readonly`
 // attribute. Shims taking only the handle + value types are `safe fn`; the
@@ -19,7 +29,8 @@ unsafe extern "C" {
     safe fn us_quic_stream_socket(s: &mut Stream) -> *mut Socket;
     safe fn us_quic_stream_shutdown(s: &mut Stream);
     safe fn us_quic_stream_close(s: &mut Stream);
-    safe fn us_quic_stream_reset(s: &mut Stream);
+    safe fn us_quic_stream_reset(s: &mut Stream, error_code: u64);
+    safe fn us_quic_stream_peer_reset(s: &mut Stream, code: &mut u64) -> c_int;
     safe fn us_quic_stream_header_count(s: &mut Stream) -> c_uint;
     safe fn us_quic_stream_header(s: &mut Stream, i: c_uint) -> *const Header;
     safe fn us_quic_stream_ext(s: &mut Stream) -> *mut c_void;
@@ -52,8 +63,17 @@ impl Stream {
         us_quic_stream_close(self)
     }
 
-    pub fn reset(&mut self) {
-        us_quic_stream_reset(self)
+    /// `RESET_STREAM(error_code)` and close. Unlike [`Stream::close`] the peer
+    /// sees no FIN, which in HTTP/3 would mark the message complete.
+    pub fn reset(&mut self, error_code: H3ErrorCode) {
+        us_quic_stream_reset(self, error_code as u64)
+    }
+
+    /// The error code of the `RESET_STREAM` the peer sent for our read half.
+    /// `on_stream_close` then follows with no `fin` from `on_stream_data`.
+    pub fn peer_reset(&mut self) -> Option<u64> {
+        let mut code = 0u64;
+        (us_quic_stream_peer_reset(self, &mut code) != 0).then_some(code)
     }
 
     pub fn header_count(&mut self) -> c_uint {

test/js/bun/http/serve-http3.test.ts

@@ -1891,6 +1891,151 @@ describe("Bun.serve HTTP/3 request validation", () => {
   });
 });
 
+// A Response body stream that fails after the status is committed cannot be
+// replaced by error(). The transport has to tell the client that the message
+// is incomplete: HTTP/1 closes the socket without the last chunk. In HTTP/3 a
+// FIN is a complete message, so the stream has to end with
+// RESET_STREAM(H3_INTERNAL_ERROR) instead.
+describe("Bun.serve HTTP/3 response body error", () => {
+  // The body source of /mid-body-error sends one chunk and then waits. It
+  // errors only once the client, which by then holds the status and that
+  // chunk, requests /release. No timing assumption orders the wire.
+  const script = `
+    const tls = ${JSON.stringify(tls)};
+    const { promise: released, resolve: release } = Promise.withResolvers();
+    let failedBeforeFirstChunk = 0;
+    const server = Bun.serve({
+      port: 0, tls, http3: true, http1: false,
+      routes: {
+        "/ok": () => new Response("ok"),
+        "/release": () => {
+          release();
+          return new Response("released");
+        },
+        "/mid-body-error": () => {
+          let pulls = 0;
+          return new Response(new ReadableStream({
+            async pull(c) {
+              if (pulls++ === 0) {
+                c.enqueue(new Uint8Array(1024).fill(65));
+                return;
+              }
+              await released;
+              c.error(new Error("boom"));
+            },
+          }));
+        },
+        "/error-before-first-chunk": () => {
+          failedBeforeFirstChunk++;
+          return new Response(new ReadableStream({
+            async pull(c) {
+              await Promise.resolve();
+              c.error(new Error("boom"));
+            },
+          }));
+        },
+        "/error-before-first-chunk/count": () => new Response(String(failedBeforeFirstChunk)),
+      },
+    });
+    console.error("PORT=" + server.port);
+    process.stdin.on("data", () => {});
+    ${STOP_ON_STDIN_END}
+  `;
+
+  const release = (port: number) => fetchH3(port, "/release").then(r => r.text());
+
+  // Raw HTTP/3 client: reads the first chunk, asks the server to fail the
+  // body, and reports how the response stream ended on the wire.
+  async function h3ReadBody(port: number, path: string) {
+    await using endpoint = new QuicEndpoint();
+    const client = await connect(`127.0.0.1:${port}`, {
+      endpoint,
+      servername: "localhost",
+      verifyPeer: "manual",
+      transportParams: { maxIdleTimeout: 5 },
+      onerror() {},
+    });
+    await client.opened;
+    let status = "";
+    const stream = await client.createBidirectionalStream({
+      headers: requestHeaders(path),
+      onheaders(received: Record<string, string>) {
+        status = received[":status"];
+      },
+    });
+    // `closed` rejects with the RESET_STREAM error code when the peer resets.
+    const closed = stream.closed.then(
+      () => "fin",
+      (e: Error & { code?: string; errorCode?: bigint }) => ({ code: e.code, errorCode: Number(e.errorCode) }),
+    );
+    let firstChunkReceived = false;
+    // Which error the iterator throws depends on whether the stream is
+    // already closed when it pulls again, so only a clean end is recorded.
+    let bodyEndedCleanly = true;
+    try {
+      for await (const _ of stream as AsyncIterable<Uint8Array[]>) {
+        if (!firstChunkReceived) {
+          firstChunkReceived = true;
+          await release(port);
+        }
+      }
+    } catch {
+      bodyEndedCleanly = false;
+    }
+    const result = { status, firstChunkReceived, bodyEndedCleanly, closed: await closed };
+    client.close().catch(() => {});
+    return result;
+  }
+
+  test("after the first chunk: fetch() sees the status, then the body read rejects", async () => {
+    await withCustomServer(script, async (port, _send, waitForStderr) => {
+      const res = await fetchH3(port, "/mid-body-error");
+      const reader = res.body!.getReader();
+      const first = await reader.read();
+      await release(port);
+      const rest = await reader.read().then(
+        ({ done }) => ({ resolved: done ? "done" : "chunk" }),
+        (e: Error & { code?: string }) => ({ rejected: e.code }),
+      );
+      // The handler's error still reaches the server's error reporting.
+      await waitForStderr(/boom/);
+      // Only the one stream was reset; the connection stays usable.
+      const after = await fetchH3(port, "/ok").then(r => r.text());
+      expect({ status: res.status, firstChunkReceived: !first.done, rest, after }).toEqual({
+        status: 200,
+        firstChunkReceived: true,
+        rest: { rejected: "HTTP3StreamReset" },
+        after: "ok",
+      });
+    });
+  });
+
+  test("after the first chunk: the stream ends with RESET_STREAM(H3_INTERNAL_ERROR), not FIN", async () => {
+    await withCustomServer(script, async port => {
+      const result = await h3ReadBody(port, "/mid-body-error");
+      expect(result).toEqual({
+        status: "200",
+        firstChunkReceived: true,
+        bodyEndedCleanly: false,
+        closed: { code: "ERR_QUIC_APPLICATION_ERROR", errorCode: 258 },
+      });
+    });
+  });
+
+  // The reset says that the handler ran (H3_INTERNAL_ERROR, not
+  // H3_REQUEST_REJECTED), so fetch() must not send the request a second time.
+  test("before the first chunk: fetch() rejects and the handler runs once", async () => {
+    await withCustomServer(script, async port => {
+      const outcome = await fetchH3(port, "/error-before-first-chunk", { method: "POST", body: "payload" }).then(
+        res => ({ resolved: res.status }),
+        (e: Error & { code?: string }) => ({ rejected: e.code }),
+      );
+      const handlerRuns = await fetchH3(port, "/error-before-first-chunk/count").then(r => r.text());
+      expect({ outcome, handlerRuns }).toEqual({ outcome: { rejected: "HTTP3StreamReset" }, handlerRuns: "1" });
+    });
+  });
+});
+
 // RFC 9110 section 10.1.1: a client that sent Expect: 100-continue can hold the
 // request content until it has the 100. The handler answers only when it has
 // all of the content, so the client ends the request only if the 100 arrived