quic: apply multiple fixes to flow control signaling

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode/Opus
PR-URL: #65309
Reviewed-By: Tim Perry <pimterry@gmail.com>
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>

Author: jasnell

lib/internal/blob.js

@@ -604,12 +604,28 @@ function createBlobReaderStream(reader) {
   }, { highWaterMark: 0 });
 }
 
-// Maximum number of chunks to collect in a single batch to prevent
-// unbounded memory growth when the DataQueue has a large burst of data.
+// Upper bound on the number of chunks collected in a single batch. This is
+// only a cap on the length of the yielded array -- the primary limit is the
+// byte budget below, since under a byte-budget backpressure model the size of
+// a batch is what matters, not how many pieces it arrives in.
 const kMaxBatchChunks = 16;
 
+// Default number of bytes to collect in a single batch. Entries in the
+// DataQueue can each be as large as the peer's flow control window, so a
+// purely count-based limit could produce enormous batches (16 entries of
+// 1 MB each).
+//
+// This matters for more than just the size of the yielded array. Consumers
+// like QUIC return flow control credit from the reader's pull path -- once
+// per pull, not once per batch -- so every pull this loop performs invites
+// the peer to send that many more bytes. Pulling greedily therefore grants
+// credit for data the consumer has not looked at yet. Bounding the loop by
+// bytes limits how far ahead of actual consumption that credit can run,
+// which is what keeps the amount of data buffered in JS bounded.
+const kDefaultMaxBatchBytes = 65536;
+
 async function* createBlobReaderIterable(reader, options = kEmptyObject) {
-  const { getReadError } = options;
+  const { getReadError, maxBatchBytes = kDefaultMaxBatchBytes } = options;
   let wakeup = PromiseWithResolvers();
   let immediate;
   let fin = false;
@@ -624,6 +640,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
   try {
     while (true) {
       const batch = [];
+      let batchBytes = 0;
       let blocked = false;
       let eos = false;
       let error = null;
@@ -652,8 +669,15 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
           blocked = true;
           break;
         }
-        ArrayPrototypePush(batch, new Uint8Array(pullResult.buffer));
-        if (batch.length >= kMaxBatchChunks) break;
+        const chunk = new Uint8Array(pullResult.buffer);
+        ArrayPrototypePush(batch, chunk);
+        // Stop collecting once the batch is large enough. The byte budget is
+        // the primary limit; the chunk count is a secondary bound so that a
+        // long run of tiny chunks cannot produce an unwieldy array.
+        batchBytes += chunk.byteLength;
+        if (batchBytes >= maxBatchBytes || batch.length >= kMaxBatchChunks) {
+          break;
+        }
       }
 
       if (batch.length > 0) {

src/dataqueue/queue.cc

@@ -174,14 +174,35 @@ class DataQueueImpl final : public DataQueue,
     backpressure_listeners_.erase(listener);
   }
 
+  // Both notifications below can re-enter this DataQueue: a listener may end
+  // up calling into JavaScript, which can destroy the owner of a listener and
+  // so mutate backpressure_listeners_ (or drop the last reference to this
+  // queue) while we are iterating. Hold a reference, iterate a snapshot, and
+  // re-check membership so a listener removed mid-notification is not called.
   void NotifyBackpressure(size_t amount) {
     if (idempotent_) return;
-    for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
+    if (backpressure_listeners_.empty()) return;
+    auto self = shared_from_this();
+    std::vector<BackpressureListener*> listeners(
+        backpressure_listeners_.begin(), backpressure_listeners_.end());
+    for (auto* listener : listeners) {
+      if (backpressure_listeners_.contains(listener)) {
+        listener->EntryRead(amount);
+      }
+    }
   }
 
   void NotifyBeforePull() {
     if (idempotent_) return;
-    for (auto& listener : backpressure_listeners_) listener->BeforePull();
+    if (backpressure_listeners_.empty()) return;
+    auto self = shared_from_this();
+    std::vector<BackpressureListener*> listeners(
+        backpressure_listeners_.begin(), backpressure_listeners_.end());
+    for (auto* listener : listeners) {
+      if (backpressure_listeners_.contains(listener)) {
+        listener->BeforePull();
+      }
+    }
   }
 
   bool HasBackpressureListeners() const noexcept {

src/quic/application.cc

@@ -254,6 +254,12 @@ void Session::Application::ReceiveStreamReset(Stream* stream,
   stream->ReceiveStreamReset(final_size, std::move(error));
 }
 
+void Session::Application::ReturnConnectionCredit(size_t datalen) {
+  if (datalen == 0 || session().is_destroyed()) return;
+  Session::SendPendingDataScope send_scope(&session());
+  session().ExtendOffset(datalen);
+}
+
 // ============================================================================
 // The DefaultApplication is the default implementation of Session::Application
 // that is used for all unrecognized ALPN identifiers.
@@ -316,6 +322,22 @@ class DefaultApplication final : public Session::Application {
                          void* stream_user_data) override {
     BaseObjectPtr<Stream> stream;
     if (stream_user_data == nullptr) {
+      // A locally-initiated stream only exists because we created it, so a
+      // missing Stream means we already destroyed it. Data the peer had put in
+      // flight must not resurrect it as a bogus "incoming" stream. Discard it
+      // and return its credit instead. The is_destroyed() check must come
+      // first: an earlier callback in this same ngtcp2 batch may have
+      // destroyed the session, after which none of this may be touched.
+      if (!session().is_destroyed() &&
+          ngtcp2_conn_is_local_stream(session(), id)) {
+        Debug(&session(),
+              "Discarding %zu bytes for destroyed local stream %" PRIi64,
+              datalen,
+              id);
+        ReturnConnectionCredit(datalen);
+        return true;
+      }
+
       // This is the first time we're seeing this stream. Implicitly create it.
       stream = session().CreateStream(id);
       if (!stream || session().is_destroyed()) [[unlikely]] {
@@ -324,9 +346,11 @@ class DefaultApplication final : public Session::Application {
         return false;
       }
 
-      // The stream was created, but was immediately destroyed because there's
-      // no onstream handler.
+      // The stream was created but immediately destroyed, either because there
+      // is no onstream handler or because the handler destroyed it. Nothing
+      // will consume the data, so discard it and return its credit.
       if (stream->is_destroyed()) [[unlikely]] {
+        ReturnConnectionCredit(datalen);
         return true;
       }
     } else {

src/quic/application.h

@@ -98,9 +98,11 @@ class Session::Application : public MemoryRetainer {
   virtual bool ReceiveStreamOpen(stream_id id) = 0;
 
   // Session will forward all received stream data immediately on to the
-  // Application. The only additional processing the Session does is to
-  // automatically adjust the session-level flow control window. It is up to
-  // the Application to do the same for the Stream-level flow control.
+  // Application without any additional processing. Every byte delivered here
+  // is charged against both the session-level and the stream-level receive
+  // window, and it is up to the Application to return that credit (see
+  // ReturnConnectionCredit and Stream::ReturnFlowControlCredit) once the
+  // bytes have been consumed or discarded.
   virtual bool ReceiveStreamData(stream_id id,
                                  const uint8_t* data,
                                  size_t datalen,
@@ -266,6 +268,12 @@ class Session::Application : public MemoryRetainer {
     return *session_;
   }
 
+  // Returns the connection-level flow control credit for `datalen` bytes that
+  // were delivered to the Application but discarded without ever reaching a
+  // Stream. Dropping them silently would permanently shrink the session's
+  // shared receive window.
+  void ReturnConnectionCredit(size_t datalen);
+
  private:
   Session* session_ = nullptr;
 };

src/quic/http3.cc

@@ -1091,6 +1091,13 @@ class Http3ApplicationImpl final : public Session::Application {
     if (auto stream = session->FindStream(id)) {
       return stream;
     }
+    // No record of a locally-initiated stream means we already destroyed it,
+    // and frames still in flight must not bring it back to life. See
+    // DefaultApplication::ReceiveStreamData for the same guard on the raw
+    // QUIC path.
+    if (!session->is_destroyed() && ngtcp2_conn_is_local_stream(*session, id)) {
+      return {};
+    }
     if (auto stream = session->CreateStream(id)) {
       return stream;
     }
@@ -1232,6 +1239,23 @@ class Http3ApplicationImpl final : public Session::Application {
       return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
     auto& session = app.session();
+
+    // DATA frames for a request stream the application already destroyed can
+    // still arrive. Drop the payload rather than resurrecting the stream or
+    // tearing down the connection, but return its credit: unlike framing
+    // bytes, DATA payload is not included in the count nghttp3 reports to
+    // ReceiveStreamData, so we own it. The is_destroyed() check must come
+    // first, see DefaultApplication::ReceiveStreamData.
+    if (!session.is_destroyed() && !session.FindStream(id) &&
+        ngtcp2_conn_is_local_stream(session, id)) {
+      Debug(&session,
+            "HTTP/3 discarding %zu bytes for destroyed local stream %" PRIi64,
+            datalen,
+            id);
+      app.ReturnConnectionCredit(datalen);
+      return NGTCP2_SUCCESS;
+    }
+
     if (auto stream = FindOrCreateStream(conn, &session, id)) [[likely]] {
       stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
       return NGTCP2_SUCCESS;

src/quic/streams.cc

@@ -1498,12 +1498,37 @@ void Stream::EndWriting() {
   if (!is_pending()) session_->ResumeStream(id());
 }
 
+void Stream::ReturnFlowControlCredit(uint64_t amount, CreditScope scope) {
+  if (amount == 0) return;
+  // The stream can outlive a destroyed session (the JS side may still hold a
+  // reader over the inbound queue), leaving no window to extend.
+  if (!session_ || session_->is_destroyed()) return;
+  // Extending a window queues MAX_STREAM_DATA / MAX_DATA. The scope flushes
+  // them; inside an ngtcp2 callback the flush is a no-op and they go out with
+  // the next scheduled send instead.
+  Session::SendPendingDataScope send_scope(&session());
+  if (scope == CreditScope::STREAM_AND_CONNECTION) {
+    // Receiving data requires an id, so this should always hold.
+    DCHECK(!is_pending());
+    session().Consume(id(), amount);
+  } else {
+    session().ExtendOffset(amount);
+  }
+}
+
+void Stream::CreditConsumedBytes(uint64_t amount) {
+  // Clamped because Destroy() returns the outstanding credit in bulk and the
+  // flush that triggers can re-enter JS, which may then report some of those
+  // same bytes as read. Never give the peer more credit than we took.
+  amount = std::min(uncredited_bytes_, amount);
+  uncredited_bytes_ -= amount;
+  ReturnFlowControlCredit(amount, CreditScope::STREAM_AND_CONNECTION);
+}
+
 void Stream::EntryRead(size_t amount) {
   // Called when the JS consumer reads data from the inbound DataQueue.
   // Extend the flow control window so the sender can transmit more.
-  if (session().is_destroyed()) return;
-  Session::SendPendingDataScope send_scope(&session());
-  session().Consume(id(), amount);
+  CreditConsumedBytes(amount);
 }
 
 void Stream::BeforePull() {
@@ -1516,16 +1541,26 @@ void Stream::BeforePull() {
 
 void Stream::FlushAccumulation() {
   if (!recv_accumulator_ || recv_accumulator_->available() == 0) return;
+  size_t flushed = recv_accumulator_->available();
   auto entry = recv_accumulator_->Flush(env());
-  if (entry) {
-    inbound_->append(std::move(entry));
+  // Flush() always drains the accumulator, so the stat is reset either way.
+  STAT_SET(Stats, bytes_accumulated, 0);
+  if (entry && inbound_->append(std::move(entry)).value_or(false)) {
     // Notify the reader that data is now available in the DataQueue.
     // This is the only place we notify — not on every ReceiveData call —
     // so the reader only wakes up when there is a well-sized entry to
     // consume.
     if (reader_) reader_->NotifyPull();
+    return;
   }
-  STAT_SET(Stats, bytes_accumulated, 0);
+  // Should be unreachable: append() only fails once the queue has been capped,
+  // EndReadable() flushes before capping, and ReceiveData() accumulates
+  // nothing once read_ended is set. Reaching here means received stream data
+  // is being dropped on the floor, so say so and at least do not also leak
+  // the flow control credit for it.
+  DCHECK(false);
+  Debug(this, "Inbound queue rejected %zu accumulated bytes", flushed);
+  CreditConsumedBytes(flushed);
 }
 
 int Stream::DoPull(bob::Next<ngtcp2_vec> next,
@@ -1651,6 +1686,16 @@ void Stream::Destroy(QuicError error) {
   // the ring buffer memory.
   recv_accumulator_.reset();
 
+  // Data that was received but never consumed still holds connection-level
+  // flow control credit, and EntryRead() will never fire for it once the
+  // listener is detached below. Leaking it would permanently shrink the
+  // session's shared receive window and, over enough streams, deadlock the
+  // connection. Zero the counter first: returning credit flushes packets,
+  // which can re-enter JS and report some of these bytes as read.
+  const uint64_t outstanding = uncredited_bytes_;
+  uncredited_bytes_ = 0;
+  ReturnFlowControlCredit(outstanding, CreditScope::CONNECTION_ONLY);
+
   // We reset the inbound here also. However, it's important to note that
   // the JavaScript side could still have a reader on the inbound DataQueue,
   // which may keep that data alive a bit longer.
@@ -1693,6 +1738,13 @@ void Stream::ReceiveData(const uint8_t* data,
   Debug(this, "Receiving %zu bytes of data", len);
   if (state()->read_ended == 1 || len == 0) {
     if (flags.fin) EndReadable();
+    // Nothing will ever consume these bytes, so return the connection-level
+    // credit ngtcp2 charged for them. The stream window is deliberately left
+    // alone: there is no point inviting more data onto a stream we have
+    // stopped reading. Reachable when, for example, HTTP/3 replays DATA
+    // payload it had buffered for QPACK head-of-line blocking after the
+    // readable side was shut down.
+    ReturnFlowControlCredit(len, CreditScope::CONNECTION_ONLY);
     return;
   }
 
@@ -1701,6 +1753,9 @@ void Stream::ReceiveData(const uint8_t* data,
   STAT_SET(Stats, max_offset_received, STAT_GET(Stats, bytes_received));
   STAT_RECORD_TIMESTAMP(Stats, received_at);
 
+  // These bytes now hold inbound flow control credit. See uncredited_bytes_.
+  uncredited_bytes_ += len;
+
   // Lazy-allocate the receive accumulation buffer on first data-carrying
   // call. Streams that never receive data (write-only, immediately reset)
   // pay zero cost.

src/quic/streams.h

@@ -395,6 +395,22 @@ class Stream final : public AsyncWrap,
   // inbound DataQueue as a single right-sized entry.
   void FlushAccumulation();
 
+  // Every byte ngtcp2 delivers is charged against both the stream-level and
+  // the connection-level receive window until we hand the credit back.
+  enum class CreditScope : uint8_t {
+    // The stream is still readable, so the peer may usefully send more on it.
+    STREAM_AND_CONNECTION,
+    // The stream is finished, so only the session-wide window is extended.
+    CONNECTION_ONLY,
+  };
+
+  // Returns `amount` bytes of inbound flow control credit to the peer.
+  void ReturnFlowControlCredit(uint64_t amount, CreditScope scope);
+
+  // Returns credit for bytes that have left our custody, either read by the
+  // consumer or dropped before reaching one.
+  void CreditConsumedBytes(uint64_t amount);
+
   // Gets a reader for the data received for this stream from the peer,
   BaseObjectPtr<Blob::Reader> get_reader();
 
@@ -459,6 +475,14 @@ class Stream final : public AsyncWrap,
   BaseObjectWeakPtr<Blob::Reader> reader_;
   std::unique_ptr<RecvAccumulator> recv_accumulator_;
 
+  // Bytes delivered to ReceiveData() that still hold inbound flow control
+  // credit. Returned incrementally as the consumer reads them, and in bulk
+  // when the stream is destroyed -- otherwise abandoning a stream with unread
+  // data would permanently shrink the session's receive window. Data still
+  // buffered inside nghttp3 is deliberately not counted: nghttp3 returns that
+  // credit itself through its deferred_consume callback.
+  uint64_t uncredited_bytes_ = 0;
+
   // If the stream cannot be opened yet, it will be created in a pending state.
   // Once the owning session is able to, it will complete opening of the stream
   // and the stream id will be assigned.