diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index c50a8e6d5..a86563fc2 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -262,6 +262,17 @@ class QuicWebTransportSession( ) : WebTransportSession { override val isOpen: Boolean get() = state.isOpen + /** + * Diagnostic-only passthrough to + * [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot]. + * Used by `SendTraceScenario` to dump the connection's + * flow-control accounting at known points (pre-pump, mid-pump, + * post-pump-grace) when investigating loss patterns. Not part of + * the [WebTransportSession] common interface — callers downcast + * explicitly when they need it. + */ + suspend fun quicFlowControlSnapshot(): com.vitorpamplona.quic.connection.QuicFlowControlSnapshot = state.connection.flowControlSnapshot() + override suspend fun openBidiStream(): WebTransportBidiStream { val s = state.openBidiStream() return QuicBidiStreamAdapter(s, state.driver) diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt index 22266fcf9..ae050352f 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt @@ -386,7 +386,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { ) = runBlocking { NostrNestsHarness.assumeNestsInterop() val harness = harnessOrNull ?: return@runBlocking - withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope -> + withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot -> val result = SendTraceScenario.run( scope = scope, @@ -395,6 +395,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { speakerPubkeyHex = hostPub, scenario = scenario, pumpScope = pumpScope, + flowControlSnapshot = flowControlSnapshot, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -410,6 +411,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { listeners: List, speakerPubkeyHex: String, pumpScope: CoroutineScope, + flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -468,7 +470,17 @@ class NostrNestsSustainedSendOutcomesInteropTest { InteropDebug.assertListenerReached(scope, "Connected", listener.state.value) listeners += listener } - block(publisher, listeners, hostSigner.pubKey, pumpScope) + // Diagnostics passthrough mirrors withProdSpeakerAndListeners + // — see that helper's comment for rationale. + val quicSpeakerWt = + speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + block( + publisher, + listeners, + hostSigner.pubKey, + pumpScope, + { quicSpeakerWt.quicFlowControlSnapshot() }, + ) } finally { for (listener in listeners) { runCatching { listener.close() } diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt index cbc0c9366..938dfc229 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt @@ -944,7 +944,7 @@ class NostrnestsProdAudioTransmissionTest { expectAllReceived: Boolean = false, ) = runBlocking { assumeProd() - withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope -> + withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot -> val result = SendTraceScenario.run( scope = scope, @@ -953,6 +953,7 @@ class NostrnestsProdAudioTransmissionTest { speakerPubkeyHex = hostPub, scenario = scenario, pumpScope = pumpScope, + flowControlSnapshot = flowControlSnapshot, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -976,6 +977,7 @@ class NostrnestsProdAudioTransmissionTest { listeners: List, speakerPubkeyHex: String, pumpScope: CoroutineScope, + flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -1030,7 +1032,19 @@ class NostrnestsProdAudioTransmissionTest { listeners += listener } - block(publisher, listeners, hostSigner.pubKey, pumpScope) + // Diagnostics passthrough: cast the speaker WT to the + // concrete QUIC adapter so the scenario can read the + // underlying connection's flow-control snapshot. See + // `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + val quicSpeakerWt = + speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + block( + publisher, + listeners, + hostSigner.pubKey, + pumpScope, + { quicSpeakerWt.quicFlowControlSnapshot() }, + ) } finally { for (listener in listeners) { runCatching { listener.close() } diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt index 4d0b8e8b3..836d14dd1 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt @@ -177,6 +177,16 @@ data class ScenarioResult( * - Tear down all of those after [run] returns. */ object SendTraceScenario { + /** + * @param flowControlSnapshot optional supplier that returns the + * speaker-side QUIC connection's flow-control state at the + * moment of the call. When provided, the scenario captures and + * logs three snapshots — before the pump, immediately after the + * pump completes, and after the receive-grace window — so a + * failure mode like "data piled up but conn-level credit ran + * out" can be diagnosed from the test report alone. See + * [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot]. + */ suspend fun run( scope: String, publisher: MoqLitePublisherHandle, @@ -184,11 +194,25 @@ object SendTraceScenario { speakerPubkeyHex: String, scenario: Scenario, pumpScope: CoroutineScope, + flowControlSnapshot: (suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot)? = null, ): ScenarioResult { require(listeners.size == scenario.parallelSubscriptions) { "expected ${scenario.parallelSubscriptions} listener(s), got ${listeners.size}" } InteropDebug.checkpoint(scope, "scenario=$scenario speaker=${speakerPubkeyHex.take(8)}…") + flowControlSnapshot?.invoke()?.let { snap -> + InteropDebug.checkpoint( + scope, + "fc-pre: peerInitMaxData=${snap.peerInitialMaxData} " + + "peerInitMaxStreamDataUni=${snap.peerInitialMaxStreamDataUni} " + + "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " + + "sendCredit=${snap.sendConnectionFlowCredit} consumed=${snap.sendConnectionFlowConsumed} " + + "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " + + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", + ) + } val sendOutcomes = BooleanArray(scenario.frameCount) val sendDurationsMicros = LongArray(scenario.frameCount) @@ -283,6 +307,18 @@ object SendTraceScenario { "(target=${scenario.frameCount * scenario.cadenceMs}ms) " + "sendTrue=${sendOutcomes.count { it }}/${scenario.frameCount}", ) + flowControlSnapshot?.invoke()?.let { snap -> + InteropDebug.checkpoint( + scope, + "fc-post-pump: sendCredit=${snap.sendConnectionFlowCredit} " + + "consumed=${snap.sendConnectionFlowConsumed} " + + "(remaining=${snap.sendConnectionFlowCredit - snap.sendConnectionFlowConsumed}) " + + "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " + + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", + ) + } // Wait for collectors. If they hit `take(N)` they exit naturally; // otherwise the per-collector withTimeoutOrNull cancels them. @@ -292,6 +328,18 @@ object SendTraceScenario { } if (job.isActive) job.cancelAndJoin() } + flowControlSnapshot?.invoke()?.let { snap -> + InteropDebug.checkpoint( + scope, + "fc-post-grace: sendCredit=${snap.sendConnectionFlowCredit} " + + "consumed=${snap.sendConnectionFlowConsumed} " + + "(remaining=${snap.sendConnectionFlowCredit - snap.sendConnectionFlowConsumed}) " + + "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " + + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", + ) + } return ScenarioResult( scenario = scenario, diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index 5b9fcc4f4..edb04fa16 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -386,6 +386,59 @@ class QuicConnection( /** Snapshot of peer-granted uni cap. */ fun peerMaxStreamsUniSnapshot(): Long = peerMaxStreamsUni + /** + * Coherent point-in-time snapshot of the connection's flow-control + * accounting. Acquires [lock] internally so the fields are read + * atomically with respect to the read / send / parse paths. + * + * Diagnostic-only: meant for tests + dev tooling investigating + * the production "stream cliff" symptom where `publisher.send` + * keeps returning `true` past frame ~99 but no data reaches the + * listener. Surface area is the smallest set that lets a caller + * answer: + * + * - what did the peer grant at handshake? + * - has the peer extended the cap since? (delta = current - initial) + * - have we hit the cap? (consumed vs credit) + * - is data piling up unsent on any stream? (sum readableBytes + * over local-initiated streams) + * + * See `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + */ + suspend fun flowControlSnapshot(): QuicFlowControlSnapshot = + lock.withLock { + val tp = peerTransportParameters + // Sum bytes the application has enqueued but the writer + // hasn't yet handed to a STREAM frame. A non-zero value + // after the pump is the smoking gun for "data stuck in + // local send buffer due to flow control". + var pending = 0L + var pendingStreamCount = 0 + for (stream in streamsList) { + val pendingOnStream = stream.send.readableBytes.toLong() + if (pendingOnStream > 0) { + pending += pendingOnStream + pendingStreamCount += 1 + } + } + QuicFlowControlSnapshot( + peerInitialMaxData = tp?.initialMaxData, + peerInitialMaxStreamDataUni = tp?.initialMaxStreamDataUni, + peerInitialMaxStreamDataBidiRemote = tp?.initialMaxStreamDataBidiRemote, + peerInitialMaxStreamsUni = tp?.initialMaxStreamsUni, + peerInitialMaxStreamsBidi = tp?.initialMaxStreamsBidi, + sendConnectionFlowCredit = sendConnectionFlowCredit, + sendConnectionFlowConsumed = sendConnectionFlowConsumed, + peerMaxStreamsUniCurrent = peerMaxStreamsUni, + peerMaxStreamsBidiCurrent = peerMaxStreamsBidi, + nextLocalUniIndex = nextLocalUniIndex, + nextLocalBidiIndex = nextLocalBidiIndex, + totalEnqueuedNotSentBytes = pending, + streamsWithPendingBytes = pendingStreamCount, + totalStreamsTracked = streamsList.size, + ) + } + suspend fun pollIncomingPeerStream(): QuicStream? = lock.withLock { newPeerStreams.removeFirstOrNull() } /** @@ -595,3 +648,83 @@ class QuicConnectionClosedException( class QuicStreamLimitException( message: String, ) : RuntimeException(message) + +/** + * Diagnostic snapshot of [QuicConnection]'s flow-control accounting at + * a single moment. Returned by [QuicConnection.flowControlSnapshot]. + * + * Reading rules of thumb for the production-cliff investigation + * (see `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`): + * + * - **`sendConnectionFlowConsumed >= peerInitialMaxData`** with + * `sendConnectionFlowCredit == peerInitialMaxData` ⇒ peer never + * extended `MAX_DATA`; we wedged on connection-level send credit. + * - **`sendConnectionFlowCredit > peerInitialMaxData`** ⇒ peer + * DID extend; suspicion shifts to per-stream credit or downstream. + * - **`totalEnqueuedNotSentBytes > 0` after a quiescent period** ⇒ + * application wrote bytes that the writer couldn't put on the + * wire (per-stream or connection budget exhausted, or peer + * stopped reading). + * - **`nextLocalUniIndex >= peerMaxStreamsUniCurrent`** ⇒ at the + * stream-id cap; would block in [openUniStream] (currently + * throws [QuicStreamLimitException]). + */ +data class QuicFlowControlSnapshot( + /** + * Peer's `initial_max_data` transport parameter (RFC 9000 §18.2) + * — the connection-level send budget the peer initially granted. + * `null` means the peer's TPs hadn't been parsed yet at snapshot + * time (i.e. handshake hadn't completed). + */ + val peerInitialMaxData: Long?, + /** Peer's `initial_max_stream_data_uni` transport parameter. */ + val peerInitialMaxStreamDataUni: Long?, + /** Peer's `initial_max_stream_data_bidi_remote` transport parameter. */ + val peerInitialMaxStreamDataBidiRemote: Long?, + /** Peer's `initial_max_streams_uni`. */ + val peerInitialMaxStreamsUni: Long?, + /** Peer's `initial_max_streams_bidi`. */ + val peerInitialMaxStreamsBidi: Long?, + /** + * Current connection-level send credit. Starts at + * [peerInitialMaxData] when the handshake completes; raised by + * inbound `MAX_DATA` frames (RFC 9000 §19.9). + */ + val sendConnectionFlowCredit: Long, + /** + * Total stream-frame bytes we've already pushed past the writer + * against [sendConnectionFlowCredit]. When this catches up to + * the credit, the writer stops draining stream data until a + * fresh `MAX_DATA` arrives. + */ + val sendConnectionFlowConsumed: Long, + /** + * Current peer-granted unidirectional stream concurrency cap. + * Starts at [peerInitialMaxStreamsUni]; raised by inbound + * `MAX_STREAMS_UNI` frames. + */ + val peerMaxStreamsUniCurrent: Long, + /** Bidi counterpart of [peerMaxStreamsUniCurrent]. */ + val peerMaxStreamsBidiCurrent: Long, + /** + * Number of client-initiated uni streams [openUniStream] has + * allocated locally. The next allocation would use this value + * as the index; if it equals or exceeds + * [peerMaxStreamsUniCurrent], the next [openUniStream] throws + * [QuicStreamLimitException]. + */ + val nextLocalUniIndex: Long, + /** Bidi counterpart of [nextLocalUniIndex]. */ + val nextLocalBidiIndex: Long, + /** + * Sum of bytes enqueued to any stream's send buffer but NOT yet + * encoded into a STREAM frame. A non-trivial value after a + * quiescent period indicates flow control (per-stream or + * connection-level) is starving the writer. + */ + val totalEnqueuedNotSentBytes: Long, + /** Number of streams contributing to [totalEnqueuedNotSentBytes]. */ + val streamsWithPendingBytes: Int, + /** Number of streams currently tracked (alive + closed-but-retained). */ + val totalStreamsTracked: Int, +) diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/FlowControlSnapshotTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/FlowControlSnapshotTest.kt new file mode 100644 index 000000000..260db3439 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/FlowControlSnapshotTest.kt @@ -0,0 +1,184 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.quic.connection + +import com.vitorpamplona.quic.tls.InProcessTlsServer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Verifies the diagnostic [QuicConnection.flowControlSnapshot] surface: + * + * - Before the handshake, peer-TP fields are null and the credit / + * index counters are zero. + * - After the handshake completes against an [InProcessTlsServer] + * advertising specific transport parameters, the snapshot reflects + * exactly those values. + * - Allocating uni / bidi streams advances the local index counters. + * - Enqueueing bytes to a stream's send buffer (without a writer + * drain) shows up as `totalEnqueuedNotSentBytes`. + * + * Used by `nestsClient/.../SendTraceScenario` to dump the connection's + * flow-control state at known points during a sweep run; the cliff + * investigation in + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md` + * relies on these fields to localise where data piles up. + */ +class FlowControlSnapshotTest { + @Test + fun snapshot_before_handshake_has_null_peer_tps_and_zero_counters() { + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = + com.vitorpamplona.quic.tls + .PermissiveCertificateValidator(), + ) + val snap = client.flowControlSnapshot() + assertNull(snap.peerInitialMaxData, "no peer TPs before handshake") + assertNull(snap.peerInitialMaxStreamDataUni) + assertNull(snap.peerInitialMaxStreamsUni) + assertEquals(0L, snap.sendConnectionFlowCredit) + assertEquals(0L, snap.sendConnectionFlowConsumed) + assertEquals(0L, snap.peerMaxStreamsUniCurrent) + assertEquals(0L, snap.peerMaxStreamsBidiCurrent) + assertEquals(0L, snap.nextLocalUniIndex) + assertEquals(0L, snap.nextLocalBidiIndex) + assertEquals(0L, snap.totalEnqueuedNotSentBytes) + assertEquals(0, snap.streamsWithPendingBytes) + assertEquals(0, snap.totalStreamsTracked) + } + } + + @Test + fun snapshot_after_handshake_reflects_peer_transport_parameters() { + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = + com.vitorpamplona.quic.tls + .PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 1_500_000, + initialMaxStreamDataBidiLocal = 200_000, + initialMaxStreamDataBidiRemote = 250_000, + initialMaxStreamDataUni = 300_000, + initialMaxStreamsBidi = 17, + initialMaxStreamsUni = 19, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + + val snap = client.flowControlSnapshot() + assertEquals(1_500_000L, snap.peerInitialMaxData) + assertEquals(300_000L, snap.peerInitialMaxStreamDataUni) + assertEquals(250_000L, snap.peerInitialMaxStreamDataBidiRemote) + assertEquals(19L, snap.peerInitialMaxStreamsUni) + assertEquals(17L, snap.peerInitialMaxStreamsBidi) + // Initial credit equals the initial cap until inbound MAX_DATA + // raises it — no traffic has been sent yet. + assertEquals(1_500_000L, snap.sendConnectionFlowCredit) + assertEquals(0L, snap.sendConnectionFlowConsumed) + assertEquals(19L, snap.peerMaxStreamsUniCurrent) + assertEquals(17L, snap.peerMaxStreamsBidiCurrent) + assertEquals(0L, snap.nextLocalUniIndex) + assertEquals(0L, snap.nextLocalBidiIndex) + } + } + + @Test + fun snapshot_tracks_enqueued_bytes_and_local_stream_indexes() { + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = QuicConnectionConfig(), + tlsCertificateValidator = + com.vitorpamplona.quic.tls + .PermissiveCertificateValidator(), + ) + val serverScid = ConnectionId.random(8) + val tlsServer = + InProcessTlsServer( + transportParameters = + TransportParameters( + initialMaxData = 1_000_000, + initialMaxStreamDataBidiLocal = 100_000, + initialMaxStreamDataBidiRemote = 100_000, + initialMaxStreamDataUni = 100_000, + initialMaxStreamsBidi = 10, + initialMaxStreamsUni = 10, + initialSourceConnectionId = serverScid.bytes, + originalDestinationConnectionId = client.destinationConnectionId.bytes, + ).encode(), + ) + val pipe = + InMemoryQuicPipe( + client = client, + initialDcid = client.destinationConnectionId.bytes, + serverScid = serverScid, + tlsServer = tlsServer, + ) + client.start() + pipe.drive(maxRounds = 16) + assertEquals(QuicConnection.Status.CONNECTED, client.status) + + val s1 = client.openUniStream() + val s2 = client.openUniStream() + client.openBidiStream() + s1.send.enqueue(ByteArray(100)) + s2.send.enqueue(ByteArray(50)) + + val snap = client.flowControlSnapshot() + assertEquals(2L, snap.nextLocalUniIndex) + assertEquals(1L, snap.nextLocalBidiIndex) + assertTrue( + snap.totalStreamsTracked >= 3, + "3 client-initiated streams should be tracked, saw ${snap.totalStreamsTracked}", + ) + assertEquals(150L, snap.totalEnqueuedNotSentBytes) + assertEquals(2, snap.streamsWithPendingBytes) + } + } +}