http: couple fetch() receive backpressure to JS body consumption (h1/h2/h3)

src/bun.js/webcore/Body.zig

@@ -76,6 +76,7 @@ pub const PendingValue = struct {
     onStartStreaming: ?*const fn (ctx: *anyopaque) jsc.WebCore.DrainResult = null,
     onReadableStreamAvailable: ?*const fn (ctx: *anyopaque, globalThis: *jsc.JSGlobalObject, readable: jsc.WebCore.ReadableStream) void = null,
     onStreamCancelled: ?*const fn (ctx: ?*anyopaque) void = null,
+    onStreamConsumed: ?*const fn (ctx: ?*anyopaque, bytes: usize) void = null,
     size_hint: Blob.SizeType = 0,
 
     deinit: bool = false,
@@ -519,6 +520,12 @@ pub const Value = union(Tag) {
                         reader.cancel_ctx = task;
                     }
                 }
+                if (locked.onStreamConsumed) |onConsumed| {
+                    if (locked.task) |task| {
+                        reader.drain_handler = onConsumed;
+                        reader.drain_ctx = task;
+                    }
+                }
 
                 reader.context.setup();
 
@@ -1034,6 +1041,19 @@ pub const Value = union(Tag) {
             .globalThis = globalThis,
         });
 
+        if (locked.onStreamCancelled) |onCancelled| {
+            if (locked.task) |task| {
+                reader.cancel_handler = onCancelled;
+                reader.cancel_ctx = task;
+            }
+        }
+        if (locked.onStreamConsumed) |onConsumed| {
+            if (locked.task) |task| {
+                reader.drain_handler = onConsumed;
+                reader.drain_ctx = task;
+            }
+        }
+
         reader.context.setup();
 
         if (drain_result == .estimated_size) {

src/bun.js/webcore/ByteStream.zig

@@ -76,6 +76,16 @@ pub fn unpipeWithoutDeref(this: *@This()) void {
     this.pipe.onPipe = null;
 }
 
+/// Report `n` bytes delivered to the JS consumer (reader fulfilled, pipe
+/// target, or buffer-action sink) so the producer can release backpressure.
+/// Bytes parked in `this.buffer` awaiting a future `onPull` are *not*
+/// reported until that pull happens.
+inline fn didDrain(this: *@This(), n: usize) void {
+    if (n == 0) return;
+    const source = this.parent();
+    if (source.drain_handler) |handler| handler(source.drain_ctx, n);
+}
+
 pub fn onData(
     this: *@This(),
     stream: streams.Result,
@@ -98,13 +108,18 @@ pub fn onData(
     this.has_received_last_chunk = stream.isDone();
 
     if (this.pipe.ctx) |ctx| {
+        this.didDrain(stream.slice().len);
         this.pipe.onPipe.?(ctx, stream, allocator);
         return;
     }
 
     const chunk = stream.slice();
 
     if (this.buffer_action) |*action| {
+        // Buffer-action consumers (`readableStreamToText` etc.) explicitly
+        // want the whole body; treat every append as consumed so the
+        // producer isn't throttled waiting for a pull that never comes.
+        this.didDrain(chunk.len);
         if (stream == .err) {
             defer {
                 this.buffer.clearAndFree();
@@ -206,6 +221,7 @@ pub fn onData(
 
         log("ByteStream.onData pending.run()", .{});
 
+        this.didDrain(to_copy.len);
         this.pending.run();
 
         return;
@@ -306,6 +322,7 @@ pub fn onPull(this: *@This(), buffer: []u8, view: jsc.JSValue) streams.Result {
         if (this.has_received_last_chunk and remaining_in_buffer.len == 0) {
             this.buffer.clearAndFree();
             this.done = true;
+            this.didDrain(to_write);
 
             return .{
                 .into_array_and_done = .{
@@ -315,6 +332,7 @@ pub fn onPull(this: *@This(), buffer: []u8, view: jsc.JSValue) streams.Result {
             };
         }
 
+        this.didDrain(to_write);
         return .{
             .into_array = .{
                 .value = view,
@@ -389,6 +407,12 @@ pub fn deinit(this: *@This()) void {
 
 pub fn drain(this: *@This()) bun.ByteList {
     if (this.buffer.items.len > 0) {
+        // Bytes placed here before the JS stream was constructed (e.g.
+        // the `.owned` drain_result from `onStartStreaming`) are handed
+        // out via `handle.drain()` without going through `onPull`; report
+        // them so they don't become a permanent `unacked_bytes` floor in
+        // the HTTP/2 per-stream window accounting.
+        this.didDrain(this.buffer.items.len);
         return bun.ByteList.moveFromList(&this.buffer);
     }
     return .{};

src/bun.js/webcore/ReadableStream.zig

@@ -432,6 +432,12 @@ pub fn NewSource(
         close_jsvalue: jsc.Strong.Optional = .empty,
         cancel_handler: ?*const fn (?*anyopaque) void = null,
         cancel_ctx: ?*anyopaque = null,
+        /// Invoked whenever `Context` hands bytes to the JS reader (as
+        /// opposed to parking them in an internal buffer). Used by fetch
+        /// to release response-body receive backpressure (h2 per-stream
+        /// WINDOW_UPDATE, h1 socket resume, h3 `wantRead`).
+        drain_handler: ?*const fn (?*anyopaque, usize) void = null,
+        drain_ctx: ?*anyopaque = null,
         globalThis: *JSGlobalObject = undefined,
         this_jsvalue: jsc.JSValue = .zero,
         is_closed: bool = false,

src/bun.js/webcore/fetch/FetchTasklet.zig

@@ -872,6 +872,13 @@ pub const FetchTasklet = struct {
 
         if (this.http) |http_| {
             http_.enableResponseBodyStreaming();
+            // Both Body.toReadableStream and Body.tee wire `drain_handler`
+            // on the ByteStream they construct after this returns, so
+            // `scheduleResponseBodyConsumed` will fire for every reader
+            // pull. Arm the signal so the transport gates receive
+            // flow-control on those reports instead of on receipt (h2
+            // per-stream WINDOW_UPDATE, h1 socket pause, h3 wantRead).
+            this.signal_store.body_consumption_tracked.store(true, .release);
 
             // If the server sent the headers and the response body in two separate socket writes
             // and if the server doesn't close the connection by itself
@@ -925,6 +932,8 @@ pub const FetchTasklet = struct {
                 const source = readable.ptr.Bytes.parent();
                 source.cancel_handler = null;
                 source.cancel_ctx = null;
+                source.drain_handler = null;
+                source.drain_ctx = null;
             }
         }
     }
@@ -935,6 +944,18 @@ pub const FetchTasklet = struct {
         this.ignoreRemainingResponseBody();
     }
 
+    /// ByteStream delivered `bytes` to the JS reader. Forward to the
+    /// HTTP thread so the transport can release response-body receive
+    /// backpressure: HTTP/2 emits per-stream WINDOW_UPDATE, HTTP/1.1
+    /// resumes a paused socket read, HTTP/3 re-enables
+    /// `lsquic_stream_wantread`.
+    fn onStreamConsumedCallback(ctx: ?*anyopaque, bytes: usize) void {
+        const this = bun.cast(*FetchTasklet, ctx.?);
+        if (this.signal_store.aborted.load(.monotonic)) return;
+        const http_ = this.http orelse return;
+        bun.http.http_thread.scheduleResponseBodyConsumed(http_.async_http_id, bytes);
+    }
+
     fn toBodyValue(this: *FetchTasklet) Body.Value {
         if (this.getAbortError()) |err| {
             return .{ .Error = err };
@@ -948,6 +969,7 @@ pub const FetchTasklet = struct {
                     .onStartStreaming = FetchTasklet.onStartStreamingHTTPResponseBodyCallback,
                     .onReadableStreamAvailable = FetchTasklet.onReadableStreamAvailable,
                     .onStreamCancelled = FetchTasklet.onStreamCancelledCallback,
+                    .onStreamConsumed = FetchTasklet.onStreamConsumedCallback,
                 },
             };
             return response;
@@ -998,6 +1020,22 @@ pub const FetchTasklet = struct {
         if (this.http) |http_| {
             http_.enableResponseBodyStreaming();
         }
+        // `drain_handler` is about to be cleared and incoming chunks will
+        // be dropped without ever reaching the ByteStream, so no more
+        // `scheduleResponseBodyConsumed` reports. Disarm the tracking
+        // signal so the HTTP/2 client falls back to receipt-based
+        // per-stream WINDOW_UPDATE and the abandoned body can drain
+        // instead of stalling the stream at the initial window. The
+        // sentinel consume message both wakes the HTTP thread (so
+        // `replenishWindow` re-runs — its only other trigger is inbound
+        // DATA, and a server that has exhausted the window sends none)
+        // and saturates `consumed_bytes` to `unacked_bytes` so the first
+        // re-run releases whatever is already outstanding regardless of
+        // which order the atomic store and the queue drain land in.
+        this.signal_store.body_consumption_tracked.store(false, .release);
+        if (this.http) |http_| {
+            bun.http.http_thread.scheduleResponseBodyConsumed(http_.async_http_id, std.math.maxInt(u32));
+        }
         // we should not keep the process alive if we are ignoring the body
         const vm = this.javascript_vm;
         this.poll_ref.unref(vm);

src/deps/uws/quic/Stream.zig

@@ -35,6 +35,16 @@ pub const Stream = opaque {
         us_quic_stream_want_write(s, @intFromBool(want));
     }
 
+    extern fn us_quic_stream_want_read(s: *Stream, want: c_int) void;
+    /// Toggle lsquic's `on_read` callback for this stream. When `false`,
+    /// bytes accumulate inside lsquic's receive buffer and it withholds
+    /// `MAX_STREAM_DATA` credit — the QUIC-level equivalent of pausing a
+    /// TCP socket read. Used by the HTTP/3 client to backpressure the
+    /// server when the JS `ReadableStream` reader has stalled.
+    pub fn wantRead(s: *Stream, want: bool) void {
+        us_quic_stream_want_read(s, @intFromBool(want));
+    }
+
     extern fn us_quic_stream_send_headers(s: *Stream, h: [*]const Header, n: c_uint, end_stream: c_int) c_int;
     pub fn sendHeaders(s: *Stream, headers: []const Header, end_stream: bool) c_int {
         return us_quic_stream_send_headers(s, headers.ptr, @intCast(headers.len), @intFromBool(end_stream));

src/http.zig

@@ -301,6 +301,20 @@
     this.start_(true);
 }
 
+/// Response-body receive backpressure for streaming consumers
+/// (`signals.body_consumption_tracked` set). Once the wire bytes handed
+/// to JS but not yet reported drained via `scheduleResponseBodyConsumed`
+/// exceed `receive_body_high_water`, the transport applies backpressure:
+/// HTTP/1.1 pauses the socket read (`us_socket_pause` → TCP rwnd closes),
+/// HTTP/3 stops `lsquic_stream_read` (`wantRead(false)` → lsquic withholds
+/// `MAX_STREAM_DATA`). HTTP/2 uses a different mechanism (per-stream
+/// WINDOW_UPDATE gated on the same consumption reports — see
+/// `h2_client/ClientSession.replenishWindow`) so its limit is governed by
+/// `H2.local_initial_window_size` instead. Resumed once consumption
+/// reports bring the outstanding count below `receive_body_low_water`.
+pub const receive_body_high_water: usize = 1 << 20; // 1 MiB
+pub const receive_body_low_water: usize = 256 * 1024;
+
 /// REFUSED_STREAM or graceful GOAWAY past our id: the server promises it
 /// did not process the request, so re-dispatch from the top. Only reached
 /// for `.bytes` bodies (replayable).
@@ -2031,11 +2045,29 @@
         .body => {
             this.setTimeout(socket, 5);
 
+            const before = this.state.total_body_received;
             const report_progress = this.handleResponseBody(incoming_data, false) catch |err| {
                 this.closeAndFail(err, is_ssl, socket);
                 return;
             };
 
+            // Must run before progressUpdate: a final chunk makes
+            // onAsyncHTTPCallback destroy the ThreadlocalAsyncHTTP that
+            // owns *this* HTTPClient inline on this thread, so `this`
+            // is poisoned by the time progressUpdate returns.
+            //
+            // Count the `total_body_received` delta, not
+            // `incoming_data.len`: for Transfer-Encoding: chunked the
+            // latter includes per-chunk framing (hex size + CRLFs) that
+            // never reaches JS, so `didDrain` can't credit it and it
+            // would accumulate as a permanent floor under
+            // `outstanding_body_bytes`. Once that floor passes the
+            // low-water mark, a long-lived SSE stream deadlocks on the
+            // first pause. The .body arm uses the same delta for
+            // symmetry (and to discard any trailing bytes past
+            // Content-Length).
+            this.maybePauseReceive(is_ssl, socket, this.state.total_body_received -| before);
+
             if (report_progress) {
                 this.progressUpdate(is_ssl, ctx, socket);
                 return;
@@ -2045,11 +2077,14 @@
         .body_chunk => {
             this.setTimeout(socket, 5);
 
+            const before = this.state.total_body_received;
             const report_progress = this.handleResponseBodyChunkedEncoding(incoming_data) catch |err| {
                 this.closeAndFail(err, is_ssl, socket);
                 return;
             };
 
+            this.maybePauseReceive(is_ssl, socket, this.state.total_body_received -| before);
+
             if (report_progress) {
                 this.progressUpdate(is_ssl, ctx, socket);
                 return;
@@ -2204,6 +2239,74 @@
     socket.setTimeoutMinutes(minutes);
 }
 
+/// Called from `onData` after body bytes have been appended and (if
+/// streaming) delivered to JS. Counts `n` wire bytes towards
+/// `outstanding_body_bytes` and, if the JS reader is reporting
+/// consumption and the outstanding total has crossed the high-water
+/// mark, pauses the socket read so TCP rwnd backpressures the server.
+/// The proxy-tunnel path is excluded: its inner TLS session needs the
+/// socket readable to complete handshake/close_notify regardless of the
+/// application body, and pausing the carrier socket would wedge that.
+fn maybePauseReceive(this: *HTTPClient, comptime is_ssl: bool, socket: NewHTTPContext(is_ssl).HTTPSocket, n: usize) void {
+    if (!this.signals.get(.body_consumption_tracked)) return;
+    if (this.state.stage == .done or this.state.stage == .fail) return;
+    this.state.outstanding_body_bytes +|= n;
+    // handleResponseBody just ran: if the body is complete (or a redirect
+    // is pending) the socket is about to be released to the keep-alive
+    // pool — or closed — inside progressUpdate. Leaving it paused would
+    // hand the next pooled request a socket with LIBUS_SOCKET_READABLE
+    // cleared and no consumeResponseBody to re-enable it.
+    //
+    // We can reach here with `receive_paused` already true because
+    // uSockets' repeat-recv fast path (loop.c) keeps calling recv() in
+    // the same epoll tick while the buffer comes back full, without
+    // re-consulting poll flags; a us_socket_pause() issued from a
+    // previous on_data in that loop doesn't stop the next recv(). The
+    // final chunk can therefore arrive while receive_paused is set.
+    if (this.state.isDone() or this.state.flags.is_redirect_pending) {
+        if (this.state.flags.receive_paused) {
+            this.state.flags.receive_paused = false;
+            if (!socket.isClosedOrHasError()) _ = socket.resumeStream();
+        }
+        return;
+    }
+    if (this.state.flags.receive_paused) return;
+    if (this.state.outstanding_body_bytes < receive_body_high_water) return;
+    if (this.proxy_tunnel != null) return;
+    if (socket.isClosedOrHasError()) return;
+    this.state.flags.receive_paused = true;
+    // While paused, the idle timer would otherwise fire with no socket
+    // activity to re-arm it. The response isn't "stalled" — the reader
+    // chose not to pull — so drop the timeout until we resume.
+    socket.setTimeoutMinutes(0);
+    socket.timeout(0);
+    _ = socket.pauseStream();
+    log("maybePauseReceive: paused at {d} bytes outstanding", .{this.state.outstanding_body_bytes});
+}
+
+/// HTTP-thread wake-up from `scheduleResponseBodyConsumed`: the JS reader
+/// drained `bytes` from the ByteStream. For HTTP/1.1 this may resume a
+/// socket read that `maybePauseReceive` paused. HTTP/2 and HTTP/3 have
+/// dedicated handlers that reach here via their session dispatch.
+pub fn consumeResponseBody(this: *HTTPClient, comptime is_ssl: bool, socket: NewHTTPContext(is_ssl).HTTPSocket, bytes: u32) void {
+    // `bytes` is post-decompression; clamp so a compression surplus
+    // doesn't leave `outstanding_body_bytes` underflowed — same rule as
+    // the h2 `consumed_bytes` clamp.
+    this.state.outstanding_body_bytes -|= @min(@as(usize, bytes), this.state.outstanding_body_bytes);
+    if (!this.state.flags.receive_paused) return;
+    // Disarming `body_consumption_tracked` (e.g. reader.cancel()) posts a
+    // saturating sentinel; treat that as "resume now" regardless of the
+    // low-water mark so the abandoned body can drain for keep-alive reuse.
+    const should_resume = this.state.outstanding_body_bytes <= receive_body_low_water or
+        !this.signals.get(.body_consumption_tracked);
+    if (!should_resume) return;
+    this.state.flags.receive_paused = false;
+    if (socket.isClosedOrHasError()) return;
+    _ = socket.resumeStream();
+    this.setTimeout(socket, 5);
+    log("consumeResponseBody: resumed, {d} bytes outstanding", .{this.state.outstanding_body_bytes});
+}
+
 pub fn drainResponseBody(this: *HTTPClient, comptime is_ssl: bool, socket: NewHTTPContext(is_ssl).HTTPSocket) void {
 
     // Find out if we should not send any update.
@@ -2267,6 +2370,16 @@
 
         if (this.isKeepAlivePossible() and !socket.isClosedOrHasError() and tunnel_poolable) {
             log("release socket", .{});
+            // Defensive: maybePauseReceive's isDone() branch resumes for
+            // the normal .body/.body_chunk onData path, but a
+            // close-delimited body or an early server reply can reach
+            // here without another onData. The uSockets `is_paused`
+            // flag survives state.reset(), so a paused socket released
+            // to the pool would hang the next request that adopts it.
+            if (this.state.flags.receive_paused) {
+                this.state.flags.receive_paused = false;
+                _ = socket.resumeStream();
+            }
             const tunnel = this.proxy_tunnel;
             this.proxy_tunnel = null;
             if (tunnel) |t| t.detachOwner(this);

src/http/H3Client.zig

@@ -26,15 +26,22 @@ pub const AltSvc = @import("./h3_client/AltSvc.zig");
 /// via TestingAPIs.quicLiveCounts so they must be atomic.
 pub var live_sessions = std.atomic.Value(u32).init(0);
 pub var live_streams = std.atomic.Value(u32).init(0);
+/// Cumulative response-body bytes delivered via `onStreamData` across all
+/// h3 client streams in this process. Exposed for the backpressure tests:
+/// a stalled JS reader should cap this near `receive_body_high_water` once
+/// `wantRead(false)` lands, whereas without the gate it tracks whatever
+/// the server pushes.
+pub var body_bytes_received = std.atomic.Value(u64).init(0);
 
 pub const TestingAPIs = struct {
     /// Named distinctly from H2's `liveCounts` because generate-js2native.ts
     /// mangles `[^A-Za-z]` to `_`, so `H2Client.zig` and `H3Client.zig` produce
     /// the same path prefix and the function name has to differ.
     pub fn quicLiveCounts(globalThis: *jsc.JSGlobalObject, _: *jsc.CallFrame) bun.JSError!jsc.JSValue {
-        const obj = jsc.JSValue.createEmptyObject(globalThis, 2);
+        const obj = jsc.JSValue.createEmptyObject(globalThis, 3);
         obj.put(globalThis, jsc.ZigString.static("sessions"), .jsNumber(live_sessions.load(.monotonic)));
         obj.put(globalThis, jsc.ZigString.static("streams"), .jsNumber(live_streams.load(.monotonic)));
+        obj.put(globalThis, jsc.ZigString.static("bodyBytesReceived"), .jsNumber(body_bytes_received.load(.monotonic)));
         return obj;
     }
 };