http3: deliver every header block that lsquic decodes while a stream is read

packages/bun-usockets/src/node_quic_shim.c

@@ -432,6 +432,9 @@ void us_nq_settings_init(struct lsquic_engine_settings *s, int is_server,
      * grant while that stream has unsent data (lsquic_qdh_arm_if_unsent). */
     s->es_qpack_enc_max_size = 0;
     s->es_qpack_enc_max_blocked = 0;
+    /* lsquic then parses nothing past an unclaimed header block. A client's
+     * default of 2 lets the body reach JS ahead of the `onheaders` before it. */
+    s->es_max_header_sets = 1;
 }
 
 #define NQ_SET(field, ctype) \

packages/bun-usockets/src/quic.c

@@ -126,6 +126,8 @@ struct us_quic_stream_s {
     struct us_quic_hset *hset;
     int headers_delivered;
     int fin_delivered;
+    /* Set by us_quic_flush_from_on_write, cleared by on_write. */
+    int flush_on_write;
     /* ext follows */
 };
 
@@ -673,27 +675,29 @@ static lsquic_stream_ctx_t *us_quic_on_new_stream(void *if_ctx, lsquic_stream_t
     return (lsquic_stream_ctx_t *) s;
 }
 
+/* Hands the next header block lsquic has decoded to on_stream_headers.
+ * Returns 1 for a block, 0 for none, -1 if on_stream_headers closed the stream. */
+static int us_quic_deliver_hset(lsquic_stream_t *stream, us_quic_stream_t *s) {
+    struct us_quic_hset *hset = (struct us_quic_hset *) lsquic_stream_get_hset(stream);
+    if (!hset) return 0;
+    us_quic_hset_finalize(hset);
+    us_quic_hset_free(s->hset);
+    s->hset = hset;
+    s->headers_delivered = 1;
+    if (s->ctx->on_stream_headers) s->ctx->on_stream_headers(s);
+    return s->stream ? 1 : -1;
+}
+
 static void us_quic_on_read(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
     us_quic_stream_t *s = (us_quic_stream_t *) h;
     us_quic_socket_context_t *ctx = s->ctx;
 
     /* lsquic queues a fresh hset for every HEADERS block (1xx interims,
      * the final response, trailers). lsquic_stream_get_hset returns the
-     * next undelivered one and lsquic_stream_read won't drain DATA past
-     * an unconsumed hset, so re-dispatch on_stream_headers each time
-     * instead of latching after the first. */
-    {
-        struct us_quic_hset *hset = (struct us_quic_hset *) lsquic_stream_get_hset(stream);
-        if (hset) {
-            us_quic_hset_finalize(hset);
-            us_quic_hset_free(s->hset);
-            s->hset = hset;
-            s->headers_delivered = 1;
-            if (ctx->on_stream_headers) ctx->on_stream_headers(s);
-            /* on_stream_headers may have closed us */
-            if (!s->stream) return;
-        }
-    }
+     * next undelivered one and lsquic_stream_read fails while one is
+     * queued, so re-dispatch on_stream_headers each time instead of
+     * latching after the first. */
+    if (us_quic_deliver_hset(stream, s) < 0) return;
 
     ssize_t r;
     while ((r = lsquic_stream_read(stream, ctx->read_buf, US_QUIC_READ_BUF)) > 0) {
@@ -702,6 +706,10 @@ static void us_quic_on_read(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
         if (!s->stream) return;
     }
     if (r == 0 && !s->fin_delivered) {
+        /* lsquic returns 0 even if this read decoded one more header block. */
+        int delivered;
+        while ((delivered = us_quic_deliver_hset(stream, s)) > 0) {}
+        if (delivered < 0) return;
         s->fin_delivered = 1;
         lsquic_stream_wantread(stream, 0);
         lsquic_stream_shutdown(stream, 0);
@@ -712,7 +720,13 @@ static void us_quic_on_read(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
 static void us_quic_on_write(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
     us_quic_stream_t *s = (us_quic_stream_t *) h;
     lsquic_stream_wantwrite(stream, 0);
+    /* Taken before the callback: a block that on_stream_writable sends (the
+     * client's request) is for the next on_write, see the helper. */
+    int flush = s->flush_on_write;
+    s->flush_on_write = 0;
     if (s->ctx->on_stream_writable) s->ctx->on_stream_writable(s);
+    /* Last, so that it also covers what on_stream_writable wrote. */
+    if (flush) lsquic_stream_flush(stream);
 }
 
 static void us_quic_on_close(lsquic_stream_t *stream, lsquic_stream_ctx_t *h) {
@@ -1122,6 +1136,14 @@ void us_quic_stream_want_write(us_quic_stream_t *s, int want) {
     if (s->stream) lsquic_stream_wantwrite(s->stream, want);
 }
 
+/* lsquic_stream_send_headers only buffers, and a flush is a no-op while lsquic
+ * holds the block back (new connection). on_write runs after lsquic wrote it,
+ * and after a handler that answered at once put its response behind a 1xx. */
+static void us_quic_flush_from_on_write(us_quic_stream_t *s) {
+    s->flush_on_write = 1;
+    lsquic_stream_wantwrite(s->stream, 1);
+}
+
 int us_quic_stream_send_informational(us_quic_stream_t *s, const char *status3) {
     if (!s->stream) return -1;
     char buf[10];
@@ -1130,7 +1152,12 @@ int us_quic_stream_send_informational(us_quic_stream_t *s, const char *status3)
     struct lsxpack_header xh;
     lsxpack_header_set_offset2(&xh, buf, 0, 7, 7, 3);
     lsquic_http_headers_t lh = { .count = 1, .headers = &xh };
-    return lsquic_stream_send_headers(s->stream, &lh, 0);
+    int r = lsquic_stream_send_headers(s->stream, &lh, 0);
+    if (r == 0) {
+        us_quic_flush_from_on_write(s);
+        s->ctx->pending_write_bytes++;
+    }
+    return r;
 }
 
 int us_quic_stream_send_headers(us_quic_stream_t *s,
@@ -1175,6 +1202,7 @@ int us_quic_stream_send_headers(us_quic_stream_t *s,
     if (buf != stackbuf) us_free(buf);
     if (xh != stackh) us_free(xh);
     if (end_stream && r == 0) lsquic_stream_shutdown(s->stream, 1);
+    if (!end_stream && r == 0) us_quic_flush_from_on_write(s);
     /* Mark the context dirty so drainQuicIfNecessary picks up header-only
      * responses (204/304) that never call us_quic_stream_write. */
     if (r == 0) s->ctx->pending_write_bytes += (unsigned int) total + 1;

src/runtime/node/quic/stream.rs

@@ -1023,49 +1023,56 @@ pub(super) unsafe extern "C" fn on_stream_read(ctx: *mut c_void, s: *mut lsquic:
     }
     // SAFETY: `ctx` is the live QuicStream we returned from on_new_stream.
     let qs = unsafe { &*ctx.cast::<QuicStream>() };
-    if let Some(hset) = stream.take_header_set() {
-        let pairs = hset.pairs();
-        /// RFC 9114 §8.1 H3_MESSAGE_ERROR — malformed message (a request
-        /// carrying :status). Matches lsquic's `HEC_MESSAGE_ERROR`
-        /// (lsquic_hq.h:82); 0x105 is H3_FRAME_UNEXPECTED, a different code.
-        const H3_MESSAGE_ERROR: u64 = 0x10e;
-        let has_status = pairs
-            .as_chunks::<2>()
-            .0
-            .iter()
-            .find(|kv| kv[0] == b":status")
-            .map(|kv| kv[1].len() == 3 && kv[1][0] == b'1');
-        // A :status in a request is malformed: node's nghttp3 resets the
-        // stream (RFC 9114 §4.1.2), and routing it to `oninfo` would leave
-        // the request unanswered until the idle timeout.
-        let peer_is_client = qs.session_ref().is_some_and(|s| s.is_server());
-        if peer_is_client && has_status.is_some() {
-            if let Some(s) = qs.ls() {
-                // reset() only ends the read side when the peer already
-                // FIN'd/RST'd, so STOP_SENDING is what stops a malformed
-                // request streaming a body into a stream nothing will answer.
-                s.reset(H3_MESSAGE_ERROR);
-                s.stop_sending(H3_MESSAGE_ERROR);
+    // Returns false if a malformed header block reset the stream.
+    let claim_header_sets = || -> bool {
+        while let Some(hset) = stream.take_header_set() {
+            let pairs = hset.pairs();
+            /// RFC 9114 §8.1 H3_MESSAGE_ERROR — malformed message (a request
+            /// carrying :status). Matches lsquic's `HEC_MESSAGE_ERROR`
+            /// (lsquic_hq.h:82); 0x105 is H3_FRAME_UNEXPECTED, a different code.
+            const H3_MESSAGE_ERROR: u64 = 0x10e;
+            let has_status = pairs
+                .as_chunks::<2>()
+                .0
+                .iter()
+                .find(|kv| kv[0] == b":status")
+                .map(|kv| kv[1].len() == 3 && kv[1][0] == b'1');
+            // A :status in a request is malformed: node's nghttp3 resets the
+            // stream (RFC 9114 §4.1.2), and routing it to `oninfo` would leave
+            // the request unanswered until the idle timeout.
+            let peer_is_client = qs.session_ref().is_some_and(|s| s.is_server());
+            if peer_is_client && has_status.is_some() {
+                if let Some(s) = qs.ls() {
+                    // reset() only ends the read side when the peer already
+                    // FIN'd/RST'd, so STOP_SENDING is what stops a malformed
+                    // request streaming a body into a stream nothing will answer.
+                    s.reset(H3_MESSAGE_ERROR);
+                    s.stop_sending(H3_MESSAGE_ERROR);
+                }
+                qs.mark_reset(H3_MESSAGE_ERROR);
+                return false;
+            }
+            // 1xx interim responses are HINTS (RFC 9114 §4.1).
+            let is_interim = has_status.unwrap_or(false);
+            let kind = if is_interim {
+                QUIC_STREAM_HEADERS_KIND_HINTS
+            } else if qs.mark_headers_received() {
+                QUIC_STREAM_HEADERS_KIND_INITIAL
+            } else {
+                QUIC_STREAM_HEADERS_KIND_TRAILING
+            };
+            if let Some(session) = qs.session_ref() {
+                session.push_event(SessionEvent::StreamHeaders {
+                    stream: ctx.cast(),
+                    pairs,
+                    kind,
+                });
             }
-            qs.mark_reset(H3_MESSAGE_ERROR);
-            return;
-        }
-        // 1xx interim responses are HINTS (RFC 9114 §4.1).
-        let is_interim = has_status.unwrap_or(false);
-        let kind = if is_interim {
-            QUIC_STREAM_HEADERS_KIND_HINTS
-        } else if qs.mark_headers_received() {
-            QUIC_STREAM_HEADERS_KIND_INITIAL
-        } else {
-            QUIC_STREAM_HEADERS_KIND_TRAILING
-        };
-        if let Some(session) = qs.session_ref() {
-            session.push_event(SessionEvent::StreamHeaders {
-                stream: ctx.cast(),
-                pairs,
-                kind,
-            });
         }
+        true
+    };
+    if !claim_header_sets() {
+        return;
     }
     if stream.received_early_data() {
         qs.with_state(|s| s.received_early_data = 1);
@@ -1082,6 +1089,10 @@ pub(super) unsafe extern "C" fn on_stream_read(ctx: *mut c_void, s: *mut lsquic:
                 got_any = true;
             }
             0 => {
+                // lsquic returns 0 even if this read decoded one more header block.
+                if !claim_header_sets() {
+                    break;
+                }
                 qs.push_inbound(&[], true);
                 stream.want_read(false);
                 let write_done = qs.with_state(|s| s.fin_sent != 0 || s.write_ended != 0);

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

@@ -1762,6 +1762,79 @@ describe("Bun.serve HTTP/3 request validation", () => {
   });
 });
 
+// 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
+// on its own. The client waits for the event, with no timer: a server that
+// keeps the 100 until the final response makes the test time out.
+describe.concurrent("Bun.serve HTTP/3 sends the automatic 100 Continue ahead of the final response", () => {
+  async function exchange(server: { port: number }, { handshakeFirst }: { handshakeFirst: boolean }) {
+    // A session or a stream that ends before a response fails the test with
+    // its reason. Once a response is in, these rejections do nothing.
+    const firstResponse = Promise.withResolvers<void>();
+    const endedEarly = (what: string) => () => firstResponse.reject(new Error(`${what} closed before a response`));
+    await using endpoint = new QuicEndpoint();
+    const client = await connect(`127.0.0.1:${server.port}`, {
+      endpoint,
+      servername: "localhost",
+      verifyPeer: "manual",
+      transportParams: { maxIdleTimeout: 5 },
+      onerror: firstResponse.reject,
+    });
+    client.closed.then(endedEarly("the session"), firstResponse.reject);
+    if (handshakeFirst) await client.opened;
+
+    const seen: string[] = [];
+    const stream = await client.createBidirectionalStream({
+      oninfo(received: Record<string, string>) {
+        seen.push("info " + received[":status"]);
+        firstResponse.resolve();
+      },
+      onheaders(received: Record<string, string>) {
+        seen.push("headers " + received[":status"]);
+        firstResponse.resolve();
+      },
+    });
+    stream.closed.then(endedEarly("the stream"), firstResponse.reject);
+    const writer = stream.writer;
+    stream.sendHeaders(requestHeaders("/", { ":method": "POST", expect: "100-continue" }));
+    await firstResponse.promise;
+    seen.push("client sends the content");
+    writer.writeSync(new TextEncoder().encode("request-content"));
+    writer.endSync();
+
+    let body = "";
+    for await (const batch of stream as AsyncIterable<Uint8Array[]>) {
+      for (const chunk of batch) body += Buffer.from(chunk).toString("latin1");
+    }
+    if (!client.destroyed) client.close().catch(() => {});
+    return [...seen, "body " + body];
+  }
+
+  const expected = ["info 100", "client sends the content", "headers 200", "body content:request-content"];
+  const serve = () =>
+    Bun.serve({
+      port: 0,
+      tls,
+      http3: true,
+      http1: false,
+      fetch: async req => new Response("content:" + (await req.text())),
+    });
+
+  test("on a connection that completed its handshake", async () => {
+    await using server = serve();
+    expect(await exchange(server, { handshakeFirst: true })).toEqual(expected);
+  });
+
+  // The request leaves with the client's handshake. The server then still has
+  // the first byte of its QPACK encoder stream to send, and lsquic holds the 100
+  // header block back until that byte is out.
+  test("on the first request of a connection, sent with the handshake", async () => {
+    await using server = serve();
+    expect(await exchange(server, { handshakeFirst: false })).toEqual(expected);
+  });
+});
+
 // The HTTP/3 twin of the HTTP/1 cases in websocket-server.test.ts: ws.close()
 // runs close() before it returns, and a request handler that calls it must still
 // run to completion before the nextTick and promise callbacks it queued. The