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

src/http/H3Client.zig

@@ -26,6 +26,12 @@ 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 = @import("../http_jsc/headers_jsc.zig").H3TestingAPIs;
 

src/http/HTTPThread.zig

@@ -34,10 +34,12 @@ has_pending_queued_abort: bool = false,
 queued_shutdowns: std.ArrayListUnmanaged(ShutdownMessage) = std.ArrayListUnmanaged(ShutdownMessage){},
 queued_writes: std.ArrayListUnmanaged(WriteMessage) = std.ArrayListUnmanaged(WriteMessage){},
 queued_response_body_drains: std.ArrayListUnmanaged(DrainMessage) = std.ArrayListUnmanaged(DrainMessage){},
+queued_response_body_consumed: std.ArrayListUnmanaged(ConsumeMessage) = std.ArrayListUnmanaged(ConsumeMessage){},
 
 queued_shutdowns_lock: bun.Mutex = .{},
 queued_writes_lock: bun.Mutex = .{},
 queued_response_body_drains_lock: bun.Mutex = .{},
+queued_response_body_consumed_lock: bun.Mutex = .{},
 
 queued_threadlocal_proxy_derefs: std.ArrayListUnmanaged(*ProxyTunnel) = std.ArrayListUnmanaged(*ProxyTunnel){},
 
@@ -117,6 +119,10 @@ const WriteMessage = struct {
 const DrainMessage = struct {
     async_http_id: u32,
 };
+const ConsumeMessage = struct {
+    async_http_id: u32,
+    bytes: u32,
+};
 const ShutdownMessage = struct {
     async_http_id: u32,
 };
@@ -507,9 +513,52 @@ fn drainQueuedHTTPResponseBodyDrains(this: *@This()) void {
     }
 }
 
+fn drainQueuedHTTPResponseBodyConsumed(this: *@This()) void {
+    while (true) {
+        var queued = brk: {
+            this.queued_response_body_consumed_lock.lock();
+            defer this.queued_response_body_consumed_lock.unlock();
+            const q = this.queued_response_body_consumed;
+            this.queued_response_body_consumed = .{};
+            break :brk q;
+        };
+        defer queued.deinit(bun.default_allocator);
+
+        for (queued.items) |msg| {
+            if (bun.http.socket_async_http_abort_tracker.get(msg.async_http_id)) |socket_ptr| {
+                switch (socket_ptr) {
+                    inline .SocketTLS, .SocketTCP => |socket, tag| {
+                        const is_tls = tag == .SocketTLS;
+                        const HTTPContext = HTTPThread.NewHTTPContext(comptime is_tls);
+                        const tagged = HTTPContext.getTaggedFromSocket(socket);
+                        if (tagged.get(HTTPClient)) |client| {
+                            // HTTP/1.1: may resume a paused socket read.
+                            client.consumeResponseBody(comptime is_tls, socket, msg.bytes);
+                        }
+                        if (tagged.get(bun.http.H2.ClientSession)) |session| {
+                            // HTTP/2: releases per-stream WINDOW_UPDATE.
+                            session.consumeResponseBodyByHttpId(msg.async_http_id, msg.bytes);
+                        }
+                    },
+                }
+            } else {
+                // HTTP/3: QUIC streams aren't in the TCP-socket tracker;
+                // dispatch via the session registry. May resume a
+                // lsquic `wantRead(0)` pause.
+                bun.http.H3.ClientContext.consumeResponseBodyByHttpId(msg.async_http_id, msg.bytes);
+            }
+        }
+        if (queued.items.len == 0) {
+            break;
+        }
+        threadlog("drained {d} queued consumes", .{queued.items.len});
+    }
+}
+
 fn drainEvents(this: *@This()) void {
     // Process any pending writes **before** aborting.
     this.drainQueuedHTTPResponseBodyDrains();
+    this.drainQueuedHTTPResponseBodyConsumed();
     this.drainQueuedWrites();
     this.drainQueuedShutdowns();
     bun.http.H3.PendingConnect.drainResolved();
@@ -661,6 +710,36 @@ pub fn scheduleResponseBodyDrain(this: *@This(), async_http_id: u32) void {
         this.loop.loop.wakeup();
 }
 
+/// JS-thread → HTTP-thread notice that the `ReadableStream` reader for
+/// `async_http_id` has drained `bytes` from its buffer. Consecutive messages
+/// for the same id are coalesced under the lock so a tight `read()` loop
+/// posts one entry per wake instead of one per pull.
+pub fn scheduleResponseBodyConsumed(this: *@This(), async_http_id: u32, bytes: usize) void {
+    const n: u32 = @truncate(@min(bytes, @as(usize, std.math.maxInt(u32))));
+    if (n == 0) return;
+    const appended = brk: {
+        this.queued_response_body_consumed_lock.lock();
+        defer this.queued_response_body_consumed_lock.unlock();
+        const items = this.queued_response_body_consumed.items;
+        if (items.len > 0 and items[items.len - 1].async_http_id == async_http_id) {
+            items[items.len - 1].bytes +|= n;
+            break :brk false;
+        }
+        this.queued_response_body_consumed.append(bun.default_allocator, .{
+            .async_http_id = async_http_id,
+            .bytes = n,
+        }) catch |err| bun.handleOom(err);
+        break :brk true;
+    };
+    // Only wake on the append path: if we coalesced into an existing
+    // entry, the HTTP thread was already woken for that entry's
+    // original append and will see the updated `bytes` when it swaps
+    // the queue under the same lock. A tight `read()` loop over a
+    // multi-GiB body otherwise issues one eventfd write per pull.
+    if (appended and this.has_awoken.load(.monotonic))
+        this.loop.loop.wakeup();
+}
+
 pub fn scheduleShutdown(this: *@This(), http: *AsyncHTTP) void {
     threadlog("scheduleShutdown {d}", .{http.async_http_id});
     {

src/http/InternalState.zig

@@ -27,6 +27,13 @@ body_out_str: ?*MutableString = null,
 compressed_body: MutableString = undefined,
 content_length: ?usize = null,
 total_body_received: usize = 0,
+/// Post-dechunk/decompress body bytes (`body_out_str` delta) handed to
+/// JS via `progressUpdate` that haven't been reported drained via
+/// `scheduleResponseBodyConsumed`. Drives the HTTP/1.1 read-side pause
+/// once over `receive_body_high_water`. HTTP/2 and HTTP/3 track the
+/// equivalent per-stream (`h2_client/Stream.unacked_bytes`,
+/// `h3_client/Stream.outstanding_body_bytes`).
+outstanding_body_bytes: usize = 0,
 request_body: []const u8 = "",
 original_request_body: HTTPRequestBody = .{ .bytes = "" },
 request_sent_len: usize = 0,
@@ -42,7 +49,13 @@ pub const InternalStateFlags = packed struct(u8) {
     is_redirect_pending: bool = false,
     is_libdeflate_fast_path_disabled: bool = false,
     resend_request_body_on_redirect: bool = false,
-    _padding: u2 = 0,
+    /// The HTTP/1.1 socket read has been paused (`us_socket_pause`)
+    /// because `outstanding_body_bytes` is above `receive_body_high_water`.
+    /// Cleared when consumption reports bring it below
+    /// `receive_body_low_water`. HTTP/3's equivalent lives on
+    /// `h3_client/Stream.read_paused`.
+    receive_paused: bool = false,
+    _padding: u1 = 0,
 };
 
 pub fn init(body: HTTPRequestBody, body_out_str: *MutableString) InternalState {

src/http/Signals.zig

@@ -2,6 +2,16 @@
 
 header_progress: ?*std.atomic.Value(bool) = null,
 response_body_streaming: ?*std.atomic.Value(bool) = null,
+/// Distinct from `response_body_streaming`: set only while a JS consumer
+/// is wired to report drained bytes via `scheduleResponseBodyConsumed`.
+/// `response_body_streaming` is also set by paths that never report
+/// consumption (S3 streaming download, abandoned bodies via
+/// `ignoreRemainingResponseBody`); gating flow-control on that would
+/// deadlock those streams. All three transports key receive-side
+/// backpressure on this signal — not `response_body_streaming` — to
+/// decide whether flow control is consumption-gated or receipt-based
+/// (h1 `maybePauseReceive`, h2 `replenishWindow`, h3 `onStreamData`).
+body_consumption_tracked: ?*std.atomic.Value(bool) = null,
 aborted: ?*std.atomic.Value(bool) = null,
 cert_errors: ?*std.atomic.Value(bool) = null,
 upgraded: ?*std.atomic.Value(bool) = null,
@@ -12,13 +22,15 @@
 pub const Store = struct {
     header_progress: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     response_body_streaming: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
+    body_consumption_tracked: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     aborted: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     cert_errors: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     upgraded: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
     pub fn to(this: *Store) Signals {
         return .{
             .header_progress = &this.header_progress,
             .response_body_streaming = &this.response_body_streaming,
+            .body_consumption_tracked = &this.body_consumption_tracked,
             .aborted = &this.aborted,
             .cert_errors = &this.cert_errors,
             .upgraded = &this.upgraded,

src/http/h2_client/ClientSession.zig

@@ -348,6 +348,26 @@ pub fn drainResponseBodyByHttpId(this: *ClientSession, async_http_id: u32) void
     }
 }
 
+/// HTTP-thread wake-up from `scheduleResponseBodyConsumed`: the JS reader
+/// drained `bytes` from the ByteStream. Bump the stream's consumption
+/// counter and release any per-stream window credit that has become
+/// available.
+pub fn consumeResponseBodyByHttpId(this: *ClientSession, async_http_id: u32, bytes: u32) void {
+    this.ref();
+    defer this.deref();
+    for (this.streams.values()) |stream| {
+        const client = stream.client orelse continue;
+        if (client.async_http_id != async_http_id) continue;
+        // `bytes` is decompressed; clamp the running total to wire bytes
+        // still outstanding so a compression surplus isn't banked to
+        // credit later DATA the reader hasn't touched.
+        stream.consumed_bytes = @min(stream.consumed_bytes +| bytes, stream.unacked_bytes);
+        this.replenishWindow();
+        if (this.write_buffer.isNotEmpty()) _ = this.flush() catch |err| this.failAll(err);
+        return;
+    }
+}
+
 /// HTTP-thread wake-up from `scheduleRequestWrite`: new body bytes (or
 /// end-of-body) are available in the ThreadSafeStreamBuffer.
 pub fn streamBodyByHttpId(this: *ClientSession, async_http_id: u32, ended: bool) void {
@@ -372,16 +392,31 @@ pub fn writeWindowUpdate(this: *ClientSession, stream_id: u32, increment: u31) v
 
 fn replenishWindow(this: *ClientSession) void {
     const threshold = local_initial_window_size / 2;
+    // Connection-level credit stays receipt-based so one stream whose JS
+    // reader is stalled doesn't starve siblings of the shared window.
     if (this.conn_unacked_bytes >= threshold) {
         this.writeWindowUpdate(0, @intCast(this.conn_unacked_bytes));
         this.conn_unacked_bytes = 0;
     }
     var it = this.streams.iterator();
     while (it.next()) |e| {
         const s = e.value_ptr.*;
-        if (s.unacked_bytes >= threshold and !s.remoteClosed()) {
-            this.writeWindowUpdate(s.id, @intCast(s.unacked_bytes));
-            s.unacked_bytes = 0;
+        if (s.remoteClosed()) continue;
+        // `body_consumption_tracked` is set only while a JS consumer is
+        // reporting drained bytes via `scheduleResponseBodyConsumed`
+        // (fetch `res.body` with a `drain_handler` wired). It is *not*
+        // set for buffering consumers (`await res.text()`), S3 streaming
+        // downloads, or once the body is abandoned via
+        // `ignoreRemainingResponseBody` — those stay receipt-based so
+        // the transfer completes. When tracked, credit only what JS has
+        // actually drained, clamped to wire bytes received so a
+        // decompressed body can't inflate the window past what was sent.
+        const tracked = if (s.client) |c| c.signals.get(.body_consumption_tracked) else false;
+        const avail: u32 = if (tracked) @min(s.consumed_bytes, s.unacked_bytes) else s.unacked_bytes;
+        if (avail >= threshold) {
+            this.writeWindowUpdate(s.id, @intCast(avail));
+            s.unacked_bytes -= avail;
+            s.consumed_bytes -|= avail;
         }
     }
 }

src/http/h2_client/Stream.zig

@@ -35,8 +35,17 @@ headers_end_stream: bool = false,
 /// or final status arrives.
 awaiting_continue: bool = false,
 fatal_error: ?anyerror = null,
-/// DATA bytes consumed since the last WINDOW_UPDATE for this stream.
+/// DATA bytes received since the last per-stream WINDOW_UPDATE. For
+/// consumers without `body_consumption_tracked` set this alone drives the
+/// credit; for tracked consumers it is the ceiling on what
+/// `consumed_bytes` may release.
 unacked_bytes: u32 = 0,
+/// Bytes the JS `ReadableStream` reader has actually drained, reported via
+/// `scheduleResponseBodyConsumed`. Only consulted when
+/// `body_consumption_tracked` is true; `replenishWindow` credits
+/// `min(consumed_bytes, unacked_bytes)` so a stalled reader withholds the
+/// per-stream window and a compressed body can't over-credit.
+consumed_bytes: u32 = 0,
 /// Σ DATA payload bytes (post-padding) for §8.1.1 Content-Length check —
 /// `total_body_received` is clamped at content_length so it can't catch
 /// overshoot.

src/http/h3_client/ClientContext.zig

@@ -101,6 +101,13 @@ pub fn streamBodyByHttpId(async_http_id: u32, ended: bool) void {
     for (this.sessions.items) |s| s.streamBodyByHttpId(async_http_id, ended);
 }
 
+pub fn consumeResponseBodyByHttpId(async_http_id: u32, bytes: u32) void {
+    const this = instance orelse return;
+    for (this.sessions.items) |s| {
+        if (s.consumeResponseBodyByHttpId(async_http_id, bytes)) return;
+    }
+}
+
 const log = bun.Output.scoped(.h3_client, .hidden);
 
 const ClientSession = @import("./ClientSession.zig");

src/http/h3_client/ClientSession.zig

@@ -78,6 +78,30 @@ pub fn streamBodyByHttpId(this: *ClientSession, async_http_id: u32, ended: bool)
     }
 }
 
+/// HTTP-thread wake-up from `scheduleResponseBodyConsumed`: the JS reader
+/// drained `bytes` from the ByteStream. Decrement the outstanding count
+/// and, if the stream's `on_read` was paused for backpressure, re-enable
+/// it so lsquic resumes draining the QUIC receive buffer and issuing
+/// `MAX_STREAM_DATA` credit.
+pub fn consumeResponseBodyByHttpId(this: *ClientSession, async_http_id: u32, bytes: u32) bool {
+    for (this.pending.items) |stream| {
+        const client = stream.client orelse continue;
+        if (client.async_http_id != async_http_id) continue;
+        stream.outstanding_body_bytes -|= @min(@as(usize, bytes), stream.outstanding_body_bytes);
+        if (!stream.read_paused) return true;
+        const should_resume = stream.outstanding_body_bytes <= bun.http.receive_body_low_water or
+            !client.signals.get(.body_consumption_tracked);
+        if (!should_resume) return true;
+        stream.read_paused = false;
+        if (stream.qstream) |qs| qs.wantRead(true);
+        // lsquic's `process_conns` runs from the loop's post-tick hook,
+        // so the re-enabled `on_read` fires on the very next iteration
+        // (this handler runs from `drainEvents`, before that tick).
+        return true;
+    }
+    return false;
+}
+
 pub fn detach(this: *ClientSession, stream: *Stream) void {
     if (stream.client) |cl| cl.h3 = null;
     stream.client = null;

src/http/h3_client/Stream.zig

@@ -24,6 +24,12 @@ pending_body: []const u8 = "",
 request_body_done: bool = false,
 is_streaming_body: bool = false,
 headers_delivered: bool = false,
+/// Wire bytes delivered to JS via `deliver()` that haven't been reported
+/// drained via `scheduleResponseBodyConsumed`. Once over
+/// `receive_body_high_water` we stop `lsquic_stream_wantread` so lsquic
+/// withholds `MAX_STREAM_DATA` and the server backpressures.
+outstanding_body_bytes: usize = 0,
+read_paused: bool = false,
 
 pub fn deinit(this: *Stream) void {
     this.decoded_headers.deinit(bun.default_allocator);

src/http/h3_client/callbacks.zig

@@ -120,8 +120,26 @@ fn onStreamData(s: *quic.Stream, data: [*]const u8, len: c_uint, fin: c_int) cal
     const stream = s.ext(Stream).* orelse return;
     if (len > 0) {
         bun.handleOom(stream.body_buffer.appendSlice(bun.default_allocator, data[0..len]));
+        _ = H3.body_bytes_received.fetchAdd(len, .monotonic);
     }
     stream.session.deliver(stream, fin != 0);
+    // `deliver` may have detached (fin/error); re-resolve from the quic
+    // stream's ext slot before touching the Stream again.
+    const still = s.ext(Stream).* orelse return;
+    if (fin != 0 or still.read_paused) return;
+    const client = still.client orelse return;
+    // Only count bytes that arrived while a JS reader is wired to
+    // report consumption. Pre-armed bytes would otherwise become a
+    // permanent floor under `outstanding_body_bytes` (their `didDrain`
+    // credit saturates against nothing counted), and if that floor is
+    // above `receive_body_low_water` the first wantRead(false) pause
+    // never resumes. Matches h1's `maybePauseReceive` early-return.
+    if (!client.signals.get(.body_consumption_tracked)) return;
+    still.outstanding_body_bytes +|= len;
+    if (still.outstanding_body_bytes < bun.http.receive_body_high_water) return;
+    still.read_paused = true;
+    s.wantRead(false);
+    log("stream read paused at {d} bytes outstanding", .{still.outstanding_body_bytes});
 }
 
 fn onStreamWritable(s: *quic.Stream) callconv(.c) void {