From e3c1b8685352994d37006593dfb41faf72b86204 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sat, 15 Aug 2026 12:57:31 -0700 Subject: [PATCH] quic: apply multiple fixes to flow control signaling Signed-off-by: James M Snell Assisted-by: Opencode/Opus --- lib/internal/blob.js | 34 ++++- src/dataqueue/queue.cc | 25 +++- src/quic/application.cc | 28 ++++ src/quic/http3.cc | 33 +++++ src/quic/streams.cc | 69 ++++++++-- src/quic/streams.h | 30 +++++ test/common/quic.mjs | 47 +++++++ ...-flow-control-bidi-simultaneous-volume.mjs | 98 ++++++++++++++ ...st-quic-flow-control-concurrent-volume.mjs | 126 ++++++++++++++++++ .../test-quic-flow-control-credit-reclaim.mjs | 98 ++++++++++++++ ...st-quic-flow-control-large-constrained.mjs | 121 +++++++++++++++++ ...quic-flow-control-slow-consumer-volume.mjs | 116 ++++++++++++++++ .../test-quic-h3-flow-control-volume.mjs | 120 +++++++++++++++++ ...st-quic-h3-stream-destroy-no-resurrect.mjs | 97 ++++++++++++++ .../test-quic-stream-destroy-no-resurrect.mjs | 80 +++++++++++ 15 files changed, 1106 insertions(+), 16 deletions(-) create mode 100644 test/parallel/test-quic-flow-control-bidi-simultaneous-volume.mjs create mode 100644 test/parallel/test-quic-flow-control-concurrent-volume.mjs create mode 100644 test/parallel/test-quic-flow-control-credit-reclaim.mjs create mode 100644 test/parallel/test-quic-flow-control-large-constrained.mjs create mode 100644 test/parallel/test-quic-flow-control-slow-consumer-volume.mjs create mode 100644 test/parallel/test-quic-h3-flow-control-volume.mjs create mode 100644 test/parallel/test-quic-h3-stream-destroy-no-resurrect.mjs create mode 100644 test/parallel/test-quic-stream-destroy-no-resurrect.mjs diff --git a/lib/internal/blob.js b/lib/internal/blob.js index 41c7f61da790..b2ea9f14de20 100644 --- a/lib/internal/blob.js +++ b/lib/internal/blob.js @@ -570,18 +570,35 @@ 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(); reader.setWakeup(wakeup.resolve); try { while (true) { const batch = []; + let batchBytes = 0; let blocked = false; let eos = false; let error = null; @@ -610,8 +627,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) { diff --git a/src/dataqueue/queue.cc b/src/dataqueue/queue.cc index 8516362b7a87..ba7348ee8608 100644 --- a/src/dataqueue/queue.cc +++ b/src/dataqueue/queue.cc @@ -174,14 +174,35 @@ 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 -- removing it from backpressure_listeners_ while we are still + // iterating. 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; + std::vector 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; + std::vector listeners( + backpressure_listeners_.begin(), backpressure_listeners_.end()); + for (auto* listener : listeners) { + if (backpressure_listeners_.contains(listener)) { + listener->BeforePull(); + } + } } bool HasBackpressureListeners() const noexcept { diff --git a/src/quic/application.cc b/src/quic/application.cc index 79a3263b8537..177ce3d30a85 100644 --- a/src/quic/application.cc +++ b/src/quic/application.cc @@ -311,6 +311,34 @@ class DefaultApplication final : public Session::Application { void* stream_user_data) override { BaseObjectPtr 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]] { diff --git a/src/quic/http3.cc b/src/quic/http3.cc index b6d876af60f6..86ccfb00cb9b 100644 --- a/src/quic/http3.cc +++ b/src/quic/http3.cc @@ -1055,6 +1055,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; } @@ -1193,6 +1201,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; diff --git a/src/quic/streams.cc b/src/quic/streams.cc index c2691362447a..640efdca1ed9 100644 --- a/src/quic/streams.cc +++ b/src/quic/streams.cc @@ -1499,12 +1499,29 @@ 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::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); + uncredited_bytes_ -= std::min(uncredited_bytes_, amount); + ReturnFlowControlCredit(amount, CreditScope::STREAM_AND_CONNECTION); } void Stream::BeforePull() { @@ -1517,14 +1534,24 @@ 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)); - // 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(); + auto appended = inbound_->append(std::move(entry)); + if (appended.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(); + } else { + // The queue rejected the entry (it is capped and this data would push + // it past the final size) so the bytes have been dropped. They will + // never reach a reader, which means EntryRead() will never fire for + // them -- return their flow control credit here instead of leaking it. + uncredited_bytes_ -= std::min(uncredited_bytes_, flushed); + ReturnFlowControlCredit(flushed, CreditScope::STREAM_AND_CONNECTION); + } } STAT_SET(Stats, bytes_accumulated, 0); } @@ -1652,6 +1679,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 +1728,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 +1745,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. diff --git a/src/quic/streams.h b/src/quic/streams.h index f18702ae75ab..df8b4a550353 100644 --- a/src/quic/streams.h +++ b/src/quic/streams.h @@ -395,6 +395,26 @@ 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); + // Gets a reader for the data received for this stream from the peer, BaseObjectPtr get_reader(); @@ -458,6 +478,16 @@ class Stream final : public AsyncWrap, BaseObjectWeakPtr reader_; std::unique_ptr recv_accumulator_; + // Number of received bytes that are still holding inbound flow control + // credit -- that is, bytes that ngtcp2 has delivered to us but that have + // not yet been handed to the JavaScript consumer (they are sitting in + // recv_accumulator_ or in the inbound_ DataQueue). This is incremented + // in ReceiveData() and decremented in EntryRead(). 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. + 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. diff --git a/test/common/quic.mjs b/test/common/quic.mjs index d05ee634f5e5..dc4b094cf900 100644 --- a/test/common/quic.mjs +++ b/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, }; diff --git a/test/parallel/test-quic-flow-control-bidi-simultaneous-volume.mjs b/test/parallel/test-quic-flow-control-bidi-simultaneous-volume.mjs new file mode 100644 index 000000000000..31f5c7b17518 --- /dev/null +++ b/test/parallel/test-quic-flow-control-bidi-simultaneous-volume.mjs @@ -0,0 +1,98 @@ +// 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. + +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(); diff --git a/test/parallel/test-quic-flow-control-concurrent-volume.mjs b/test/parallel/test-quic-flow-control-concurrent-volume.mjs new file mode 100644 index 000000000000..8e81b8314988 --- /dev/null +++ b/test/parallel/test-quic-flow-control-concurrent-volume.mjs @@ -0,0 +1,126 @@ +// 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. + +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(); diff --git a/test/parallel/test-quic-flow-control-credit-reclaim.mjs b/test/parallel/test-quic-flow-control-credit-reclaim.mjs new file mode 100644 index 000000000000..f5dcedad1cce --- /dev/null +++ b/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'); diff --git a/test/parallel/test-quic-flow-control-large-constrained.mjs b/test/parallel/test-quic-flow-control-large-constrained.mjs new file mode 100644 index 000000000000..db73141528b1 --- /dev/null +++ b/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(); +} diff --git a/test/parallel/test-quic-flow-control-slow-consumer-volume.mjs b/test/parallel/test-quic-flow-control-slow-consumer-volume.mjs new file mode 100644 index 000000000000..52e6a92512ef --- /dev/null +++ b/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(); diff --git a/test/parallel/test-quic-h3-flow-control-volume.mjs b/test/parallel/test-quic-h3-flow-control-volume.mjs new file mode 100644 index 000000000000..e422d1a411ad --- /dev/null +++ b/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(); diff --git a/test/parallel/test-quic-h3-stream-destroy-no-resurrect.mjs b/test/parallel/test-quic-h3-stream-destroy-no-resurrect.mjs new file mode 100644 index 000000000000..c9e05eb245b1 --- /dev/null +++ b/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(); diff --git a/test/parallel/test-quic-stream-destroy-no-resurrect.mjs b/test/parallel/test-quic-stream-destroy-no-resurrect.mjs new file mode 100644 index 000000000000..50687472a64c --- /dev/null +++ b/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();