diagnostic: snapshot listener-side flow control too, not just speaker

The previous round of fc-* lines exclusively reported the speaker's QUIC connection state. That hid the real story for the residual stream loss: each Opus frame the relay forwards arrives as a fresh peer-initiated uni stream on the LISTENER side, so the listener's peerInitiatedUni counter and advertisedMaxStreamsUni evolution are what matter — the speaker's view is stable at peerInitiatedUni=1 for the whole run (only the WT control stream is relay-initiated on that side). Changes: - MoqLiteSession.transport changed from `private` to `internal val` so test code can downcast. - MoqLiteNestsListener exposes an `internal val transport` getter that returns the underlying WebTransportSession via the session's new internal accessor. - SendTraceScenario.run takes a new optional `listenerFlowControlSnapshots: List<suspend () -> QuicFlowControlSnapshot>`. When supplied, it emits `fc-listener[idx]-pre / -post-pump / -post-grace` lines in the same shape as the speaker side. - Refactored the three checkpoint emit sites into a shared `logFcSnapshot(scope, label, snap, includeRcvBuf)` helper so the speaker and listener lines print identical column layouts. - withProdSpeakerAndListeners + withHarnessSpeakerAndListeners cast each listener to MoqLiteNestsListener (the only impl in the tree) and downcast its WebTransportSession to QuicWebTransportSession, yielding a per-listener snapshot supplier list to the scenario block. What the next sweep tells us: for each cliffed scenario, fc-listener[0] should now show whether the listener is the bottleneck (e.g. peerInitiatedUni stuck below the relay's intent, or udpDatagrams not matching the relay's send count) or whether the relay just stopped forwarding (peerInitiatedUni climbs but matches the partial received count). All :quic and :nestsClient JVM tests pass.
2026-05-01 15:22:58 +00:00
parent 46af65591a
commit e27abf49a0
5 changed files with 118 additions and 51 deletions
@@ -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)
@@ -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()
@@ -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) {
@@ -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) {
@@ -205,28 +205,29 @@ 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} " +
-                    "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}",
-            )
+            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)
@@ -332,20 +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} " +
-                    "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}",
-            )
+            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;
@@ -357,20 +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} " +
-                    "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}",
-            )
+            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(
@@ -390,6 +375,43 @@ object SendTraceScenario {
        )
    }

+    /**
+     * 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,