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 aa9c9231e..7bd7d40b8 100644 --- a/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md +++ b/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md @@ -1,11 +1,37 @@ # QUIC stream cliff against nostrnests.com — investigation plan -**Status: PARTIALLY FIXED.** Original 99-frame production cliff is -gone (verified `sweep_frames_200` at `received=200/200 missing=[]`, -plus baseline / cad / payload sweeps now 100/100). Residual loss -appears in extreme scenarios (very long broadcasts, high payload -sizes, mid-stream pauses, very fast bursts) — see "Residual loss -after the MAX_STREAMS fix" section below. +**Status: PRODUCTION-FIXED via two-layer fix.** + + 1. `:quic` now emits `MAX_STREAMS_UNI` extensions to widen the + listener's peer-initiated stream-id cap (commit `d391ae1d`). + Closes the original 99-frame cliff for short broadcasts. + + 2. `NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP = 5` packs five + Opus frames per moq-lite group, dropping client uni-stream + creation from 50/sec to 10/sec — comfortably under the + production nostrnests relay's sustained per-subscriber forward + ceiling of ~40 streams/sec. Closes the long-broadcast residual + where the relay's per-subscriber queue would overflow after 6–14 + seconds of continuous push and silently terminate forwarding to + that subscriber. + +Listener-side flow-control snapshots in the round-2 sweep confirmed +the residual was strictly relay-side: every uni stream the relay +opened to the listener delivered cleanly to the application +(`peerInitiatedUni == received + 1`, where `+1` is the WT control +stream); `pendingBytes == 0`; cap headroom unused. The relay simply +stopped opening new streams to the listener once its per-subscriber +buffer overflowed. Reducing the stream-creation rate to 10/sec keeps +the relay's queue from ever filling. + +Late-join latency cost: ≤ 100 ms initial audio gap (one group +boundary) instead of ≤ 20 ms with `framesPerGroup = 1`. Imperceptible +for live audio. + +The remaining open work is **upstream**: file an issue at +`kixelated/moq` describing the per-subscriber forward queue +limitation so future relay versions can either lift the ceiling or +expose it as a config knob. **Original root cause: FIXED.** Our `:quic` client never emitted `MAX_STREAMS_*` frames to extend the peer-initiated stream-id cap. @@ -57,52 +83,79 @@ matches frame count) — so the loss is on the relay→listener path. (no late delivery during grace), confirming the listener never recovers once it stalls. -## Next investigation: relay-side per-subscription buffer +## Round 2 — chasing the residual -The post-fix profile suggests a *relay-side* per-subscriber forward -buffer or rate-limiter that drops new uni streams once the listener -falls behind. Hypotheses to test: +After the original `MAX_STREAMS_UNI` fix, residual loss appeared in +sustained / bursty scenarios. Three round-2 changes against +likely contributors: -1. **Listener QUIC RX coalescing.** Our `:quic` reads packets from - the UDP socket one at a time. Under high stream-creation rate - the kernel's UDP receive buffer might drop datagrams before we - pick them up. `recv` overhead per packet (we use a fresh - `ByteBuffer` per call in `UdpSocket.kt:60-70`) compounds. Worth - profiling with `dropwatch` or just `cat /proc/net/snmp | grep -i - udp` (RcvbufErrors counter) during a stuck run. +### 2a. Test-side stdout serialisation (hypothesis 2) -2. **Receive-side `MAX_STREAM_DATA` extension.** Symmetric to the - `MAX_STREAMS` fix: if our writer's threshold for re-crediting - per-stream data is too conservative under high stream count, - the relay's per-stream send credit to us could starve. Current - threshold is "consumed > limit - window/2" with `window = - 1 MB`. With 80-byte audio frames we'd never hit it on a single - stream, but for the 16 KB payload test, 8 KB consumed per - stream is well under the 500 KB threshold — so this isn't the - primary cause. +`SendTraceScenario.Scenario.verbosePerFrame` defaulted to `true`, +which logged a per-frame `tx i=…` and `rx[idx] gid=…` line via +`InteropDebug.checkpoint`. Under JUnit's stdout capture, that's a +synchronous write per event. At 50 frames/sec sustained the +capture thread can serialise the receive coroutine and starve the +QUIC read loop, biasing the `received` count downward. **Default +flipped to `false`**; specific tests opt in for debugging. -3. **Application-thread stalling on the listener side.** Our test - `SendTraceScenario` collector logs every `rx[…]` line via - `InteropDebug.checkpoint`, which writes to stdout under a JUnit - capture. Under 50+ stream/sec rates the stdout flush may - serialize the receive coroutine. Worth disabling - `verbosePerFrame` and re-running long scenarios — if numbers - improve dramatically, the receive side is application-rate - limited, not transport-bound. +### 2b. Test-side `CopyOnWriteArrayList` add cost -4. **moq-rs relay's `--cluster-replication` / per-subscriber queue - policy.** Some upstream knobs control how many in-flight uni - streams the relay queues per subscriber before dropping new - ones. Worth checking `moq-relay --help | grep -i buffer` or - filing an issue at `kixelated/moq` for the relay's behaviour - spec under saturation. +The collector accumulated arrivals in a `CopyOnWriteArrayList`, +which is O(N) per add — copying the entire underlying array on +every `+=`. For 1500-frame runs that's ~1.1 M element copies and +~35 MB of memory ops cumulative across the run. Replaced with +`Collections.synchronizedList(ArrayList(capacityHint))` for O(1) +amortized adds. Snapshot at end via +`synchronized(sink) { ArrayList(sink) }`. -The MAIN production bug (audio cuts out mid-broadcast at frame ~99 -in two-user rooms) is FIXED. The residual is a separate -investigation that mostly affects extreme scenarios — long -broadcasts, large payloads, fan-out asymmetry, fast bursts — and -should be tracked as a separate plan once one of the hypotheses -above is confirmed. +### 2c. Listener-side kernel UDP receive buffer (hypothesis 1) + +`UdpSocket.connect` did not configure `SO_RCVBUF`, so the kernel's +default applied (~200 KB on Linux/macOS, similarly small on +Android). At MTU-sized datagrams that's room for ~130 packets. A +multi-second moq-lite broadcast that the relay fans out across +multiple subscribers can transiently exceed this — anything +queued past `rmem` is silently dropped by the kernel and never +reaches our QUIC stack, manifesting as "subscription stops +mid-broadcast even though `publisher.send` keeps returning true". +Bumped to 4 MiB at socket-bind time (Linux doubles + clamps via +`rmem_max`, so the effective cap is whatever the kernel allows up +to `min(8 MiB, rmem_max)`). + +Added lifetime UDP datagram counters (`receivedDatagramCount`, +`receivedByteCount`, `receiveBufferSizeBytes`) to the `UdpSocket` +expect surface; the driver wires them into the existing +`QuicConnection.flowControlSnapshot` via a `udpStatsSupplier` hook. +The `fc-pre / fc-post-pump / fc-post-grace` lines now include +`udpDatagrams=… udpBytes=… udpRcvBuf=…` so a future sweep can +correlate "frames missing on subscription" against "datagrams the +kernel actually delivered". + +### Hypotheses still open after round 2 + +3. **moq-rs relay per-subscriber queue policy.** Even if our + listener kernel + collector are infinitely fast, the relay + itself may have an in-flight uni-stream cap per subscriber that + drops new groups once the listener falls behind. The + asymmetric loss in `sweep_3subs sub[2]` (77/100 while sub[0] + and sub[1] got 100/100) is consistent with a per-subscriber + queue rather than a connection- or relay-wide limit. Worth + filing an issue at `kixelated/moq` once we have one more sweep + confirming the round-2 changes don't already close the gap. + +4. **Receive-side `MAX_STREAM_DATA` thresholds.** Probably not + the cause for 80-byte audio frames (1 MB per-stream window vs + ~80 bytes used per stream) but may bite the 16 KB-payload + scenario. Worth a follow-up only if `sweep_payload_16kb` + doesn't recover. + +## Re-running after round 2 + +The acceptance criterion is the same: every sweep row showing +`received=N/N missing=[]`, except the explicit late-join scenarios +(`sweep_late_subscribe_after_25` / `_after_50`) which legitimately +report `received < N` due to from-latest semantics. **Reproduces against:** `https://moq.nostrnests.com:4443` (production relay). Not consistently reproducible against the local diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 07a1be7ef..88e9b837b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -65,6 +65,16 @@ class MoqLiteNestsListener internal constructor( ) : NestsListener { override val state: StateFlow = mutableState.asStateFlow() + /** + * `internal` accessor for diagnostics: a test downcasts the + * returned [WebTransportSession] to a platform-specific type + * (e.g. `QuicWebTransportSession`) to read flow-control counters + * from the underlying QUIC connection. Not part of the public + * [com.vitorpamplona.nestsclient.NestsListener] surface. + */ + internal val transport + get() = session.transport + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK) override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index 2bbd865c2..409949979 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -49,27 +49,42 @@ class NestMoqLiteBroadcaster( * varint-prefixed frames; [MoqLitePublisherHandle.endGroup] FINs * the stream and the next [send] starts a fresh group. * - * **Default = 1 (one Opus frame per group).** Matches the JS - * reference broadcaster's wire shape and gives any late-joining - * subscriber a sub-20 ms initial audio gap (moq-lite "from-latest" - * semantics — new subscribers pick up at the next group boundary). + * **Default = 5 (≈ 100 ms of audio per group → 10 streams/sec).** * - * Earlier versions defaulted to 5 as a mitigation for a production - * cliff at frame ~99: the relay would only forward the first ~100 - * uni streams to a listener for the lifetime of the connection. - * The actual root cause was on the listener's *receive* side — - * our `:quic` never emitted `MAX_STREAMS_UNI` frames to extend - * the peer-initiated stream-id cap, so the relay's initial 100 - * uni-stream allowance was the lifetime maximum. Once - * [com.vitorpamplona.quic.connection.QuicConnectionWriter] - * started emitting periodic `MAX_STREAMS_*` extensions, packing - * stopped being necessary. See - * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + * The story: * - * Set to a higher value to amortise per-stream overhead (CPU, - * relay bookkeeping) at the cost of late-join latency: with - * `framesPerGroup = 5` the late-join initial gap is ≤ 100 ms - * instead of ≤ 20 ms. + * Round 1 found that the production relay's lifetime peer-uni + * stream cap of 100 was bitten by our `:quic` never emitting + * `MAX_STREAMS_UNI` extensions; that fix landed and short + * broadcasts (≤ 100 frames) now deliver 100/100. + * + * Round 2 surfaced a *separate* residual cliff on **sustained** + * broadcasts: the production relay's per-subscriber forward + * pipeline runs at ≈ 40 streams/sec sustained. With one Opus + * frame per group = 50 streams/sec, the relay falls behind by + * ~9 streams/sec; once its per-subscriber buffer fills (after + * 6–14 seconds of sustained push) it stops forwarding to that + * subscriber entirely. Production sweeps reproduced this on + * `sweep_30s` (610/1500), `sweep_120s` (227/6000), + * `sweep_frames400` (36/400 in some runs, 400/400 in others — + * highly variance-prone). Listener-side flow-control snapshots + * confirm every stream the relay forwarded reached the + * application; the relay simply stops opening new uni streams + * once its queue overflows. + * + * Packing 5 frames per group cuts stream-creation rate to 10/sec, + * comfortably below the relay's sustained-forward ceiling. Sweep + * results with `framesPerGroup = 5` show 100/100 across every + * scenario including 30 s and 120 s broadcasts. Late-join gap + * grows from ≤ 20 ms (one frame per group) to ≤ 100 ms (five + * frames per group) — imperceptible for live audio rooms. + * + * See `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + * + * Set to 1 to match the JS reference broadcaster's wire shape + * exactly — fine for short / bursty broadcasts and useful when + * pointing at a relay deployment without the per-subscriber + * forward limit (e.g. self-hosted moq-rs with tuned config). */ private val framesPerGroup: Int = DEFAULT_FRAMES_PER_GROUP, ) { @@ -198,10 +213,14 @@ class NestMoqLiteBroadcaster( companion object { /** - * Default moq-lite group size = 1 Opus frame per group, matching - * the JS reference broadcaster's wire shape. See [framesPerGroup] - * kdoc for the full rationale + history. + * Default moq-lite group size = 5 Opus frames ≈ 100 ms of audio. + * Picked to keep the QUIC uni-stream creation rate + * (10 streams/sec at 20 ms cadence) under the production + * nostrnests relay's sustained per-subscriber forward + * ceiling (~40 streams/sec) while still giving late-joining + * subscribers a sub-100 ms initial audio gap. See + * [framesPerGroup] kdoc for the full rationale + history. */ - const val DEFAULT_FRAMES_PER_GROUP: Int = 1 + const val DEFAULT_FRAMES_PER_GROUP: Int = 5 } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 94f043bf8..c96fdd72a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -65,7 +65,14 @@ import kotlinx.coroutines.sync.withLock * layout. */ class MoqLiteSession internal constructor( - private val transport: WebTransportSession, + /** + * `internal` (not `private`) so test code in the same module can + * downcast to a platform [WebTransportSession] (typically + * `QuicWebTransportSession`) and read diagnostic flow-control + * counters from the underlying QUIC connection. Production code + * paths inside this file continue to use `transport` as before. + */ + internal val transport: WebTransportSession, private val scope: CoroutineScope, ) { private val state = Mutex() 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 ae050352f..3ae5197f7 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, flowControlSnapshot -> + withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot, listenerFlowControlSnapshots -> val result = SendTraceScenario.run( scope = scope, @@ -396,6 +396,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { scenario = scenario, pumpScope = pumpScope, flowControlSnapshot = flowControlSnapshot, + listenerFlowControlSnapshots = listenerFlowControlSnapshots, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -412,6 +413,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { speakerPubkeyHex: String, pumpScope: CoroutineScope, flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, + listenerFlowControlSnapshots: List com.vitorpamplona.quic.connection.QuicFlowControlSnapshot>, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -474,12 +476,22 @@ class NostrNestsSustainedSendOutcomesInteropTest { // — see that helper's comment for rationale. val quicSpeakerWt = speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val listenerSnapshots = + listeners.map { listener -> + val listenerWt = + (listener as com.vitorpamplona.nestsclient.MoqLiteNestsListener) + .transport as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val supplier: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot = + { listenerWt.quicFlowControlSnapshot() } + supplier + } block( publisher, listeners, hostSigner.pubKey, pumpScope, { quicSpeakerWt.quicFlowControlSnapshot() }, + listenerSnapshots, ) } finally { for (listener in listeners) { 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 938dfc229..94b3880f0 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, flowControlSnapshot -> + withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot, listenerFlowControlSnapshots -> val result = SendTraceScenario.run( scope = scope, @@ -954,6 +954,7 @@ class NostrnestsProdAudioTransmissionTest { scenario = scenario, pumpScope = pumpScope, flowControlSnapshot = flowControlSnapshot, + listenerFlowControlSnapshots = listenerFlowControlSnapshots, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -978,6 +979,7 @@ class NostrnestsProdAudioTransmissionTest { speakerPubkeyHex: String, pumpScope: CoroutineScope, flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, + listenerFlowControlSnapshots: List com.vitorpamplona.quic.connection.QuicFlowControlSnapshot>, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -1032,18 +1034,32 @@ class NostrnestsProdAudioTransmissionTest { listeners += listener } - // 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`. + // Diagnostics passthrough: cast the speaker WT and each + // listener's underlying WT to the concrete QUIC adapter + // so the scenario can read flow-control snapshots from + // both sides. The cliff investigation + // (`nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`) + // needs the listener-side numbers because the + // peer-initiated stream count is what bites us at the + // audience. val quicSpeakerWt = speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val listenerSnapshots = + listeners.map { listener -> + val listenerWt = + (listener as com.vitorpamplona.nestsclient.MoqLiteNestsListener) + .transport as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val supplier: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot = + { listenerWt.quicFlowControlSnapshot() } + supplier + } block( publisher, listeners, hostSigner.pubKey, pumpScope, { quicSpeakerWt.quicFlowControlSnapshot() }, + listenerSnapshots, ) } finally { for (listener in listeners) { 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 836d14dd1..e3008a2b5 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt @@ -94,8 +94,18 @@ data class Scenario( * subscriber over an independent uni stream. */ val parallelSubscriptions: Int = 1, - /** Larger payloads (e.g. 4 KB) test stream-write vs stream-creation cost. */ - val verbosePerFrame: Boolean = true, + /** + * Emit a `tx i=…` log line per send and an `rx[idx] gid=…` log + * line per arrival. Useful for debugging a specific scenario but + * produces a lot of stdout — at 50 frames/sec the JUnit capture + * thread can serialise the receive coroutine and starve the QUIC + * read loop, biasing the recorded `received` count downward. + * + * Default `false`. Opt in for a known-short test if you need the + * per-frame timeline. The summary line (`sub[…] received=N/M …`) + * always prints regardless. + */ + val verbosePerFrame: Boolean = false, ) /** @@ -195,31 +205,46 @@ object SendTraceScenario { scenario: Scenario, pumpScope: CoroutineScope, flowControlSnapshot: (suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot)? = null, + /** + * Optional listener-side snapshot suppliers, one per parallel + * subscriber. When provided, the scenario emits + * `fc-listener[idx]-pre/post-pump/post-grace` lines alongside + * the speaker-side `fc-*` lines so a test report can correlate + * loss against the audience-side QUIC state — particularly + * `peerInitiatedUni` (lifetime count of relay-opened uni + * streams the listener accepted) and `udpDatagrams` (datagrams + * the kernel actually delivered). + */ + listenerFlowControlSnapshots: List com.vitorpamplona.quic.connection.QuicFlowControlSnapshot> = emptyList(), ): 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}", - ) + logFcSnapshot(scope, "fc-pre", snap, includeRcvBuf = true) + } + listenerFlowControlSnapshots.forEachIndexed { idx, supplier -> + supplier().let { snap -> + logFcSnapshot(scope, "fc-listener[$idx]-pre", snap, includeRcvBuf = true) + } } val sendOutcomes = BooleanArray(scenario.frameCount) val sendDurationsMicros = LongArray(scenario.frameCount) val endGroupErrors = arrayOfNulls(scenario.frameCount) - val arrivalsPerSubscriber = + // Pre-sized synchronized ArrayList per subscriber. + // CopyOnWriteArrayList (the previous choice) is O(N) per add — + // at 50 frames/sec sustained, the cumulative copying becomes + // a non-trivial back-pressure on the receive coroutine and + // biases the recorded `received` count downward on long runs + // (sweep_30s went from 99/1500 to 653/1500 with the + // MAX_STREAMS fix; the residual gap is largely this and + // verbosePerFrame stdout). Synchronized ArrayList with a + // capacity hint is O(1) amortized. + val arrivalsPerSubscriber: List> = List(scenario.parallelSubscriptions) { - java.util.concurrent.CopyOnWriteArrayList() + java.util.Collections.synchronizedList(ArrayList(scenario.frameCount)) } val collectStart = System.currentTimeMillis() @@ -308,16 +333,12 @@ object SendTraceScenario { "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}", - ) + logFcSnapshot(scope, "fc-post-pump", snap, includeRcvBuf = false) + } + listenerFlowControlSnapshots.forEachIndexed { idx, supplier -> + supplier().let { snap -> + logFcSnapshot(scope, "fc-listener[$idx]-post-pump", snap, includeRcvBuf = false) + } } // Wait for collectors. If they hit `take(N)` they exit naturally; @@ -329,16 +350,12 @@ 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}", - ) + logFcSnapshot(scope, "fc-post-grace", snap, includeRcvBuf = false) + } + listenerFlowControlSnapshots.forEachIndexed { idx, supplier -> + supplier().let { snap -> + logFcSnapshot(scope, "fc-listener[$idx]-post-grace", snap, includeRcvBuf = false) + } } return ScenarioResult( @@ -346,20 +363,62 @@ object SendTraceScenario { sendOutcomes = sendOutcomes, sendDurationsMicros = sendDurationsMicros, endGroupErrors = endGroupErrors, - arrivalsPerSubscriber = arrivalsPerSubscriber.map { it.toList() }, + // Snapshot under lock — synchronizedList iteration must + // be guarded explicitly per the JDK contract. + arrivalsPerSubscriber = + arrivalsPerSubscriber.map { sink -> + synchronized(sink) { ArrayList(sink) } + }, pumpStartedAtMs = pumpStart - collectStart, pumpDurationMs = pumpDuration, collectStartedAtMs = collectStart, ) } + /** + * Format and log a [QuicFlowControlSnapshot] in a single line under + * the given checkpoint label. Used by both the speaker-side + * (`fc-pre`, `fc-post-pump`, `fc-post-grace`) and the listener-side + * (`fc-listener[idx]-…`) emission paths so the line shape matches + * exactly between them. + * + * `includeRcvBuf` is `true` only on the pre-pump checkpoint — + * the buffer size is set once at bind time and doesn't change, so + * repeating it on later snapshots wastes column space. + */ + private fun logFcSnapshot( + scope: String, + label: String, + snap: com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, + includeRcvBuf: Boolean, + ) { + InteropDebug.checkpoint( + scope, + "$label: peerInitMaxData=${snap.peerInitialMaxData} " + + "peerInitMaxStreamDataUni=${snap.peerInitialMaxStreamDataUni} " + + "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " + + "sendCredit=${snap.sendConnectionFlowCredit} " + + "consumed=${snap.sendConnectionFlowConsumed} " + + "(remaining=${snap.sendConnectionFlowCredit - snap.sendConnectionFlowConsumed}) " + + "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " + + "advertisedMaxStreamsUni=${snap.advertisedMaxStreamsUni} " + + "peerInitiatedUni=${snap.peerInitiatedUniCount} " + + "udpDatagrams=${snap.udp?.receivedDatagrams} " + + "udpBytes=${snap.udp?.receivedBytes} " + + (if (includeRcvBuf) "udpRcvBuf=${snap.udp?.receiveBufferSizeBytes} " else "") + + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", + ) + } + private fun spawnCollector( scope: String, subscriberIndex: Int, listener: NestsListener, speakerPubkeyHex: String, scenario: Scenario, - sink: java.util.concurrent.CopyOnWriteArrayList, + sink: MutableList, collectStart: Long, pumpScope: CoroutineScope, ): Job = 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 3d7427701..041974e02 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -172,6 +172,18 @@ class QuicConnection( /** Bidi counterpart of [advertisedMaxStreamsUni]. */ internal var advertisedMaxStreamsBidi: Long = config.initialMaxStreamsBidi + /** + * Optional supplier of underlying UDP-socket counters. Wired by the + * platform-specific driver since `UdpSocket`'s counters are + * JVM-side fields the commonMain side can't see directly. + * Diagnostic-only: surfaces in [flowControlSnapshot] so a test + * can correlate "frames lost on the wire" against "datagrams the + * kernel actually delivered to the application". Null when no + * driver is attached (in-process tests etc). + */ + @Volatile + internal var udpStatsSupplier: (() -> UdpSocketStats)? = null + /** * Round-robin starting index for the writer's stream-drain iteration. * Without rotation, streams created earlier always drain first under MTU @@ -451,6 +463,7 @@ class QuicConnection( pendingStreamCount += 1 } } + val udp = udpStatsSupplier?.invoke() QuicFlowControlSnapshot( peerInitialMaxData = tp?.initialMaxData, peerInitialMaxStreamDataUni = tp?.initialMaxStreamDataUni, @@ -463,9 +476,14 @@ class QuicConnection( peerMaxStreamsBidiCurrent = peerMaxStreamsBidi, nextLocalUniIndex = nextLocalUniIndex, nextLocalBidiIndex = nextLocalBidiIndex, + advertisedMaxStreamsUni = advertisedMaxStreamsUni, + advertisedMaxStreamsBidi = advertisedMaxStreamsBidi, + peerInitiatedUniCount = peerInitiatedUniCount, + peerInitiatedBidiCount = peerInitiatedBidiCount, totalEnqueuedNotSentBytes = pending, streamsWithPendingBytes = pendingStreamCount, totalStreamsTracked = streamsList.size, + udp = udp, ) } @@ -766,4 +784,50 @@ data class QuicFlowControlSnapshot( val streamsWithPendingBytes: Int, /** Number of streams currently tracked (alive + closed-but-retained). */ val totalStreamsTracked: Int, + /** + * Outbound peer-initiated stream cap we've currently advertised. + * Starts at `config.initialMaxStreams*` and grows as the writer + * emits `MAX_STREAMS_*` frames. If this stays at the initial cap + * while [peerInitiatedUniCount] climbs into the same range, the + * peer's stream-id allowance against us is starving — see + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + */ + val advertisedMaxStreamsUni: Long, + /** Bidi counterpart of [advertisedMaxStreamsUni]. */ + val advertisedMaxStreamsBidi: Long, + /** + * Lifetime count of peer-initiated unidirectional streams accepted + * by `getOrCreatePeerStreamLocked`. Compared against + * [advertisedMaxStreamsUni] by the writer to decide when to extend + * the cap. + */ + val peerInitiatedUniCount: Long, + /** Bidi counterpart of [peerInitiatedUniCount]. */ + val peerInitiatedBidiCount: Long, + /** + * Underlying UDP-socket counters when the connection has a + * platform driver attached. Null otherwise (in-process tests + * have no real socket). Lets a test answer "did the kernel + * actually deliver these datagrams to us?" by diffing + * [UdpSocketStats.receivedDatagrams] over the test window. + */ + val udp: UdpSocketStats?, +) + +/** + * Lifetime UDP-socket counters surfaced through + * [QuicFlowControlSnapshot]. The driver populates this from its + * platform-specific `UdpSocket` impl. + */ +data class UdpSocketStats( + /** Number of `recv()` calls that returned a non-null datagram. */ + val receivedDatagrams: Long, + /** Sum of payload bytes returned by [receivedDatagrams] calls. */ + val receivedBytes: Long, + /** + * Effective `SO_RCVBUF` value the kernel reports. On Linux the + * application requested value is *doubled* and then capped at + * `rmem_max`, so this is what the kernel actually allocates. + */ + val receiveBufferSizeBytes: Int, ) diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt index e3baf181d..9fd62b324 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionDriver.kt @@ -73,6 +73,17 @@ class QuicConnectionDriver( fun start() { connection.start() + // Wire the diagnostic UDP-stats supplier so + // QuicConnection.flowControlSnapshot can surface + // kernel-delivered datagram counters alongside the QUIC + // flow-control fields. + connection.udpStatsSupplier = { + UdpSocketStats( + receivedDatagrams = socket.receivedDatagramCount, + receivedBytes = socket.receivedByteCount, + receiveBufferSizeBytes = socket.receiveBufferSizeBytes, + ) + } readJob = scope.launch { readLoop() } sendJob = scope.launch { sendLoop() } // Initial nudge so the ClientHello goes out immediately. diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt index c60e280a2..3d45ecf48 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt @@ -49,6 +49,25 @@ expect class UdpSocket { /** Local port the OS assigned to the socket. */ val localPort: Int + /** + * Lifetime count of datagrams successfully returned by [receive]. + * Diagnostic-only — surfaces in + * [com.vitorpamplona.quic.connection.QuicFlowControlSnapshot.udp] + * so a test can correlate apparent stream loss against the + * datagrams the kernel actually delivered to the application. + */ + val receivedDatagramCount: Long + + /** Sum of payload bytes returned by [receive]. */ + val receivedByteCount: Long + + /** + * Effective `SO_RCVBUF` value the kernel reports. On Linux the + * application-requested value is doubled and then capped at + * `rmem_max`, so this is what the kernel actually allocates. + */ + val receiveBufferSizeBytes: Int + companion object { /** Open a UDP socket connected to [host]:[port]. Throws on resolution / bind / connect failure. */ suspend fun connect( diff --git a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt index fdd6420d5..e0967bc39 100644 --- a/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt +++ b/quic/src/jvmAndroid/kotlin/com/vitorpamplona/quic/transport/UdpSocket.kt @@ -24,10 +24,12 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import java.net.InetAddress import java.net.InetSocketAddress +import java.net.StandardSocketOptions import java.nio.ByteBuffer import java.nio.channels.ClosedChannelException import java.nio.channels.DatagramChannel import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.atomic.AtomicLong /** * JVM/Android UDP socket using blocking [DatagramChannel] dispatched onto @@ -49,9 +51,31 @@ actual class UdpSocket private constructor( // 64 KiB buffer was wasteful per connection. private val readBuf = ByteBuffer.allocate(2048) + /** + * Lifetime UDP datagram counters. Diagnostic-only. Useful for + * correlating apparent stream-loss against actual receive-side + * activity — if [receivedDatagramCount] plateaus while the + * speaker is still pumping, the loss is on the wire / kernel and + * not in the QUIC stack. Bytes counter feeds the same diff + * against + * [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot]'s + * `sendConnectionFlowConsumed` for the speaker side. + */ + private val receivedDatagrams: AtomicLong = AtomicLong(0L) + private val receivedBytes: AtomicLong = AtomicLong(0L) + actual val localPort: Int get() = (channel.localAddress as InetSocketAddress).port + actual val receivedDatagramCount: Long + get() = receivedDatagrams.get() + + actual val receivedByteCount: Long + get() = receivedBytes.get() + + actual val receiveBufferSizeBytes: Int + get() = channel.getOption(StandardSocketOptions.SO_RCVBUF) + actual suspend fun send(payload: ByteArray): Int = withContext(Dispatchers.IO) { if (closed.get()) throw ClosedChannelException() @@ -70,6 +94,8 @@ actual class UdpSocket private constructor( readBuf.flip() val out = ByteArray(readBuf.remaining()) readBuf.get(out) + receivedDatagrams.incrementAndGet() + receivedBytes.addAndGet(out.size.toLong()) out } catch (_: ClosedChannelException) { null @@ -96,6 +122,22 @@ actual class UdpSocket private constructor( val remote = InetSocketAddress(address, port) val channel = DatagramChannel.open() channel.configureBlocking(true) + // Bump SO_RCVBUF before bind so the kernel allocates a + // generous queue. Default rmem (~200 KB on Linux, + // similar elsewhere) holds barely 130 ~1500-byte + // datagrams — which is *exactly* the cap MoQ-over-WT + // listeners brush against once the relay is fanning + // out a multi-second broadcast (one peer-uni stream per + // group, multiple subscribers, occasional reorder / + // retransmit). Under the burst that follows handshake + // settle, anything queued past rmem is silently dropped + // by the kernel, manifesting downstream as + // "subscription stops mid-broadcast even though + // publisher.send keeps returning true". 4 MiB gives ~30 s + // of headroom at sustained 1 KB/frame audio rates. + runCatching { + channel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024 * 1024) + } channel.bind(InetSocketAddress(0)) // ephemeral // We use receive()/send(addr) instead of channel.connect() so that // sendDatagram-style flows can still be implemented on the same