quic: apply multiple fixes to flow control signaling

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode/Opus

Author: jasnell

lib/internal/blob.js

@@ -610,12 +610,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;
@@ -630,6 +646,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
   try {
     while (true) {
       const batch = [];
+      let batchBytes = 0;
       let blocked = false;
       let eos = false;
       let error = null;
@@ -658,8 +675,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,39 @@ class DataQueueImpl final : public DataQueue,
     backpressure_listeners_.erase(listener);
   }
 
+  // Both notifications can re-enter this DataQueue. A listener may, for
+  // instance, extend a QUIC flow control window, which flushes packets, which
+  // can call into JavaScript and end up destroying the stream that owns the
+  // listener -- dropping its reference to this queue and removing itself from
+  // backpressure_listeners_ while we are still iterating. Hold a reference so
+  // this instance cannot be freed underneath us, iterate over a snapshot so
+  // that mutation is safe, and re-check membership before each call so a
+  // listener removed earlier in the same notification is not invoked after
+  // the fact.
   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

@@ -316,6 +316,34 @@ class DefaultApplication final : public Session::Application {
                          void* stream_user_data) override {
     BaseObjectPtr<Stream> stream;
     if (stream_user_data == nullptr) {
+      // A locally-initiated stream can only ever come into existence because
+      // we created it, so a missing Stream means we already destroyed it.
+      // Data the peer had already put in flight must not resurrect it:
+      // re-creating it here would hand the application a bogus "incoming"
+      // stream for a stream it just destroyed, and would do so again for
+      // every frame still in flight.
+      //
+      // Discard the data instead, but return the connection-level flow
+      // control credit for it. ngtcp2 has delivered these bytes to us, so we
+      // own their credit; dropping them silently would shrink the session's
+      // shared receive window for good.
+      // Note the is_destroyed() check has to come first: a prior callback in
+      // this same ngtcp2 batch may have destroyed the session, and neither the
+      // ngtcp2 connection nor the flow control helpers below may be touched
+      // once that has happened.
+      if (!session().is_destroyed() &&
+          ngtcp2_conn_is_local_stream(session(), id)) {
+        Debug(&session(),
+              "Discarding %zu bytes for destroyed local stream %" PRIi64,
+              datalen,
+              id);
+        if (datalen > 0) {
+          Session::SendPendingDataScope send_scope(&session());
+          session().ExtendOffset(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]] {

src/quic/http3.cc

@@ -1083,6 +1083,14 @@ class Http3ApplicationImpl final : public Session::Application {
     if (auto stream = session->FindStream(id)) {
       return stream;
     }
+    // A locally-initiated stream can only exist because we created it, so if
+    // we have no record of it the application already destroyed it. Frames the
+    // peer had already put 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;
     }
@@ -1224,6 +1232,31 @@ class Http3ApplicationImpl final : public Session::Application {
       return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
     auto& session = app.session();
+
+    // If the application destroyed a request stream it initiated, DATA frames
+    // the peer had already sent can still arrive. Ignore that payload rather
+    // than resurrecting the stream or tearing down the connection, but return
+    // its connection-level flow control credit: nghttp3 hands DATA payload to
+    // us uncredited (it is excluded from the framing bytes credited by the
+    // caller), so dropping it silently would permanently shrink the session's
+    // shared receive window.
+    // The is_destroyed() check has to come first: an earlier nghttp3 callback
+    // in this same batch may have destroyed the session (for example because a
+    // JS callback threw), and neither the ngtcp2 connection nor the flow
+    // control helpers below may be touched afterwards.
+    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);
+      if (datalen > 0) {
+        Session::SendPendingDataScope send_scope(&session);
+        session.ExtendOffset(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

@@ -1499,12 +1499,33 @@ void Stream::EndWriting() {
   if (!is_pending()) session_->ResumeStream(id());
 }
 
+void Stream::ReturnFlowControlCredit(uint64_t amount, CreditScope scope) {
+  if (amount == 0) return;
+  // The stream may outlive a destroyed session (the JS side can still hold a
+  // reader over the inbound queue), in which case there is no window left to
+  // extend.
+  if (!session_ || session_->is_destroyed()) return;
+  // Extending a window queues MAX_STREAM_DATA / MAX_DATA frames. The scope
+  // ensures they get flushed to the peer. When we are inside an ngtcp2
+  // callback the flush is a no-op (can_send_packets() is false) and the
+  // frames go out with the next scheduled send instead.
+  Session::SendPendingDataScope send_scope(&session());
+  if (scope == CreditScope::STREAM_AND_CONNECTION && !is_pending()) {
+    session().Consume(id(), amount);
+  } else {
+    session().ExtendOffset(amount);
+  }
+}
+
+void Stream::CreditConsumedBytes(uint64_t amount) {
+  uncredited_bytes_ -= std::min(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() {
@@ -1517,16 +1538,23 @@ 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);
+  // The bytes did not make it into the queue (it is capped and this data
+  // would push it past the final size), so they will never reach a reader
+  // and EntryRead() will never fire for them. Return their credit here
+  // instead of leaking it.
+  CreditConsumedBytes(flushed);
 }
 
 int Stream::DoPull(bob::Next<ngtcp2_vec> next,
@@ -1652,6 +1680,16 @@ void Stream::Destroy(QuicError error) {
   // the ring buffer memory.
   recv_accumulator_.reset();
 
+  // Any data that was received but never consumed is still holding inbound
+  // flow control credit. Once the backpressure listener is detached below,
+  // EntryRead() will never fire for it again, so return that credit now.
+  // The stream-level window is irrelevant at this point (the stream is going
+  // away) but the connection-level window is shared by the whole session:
+  // leaking it here would permanently shrink the session's receive window
+  // and, over enough streams, deadlock the connection.
+  ReturnFlowControlCredit(uncredited_bytes_, CreditScope::CONNECTION_ONLY);
+  uncredited_bytes_ = 0;
+
   // 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.
@@ -1691,6 +1729,15 @@ 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();
+    // These bytes are being discarded, but ngtcp2 already charged them
+    // against both receive windows when it delivered them to us. Nothing
+    // downstream will ever consume them, so give the credit back now.
+    // This is reachable, for instance, when HTTP/3 replays DATA payload
+    // that it had buffered for QPACK head-of-line blocking after the
+    // readable side was already shut down.
+    if (len > 0) {
+      ReturnFlowControlCredit(len, CreditScope::STREAM_AND_CONNECTION);
+    }
     return;
   }
 
@@ -1699,6 +1746,11 @@ 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. The credit is returned
+  // incrementally as the JS consumer reads them (EntryRead), and any
+  // remainder is returned when the stream is destroyed.
+  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,31 @@ class Stream final : public AsyncWrap,
   // inbound DataQueue as a single right-sized entry.
   void FlushAccumulation();
 
+  // Which receive windows a flow control credit return applies to.
+  enum class CreditScope : uint8_t {
+    // Extend both the stream-level and the connection-level window. This is
+    // the normal case: the stream is still alive and the peer may send more
+    // data on it.
+    STREAM_AND_CONNECTION,
+    // Extend only the connection-level window. Used when the stream is going
+    // away or has no id yet, where a MAX_STREAM_DATA would be pointless or
+    // impossible, but the connection-level window is shared by the whole
+    // session and must never be leaked.
+    CONNECTION_ONLY,
+  };
+
+  // Returns `amount` bytes of inbound flow control credit to the peer.
+  // Every byte that ngtcp2 delivers to us is charged against both the
+  // stream-level and the connection-level receive windows, and it is the
+  // application's responsibility to give that credit back once those bytes
+  // have either been consumed or discarded.
+  void ReturnFlowControlCredit(uint64_t amount, CreditScope scope);
+
+  // Drops `amount` bytes from uncredited_bytes_ (saturating at zero) and
+  // returns their credit to both receive windows. Used when bytes leave 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();
 
@@ -458,6 +483,15 @@ class Stream final : public AsyncWrap,
   BaseObjectWeakPtr<Blob::Reader> reader_;
   std::unique_ptr<RecvAccumulator> recv_accumulator_;
 
+  // Number of bytes delivered to ReceiveData() that have not yet been handed
+  // to the JavaScript consumer and so still hold inbound flow control credit.
+  // Any remainder is returned to the connection-level window 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 here: 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.

test/common/quic.mjs

@@ -53,9 +53,56 @@ async function connect(address, options = {}) {
   return quic.connect(address, { alpn, verifyPeer, ...rest });
 }
 
+/**
+ * Build a deterministic payload whose content depends on absolute position.
+ *
+ * Flow control bugs frequently show up as duplicated, dropped, or reordered
+ * regions rather than as a wrong total length, so the pattern deliberately
+ * varies over a long period (not a repeating 256-byte ramp) to make such
+ * damage detectable by `hashBytes` below.
+ * @param {number} size Number of bytes to generate.
+ * @param {number} [seed] Offsets the pattern so callers can build distinct
+ *   payloads of the same length.
+ * @returns {Uint8Array}
+ */
+function makePayload(size, seed = 0) {
+  const out = new Uint8Array(size);
+  let state = (seed * 2654435761 + 1) >>> 0;
+  for (let i = 0; i < size; i++) {
+    // xorshift32 -- cheap, deterministic, and position sensitive.
+    state ^= state << 13; state >>>= 0;
+    state ^= state >>> 17;
+    state ^= state << 5; state >>>= 0;
+    out[i] = state & 0xff;
+  }
+  return out;
+}
+
+/**
+ * Order-sensitive FNV-1a 32-bit hash.
+ *
+ * Note this is deliberately not a simple additive checksum: addition is
+ * commutative, so it cannot distinguish correctly ordered data from
+ * reordered data. Flow control errors can reorder or duplicate regions
+ * while preserving the byte total, so verification needs to be sensitive to
+ * position.
+ * @param {Uint8Array} buf
+ * @returns {number} Hash as an unsigned 32-bit integer.
+ */
+function hashBytes(buf) {
+  let h = 0x811c9dc5;
+  for (let i = 0; i < buf.byteLength; i++) {
+    h ^= buf[i];
+    h = Math.imul(h, 0x01000193) >>> 0;
+  }
+  return h >>> 0;
+}
+
 export {
   key,
   cert,
   listen,
   connect,
+  makePayload,
+  hashBytes,
 };