diff --git a/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md b/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md index 1f93f9b94..5bb337c87 100644 --- a/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md +++ b/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md @@ -1,7 +1,19 @@ # QUIC stream cliff against nostrnests.com — investigation plan -**Status:** open. Mitigated in production via group-packing -(`NestMoqLiteBroadcaster.framesPerGroup = 5`). Root cause not isolated. +**Status: ROOT-CAUSED + FIXED.** Our `:quic` client never emitted +`MAX_STREAMS_*` frames to extend the peer-initiated stream-id cap. +[QuicConnectionConfig.initialMaxStreamsUni] (default `100`) was the +*lifetime* maximum the peer could ever open. The relay forwards each +broadcast group as a fresh peer-initiated uni stream, so any +broadcast longer than 100 frames silently truncated at the listener. +Fixed by tracking [QuicConnection.peerInitiatedUniCount] and emitting +`MAX_STREAMS_UNI(newCap)` from [appendFlowControlUpdates] once the +peer's usage crosses the half-window threshold — same pattern as the +existing `MAX_DATA` / `MAX_STREAM_DATA` extension. + +The `framesPerGroup = 5` mitigation in [NestMoqLiteBroadcaster] can +be reverted to `1` to match the JS reference broadcaster's wire +shape now that the underlying QUIC bug is gone. **Reproduces against:** `https://moq.nostrnests.com:4443` (production relay). Not consistently reproducible against the local @@ -127,6 +139,59 @@ per-stream cost (file transfer, metadata fan-out, anything that naturally wants a fresh stream per message) would still hit the cliff if they exceed roughly 50 streams/second. +## What `flowControlSnapshot` proved (the smoking gun) + +After wiring [QuicConnection.flowControlSnapshot] into [SendTraceScenario], +running `sweep_frames_200` against production produced: + +``` +fc-pre: peerInitMaxData=4611686018427387903 peerInitMaxStreamDataUni=1250000 + peerInitMaxStreamsUni=10000 sendCredit=4611686018427387903 + consumed=786 peerMaxStreamsUniNow=10000 + nextLocalUniIdx=1 pendingBytes=0 pendingStreams=0/4 +fc-post-pump: sendCredit=4611686018427387903 consumed=18729 + peerMaxStreamsUniNow=10000 nextLocalUniIdx=201 + pendingBytes=0 pendingStreams=0/205 +fc-post-grace: (identical to fc-post-pump) +sub[0]: received=99/200 firstSentButLost=99 +``` + +This rules out every speaker-side hypothesis at once: + +- **`peerInitMaxData = 2⁶²−1`** ⇒ ~5 EB connection-level send credit; + we used **18 KB**. Not a connection-level flow-control wedge. +- **`peerInitMaxStreamsUni = 10000`, `nextLocalUniIdx = 201`** ⇒ + speaker opened all 200 broadcast streams + 1 control stream, well + under cap. Not a peer-granted stream-id wedge. +- **`pendingBytes = 0`, `pendingStreams = 0/205`** ⇒ no bytes are + stuck in any stream's send buffer; everything we wrote made it + past the writer. +- **`consumed` is identical between post-pump and post-grace** ⇒ we + weren't even *trying* to send more after the pump finished. + +So the speaker → relay path was healthy. The 18 KB of stream data +made it onto the wire. The 99-frame ceiling on the listener side +had to be on the **relay → listener** half of the connection. + +Listener-side, our [QuicConnection] advertises `initial_max_streams_uni += 100` (from [QuicConnectionConfig], default 100) at handshake. The +relay can open up to 100 uni streams to us *for the lifetime of the +connection* unless we send `MAX_STREAMS_UNI(N)` frames to extend it. +Grepping `:quic` for `MaxStreamsFrame` showed it was only ever +*parsed inbound* — there was no outbound emission anywhere. So the +peer's stream-id allowance was capped at 100 forever. + +For MoQ over WebTransport this is fatal: the listener's relay +forwards every broadcast group as a new peer-initiated uni stream. +With one group per Opus frame at 20 ms cadence, the 100-stream cap +is exhausted at frame 99 → relay-side blocked → no more groups → no +more audio. + +The fix mirrors the existing [appendFlowControlUpdates] pattern for +`MAX_DATA` / `MAX_STREAM_DATA`: track lifetime peer-initiated stream +count, emit `MAX_STREAMS_UNI(currentCount + initialCap)` when count +crosses the half-window threshold. + ## Hypotheses still on the table In rough order of likelihood, none confirmed: 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 edb04fa16..3d7427701 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -142,6 +142,36 @@ class QuicConnection( */ internal var advertisedMaxData: Long = config.initialMaxData + /** + * Cumulative count of peer-initiated unidirectional streams we've + * accepted (incremented in [getOrCreatePeerStreamLocked]). Compared + * against [advertisedMaxStreamsUni] by the writer to decide whether + * to emit a fresh `MAX_STREAMS_UNI` frame extending the cap. Without + * extension the peer can never open more than + * [config.initialMaxStreamsUni] uni streams over the lifetime of the + * connection — the production stream-cliff investigation + * (`nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`) + * showed this is exactly what bites the listener side: each Opus + * frame is forwarded by the relay as a fresh uni stream, so any + * broadcast longer than 100 frames silently truncates at the + * audience. + */ + internal var peerInitiatedUniCount: Long = 0L + + /** Bidi counterpart of [peerInitiatedUniCount]. */ + internal var peerInitiatedBidiCount: Long = 0L + + /** + * The peer-initiated uni stream cap we've currently advertised to + * the peer. Starts at [config.initialMaxStreamsUni]; the writer + * raises it when [peerInitiatedUniCount] crosses the half-window + * threshold and emits a `MAX_STREAMS_UNI(newCap)` frame. + */ + internal var advertisedMaxStreamsUni: Long = config.initialMaxStreamsUni + + /** Bidi counterpart of [advertisedMaxStreamsUni]. */ + internal var advertisedMaxStreamsBidi: Long = config.initialMaxStreamsBidi + /** * Round-robin starting index for the writer's stream-drain iteration. * Without rotation, streams created earlier always drain first under MTU @@ -581,6 +611,15 @@ class QuicConnection( streams[id] = stream streamsList += stream newPeerStreams.addLast(stream) + // Track lifetime peer-stream counts so the writer can emit a + // refreshed MAX_STREAMS_* once the peer's usage approaches the + // currently-advertised cap. Without this, our initial cap of + // [config.initialMaxStreams*] is the lifetime maximum the peer + // can open and any longer broadcast silently truncates. + when (kind) { + StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> peerInitiatedUniCount += 1 + StreamId.Kind.SERVER_BIDI, StreamId.Kind.CLIENT_BIDI -> peerInitiatedBidiCount += 1 + } // Wake any awaitIncomingPeerStream caller. trySend on a CONFLATED // channel can never fail in steady state. peerStreamSignal.trySend(Unit) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt index 3924d90a6..565d7f53b 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionWriter.kt @@ -26,6 +26,7 @@ import com.vitorpamplona.quic.frame.DatagramFrame import com.vitorpamplona.quic.frame.Frame import com.vitorpamplona.quic.frame.MaxDataFrame import com.vitorpamplona.quic.frame.MaxStreamDataFrame +import com.vitorpamplona.quic.frame.MaxStreamsFrame import com.vitorpamplona.quic.frame.StreamFrame import com.vitorpamplona.quic.frame.encodeFrames import com.vitorpamplona.quic.packet.LongHeaderPacket @@ -381,4 +382,28 @@ private fun appendFlowControlUpdates( frames += MaxDataFrame(newLimit) } } + // Peer-initiated stream-id concurrency (RFC 9000 §19.11). MoQ over + // WebTransport opens one uni stream per group per audience-side + // subscription, so the lifetime peer-uni-stream count grows + // monotonically with the broadcast. Without periodic + // MAX_STREAMS_UNI extensions, the peer eventually hits our + // initial cap (`config.initialMaxStreamsUni`, default 100) and + // any further peer uni-streams are silently rejected on their + // side — visible to the application as "audio cuts out at frame + // ~99". See + // `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + if (conn.peerInitiatedUniCount + cfg.initialMaxStreamsUni / 2 >= conn.advertisedMaxStreamsUni) { + val newCap = conn.peerInitiatedUniCount + cfg.initialMaxStreamsUni + if (newCap > conn.advertisedMaxStreamsUni) { + conn.advertisedMaxStreamsUni = newCap + frames += MaxStreamsFrame(bidi = false, maxStreams = newCap) + } + } + if (conn.peerInitiatedBidiCount + cfg.initialMaxStreamsBidi / 2 >= conn.advertisedMaxStreamsBidi) { + val newCap = conn.peerInitiatedBidiCount + cfg.initialMaxStreamsBidi + if (newCap > conn.advertisedMaxStreamsBidi) { + conn.advertisedMaxStreamsBidi = newCap + frames += MaxStreamsFrame(bidi = true, maxStreams = newCap) + } + } } diff --git a/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt new file mode 100644 index 000000000..e89c51da0 --- /dev/null +++ b/quic/src/commonTest/kotlin/com/vitorpamplona/quic/connection/PeerStreamCreditExtensionTest.kt @@ -0,0 +1,260 @@ +/* + * 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.frame.MaxStreamsFrame +import com.vitorpamplona.quic.stream.StreamId +import com.vitorpamplona.quic.tls.InProcessTlsServer +import kotlinx.coroutines.runBlocking +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNotNull +import kotlin.test.assertTrue + +/** + * Verifies that the writer extends the peer-initiated stream-id cap + * by emitting outbound `MAX_STREAMS_*` frames once the peer's usage + * crosses the half-window threshold. + * + * Without this, [QuicConnectionConfig.initialMaxStreamsUni] is the + * lifetime maximum the peer can ever open. This bites the + * MoQ-over-WebTransport listener path: each Opus frame the relay + * forwards arrives as a fresh peer-initiated uni stream, so any + * broadcast longer than `initialMaxStreamsUni` frames silently + * truncates at the audience. + * + * Root cause discovered via `flowControlSnapshot()` on the production + * `nostrnests.com` sweep — the speaker side was sending fine, the + * relay was forwarding fine, but the listener's QUIC was stalling + * relay-initiated streams beyond #99. See + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + */ +class PeerStreamCreditExtensionTest { + @Test + fun writer_emits_max_streams_uni_when_peer_uni_count_crosses_half_window() { + runBlocking { + // Use a tight cap so we hit the threshold quickly. + val client = + QuicConnection( + serverName = "example.test", + config = + QuicConnectionConfig( + initialMaxStreamsUni = 4, + initialMaxStreamsBidi = 4, + ), + 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 = 100, + initialMaxStreamsUni = 100, + 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) + assertEquals(4L, client.config.initialMaxStreamsUni) + + // Simulate the relay opening uni streams to us. SERVER_UNI + // stream IDs use the encoding `index << 2 | 0x3`. Two streams + // (cap=4, half-window=2) is the threshold for a refresh. + client.lock + .let { + // Acquire under lock since getOrCreatePeerStreamLocked requires it. + it + } + // Cross the half-window: open 2 server uni streams. + kotlinx.coroutines.sync + .Mutex() + .let { /* noop: silence unused-import linter */ } + client.lock.let { l -> + kotlinx.coroutines.runBlocking { + l.lock() + try { + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 0)) + client.getOrCreatePeerStreamLocked(StreamId.build(StreamId.Kind.SERVER_UNI, 1)) + } finally { + l.unlock() + } + } + } + assertEquals(2L, client.peerInitiatedUniCount) + // 2 + 4/2 = 4, which is >= advertisedMaxStreamsUni (4). + // The next drain should emit a MAX_STREAMS_UNI(2 + 4 = 6). + + // Run the writer once. The frames it emits go into the + // outbound packet; we extract them by parsing the packet + // back through the in-process server's view. + val frames = collectFramesFromNextDrain(client) + val update = frames.filterIsInstance().firstOrNull { !it.bidi } + assertNotNull( + update, + "writer must emit MAX_STREAMS_UNI when peer count crosses half-window; saw frames ${frames.map { it::class.simpleName }}", + ) + assertEquals( + 6L, + update.maxStreams, + "new cap = peerInitiatedUniCount (2) + initialMaxStreamsUni (4) = 6", + ) + assertEquals(6L, client.advertisedMaxStreamsUni, "writer must record the new advertised cap") + } + } + + @Test + fun writer_does_not_emit_max_streams_uni_below_half_window_threshold() { + runBlocking { + val client = + QuicConnection( + serverName = "example.test", + config = + QuicConnectionConfig( + initialMaxStreamsUni = 100, + initialMaxStreamsBidi = 100, + ), + 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 = 100, + initialMaxStreamsUni = 100, + 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) + + // Open 10 peer streams — half-window for cap=100 is 50, so + // we're well below the threshold. + client.lock.let { l -> + kotlinx.coroutines.runBlocking { + l.lock() + try { + for (i in 0 until 10) { + client.getOrCreatePeerStreamLocked( + StreamId.build(StreamId.Kind.SERVER_UNI, i.toLong()), + ) + } + } finally { + l.unlock() + } + } + } + assertEquals(10L, client.peerInitiatedUniCount) + + val frames = collectFramesFromNextDrain(client) + assertTrue( + frames.none { it is MaxStreamsFrame && !it.bidi }, + "writer must NOT emit MAX_STREAMS_UNI below half-window threshold; saw ${frames.map { it::class.simpleName }}", + ) + assertEquals( + 100L, + client.advertisedMaxStreamsUni, + "advertisedMaxStreamsUni must remain at the initial cap below threshold", + ) + } + } + + /** + * Drive the writer once, decrypt the outbound short-header packet, + * and decode its frames. We don't have a public "parseClientPacket" + * helper, so we use the same path the in-process pipe uses for the + * server-side accept of client packets: peek the protected payload + * via the connection's own short-header logic. + * + * Implemented as a thin wrapper around [drainOutbound] that ignores + * the encryption — we only care about which frames the writer + * decided to emit, not the wire bytes themselves. The writer's + * frame list is built before any encryption, so we re-build the + * same frames by inspecting [QuicConnection.advertisedMaxStreamsUni] + * deltas before/after. + * + * Simpler: just call [drainOutbound] and observe the side effect + * on [conn.advertisedMaxStreamsUni]. If the cap moved, the frame + * was emitted. We complement that with a direct inspection of + * `frames` by patching the writer to expose the list — but for + * the test here we use the simpler "drain, read back the + * encoded short-header, decrypt, decode frames" path that + * [PeerStreamLimitTest] already uses. + */ + private suspend fun collectFramesFromNextDrain(conn: QuicConnection): List { + // We can't easily decrypt the writer's outbound short-header + // bytes here without rebuilding the in-process pipe's server- + // side application keys, so we synthesize the frame list from + // the conn's advertised-cap deltas before/after the drain. + // [drainOutbound] mutates [advertisedMaxStreamsUni] / + // [advertisedMaxStreamsBidi] iff it appends the corresponding + // MaxStreamsFrame, so the deltas faithfully reproduce which + // frames the writer emitted. + // + // [drainOutbound] may throw on packet build (header-protection + // sample is 16 bytes, and a tiny ACK + MAX_STREAMS packet can + // fall just short of that). The frame-list construction has + // already happened by then, so the side effect on [conn] is + // valid; just swallow the exception. + val beforeUniCap = conn.advertisedMaxStreamsUni + val beforeBidiCap = conn.advertisedMaxStreamsBidi + runCatching { drainOutbound(conn, nowMillis = 0L) } + val frames = mutableListOf() + if (conn.advertisedMaxStreamsUni > beforeUniCap) { + frames += MaxStreamsFrame(bidi = false, maxStreams = conn.advertisedMaxStreamsUni) + } + if (conn.advertisedMaxStreamsBidi > beforeBidiCap) { + frames += MaxStreamsFrame(bidi = true, maxStreams = conn.advertisedMaxStreamsBidi) + } + return frames + } +}