diff --git a/packages/bun-usockets/src/quic.c b/packages/bun-usockets/src/quic.c index d74117b4bb75..ffc7c07eab3a 100644 --- a/packages/bun-usockets/src/quic.c +++ b/packages/bun-usockets/src/quic.c @@ -1244,7 +1244,7 @@ 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 @@ -1252,13 +1252,21 @@ void lsquic_stream_maybe_reset(struct lsquic_stream *, uint64_t error_code, int) * 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; } diff --git a/packages/bun-usockets/src/quic.h b/packages/bun-usockets/src/quic.h index d64c230b6941..354c1f2fc685 100644 --- a/packages/bun-usockets/src/quic.h +++ b/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); diff --git a/src/http/h3_client/ClientSession.rs b/src/http/h3_client/ClientSession.rs index b3e013f20cb0..ee191bfe5b91 100644 --- a/src/http/h3_client/ClientSession.rs +++ b/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) { + 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 diff --git a/src/http/h3_client/callbacks.rs b/src/http/h3_client/callbacks.rs index 08496f1bfb32..648dfa0b1a9c 100644 --- a/src/http/h3_client/callbacks.rs +++ b/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::() = 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); } diff --git a/src/uws_sys/h3.rs b/src/uws_sys/h3.rs index caabd299d636..cde8c73abcee 100644 --- a/src/uws_sys/h3.rs +++ b/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(&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, diff --git a/src/uws_sys/libuwsockets_h3.cpp b/src/uws_sys/libuwsockets_h3.cpp index fa06529014b0..f1f87b3a563e 100644 --- a/src/uws_sys/libuwsockets_h3.cpp +++ b/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) diff --git a/src/uws_sys/quic.rs b/src/uws_sys/quic.rs index 53adc9c8ae67..1f58f16652ee 100644 --- a/src/uws_sys/quic.rs +++ b/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; diff --git a/src/uws_sys/quic/Stream.rs b/src/uws_sys/quic/Stream.rs index ef565b63ac60..640db21a7b5d 100644 --- a/src/uws_sys/quic/Stream.rs +++ b/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 { + let mut code = 0u64; + (us_quic_stream_peer_reset(self, &mut code) != 0).then_some(code) } pub fn header_count(&mut self) -> c_uint { diff --git a/test/js/bun/http/serve-http3.test.ts b/test/js/bun/http/serve-http3.test.ts index 9869238fc1f6..c626e81b721e 100644 --- a/test/js/bun/http/serve-http3.test.ts +++ b/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) { + 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) { + 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 diff --git a/test/js/web/fetch/fetch-http3-client.test.ts b/test/js/web/fetch/fetch-http3-client.test.ts index 936fe7ee3eab..71a12b1d4bb8 100644 --- a/test/js/web/fetch/fetch-http3-client.test.ts +++ b/test/js/web/fetch/fetch-http3-client.test.ts @@ -4,7 +4,7 @@ import { afterAll, beforeAll, describe, expect, test } from "bun:test"; import { bunEnv, bunExe, isLinux, tempDir, tls } from "harness"; import { createPrivateKey } from "node:crypto"; import { join } from "node:path"; -import { listen } from "node:quic"; +import { listen, QuicError } from "node:quic"; // In-process server with `http1: false` so the build under test binds UDP only. // A fetch that silently fell back to HTTP/1.1 would get ECONNREFUSED, which @@ -941,6 +941,135 @@ test("retries on a fresh session when a pooled session is stale (port reuse)", a } }); +// In HTTP/3 only a FIN completes a message. A response stream that goes away +// after the headers without one (the server sent RESET_STREAM, or the +// connection closed) carries a truncated body, and the body read must fail +// the way HTTP2StreamReset does for h2 instead of resolving with the bytes +// received so far. +describe("response stream ends without FIN", () => { + const listenOrigin = (onheaders: (this: any, headers: Record) => void) => + listen( + async (session: any) => { + session.onstream = (stream: any) => stream.closed.catch(() => {}); + await session.closed.catch(() => {}); + }, + { + sni: { "*": { keys: [createPrivateKey(tls.key)], certs: [Buffer.from(tls.cert)] } }, + transportParams: { maxIdleTimeout: 5 }, + onheaders, + }, + ); + + // A node:quic HTTP/3 server whose response is "200" and the body "partial". + // It runs `end(stream, session)` only once the client, which by then holds + // the status and that chunk, requests /release. No timing assumption orders + // the wire. + async function fetchFromServerThat(end: (stream: any, session: any) => void) { + const { promise: released, resolve: release } = Promise.withResolvers(); + const origin = await listenOrigin(async function (headers) { + this.sendHeaders({ ":status": "200" }); + if (headers[":path"] === "/release") { + this.writer.endSync(); + release(); + return; + } + this.writer.writeSync(new TextEncoder().encode("partial")); + await released; + end(this, this.session); + }); + try { + const base = `https://127.0.0.1:${origin.address.port}`; + const res = await fetch(`${base}/`, h3); + const reader = res.body!.getReader(); + const first = await reader.read(); + // Its response can be lost to the CONNECTION_CLOSE case, so it is not + // awaited: the first stream's read is what reports the outcome. + fetch(`${base}/release`, h3) + .then(r => r.text()) + .catch(() => {}); + const rest = await reader.read().then( + ({ done }) => ({ resolved: done ? "done" : "chunk" }), + (e: Error & { code?: string }) => ({ rejected: e.code }), + ); + return { status: res.status, first: new TextDecoder().decode(first.value), rest }; + } finally { + // Not close(): it waits for the session that fetch() keeps in its pool. + await origin.destroy(); + } + } + + test("RESET_STREAM mid-body rejects the body read with HTTP3StreamReset", async () => { + const result = await fetchFromServerThat(stream => + stream.destroy(new QuicError("body failed", { errorCode: 0x102 })), + ); + expect(result).toEqual({ status: 200, first: "partial", rest: { rejected: "HTTP3StreamReset" } }); + }); + + test("CONNECTION_CLOSE mid-body rejects the body read with ECONNRESET", async () => { + const result = await fetchFromServerThat((_stream, session) => session.close({ code: 0x100, type: "application" })); + expect(result).toEqual({ status: 200, first: "partial", rest: { rejected: "ECONNRESET" } }); + }); + + // RESET_STREAM before the response headers. Only H3_REQUEST_REJECTED (0x10B) + // promises that the server did no application processing (RFC 9114 section + // 8.1), so only that code may re-send the request, as h2 retries only + // REFUSED_STREAM. Any other code fails the request without a second copy. + async function postToServerThatResetsWith(reset: (stream: any) => void, body: string | Uint8Array = "payload") { + const requests: string[] = []; + const origin = await listenOrigin(function (headers) { + requests.push(`${headers[":method"]} ${headers[":path"]}`); + if (requests.length === 1) { + reset(this); + return; + } + this.sendHeaders({ ":status": "200" }); + this.writer.writeSync(new TextEncoder().encode("second attempt")); + this.writer.endSync(); + }); + try { + const result = await fetch(`https://127.0.0.1:${origin.address.port}/submit`, { + ...h3, + method: "POST", + body, + }).then( + async res => ({ resolved: `${res.status} ${await res.text()}` }), + (e: Error & { code?: string }) => ({ rejected: e.code }), + ); + return { result, requests }; + } finally { + await origin.destroy(); + } + } + + // destroy() sends STOP_SENDING next to RESET_STREAM, both with the code. + const destroyWith = (errorCode: number) => (stream: any) => stream.destroy(new QuicError("rejected", { errorCode })); + + test("RESET_STREAM(H3_REQUEST_REJECTED) before the headers re-sends the request once", async () => { + expect(await postToServerThatResetsWith(destroyWith(0x10b))).toEqual({ + result: { resolved: "200 second attempt" }, + requests: ["POST /submit", "POST /submit"], + }); + }); + + // RESET_STREAM alone, while the client still sends the body: its own send + // half is open and no STOP_SENDING repeats the code. The code the client + // reads at close must still be the server's. + test("RESET_STREAM(H3_REQUEST_REJECTED) alone during the body upload re-sends the request once", async () => { + const resetOnly = (stream: any) => stream.resetStream(0x10b); + expect(await postToServerThatResetsWith(resetOnly, Buffer.alloc(8 * 1024 * 1024, "a"))).toEqual({ + result: { resolved: "200 second attempt" }, + requests: ["POST /submit", "POST /submit"], + }); + }); + + test("RESET_STREAM(H3_INTERNAL_ERROR) before the headers fails without a second copy of the request", async () => { + expect(await postToServerThatResetsWith(destroyWith(0x102))).toEqual({ + result: { rejected: "HTTP3StreamReset" }, + requests: ["POST /submit"], + }); + }); +}); + // Same retry, but the reconnect fails inside `connect()` itself. A QUIC // connect to a resolved hostname probes each address with a throwaway UDP // `connect(2)` (quic.c, us_quic_connect_result) and gives up when no entry is