Merge pull request #2687 from vitorpamplona/claude/audio-transmission-tests-zKzlB

Final fix on missed audio frames.

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

nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt

@@ -65,6 +65,16 @@ class MoqLiteNestsListener internal constructor(
 ) : NestsListener {
     override val state: StateFlow<NestsListenerState> = 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)

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

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()

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<suspend () -> 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) {

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<suspend () -> 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) {

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<suspend () -> 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<String>(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<MutableList<FrameArrival>> =
             List(scenario.parallelSubscriptions) {
-                java.util.concurrent.CopyOnWriteArrayList<FrameArrival>()
+                java.util.Collections.synchronizedList(ArrayList<FrameArrival>(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<FrameArrival>,
+        sink: MutableList<FrameArrival>,
         collectStart: Long,
         pumpScope: CoroutineScope,
     ): Job =

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

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.

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(

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