Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions packages/bun-usockets/src/quic.c
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,8 @@ struct us_quic_stream_s {
int fin_delivered;
/* Set by us_quic_flush_from_on_write, cleared by on_write. */
int flush_on_write;
/* The code of the peer's RESET_STREAM or STOP_SENDING, -1 until it sends one. */
int64_t peer_reset_code;
/* ext follows */
};

Expand Down Expand Up @@ -682,6 +684,7 @@ static lsquic_stream_ctx_t *us_quic_on_new_stream(void *if_ctx, lsquic_stream_t
if (!s) { lsquic_stream_close(stream); return NULL; }
s->stream = stream;
s->ctx = ctx;
s->peer_reset_code = -1;
if (ctx->on_stream_open) ctx->on_stream_open(s, ctx->is_client);
lsquic_stream_wantread(stream, 1);
return (lsquic_stream_ctx_t *) s;
Expand Down Expand Up @@ -751,6 +754,8 @@ static void us_quic_on_close(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
}

static void us_quic_on_reset(lsquic_stream_t *stream, lsquic_stream_ctx_t *h, int how) {
us_quic_stream_t *s = (us_quic_stream_t *) h;
if (s && stream) s->peer_reset_code = (int64_t) lsquic_stream_get_error_code(stream);
/* how=0 → peer sent RESET_STREAM (our read half is gone): nothing left
* to deliver, close so on_stream_close fires.
* how=1 → peer sent STOP_SENDING (wants us to abort our write half):
Expand Down Expand Up @@ -1263,6 +1268,8 @@ int us_quic_stream_has_unacked(us_quic_stream_t *s) {
return s->stream ? lsquic_stream_has_unacked_data(s->stream) : 0;
}

int64_t us_quic_stream_peer_reset_code(us_quic_stream_t *s) { return s->peer_reset_code; }

void *us_quic_stream_ext(us_quic_stream_t *s) { return s + 1; }

us_quic_socket_t *us_quic_stream_socket(us_quic_stream_t *s) {
Expand Down
2 changes: 2 additions & 0 deletions packages/bun-usockets/src/quic.h
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,8 @@ 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);
int us_quic_stream_has_unacked(us_quic_stream_t *s);
/* The code of the peer's RESET_STREAM or STOP_SENDING, -1 when it sent neither. */
int64_t us_quic_stream_peer_reset_code(us_quic_stream_t *s);

void *us_quic_stream_ext(us_quic_stream_t *s);
us_quic_socket_t *us_quic_stream_socket(us_quic_stream_t *s);
Expand Down
5 changes: 4 additions & 1 deletion src/http/h3_client/ClientSession.rs
Original file line number Diff line number Diff line change
Expand Up @@ -269,7 +269,10 @@ impl ClientSession {
// `Stream.client` is a live backref while attached; `ParentRef::from`
// (NonNull → shared deref) reads the Copy `flags` field without
// forming `&mut HTTPClient` across the `detach()` below.
if bun_ptr::ParentRef::from(client_ptr).flags.h3_retried || st.is_streaming_body {
let client_ref = bun_ptr::ParentRef::from(client_ptr);
// RFC 9110 §9.2.2: the origin may have acted on a request that was written.
let replay_safe = !st.headers_sent || st.peer_rejected || client_ref.method.is_idempotent();
if client_ref.flags.h3_retried || st.is_streaming_body || !replay_safe {
return self.fail(stream, err);
}
let Some(ctx) = ClientContext::get() else {
Expand Down
3 changes: 3 additions & 0 deletions src/http/h3_client/Stream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ pub struct Stream {
pub(crate) is_streaming_body: bool,
pub(crate) headers_delivered: bool,
pub(crate) read_paused: bool,
/// The origin ended the stream with H3_REQUEST_REJECTED: it did not process the request.
pub(crate) peer_rejected: bool,
}

impl Stream {
Expand All @@ -58,6 +60,7 @@ impl Stream {
is_streaming_body: false,
headers_delivered: false,
read_paused: false,
peer_rejected: false,
}))
}

Expand Down
4 changes: 4 additions & 0 deletions src/http/h3_client/callbacks.rs
Original file line number Diff line number Diff line change
Expand Up @@ -305,11 +305,15 @@ extern "C" fn on_stream_writable(s: *mut quic::Stream) {
encode::drain_send_body(stream, s);
}

/// RFC 9114 §8.1: the server did no application processing.
const H3_REQUEST_REJECTED: u64 = 0x10B;

extern "C" fn on_stream_close(s: *mut quic::Stream) {
let s = qstream_arg(s);
let Some(stream) = stream_of(s) else { return };
*s.ext::<Stream>() = None;
stream.qstream = None;
stream.peer_rejected = s.peer_reset_code() == Some(H3_REQUEST_REJECTED);
bun_core::scoped_log!(
h3_client,
"stream_close status={} delivered={}",
Expand Down
6 changes: 6 additions & 0 deletions src/uws_sys/quic/Stream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ unsafe extern "C" {
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_peer_reset_code(s: &mut Stream) -> i64;
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;
Expand Down Expand Up @@ -56,6 +57,11 @@ impl Stream {
us_quic_stream_reset(self)
}

/// The error code of the peer's RESET_STREAM or STOP_SENDING, if it sent one.
pub fn peer_reset_code(&mut self) -> Option<u64> {
u64::try_from(us_quic_stream_peer_reset_code(self)).ok()
}

pub fn header_count(&mut self) -> c_uint {
us_quic_stream_header_count(self)
}
Expand Down
150 changes: 150 additions & 0 deletions test/js/web/fetch/fetch-http3-client.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1061,6 +1061,156 @@ test.skipIf(!isLinux || !cc)("a retry whose reconnect fails before it returns re
});
});

// Same shape as the retry above, but the in-flight request is not idempotent
// and A's handler has already received it. The origin may have acted on it, so
// the client must surface the failure instead of sending it a second time
// (RFC 9110 §9.2.2); the HTTP/1.1 and HTTP/2 clients apply the same rule.
test.each(["POST", "PATCH"])("does not replay a %s that was already sent", async method => {
const hits: string[] = [];
const received = Promise.withResolvers<void>();
const release = Promise.withResolvers<void>();
const a = Bun.serve({
port: 0,
reusePort: true,
tls,
http3: true,
http1: false,
fetch: async req => {
hits.push(`a ${req.method}`);
if (req.method === method) {
received.resolve();
await release.promise;
}
return new Response("a");
},
});
let b: Server | undefined;
try {
const port = a.port;
expect(await fetch(`https://127.0.0.1:${port}/`, h3).then(r => r.text())).toBe("a");
const inflight = fetch(`https://127.0.0.1:${port}/`, { ...h3, method, body: "payload" });
await received.promise;
b = Bun.serve({
port,
reusePort: true,
tls,
http3: true,
http1: false,
fetch: req => {
hits.push(`b ${req.method}`);
return new Response("b");
},
});
a.stop(true);
release.resolve();
const outcome = await inflight.then(r => r.text()).catch(e => e.code);
expect(outcome).toBe("HTTP3StreamReset");
// The origin is reachable: a new request lands on B, the first one never did.
expect(await fetch(`https://127.0.0.1:${port}/`, h3).then(r => r.text())).toBe("b");
expect(hits).toEqual(["a GET", `a ${method}`, "b GET"]);
} finally {
release.resolve();
void a.stop(true);
void b?.stop(true);
}
});

// A request that never left the client is safe to send again whatever its
// method. The fake server answers each Initial with a Version Negotiation
// packet that lists only a reserved version, so the handshake fails before the
// request is written. Each connection attempt uses a new connection ID.
test.each(["GET", "POST"])("re-sends a %s once when the connection failed before it was sent", async method => {
const attempts = new Set<string>();
const server = await Bun.udpSocket({
hostname: "127.0.0.1",
port: 0,
socket: {
data(socket, packet, port, address) {
if (packet.length < 7 || (packet[0] & 0x80) === 0) return;
const dcidLength = packet[5];
const dcid = packet.subarray(6, 6 + dcidLength);
const scidLength = packet[6 + dcidLength];
const scid = packet.subarray(7 + dcidLength, 7 + dcidLength + scidLength);
attempts.add(dcid.toString("hex"));
socket.send(
Buffer.concat([
Buffer.from([0xc0, 0, 0, 0, 0, scidLength]),
scid,
Buffer.from([dcidLength]),
dcid,
Buffer.from([0x0a, 0x0a, 0x0a, 0x0a]),
]),
port,
address,
);
},
},
});
try {
const init = method === "GET" ? {} : { method, body: "payload" };
const outcome = await fetch(`https://127.0.0.1:${server.port}/`, { ...h3, ...init }).catch(e => e.code);
expect(outcome).toBe("HTTP3HandshakeFailed");
expect(attempts.size).toBe(2);
} finally {
server.close();
}
});

// The origin can also end the stream itself, with the connection still up. A
// second send then reaches the same origin and its handler runs again. Only
// H3_REQUEST_REJECTED says that the origin did not process the request
// (RFC 9114 section 4.1.1). Bun.serve sends none of these, so a node:quic
// server is the origin here.
describe("the origin ended the stream with no final response", () => {
const encoder = new TextEncoder();
const listenOrigin = (endWithoutResponse: (stream: any) => void, received: string[]) =>
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 },
async onheaders(this: any, headers: Record<string, string>) {
for await (const _ of this);
received.push(headers[":method"]);
if (received.length === 1) return endWithoutResponse(this);
this.sendHeaders({ ":status": "200" });
this.writer.writeSync(encoder.encode("ok"));
this.writer.endSync();
},
},
);
const internalError = (stream: any) => stream.resetStream(0x102n);
const requestRejected = (stream: any) => stream.resetStream(0x10bn);
const interimOnly = (stream: any) => {
stream.sendInformationalHeaders({ ":status": "103" });
stream.writer.endSync();
};

test.each([
["RESET_STREAM(H3_INTERNAL_ERROR)", "POST", internalError, { outcome: "HTTP3StreamReset", received: ["POST"] }],
["RESET_STREAM(H3_INTERNAL_ERROR)", "PUT", internalError, { outcome: "ok", received: ["PUT", "PUT"] }],
["a 103 and the end of the stream", "POST", interimOnly, { outcome: "HTTP3StreamReset", received: ["POST"] }],
["a 103 and the end of the stream", "PUT", interimOnly, { outcome: "ok", received: ["PUT", "PUT"] }],
["RESET_STREAM(H3_REQUEST_REJECTED)", "POST", requestRejected, { outcome: "ok", received: ["POST", "POST"] }],
["RESET_STREAM(H3_REQUEST_REJECTED)", "PUT", requestRejected, { outcome: "ok", received: ["PUT", "PUT"] }],
])("%s: %s", async (_name, method, endWithoutResponse, expected) => {
const received: string[] = [];
const origin = await listenOrigin(endWithoutResponse, received);
try {
const outcome = await fetch(`https://127.0.0.1:${origin.address.port}/`, { ...h3, method, body: "one order" })
.then(r => r.text())
.catch(e => e.code);
expect({ outcome, received }).toEqual(expected);
} finally {
// Not close(): it waits for the session that fetch() keeps in its pool.
await origin.destroy();
}
});
});

// Subprocess so the experimental flag is process-scoped and the in-process
// server above (http1: false) doesn't interfere — this server keeps http1 on
// so the first fetch goes over TCP and reads Alt-Svc.
Expand Down
Loading