From e27abf49a0f07df1566c1067ecb4832af7893797 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 15:22:58 +0000 Subject: [PATCH] diagnostic: snapshot listener-side flow control too, not just speaker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 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. --- .../nestsclient/MoqLiteNestsListener.kt | 10 ++ .../nestsclient/moq/lite/MoqLiteSession.kt | 9 +- ...trNestsSustainedSendOutcomesInteropTest.kt | 14 ++- .../NostrnestsProdAudioTransmissionTest.kt | 26 ++++- .../nestsclient/interop/SendTraceScenario.kt | 110 +++++++++++------- 5 files changed, 118 insertions(+), 51 deletions(-) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 07a1be7ef..88e9b837b 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -65,6 +65,16 @@ class MoqLiteNestsListener internal constructor( ) : NestsListener { override val state: StateFlow = mutableState.asStateFlow() + /** + * `internal` accessor for diagnostics: a test downcasts the + * returned [WebTransportSession] to a platform-specific type + * (e.g. `QuicWebTransportSession`) to read flow-control counters + * from the underlying QUIC connection. Not part of the public + * [com.vitorpamplona.nestsclient.NestsListener] surface. + */ + internal val transport + get() = session.transport + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK) override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index 94f043bf8..c96fdd72a 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -65,7 +65,14 @@ import kotlinx.coroutines.sync.withLock * layout. */ class MoqLiteSession internal constructor( - private val transport: WebTransportSession, + /** + * `internal` (not `private`) so test code in the same module can + * downcast to a platform [WebTransportSession] (typically + * `QuicWebTransportSession`) and read diagnostic flow-control + * counters from the underlying QUIC connection. Production code + * paths inside this file continue to use `transport` as before. + */ + internal val transport: WebTransportSession, private val scope: CoroutineScope, ) { private val state = Mutex() diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt index ae050352f..3ae5197f7 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt @@ -386,7 +386,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { ) = runBlocking { NostrNestsHarness.assumeNestsInterop() val harness = harnessOrNull ?: return@runBlocking - withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot -> + withHarnessSpeakerAndListeners(scope, harness, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot, listenerFlowControlSnapshots -> val result = SendTraceScenario.run( scope = scope, @@ -396,6 +396,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { scenario = scenario, pumpScope = pumpScope, flowControlSnapshot = flowControlSnapshot, + listenerFlowControlSnapshots = listenerFlowControlSnapshots, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -412,6 +413,7 @@ class NostrNestsSustainedSendOutcomesInteropTest { speakerPubkeyHex: String, pumpScope: CoroutineScope, flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, + listenerFlowControlSnapshots: List com.vitorpamplona.quic.connection.QuicFlowControlSnapshot>, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -474,12 +476,22 @@ class NostrNestsSustainedSendOutcomesInteropTest { // — see that helper's comment for rationale. val quicSpeakerWt = speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val listenerSnapshots = + listeners.map { listener -> + val listenerWt = + (listener as com.vitorpamplona.nestsclient.MoqLiteNestsListener) + .transport as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val supplier: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot = + { listenerWt.quicFlowControlSnapshot() } + supplier + } block( publisher, listeners, hostSigner.pubKey, pumpScope, { quicSpeakerWt.quicFlowControlSnapshot() }, + listenerSnapshots, ) } finally { for (listener in listeners) { diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt index 938dfc229..94b3880f0 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt @@ -944,7 +944,7 @@ class NostrnestsProdAudioTransmissionTest { expectAllReceived: Boolean = false, ) = runBlocking { assumeProd() - withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot -> + withProdSpeakerAndListeners(scope, scenario.parallelSubscriptions) { publisher, listeners, hostPub, pumpScope, flowControlSnapshot, listenerFlowControlSnapshots -> val result = SendTraceScenario.run( scope = scope, @@ -954,6 +954,7 @@ class NostrnestsProdAudioTransmissionTest { scenario = scenario, pumpScope = pumpScope, flowControlSnapshot = flowControlSnapshot, + listenerFlowControlSnapshots = listenerFlowControlSnapshots, ) SendTraceScenario.reportAndAssert(scope, result, expectAllReceived) } @@ -978,6 +979,7 @@ class NostrnestsProdAudioTransmissionTest { speakerPubkeyHex: String, pumpScope: CoroutineScope, flowControlSnapshot: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot, + listenerFlowControlSnapshots: List com.vitorpamplona.quic.connection.QuicFlowControlSnapshot>, ) -> Unit, ) { val hostSigner = NostrSignerInternal(KeyPair()) @@ -1032,18 +1034,32 @@ class NostrnestsProdAudioTransmissionTest { listeners += listener } - // Diagnostics passthrough: cast the speaker WT to the - // concrete QUIC adapter so the scenario can read the - // underlying connection's flow-control snapshot. See - // `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. + // Diagnostics passthrough: cast the speaker WT and each + // listener's underlying WT to the concrete QUIC adapter + // so the scenario can read flow-control snapshots from + // both sides. The cliff investigation + // (`nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`) + // needs the listener-side numbers because the + // peer-initiated stream count is what bites us at the + // audience. val quicSpeakerWt = speakerWt as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val listenerSnapshots = + listeners.map { listener -> + val listenerWt = + (listener as com.vitorpamplona.nestsclient.MoqLiteNestsListener) + .transport as com.vitorpamplona.nestsclient.transport.QuicWebTransportSession + val supplier: suspend () -> com.vitorpamplona.quic.connection.QuicFlowControlSnapshot = + { listenerWt.quicFlowControlSnapshot() } + supplier + } block( publisher, listeners, hostSigner.pubKey, pumpScope, { quicSpeakerWt.quicFlowControlSnapshot() }, + listenerSnapshots, ) } finally { for (listener in listeners) { diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt index b182fa5f5..e3008a2b5 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/SendTraceScenario.kt @@ -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 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,