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,