quic: apply multiple fixes to flow control signaling

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

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,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.

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,
 };

test/parallel/test-quic-flow-control-bidi-simultaneous-volume.mjs

@@ -0,0 +1,102 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: large simultaneous transfers in both directions on one stream,
+// through constrained windows in both directions.
+//
+// Each direction of a bidirectional stream has its own independent flow
+// control window, governed by different transport parameters:
+//
+//   client -> server  is limited by the server's
+//                     initialMaxStreamDataBidiRemote
+//   server -> client  is limited by the client's
+//                     initialMaxStreamDataBidiLocal
+//
+// Running both directions at volume at the same time means credit is being
+// consumed and returned in both directions concurrently on the same stream.
+// A bug that credits the wrong direction, or that lets one direction's
+// accounting interfere with the other's, shows up here but not in the
+// one-direction-at-a-time tests.
+//
+// This exercises the raw QUIC application: the test helpers negotiate the
+// 'quic-test' ALPN, not HTTP/3. See test-quic-h3-flow-control-volume.mjs for
+// the HTTP/3 equivalent.
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect, makePayload, hashBytes } =
+  await import('../common/quic.mjs');
+const { bytes } = await import('stream/iter');
+
+const kTotal = 4 * 1024 * 1024;         // per direction
+const kStreamWindow = 16 * 1024;
+const kConnWindow = 32 * 1024;
+
+assert.ok(kTotal / kConnWindow >= 100,
+          'payload must require >=100 connection window refills');
+
+// Distinct payloads per direction so a direction mixup is detectable.
+const toServer = makePayload(kTotal, 11);
+const toClient = makePayload(kTotal, 22);
+const toServerHash = hashBytes(toServer);
+const toClientHash = hashBytes(toClient);
+
+const serverDone = Promise.withResolvers();
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    // Start sending before reading, so both directions are in flight at
+    // once. If the server read to completion first, the two directions
+    // would be sequential and the test would prove much less.
+    stream.setBody(toClient);
+
+    const received = await bytes(stream);
+    assert.strictEqual(received.byteLength, kTotal);
+    // Must be the client->server payload, not an echo of our own.
+    assert.strictEqual(hashBytes(received), toServerHash);
+
+    await stream.closed;
+    serverSession.close();
+    serverDone.resolve();
+  });
+}), {
+  transportParams: {
+    // Limits client -> server.
+    initialMaxStreamDataBidiRemote: kStreamWindow,
+    initialMaxData: kConnWindow,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+});
+
+const clientSession = await connect(serverEndpoint.address, {
+  transportParams: {
+    // Limits server -> client.
+    initialMaxStreamDataBidiLocal: kStreamWindow,
+    initialMaxData: kConnWindow,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+});
+await clientSession.opened;
+
+const stream = await clientSession.createBidirectionalStream();
+stream.setBody(toServer);
+
+// Read the server's payload while our own body is still being sent.
+const received = await bytes(stream);
+assert.strictEqual(received.byteLength, kTotal);
+// Must be the server->client payload, not an echo of our own.
+assert.strictEqual(hashBytes(received), toClientHash);
+
+// Sanity: the two directions carried genuinely different data, so the
+// assertions above could not both be satisfied by one payload echoed back.
+assert.notStrictEqual(toServerHash, toClientHash);
+
+await Promise.all([stream.closed, serverDone.promise]);
+await clientSession.close();
+await serverEndpoint.close();

test/parallel/test-quic-flow-control-concurrent-volume.mjs

@@ -0,0 +1,130 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: many concurrent streams contending for one constrained
+// connection-level flow control window, at volume.
+//
+// The connection-level window (initialMaxData) is shared by every stream on
+// the session, while each stream also has its own window. When several
+// streams are all trying to move data through a connection window smaller
+// than any single stream's payload, the connection-level credit has to be
+// recycled continuously and shared across them.
+//
+// This is the configuration most likely to expose credit accounting errors:
+// per-stream bugs that cancel out on a single stream become visible when
+// several streams draw on the same pool, and any net under-crediting
+// deadlocks every stream at once rather than just slowing one down.
+//
+// This exercises the raw QUIC application: the test helpers negotiate the
+// 'quic-test' ALPN, not HTTP/3. See test-quic-h3-flow-control-volume.mjs for
+// the HTTP/3 equivalent.
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect, makePayload, hashBytes } =
+  await import('../common/quic.mjs');
+const { bytes } = await import('stream/iter');
+
+const kStreams = 8;
+const kPerStream = 512 * 1024;          // 4 MB total
+const kStreamWindow = 16 * 1024;
+const kConnWindow = 32 * 1024;          // Shared by all 8 streams
+
+const kTotal = kStreams * kPerStream;
+
+assert.ok(kConnWindow < kPerStream,
+          'connection window must be smaller than a single stream payload');
+assert.ok(kTotal / kConnWindow >= 100,
+          'aggregate payload must require >=100 connection window refills');
+
+// Distinct payload per stream so a cross-stream mixup is detectable, not
+// just a wrong byte total.
+const payloads = [];
+const hashes = [];
+for (let i = 0; i < kStreams; i++) {
+  const p = makePayload(kPerStream, i + 1);
+  payloads.push(p);
+  hashes.push(hashBytes(p));
+}
+
+// Map each received payload back to the stream that sent it. The server has
+// no ordering guarantee across streams, so match by hash rather than by
+// arrival order.
+const remaining = new Set(hashes);
+const serverDone = Promise.withResolvers();
+let completed = 0;
+
+// Track how many streams are being read at the same time. The whole premise
+// of this test is contention for the shared connection window, so if the
+// streams end up serialized the test still passes but proves much less.
+// Assert the overlap explicitly rather than trusting it.
+let openStreams = 0;
+let maxOpenStreams = 0;
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    openStreams++;
+    maxOpenStreams = Math.max(maxOpenStreams, openStreams);
+
+    const received = await bytes(stream);
+    openStreams--;
+    assert.strictEqual(received.byteLength, kPerStream);
+
+    const h = hashBytes(received);
+    // Each payload must arrive exactly once, intact. This catches data from
+    // one stream being credited or delivered onto another.
+    assert.ok(remaining.has(h),
+              'received payload did not match any expected stream payload');
+    remaining.delete(h);
+
+    stream.writer.endSync();
+    await stream.closed;
+    if (++completed === kStreams) {
+      serverSession.close();
+      serverDone.resolve();
+    }
+  }, kStreams);
+}), {
+  transportParams: {
+    initialMaxStreamDataBidiRemote: kStreamWindow,
+    initialMaxData: kConnWindow,
+    // Allow all streams to be open at once; the point is contention.
+    initialMaxStreamsBidi: kStreams,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+});
+
+const clientSession = await connect(serverEndpoint.address);
+await clientSession.opened;
+
+// Start every stream before draining any of them, so they genuinely contend
+// for the shared connection window rather than running one after another.
+const streams = [];
+for (let i = 0; i < kStreams; i++) {
+  const stream = await clientSession.createBidirectionalStream();
+  stream.setBody(payloads[i]);
+  streams.push(stream);
+}
+
+await Promise.all(streams.map(async (stream) => {
+  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await stream.closed;
+}));
+
+await serverDone.promise;
+// Every stream's payload must have arrived exactly once.
+assert.strictEqual(remaining.size, 0);
+// All streams are created before any is drained, so all of them should be
+// open simultaneously. Anything less means they serialized and the window
+// contention this test exists to exercise did not actually happen.
+assert.strictEqual(maxOpenStreams, kStreams,
+                   'all streams must be open simultaneously to contend for ' +
+                   'the shared connection window');
+
+await clientSession.close();
+await serverEndpoint.close();

test/parallel/test-quic-flow-control-credit-reclaim.mjs

@@ -0,0 +1,98 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: connection-level flow control credit is reclaimed for inbound data
+// that is never consumed.
+//
+// Every byte QUIC delivers to us is charged against both the stream-level
+// and the connection-level receive window. The stream-level window dies with
+// the stream, but the connection-level window (`initialMaxData`) is shared by
+// every stream on the session and is only replenished when we send MAX_DATA.
+//
+// If a stream is destroyed while inbound data is still sitting unread, that
+// credit must still be returned, otherwise the session's receive window
+// shrinks a little on every such stream until the connection deadlocks.
+//
+// This test destroys many streams without reading their responses, moving far
+// more data in total than `initialMaxData` allows to be outstanding at once.
+// It only completes if the credit is reclaimed on destroy.
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect } = await import('../common/quic.mjs');
+
+const encoder = new TextEncoder();
+
+// Each response is 16 KB and the connection-level window is 64 KB, so the
+// session can only ever have 4 unread responses outstanding. Running 20
+// streams pushes 320 KB total -- 5x the window -- which can only work if the
+// window is replenished as each stream is discarded.
+const kPayloadSize = 16 * 1024;
+const kInitialMaxData = 64 * 1024;
+const kStreamCount = 20;
+
+const payload = encoder.encode('a'.repeat(kPayloadSize));
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  // One response per client stream, plus the final health-check stream.
+  serverSession.onstream = mustCall((stream) => {
+    // Respond and immediately finish. We never read the request body.
+    stream.setBody(payload);
+  }, kStreamCount + 1);
+}));
+
+const clientSession = await connect(serverEndpoint.address, {
+  transportParams: {
+    // Connection-level receive window, shared by all streams.
+    initialMaxData: kInitialMaxData,
+    // Give each individual stream enough room for a whole response so that
+    // the connection-level window is the only thing that can throttle us.
+    initialMaxStreamDataBidiLocal: kPayloadSize * 2,
+  },
+});
+await clientSession.opened;
+
+for (let i = 0; i < kStreamCount; i++) {
+  const stream = await clientSession.createBidirectionalStream({
+    body: encoder.encode('request'),
+  });
+
+  // Wait until the whole response has actually been received, so that the
+  // bytes are genuinely holding flow control credit at the point we destroy
+  // the stream. If credit is leaked, this loop is where later iterations
+  // stall: the server runs out of connection-level window and can no longer
+  // send, so bytesReceived never reaches the payload size.
+  while (stream.stats.bytesReceived < kPayloadSize) {
+    await new Promise((resolve) => setTimeout(resolve, 5));
+  }
+
+  // Destroy with the response still buffered and unread. This is the case
+  // that used to leak: the backpressure listener is detached, so nothing
+  // would ever return the credit for these bytes.
+  stream.destroy();
+}
+
+// Getting here at all is the assertion -- a leak manifests as a timeout
+// above. Verify the session is still healthy and able to move data, which
+// confirms the window was replenished rather than merely limping along.
+{
+  const stream = await clientSession.createBidirectionalStream({
+    body: encoder.encode('request'),
+  });
+  let received = 0;
+  for await (const chunks of stream) {
+    for (const chunk of chunks) received += chunk.byteLength;
+  }
+  assert.strictEqual(received, kPayloadSize);
+}
+
+await clientSession.close();
+await serverEndpoint.close();
+
+// Sanity check that the test actually exercised the intended path.
+assert.ok(kPayloadSize * kStreamCount > kInitialMaxData,
+          'test must move more data than the connection window allows');

test/parallel/test-quic-flow-control-large-constrained.mjs

@@ -0,0 +1,121 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: high volume transfer through a deliberately constrained flow
+// control window.
+//
+// The existing flow control tests either move a small amount of data through
+// a tiny window, or a large amount of data through the default (large)
+// window. Neither combination stresses the credit accounting: a bug that
+// mis-credits a few bytes per window extension is invisible over 8KB but
+// fatal over several megabytes.
+//
+// Here the payload is orders of magnitude larger than both the stream-level
+// and connection-level receive windows, so completing the transfer requires
+// hundreds of MAX_STREAM_DATA / MAX_DATA extensions. If any extension
+// under-credits, the transfer stalls and the test times out; if any
+// over-credits or misorders data, the hash check fails.
+
+import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect, makePayload, hashBytes } =
+  await import('../common/quic.mjs');
+const { bytes } = await import('stream/iter');
+
+// 4 MB through a 16 KB stream window and a 32 KB connection window. The
+// windows are capped via maxStreamWindow/maxWindow as well, otherwise
+// ngtcp2's window auto-tuning grows them and the transfer stops being
+// flow-control bound.
+const kTotal = 4 * 1024 * 1024;
+const kStreamWindow = 16 * 1024;
+const kConnWindow = 32 * 1024;
+
+// Guard: the point of this test is that the window has to be recycled many
+// times over. If someone lowers kTotal or raises the windows to speed the
+// test up, this makes the loss of coverage explicit rather than silent.
+assert.ok(kTotal / kConnWindow >= 100,
+          'payload must require >=100 connection window refills');
+assert.ok(kTotal / kStreamWindow >= 100,
+          'payload must require >=100 stream window refills');
+
+const transportParams = {
+  initialMaxStreamDataBidiRemote: kStreamWindow,
+  initialMaxStreamDataUni: kStreamWindow,
+  initialMaxData: kConnWindow,
+};
+const windowCaps = {
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+};
+
+// Bidirectional: client sends a large body, server verifies it.
+{
+  const payload = makePayload(kTotal);
+  const expectedHash = hashBytes(payload);
+  const serverDone = Promise.withResolvers();
+
+  const serverEndpoint = await listen(mustCall((serverSession) => {
+    serverSession.onstream = mustCall(async (stream) => {
+      const received = await bytes(stream);
+      assert.strictEqual(received.byteLength, kTotal);
+      // Order-sensitive hash: catches duplicated, dropped, or reordered
+      // regions that preserve the total length.
+      assert.strictEqual(hashBytes(received), expectedHash);
+      stream.writer.endSync();
+      await stream.closed;
+      serverSession.close();
+      serverDone.resolve();
+    });
+  }), { transportParams, ...windowCaps });
+
+  const clientSession = await connect(serverEndpoint.address);
+  await clientSession.opened;
+
+  const stream = await clientSession.createBidirectionalStream();
+
+  // Assert the sender actually hit the window. Without this the test could
+  // pass trivially if the windows were silently widened, and we would lose
+  // the coverage without noticing.
+  stream.onblocked = mustCallAtLeast(1);
+
+  stream.setBody(payload);
+
+  for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+  await Promise.all([stream.closed, serverDone.promise]);
+  await clientSession.close();
+  await serverEndpoint.close();
+}
+
+// Unidirectional: same volume, exercises the uni credit path, which uses
+// different transport parameters and a different stream type.
+{
+  const payload = makePayload(kTotal, 7);
+  const expectedHash = hashBytes(payload);
+  const serverDone = Promise.withResolvers();
+
+  const serverEndpoint = await listen(mustCall((serverSession) => {
+    serverSession.onstream = mustCall(async (stream) => {
+      const received = await bytes(stream);
+      assert.strictEqual(received.byteLength, kTotal);
+      assert.strictEqual(hashBytes(received), expectedHash);
+      await stream.closed;
+      serverSession.close();
+      serverDone.resolve();
+    });
+  }), { transportParams, ...windowCaps });
+
+  const clientSession = await connect(serverEndpoint.address);
+  await clientSession.opened;
+
+  const stream = await clientSession.createUnidirectionalStream();
+  stream.onblocked = mustCallAtLeast(1);
+  stream.setBody(payload);
+
+  await Promise.all([stream.closed, serverDone.promise]);
+  await clientSession.close();
+  await serverEndpoint.close();
+}

test/parallel/test-quic-flow-control-slow-consumer-volume.mjs

@@ -0,0 +1,116 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: a slow consumer keeps buffered data bounded by the flow control
+// window, over a transfer far larger than that window.
+//
+// The existing slow-consumer test only checks that a small transfer still
+// completes and that onblocked fired. The stronger property -- that the
+// receiver never buffers more than the window allows, no matter how far
+// behind the consumer falls -- needs volume to be meaningful, because it is
+// the *repeated* refusal to over-credit that keeps memory bounded.
+//
+// Credit for inbound data is only returned once the consumer actually reads
+// it, so if the receiver ever credited data it had merely buffered, the
+// sender would be free to run arbitrarily far ahead and buffered bytes would
+// grow without bound. Asserting a ceiling on maxBytesAccumulated over a
+// multi-megabyte transfer is a direct check that this does not happen.
+
+import { hasQuic, skip, mustCall, mustCallAtLeast } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect, makePayload, hashBytes } =
+  await import('../common/quic.mjs');
+const { setImmediate: yieldToLoop } = await import('node:timers/promises');
+
+const kTotal = 4 * 1024 * 1024;
+const kStreamWindow = 16 * 1024;
+const kConnWindow = 32 * 1024;
+
+// Yield to the event loop every few reads. This lets the sender run as far
+// ahead as flow control permits without adding wall-clock delay to the test.
+const kYieldEvery = 4;
+
+assert.ok(kTotal / kStreamWindow >= 100,
+          'payload must be far larger than the window to be meaningful');
+
+const payload = makePayload(kTotal, 5);
+const expectedHash = hashBytes(payload);
+
+const serverDone = Promise.withResolvers();
+let peakAccumulated = 0;
+let received = 0;
+const parts = [];
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    let reads = 0;
+    for await (const chunks of stream) {
+      for (const chunk of chunks) {
+        received += chunk.byteLength;
+        parts.push(chunk);
+      }
+
+      // maxBytesAccumulated is the high water mark of data that has been
+      // received but not yet consumed.
+      const accumulated = Number(stream.stats.maxBytesAccumulated);
+      if (accumulated > peakAccumulated) peakAccumulated = accumulated;
+
+      if (++reads % kYieldEvery === 0) await yieldToLoop();
+    }
+
+    stream.writer.endSync();
+    await stream.closed;
+    serverSession.close();
+    serverDone.resolve();
+  });
+}), {
+  transportParams: {
+    initialMaxStreamDataBidiRemote: kStreamWindow,
+    initialMaxData: kConnWindow,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+});
+
+const clientSession = await connect(serverEndpoint.address);
+await clientSession.opened;
+
+const stream = await clientSession.createBidirectionalStream();
+// The sender must actually hit the window, otherwise the consumer was not
+// slow relative to the sender and the ceiling below proves nothing.
+stream.onblocked = mustCallAtLeast(1);
+stream.setBody(payload);
+
+for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars
+await Promise.all([stream.closed, serverDone.promise]);
+
+// Integrity first: bounded buffering is only interesting if no data was lost.
+assert.strictEqual(received, kTotal);
+const assembled = new Uint8Array(received);
+let offset = 0;
+for (const part of parts) {
+  assembled.set(part, offset);
+  offset += part.byteLength;
+}
+assert.strictEqual(hashBytes(assembled), expectedHash);
+
+// Upper bound: buffered-but-unread data never exceeded the window, even
+// though the payload was 256x the window. This is the property that keeps
+// receiver memory bounded regardless of consumer speed.
+assert.ok(peakAccumulated <= kStreamWindow,
+          `buffered data (${peakAccumulated}) must stay within the flow ` +
+          `control window (${kStreamWindow})`);
+
+// Lower bound: the buffer genuinely filled up. Without this the ceiling
+// above could be satisfied trivially by a consumer that kept pace with the
+// sender, which would not test the bound at all.
+assert.ok(peakAccumulated > kStreamWindow / 2,
+          `consumer was not slow enough to exercise the bound, peak ` +
+          `buffered was only ${peakAccumulated}`);
+
+await clientSession.close();
+await serverEndpoint.close();

test/parallel/test-quic-h3-flow-control-volume.mjs

@@ -0,0 +1,120 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: HTTP/3 high volume transfer through constrained flow control
+// windows, in both directions.
+//
+// HTTP/3 credits inbound flow control differently from raw QUIC. The bytes
+// consumed by nghttp3 frame parsing are credited as soon as they are read,
+// but DATA frame *payload* is deliberately excluded from that and is instead
+// credited later, when the application actually consumes it. That makes the
+// HTTP/3 receive path a genuinely separate code path from raw QUIC rather
+// than a thin wrapper over it, so it needs its own volume coverage: a
+// mis-credit here (either double counting the payload or never crediting it)
+// would not be caught by the raw QUIC tests.
+//
+// A large request body and a large response body are both pushed through
+// windows far smaller than either, so the transfer only completes if the
+// payload is credited exactly once as it is consumed.
+
+import { hasQuic, skip, mustCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import * as fixtures from '../common/fixtures.mjs';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+const { makePayload, hashBytes } = await import('../common/quic.mjs');
+const { bytes } = await import('stream/iter');
+
+const key = createPrivateKey(fixtures.readKey('agent1-key.pem'));
+const cert = fixtures.readKey('agent1-cert.pem');
+
+const kTotal = 2 * 1024 * 1024;         // per direction
+const kStreamWindow = 16 * 1024;
+const kConnWindow = 32 * 1024;
+
+assert.ok(kTotal / kStreamWindow >= 100,
+          'payload must be far larger than the window to be meaningful');
+
+// Distinct payloads per direction so a direction mixup is detectable.
+const requestBody = makePayload(kTotal, 31);
+const responseBody = makePayload(kTotal, 41);
+const requestHash = hashBytes(requestBody);
+const responseHash = hashBytes(responseBody);
+assert.notStrictEqual(requestHash, responseHash);
+
+const serverDone = Promise.withResolvers();
+
+const serverEndpoint = await listen(mustCall(async (serverSession) => {
+  serverSession.onstream = mustCall(async (stream) => {
+    // Read the large request body. This is the path where DATA payload
+    // credit is deferred until consumption.
+    const body = await bytes(stream);
+    assert.strictEqual(body.byteLength, kTotal);
+    // Must be the request body, intact and in order.
+    assert.strictEqual(hashBytes(body), requestHash);
+
+    await stream.closed;
+    serverSession.close();
+    serverDone.resolve();
+  });
+}), {
+  sni: { '*': { keys: [key], certs: [cert] } },
+  transportParams: {
+    initialMaxStreamDataBidiRemote: kStreamWindow,
+    initialMaxData: kConnWindow,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+  onheaders: mustCall(function(headers) {
+    assert.strictEqual(headers[':method'], 'POST');
+    this.sendHeaders({ ':status': '200', 'content-type': 'application/octet-stream' });
+    // Send a large response body concurrently with reading the request
+    // body, so both directions are flow-control bound at the same time.
+    this.setBody(responseBody);
+  }),
+});
+
+const clientSession = await connect(serverEndpoint.address, {
+  servername: 'localhost',
+  verifyPeer: 'manual',
+  transportParams: {
+    initialMaxStreamDataBidiLocal: kStreamWindow,
+    initialMaxData: kConnWindow,
+  },
+  maxStreamWindow: kStreamWindow,
+  maxWindow: kConnWindow,
+});
+
+const info = await clientSession.opened;
+assert.strictEqual(info.protocol, 'h3');
+
+const headersReceived = Promise.withResolvers();
+
+const stream = await clientSession.createBidirectionalStream({
+  headers: {
+    ':method': 'POST',
+    ':path': '/upload',
+    ':scheme': 'https',
+    ':authority': 'localhost',
+  },
+  body: requestBody,
+  onheaders: mustCall(function(headers) {
+    assert.strictEqual(headers[':status'], 200);
+    headersReceived.resolve();
+  }),
+});
+
+await headersReceived.promise;
+
+const received = await bytes(stream);
+assert.strictEqual(received.byteLength, kTotal);
+// Must be the response body, intact and in order.
+assert.strictEqual(hashBytes(received), responseHash);
+
+await Promise.all([stream.closed, serverDone.promise]);
+await clientSession.close();
+await serverEndpoint.close();

test/parallel/test-quic-h3-stream-destroy-no-resurrect.mjs

@@ -0,0 +1,97 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: HTTP/3 request streams are not resurrected after being destroyed
+// with response DATA still in flight.
+//
+// This is the HTTP/3 counterpart to test-quic-stream-destroy-no-resurrect.
+// The HTTP/3 receive path finds or creates a Stream for each nghttp3
+// callback, so it has to make the same distinction: a locally-initiated
+// request stream that is no longer tracked was destroyed by the application
+// and must not be recreated when the remaining DATA frames arrive.
+//
+// It also has to keep crediting flow control for the payload it discards.
+// nghttp3 hands DATA payload to the application uncredited (it is excluded
+// from the framing bytes credited when the stream data is read), so silently
+// dropping it would permanently shrink the session's shared receive window --
+// which is why this test runs enough requests to outlast a small window.
+
+import { hasQuic, skip, mustCall, mustNotCall } from '../common/index.mjs';
+import assert from 'node:assert';
+import * as fixtures from '../common/fixtures.mjs';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+const { makePayload } = await import('../common/quic.mjs');
+
+const key = createPrivateKey(fixtures.readKey('agent1-key.pem'));
+const cert = fixtures.readKey('agent1-cert.pem');
+
+const kResponseSize = 256 * 1024;
+const kRequests = 12;
+// Deliberately smaller than the aggregate discarded payload, so the run only
+// completes if discarded DATA is still credited back.
+const kConnWindow = 256 * 1024;
+
+const responseBody = makePayload(kResponseSize, 17);
+
+assert.ok(kResponseSize * kRequests > kConnWindow * 4,
+          'aggregate response data must far exceed the connection window');
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  serverSession.onstream = mustCall((stream) => {
+    // The client destroys these early; the truncated write is expected.
+    stream.onerror = () => {};
+  }, kRequests);
+}), {
+  sni: { '*': { keys: [key], certs: [cert] } },
+  onheaders: mustCall(function() {
+    this.sendHeaders({ ':status': '200' });
+    this.setBody(responseBody);
+  }, kRequests),
+});
+
+const clientSession = await connect(serverEndpoint.address, {
+  servername: 'localhost',
+  verifyPeer: 'manual',
+  transportParams: {
+    initialMaxData: kConnWindow,
+    initialMaxStreamDataBidiLocal: kResponseSize * 2,
+  },
+});
+
+// The client opens every stream itself; the server opens none. Any onstream
+// here is a destroyed request stream being resurrected and misreported as
+// peer-initiated.
+clientSession.onstream = mustNotCall(
+  'client must not receive onstream for its own destroyed request streams');
+
+const info = await clientSession.opened;
+assert.strictEqual(info.protocol, 'h3');
+
+for (let i = 0; i < kRequests; i++) {
+  const stream = await clientSession.createBidirectionalStream({
+    headers: {
+      ':method': 'GET',
+      ':path': `/${i}`,
+      ':scheme': 'https',
+      ':authority': 'localhost',
+    },
+  });
+
+  // Take one batch of the response, then abandon the rest and destroy, so
+  // DATA frames keep arriving for a stream we no longer track.
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of stream) break;
+
+  stream.destroy();
+}
+
+// Exactly one locally-opened stream per request.
+assert.strictEqual(Number(clientSession.stats.bidiOutStreamCount), kRequests);
+
+await clientSession.close();
+await serverEndpoint.close();

test/parallel/test-quic-stream-destroy-no-resurrect.mjs

@@ -0,0 +1,80 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: destroying a locally-initiated stream with data still in flight does
+// not resurrect it.
+//
+// When a stream is destroyed, its ngtcp2 stream user data is cleared. Inbound
+// STREAM frames the peer had already put in flight then arrive for a stream
+// the receiver no longer tracks. Treating "no user data" as "a stream I have
+// not seen before" is wrong for a stream we initiated ourselves: we are the
+// only party that can create it, so the absence of a record means we
+// destroyed it, not that it is new.
+//
+// Getting this wrong is visible from JavaScript in two ways, both asserted
+// here:
+//
+//   * the session reports far more locally-opened streams than were actually
+//     opened, because a fresh Stream is created for every frame still in
+//     flight
+//   * the application receives onstream events for streams it initiated and
+//     already destroyed, which is a contract violation -- onstream is for
+//     peer-initiated streams
+//
+// It also eventually breaks the session outright: the churn exhausts the
+// local stream budget and createBidirectionalStream() starts throwing
+// ERR_QUIC_OPEN_STREAM_FAILED.
+
+import { hasQuic, skip, mustCall, mustNotCall } from '../common/index.mjs';
+import assert from 'node:assert';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+
+const { listen, connect, makePayload } = await import('../common/quic.mjs');
+
+// The payload has to be big enough that the server still has plenty in flight
+// when the client walks away, which is what creates the stale frames.
+const kPayloadSize = 256 * 1024;
+const kStreams = 12;
+
+const payload = makePayload(kPayloadSize, 13);
+
+const serverEndpoint = await listen(mustCall((serverSession) => {
+  serverSession.onstream = mustCall((stream) => {
+    // The client destroys these early, so the write is cut short by
+    // STOP_SENDING. That is the scenario under test, not a failure.
+    stream.onerror = () => {};
+    stream.setBody(payload);
+  }, kStreams);
+}));
+
+const clientSession = await connect(serverEndpoint.address);
+
+// The client never expects an incoming stream: it opens every stream itself
+// and the server opens none. Any onstream event here is a resurrected local
+// stream being mistaken for a peer-initiated one.
+clientSession.onstream = mustNotCall(
+  'client must not receive onstream for its own destroyed streams');
+
+await clientSession.opened;
+
+for (let i = 0; i < kStreams; i++) {
+  const stream = await clientSession.createBidirectionalStream({
+    body: new Uint8Array(8),
+  });
+
+  // Read a little, then abandon the rest and destroy. Breaking out of the
+  // loop leaves the server mid-send, so frames keep arriving after destroy.
+  // eslint-disable-next-line no-unused-vars
+  for await (const _ of stream) break;
+
+  stream.destroy();
+}
+
+// Exactly one locally-opened stream per iteration. Before the fix this
+// counted in the hundreds, because every stale frame created a new stream.
+assert.strictEqual(Number(clientSession.stats.bidiOutStreamCount), kStreams);
+
+await clientSession.close();
+await serverEndpoint.close();