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

src/bun.js/webcore/Body.zig

@@ -76,6 +76,7 @@
     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 @@
                         reader.cancel_ctx = task;
                     }
                 }
+                if (locked.onStreamConsumed) |onConsumed| {
+                    if (locked.task) |task| {
+                        reader.drain_handler = onConsumed;
+                        reader.drain_ctx = task;
+                    }
+                }
 
                 reader.context.setup();
 

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,

src/bun.js/webcore/ReadableStream.zig

@@ -432,6 +432,11 @@ 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
+        /// couple HTTP/2 per-stream WINDOW_UPDATE to actual body consumption.
+        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

@@ -925,6 +925,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 +937,17 @@ pub const FetchTasklet = struct {
         this.ignoreRemainingResponseBody();
     }
 
+    /// ByteStream delivered `bytes` to the JS reader. Forward to the HTTP
+    /// thread so the HTTP/2 client can release per-stream flow-control
+    /// window proportional to what JS has actually drained; no-op for
+    /// HTTP/1.1 sockets.
+    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 +961,7 @@ pub const FetchTasklet = struct {
                     .onStartStreaming = FetchTasklet.onStartStreamingHTTPResponseBodyCallback,
                     .onReadableStreamAvailable = FetchTasklet.onReadableStreamAvailable,
                     .onStreamCancelled = FetchTasklet.onStreamCancelledCallback,
+                    .onStreamConsumed = FetchTasklet.onStreamConsumedCallback,
                 },
             };
             return response;

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,
 };
@@ -510,9 +516,45 @@ 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);
+                        // Only HTTP/2 has per-stream receive windows; for
+                        // HTTP/1.1 the tagged ptr is an HTTPClient and the
+                        // message is a no-op.
+                        if (tagged.get(bun.http.H2.ClientSession)) |session| {
+                            session.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();
@@ -664,6 +706,30 @@ 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;
+    {
+        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;
+        } else {
+            this.queued_response_body_consumed.append(bun.default_allocator, .{
+                .async_http_id = async_http_id,
+                .bytes = n,
+            }) catch |err| bun.handleOom(err);
+        }
+    }
+    if (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/h2_client/ClientSession.zig

@@ -348,6 +348,26 @@
     }
 }
 
+/// 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,17 +392,29 @@
 
 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;
+        // Streaming consumer (`res.body.getReader()`): 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.
+        // Buffering consumers (`await res.text()` etc.) keep receipt-based
+        // crediting — the whole body is going into memory regardless, so
+        // withholding the window just slows the transfer.
+        const streaming = if (s.client) |c| c.signals.get(.response_body_streaming) else false;
+        const avail: u32 = if (streaming) @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
+/// buffering consumers (`response_body_streaming` false) this alone drives
+/// the credit; for streaming 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
+/// `response_body_streaming` 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.