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..630273fa9 100644 --- a/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md +++ b/nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md @@ -57,52 +57,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/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt index 836d14dd1..b182fa5f5 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, ) /** @@ -208,6 +218,11 @@ object SendTraceScenario { "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " + "sendCredit=${snap.sendConnectionFlowCredit} consumed=${snap.sendConnectionFlowConsumed} " + "peerMaxStreamsUniNow=${snap.peerMaxStreamsUniCurrent} " + + "advertisedMaxStreamsUni=${snap.advertisedMaxStreamsUni} " + + "peerInitiatedUni=${snap.peerInitiatedUniCount} " + + "udpDatagrams=${snap.udp?.receivedDatagrams} " + + "udpBytes=${snap.udp?.receivedBytes} " + + "udpRcvBuf=${snap.udp?.receiveBufferSizeBytes} " + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", @@ -217,9 +232,18 @@ object SendTraceScenario { 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() @@ -314,6 +338,10 @@ object SendTraceScenario { "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} " + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", @@ -335,6 +363,10 @@ object SendTraceScenario { "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} " + "nextLocalUniIdx=${snap.nextLocalUniIndex} " + "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked}", @@ -346,7 +378,12 @@ 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, @@ -359,7 +396,7 @@ object SendTraceScenario { 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