chase residual stream loss: bump SO_RCVBUF to 4 MiB + diagnostic UDP counters + cheaper test collector

Three targeted changes against the residual sustained-load stream loss
that survived the MAX_STREAMS_UNI fix (long broadcasts and bursty
scenarios still cliffed mid-pump). Each addresses one of the round-1
hypotheses; the next prod sweep should tell us how much each
contributes.

quic/UdpSocket:
  - Bump SO_RCVBUF to 4 MiB at bind time. The kernel default (~200 KB
    on Linux/macOS, similar on Android) holds barely 130 MTU-sized
    datagrams; a multi-second moq-lite broadcast that the relay fans
    out to several subscribers transiently overflows this. Anything
    queued past `rmem` is silently dropped by the kernel and never
    reaches our QUIC stack.
  - Add lifetime diagnostic counters (receivedDatagramCount,
    receivedByteCount, receiveBufferSizeBytes) on the expect surface
    so commonMain code can read them via the new udpStatsSupplier
    hook on QuicConnection without each platform having to surface its
    own native bookkeeping.
  - QuicConnectionDriver wires the supplier on start; the connection's
    flowControlSnapshot bundles the stats into a new
    QuicFlowControlSnapshot.udp field.

QuicConnection.flowControlSnapshot:
  - Now also surfaces advertisedMaxStreamsUni / advertisedMaxStreamsBidi
    and peerInitiatedUniCount / peerInitiatedBidiCount so a sweep can
    see the listener-side stream-cap evolution alongside speaker-side
    counters that were already there.

SendTraceScenario:
  - verbosePerFrame default flipped from true to false. The per-frame
    `tx i=…` and `rx[idx] gid=…` logs go through InteropDebug ->
    JUnit's stdout capture; at 50 frames/sec the capture thread
    serialises the receive coroutine and starves the QUIC read loop,
    biasing the recorded received count downward on long runs. Tests
    that need the per-frame timeline opt in explicitly.
  - Replace the receive-side CopyOnWriteArrayList sink (O(N) per add,
    ~1.1M element copies cumulative for a 1500-frame run) with
    Collections.synchronizedList(ArrayList(capacityHint)) — O(1)
    amortised. Snapshot under the same lock at end of run so the
    JDK's iterator-locking contract is satisfied.
  - fc-pre / fc-post-pump / fc-post-grace lines now include
    `udpDatagrams=… udpBytes=… udpRcvBuf=…` plus
    `advertisedMaxStreamsUni=… peerInitiatedUni=…` so the next sweep's
    diagnostic dump shows directly whether the kernel actually
    delivered datagrams the relay sent.

Plan doc: nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md
documents round-2 changes and lists the remaining open hypotheses
(relay-side per-subscriber queue policy, receive-side
MAX_STREAM_DATA threshold).

All :quic + :nestsClient JVM tests pass.

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