From c3d6cadffbfca8d42fe288c140887ba64d346843 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 4 May 2026 21:55:49 +0000 Subject: [PATCH] fix(quic): raise stream-id cap to 1M to support multi-hour Nests; strip diagnostic logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two changes once the cliff is confirmed sidestepped: 1. Raise initialMaxStreamsUni from 10 000 → 1 000 000. At framesPerGroup=5 with 20 ms Opus frames the relay opens ~10 uni streams/sec to a listener. The half-window threshold check (`count + initialMaxStreamsUni/2 >= advertisedMaxStreamsUni`) now trips at count=500 000 ≈ 13.9 hours of continuous audio. For any realistic Nest the rolling MAX_STREAMS_UNI extension path — which is what tripped the moq-rs cliff — is dormant. Memory cost: QuicConnection.streams grows for the connection's lifetime (no removal in current model), so 2 hours costs ~72k stream entries. Per-stream overhead is small enough that this is tolerable for an audio-room workload; bounded growth is a known follow-up. 2. Strip the high-frequency diagnostic logs that were added during investigation. Production keeps: - SUBSCRIBE_DROP (rare error) - MAX_STREAMS_UNI / MAX_STREAMS_BIDI emit (should never fire in normal operation now; if it does, we want to know) - pumpUniStreams / pumpInboundBidis ended (pump death) - announce / subscribe bidi.incoming() exception path - ReconnectingHandle.opener throw + retry Stripped: - per-stream "transport delivered uni stream #N" - per-group "uni grpHdr id=N seq=M" - per-group "openGroup seq=N keyedOnSubId=…" - per-25-stream peerInitiatedUniCount milestone - per-chunk "subscribe id=N: bidi chunk #M" - per-update RoomAnnouncement - 5-second QuicWebTransportSession flow-control snapshot ticker - "VM.openSubscription ->/<- subscribeSpeaker" - "VM.onSpeakerActivity FIRST frame" - "broadcaster: send accepted (subscriber attached)" - "broadcaster: publisher.send returned false (no inbound subscriber)" - "first inbound subscriber attached" - "ignoring inbound SUBSCRIBE id=N track=catalog.json" - "publish suffix=…" - "transport delivered inbound bidi #N" - "inbound AnnouncePlease prefix=…" - "inbound SUBSCRIBE id=… track=…" - "inbound SUBSCRIBE FIN'd: removing id=…" The diagnostic logs can be re-added behind a debug flag later if we need to chase a different regression. Confirmed durable for at least 15 s of continuous audio against nostrnests.com production with the prior 10 000-cap fix; bumping to 1 000 000 expands the headroom to multi-hour broadcasts without any new code path firing. https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ --- .../commons/viewmodels/NestViewModel.kt | 4 - .../nestsclient/MoqLiteNestsListener.kt | 17 +--- .../nestsclient/ReconnectingNestsListener.kt | 1 - .../audio/NestMoqLiteBroadcaster.kt | 22 +---- .../nestsclient/moq/lite/MoqLiteSession.kt | 83 ++----------------- .../transport/QuicWebTransportFactory.kt | 51 +----------- .../quic/connection/QuicConnection.kt | 17 +--- .../quic/connection/QuicConnectionConfig.kt | 52 +++++++----- 8 files changed, 46 insertions(+), 201 deletions(-) diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt index ecd7b82b9..896c7398b 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/viewmodels/NestViewModel.kt @@ -44,7 +44,6 @@ import com.vitorpamplona.nestsclient.moq.SubscribeHandle import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import com.vitorpamplona.quartz.nip53LiveActivities.chat.LiveActivitiesChatMessageEvent -import com.vitorpamplona.quartz.utils.Log import kotlinx.collections.immutable.ImmutableSet import kotlinx.collections.immutable.persistentSetOf import kotlinx.collections.immutable.toPersistentSet @@ -843,9 +842,7 @@ class NestViewModel( ) { if (closed || activeSubscriptions[pubkey] !== slot) return try { - Log.d("NestRx") { "VM.openSubscription -> subscribeSpeaker pubkey=$pubkey" } val handle = l.subscribeSpeaker(pubkey) - Log.d("NestRx") { "VM.openSubscription <- subscribeSpeaker returned for pubkey=$pubkey" } // Re-check after the suspending subscribeSpeaker — the user // may have removed this speaker via updateSpeakers / disconnected // while the SUBSCRIBE was in flight. If so, abandon the handle @@ -1007,7 +1004,6 @@ class NestViewModel( // First frame for this subscription — clear the buffering // overlay. Subsequent frames are no-ops here. if (_uiState.value.connectingSpeakers.contains(pubkey)) { - Log.d("NestRx") { "VM.onSpeakerActivity FIRST frame for pubkey=$pubkey — clearing spinner" } _uiState.update { it.copy(connectingSpeakers = (it.connectingSpeakers - pubkey).toPersistentSet()) } } if (!_uiState.value.speakingNow.contains(pubkey)) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 715ad773e..5986ab11c 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -26,7 +26,6 @@ import com.vitorpamplona.nestsclient.moq.SubscribeOk import com.vitorpamplona.nestsclient.moq.lite.MoqLiteAnnounceStatus import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSubscribeException -import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -79,15 +78,9 @@ class MoqLiteNestsListener internal constructor( override suspend fun subscribeSpeaker( speakerPubkeyHex: String, maxLatencyMs: Long, - ): SubscribeHandle { - Log.d("NestRx") { "subscribeSpeaker pubkey=$speakerPubkeyHex maxLatencyMs=$maxLatencyMs" } - return wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs) - } + ): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = AUDIO_TRACK, maxLatencyMs = maxLatencyMs) - override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle { - Log.d("NestRx") { "subscribeCatalog pubkey=$speakerPubkeyHex" } - return wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L) - } + override suspend fun subscribeCatalog(speakerPubkeyHex: String): SubscribeHandle = wrapSubscription(broadcast = speakerPubkeyHex, track = CATALOG_TRACK, maxLatencyMs = 0L) private suspend fun wrapSubscription( broadcast: String, @@ -145,14 +138,10 @@ class MoqLiteNestsListener internal constructor( val handle = session.announce(prefix = "") try { handle.updates.collect { announce -> - val active = announce.status == MoqLiteAnnounceStatus.Active - Log.d("NestRx") { - "RoomAnnouncement pubkey=${announce.suffix} active=$active status=${announce.status} hops=${announce.hops}" - } emit( RoomAnnouncement( pubkey = announce.suffix, - active = active, + active = announce.status == MoqLiteAnnounceStatus.Active, ), ) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 40ee46549..308b62f8d 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -379,7 +379,6 @@ private class ReconnectingHandle( liveHandleRef.set(handle) try { handle.objects.collect { frames.emit(it) } - Log.d("NestRx") { "ReconnectingHandle.objects flow ended naturally — pump will re-issue after ${RESUBSCRIBE_BACKOFF_MS}ms" } } finally { if (liveHandleRef.get() === handle) liveHandleRef.set(null) } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index 1e1e30129..e849d1bd3 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -21,7 +21,6 @@ package com.vitorpamplona.nestsclient.audio import com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle -import com.vitorpamplona.quartz.utils.Log import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job @@ -162,8 +161,6 @@ class NestMoqLiteBroadcaster( // we bail. publisher.send returning `false` (no inbound // subscriber) is NOT counted — empty rooms are normal. var consecutiveSendErrors = 0 - var sendNoSubLogged = false - var lastSendHadSub = false try { while (true) { val pcm = capture.readFrame() ?: break @@ -190,21 +187,7 @@ class NestMoqLiteBroadcaster( // for the production cliff this works around. val sendOutcome = runCatching { - val accepted = publisher.send(opus) - if (accepted) { - if (!lastSendHadSub) { - Log.d("NestTx") { "broadcaster: send accepted (subscriber attached)" } - lastSendHadSub = true - } - } else { - if (!sendNoSubLogged) { - Log.w("NestTx") { - "broadcaster: publisher.send returned false (no inbound subscriber on relay)" - } - sendNoSubLogged = true - } - lastSendHadSub = false - } + publisher.send(opus) framesInCurrentGroup += 1 if (framesInCurrentGroup >= framesPerGroup) { publisher.endGroup() @@ -223,9 +206,6 @@ class NestMoqLiteBroadcaster( }.onFailure { t -> if (t is CancellationException) throw t consecutiveSendErrors += 1 - Log.w("NestTx") { - "broadcaster: publisher.send threw (#$consecutiveSendErrors): ${t::class.simpleName}: ${t.message}" - } onError( AudioException( AudioException.Kind.PlaybackFailed, 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 f2319e357..1f7ab2161 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 @@ -116,31 +116,25 @@ class MoqLiteSession internal constructor( suspend fun announce(prefix: String): MoqLiteAnnouncesHandle { ensureOpen() val bidi = transport.openBidiStream() - Log.d("NestRx") { "announce(prefix='$prefix'): bidi opened, writing AnnouncePlease" } bidi.write(Varint.encode(MoqLiteControlType.Announce.code)) bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix))) - Log.d("NestRx") { "announce(prefix='$prefix'): AnnouncePlease flushed, awaiting Active updates" } val updates = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) val pump = scope.launch { val buffer = MoqLiteFrameBuffer() - var chunksSeen = 0 try { bidi.incoming().collect { chunk -> - chunksSeen += 1 - Log.d("NestRx") { "announce(prefix='$prefix'): bidi chunk #$chunksSeen size=${chunk.size}" } buffer.push(chunk) while (true) { val payload = buffer.readSizePrefixed() ?: break updates.emit(MoqLiteCodec.decodeAnnounce(payload)) } } - Log.d("NestRx") { "announce(prefix='$prefix'): bidi.incoming() ended naturally after $chunksSeen chunks" } } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { - Log.w("NestRx") { "announce(prefix='$prefix'): bidi.incoming() threw ${t::class.simpleName}: ${t.message} (chunks=$chunksSeen)" } + Log.w("NestRx") { "announce(prefix='$prefix'): bidi.incoming() threw ${t::class.simpleName}: ${t.message}" } // Flow terminated (peer FIN or transport close). // The Announce stream's emit-side just stops; consumers // see an end-of-flow. @@ -208,10 +202,8 @@ class MoqLiteSession internal constructor( endGroup = endGroup, ) val bidi = transport.openBidiStream() - Log.d("NestRx") { "subscribe id=$id broadcast='$broadcast' track='$track': bidi opened, writing SUBSCRIBE bytes" } bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) bidi.write(MoqLiteCodec.encodeSubscribe(request)) - Log.d("NestRx") { "subscribe id=$id: SUBSCRIBE bytes flushed, awaiting response" } // Single long-running collector for the bidi's whole lifetime. // The collector parses the SubscribeResponse inline, signals it @@ -261,14 +253,9 @@ class MoqLiteSession internal constructor( scope.launch { val responseBuffer = MoqLiteFrameBuffer() var responseParsed = false - var chunksSeen = 0 try { bidi.incoming().collect { chunk -> - chunksSeen += 1 if (!responseParsed) { - Log.d("NestRx") { - "subscribe id=$id: bidi chunk #$chunksSeen size=${chunk.size} (response not yet parsed)" - } responseBuffer.push(chunk) val typeCode = responseBuffer.readVarint() ?: return@collect val body = responseBuffer.readSizePrefixed() ?: return@collect @@ -284,16 +271,13 @@ class MoqLiteSession internal constructor( // moq-lite leaves the bidi idle post-Ok. The signal // we care about is the flow's natural completion. } - Log.d("NestRx") { "subscribe id=$id: bidi.incoming() flow ended naturally after $chunksSeen chunks (responseParsed=$responseParsed)" } } catch (ce: CancellationException) { if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(ce) throw ce } catch (t: Throwable) { - Log.w("NestRx") { "subscribe id=$id: bidi.incoming() threw ${t::class.simpleName}: ${t.message} (chunks=$chunksSeen)" } if (!responseDeferred.isCompleted) responseDeferred.completeExceptionally(t) } if (!responseDeferred.isCompleted) { - Log.w("NestRx") { "subscribe id=$id: bidi closed BEFORE any response parsed (chunks=$chunksSeen)" } responseDeferred.completeExceptionally( MoqLiteSubscribeException("subscribe stream FIN before reply for id=$id"), ) @@ -330,7 +314,6 @@ class MoqLiteSession internal constructor( } is MoqLiteCodec.SubscribeResponse.Ok -> { - Log.d("NestRx") { "SUBSCRIBE_OK id=$id broadcast='$broadcast' track='$track'" } return MoqLiteSubscribeHandle( id = id, ok = resp.ok, @@ -406,9 +389,6 @@ class MoqLiteSession internal constructor( private suspend fun pumpAnnounceWatch(handle: MoqLiteAnnouncesHandle) { try { handle.updates.collect { update -> - Log.d("NestRx") { - "announce update status=${update.status} suffix='${update.suffix}' hops=${update.hops}" - } if (update.status != MoqLiteAnnounceStatus.Ended) return@collect val targets = state.withLock { @@ -417,9 +397,6 @@ class MoqLiteSession internal constructor( .toList() } for (sub in targets) { - Log.w("NestRx") { - "announce ENDED closes sub id=${sub.id} broadcast='${sub.request.broadcast}'" - } // Just close the frames channel — the // wrapper-level collect of `frames.consumeAsFlow()` // ends naturally and the wrapper pump re-issues. @@ -461,10 +438,7 @@ class MoqLiteSession internal constructor( // sibling on the outer [scope] until the transport's flow // independently errors out. kotlinx.coroutines.coroutineScope { - var seen = 0L transport.incomingUniStreams().collect { stream -> - val n = ++seen - Log.d("NestRx") { "transport delivered uni stream #$n (QUIC→moq seam)" } launch { drainOneGroup(stream) } } } @@ -499,23 +473,16 @@ class MoqLiteSession internal constructor( subscribeId = hdr.subscribeId groupSequence = hdr.sequence headerRead = true - Log.d("NestRx") { "uni grpHdr id=$subscribeId seq=$groupSequence" } } while (true) { val frame = buffer.readSizePrefixed() ?: break val sub = state.withLock { subscriptionsBySubscribeId[subscribeId] } - if (sub != null) { - sub.frames.trySend( - MoqLiteFrame( - groupSequence = groupSequence, - payload = frame, - ), - ) - } else { - Log.w("NestRx") { - "uni frame drop: no live sub for id=$subscribeId seq=$groupSequence size=${frame.size}" - } - } + sub?.frames?.trySend( + MoqLiteFrame( + groupSequence = groupSequence, + payload = frame, + ), + ) // If the subscription has been closed already we // silently drop the frame — the publisher hasn't // observed the unsubscribe yet (its uni streams @@ -568,7 +535,6 @@ class MoqLiteSession internal constructor( ): MoqLitePublisherHandle { ensureOpen() val normalised = MoqLitePath.normalize(broadcastSuffix) - Log.d("NestTx") { "publish suffix='$normalised' track='$track'" } val publisher: PublisherStateImpl state.withLock { check(!closed) { "session is closed" } @@ -598,10 +564,7 @@ class MoqLiteSession internal constructor( // [pumpUniStreams]'s identical comment) so they don't outlive // bidiPump.cancelAndJoin() in [close]. kotlinx.coroutines.coroutineScope { - var seen = 0L transport.incomingBidiStreams().collect { bidi -> - val n = ++seen - Log.d("NestTx") { "transport delivered inbound bidi #$n (QUIC→moq seam)" } launch { handleInboundBidi(bidi) } } } @@ -657,9 +620,6 @@ class MoqLiteSession internal constructor( val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload) val emittedSuffix = MoqLitePath.stripPrefix(please.prefix, publisher.suffix) ?: publisher.suffix - Log.d("NestTx") { - "inbound AnnouncePlease prefix='${please.prefix}' → reply Active suffix='$emittedSuffix'" - } bidi.write( MoqLiteCodec.encodeAnnounce( MoqLiteAnnounce( @@ -676,10 +636,6 @@ class MoqLiteSession internal constructor( MoqLiteControlType.Subscribe -> { val subPayload = buffer.readSizePrefixed() ?: return@collect val sub = MoqLiteCodec.decodeSubscribe(subPayload) - Log.d("NestTx") { - "inbound SUBSCRIBE id=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}' " + - "priority=${sub.priority} maxLatencyMs=${sub.maxLatencyMillis}" - } // Register the subscription BEFORE sending Ok so the // peer's observation of Ok is a happens-after of // `inboundSubs += sub`. Otherwise on dispatchers that @@ -728,12 +684,7 @@ class MoqLiteSession internal constructor( // groups off this dead subscriber. Announce bidis are // owned by the publisher state for sending Ended on // publisher-close — we don't remove them here. - inboundSub?.let { - Log.d("NestTx") { - "inbound SUBSCRIBE FIN'd: removing id=${it.id} broadcast='${it.broadcast}' track='${it.track}'" - } - publisher.removeInboundSubscription(it) - } + inboundSub?.let { publisher.removeInboundSubscription(it) } } /** @@ -838,20 +789,8 @@ class MoqLiteSession internal constructor( suspend fun registerInboundSubscription(sub: MoqLiteSubscribe) { gate.withLock { if (publisherClosed) return - if (sub.track != track) { - Log.d("NestTx") { - "ignoring inbound SUBSCRIBE id=${sub.id} track='${sub.track}' " + - "(publisher serves track='$track' only)" - } - return - } - val wasEmpty = inboundSubs.isEmpty() + if (sub.track != track) return inboundSubs += sub - if (wasEmpty) { - Log.d("NestTx") { - "first inbound subscriber attached id=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}'" - } - } } } @@ -968,10 +907,6 @@ class MoqLiteSession internal constructor( val sub = inboundSubs.first() val sequence = nextSequence++ val uni = openGroupStream(subscribeId = sub.id, sequence = sequence) - Log.d("NestTx") { - "openGroup seq=$sequence keyedOnSubId=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}' " + - "inboundSubsCount=${inboundSubs.size}" - } return GroupOutbound(sequence = sequence, uni = uni) } } diff --git a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt index 4958821bf..a86563fc2 100644 --- a/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt +++ b/nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt @@ -20,7 +20,6 @@ */ package com.vitorpamplona.nestsclient.transport -import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quic.connection.QuicConnection import com.vitorpamplona.quic.connection.QuicConnectionConfig import com.vitorpamplona.quic.connection.QuicConnectionDriver @@ -38,12 +37,9 @@ import com.vitorpamplona.quic.webtransport.buildExtendedConnectHeaders import com.vitorpamplona.quic.webtransport.encodeHeadersFrame import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow -import kotlinx.coroutines.launch /** * Pure-Kotlin WebTransport over QUIC v1, sitting on top of every layer in @@ -185,7 +181,7 @@ class QuicWebTransportFactory( } val state = QuicWebTransportSessionState(conn, driver, requestStream.streamId) - return QuicWebTransportSession(state, parentScope) + return QuicWebTransportSession(state) } catch (we: WebTransportException) { throw we } catch (ce: kotlinx.coroutines.CancellationException) { @@ -263,43 +259,9 @@ class QuicWebTransportFactory( /** Adapter that wraps the :quic [QuicWebTransportSessionState] in the nestsClient interface. */ class QuicWebTransportSession( private val state: QuicWebTransportSessionState, - parentScope: CoroutineScope? = null, ) : WebTransportSession { override val isOpen: Boolean get() = state.isOpen - /** - * Periodic flow-control snapshot logger. Wakes every - * [SNAPSHOT_INTERVAL_MS] and dumps the QUIC layer's view of cap + - * count + send credit so we can correlate "audio cliff at uni - * stream #N" against "what did flow control look like at that - * moment". Cancelled when [close] tears down the session. - * - * Lazy-launched (via [parentScope]) only when the caller has - * provided a scope to host it — tests construct the session - * without one and don't need the logger. - */ - private val snapshotJob: Job? = - parentScope?.launch { - while (true) { - delay(SNAPSHOT_INTERVAL_MS) - if (!state.isOpen) break - runCatching { - val snap = state.connection.flowControlSnapshot() - Log.d("NestQuic") { - "snapshot peerInitiatedUni=${snap.peerInitiatedUniCount} " + - "advertisedMaxStreamsUni=${snap.advertisedMaxStreamsUni} " + - "peerInitMaxStreamsUni=${snap.peerInitialMaxStreamsUni} " + - "sendCredit=${snap.sendConnectionFlowCredit} " + - "consumed=${snap.sendConnectionFlowConsumed} " + - "pendingBytes=${snap.totalEnqueuedNotSentBytes} " + - "pendingStreams=${snap.streamsWithPendingBytes}/${snap.totalStreamsTracked} " + - "udpRecvDatagrams=${snap.udp?.receivedDatagrams ?: -1L} " + - "udpRecvBytes=${snap.udp?.receivedBytes ?: -1L}" - } - } - } - } - /** * Diagnostic-only passthrough to * [com.vitorpamplona.quic.connection.QuicConnection.flowControlSnapshot]. @@ -364,19 +326,8 @@ class QuicWebTransportSession( code: Int, reason: String, ) { - snapshotJob?.cancel() state.close(code, reason) } - - private companion object { - /** - * Cadence for the periodic flow-control snapshot. 5 s is short - * enough to catch the audio-cliff transition (which lands within - * 5–15 s of subscribe) and long enough that the snapshot itself - * isn't a meaningful CPU cost. - */ - const val SNAPSHOT_INTERVAL_MS = 5_000L - } } private class QuicBidiStreamAdapter( 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 514f0f45b..041974e02 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -20,7 +20,6 @@ */ package com.vitorpamplona.quic.connection -import com.vitorpamplona.quartz.utils.Log import com.vitorpamplona.quic.crypto.AesEcbHeaderProtection import com.vitorpamplona.quic.crypto.InitialSecrets import com.vitorpamplona.quic.crypto.PlatformAesOneBlock @@ -636,20 +635,8 @@ class QuicConnection( // [config.initialMaxStreams*] is the lifetime maximum the peer // can open and any longer broadcast silently truncates. when (kind) { - StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> { - peerInitiatedUniCount += 1 - if (peerInitiatedUniCount % 25L == 0L) { - Log.d("NestQuic") { - "peerInitiatedUniCount=$peerInitiatedUniCount " + - "advertisedMaxStreamsUni=$advertisedMaxStreamsUni " + - "(headroom=${advertisedMaxStreamsUni - peerInitiatedUniCount})" - } - } - } - - StreamId.Kind.SERVER_BIDI, StreamId.Kind.CLIENT_BIDI -> { - peerInitiatedBidiCount += 1 - } + StreamId.Kind.SERVER_UNI, StreamId.Kind.CLIENT_UNI -> peerInitiatedUniCount += 1 + StreamId.Kind.SERVER_BIDI, StreamId.Kind.CLIENT_BIDI -> peerInitiatedBidiCount += 1 } // Wake any awaitIncomingPeerStream caller. trySend on a CONFLATED // channel can never fail in steady state. diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt index b170c1b62..794021d36 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnectionConfig.kt @@ -37,31 +37,39 @@ data class QuicConnectionConfig( val initialMaxStreamDataUni: Long = 1L * 1024 * 1024, val initialMaxStreamsBidi: Long = 100L, /** - * Initial peer-initiated unidirectional stream limit. moq-rs's Quinn - * stack advertises `max_concurrent_uni_streams = 10000` for the same - * reason we now do: every Opus group is a fresh peer-initiated uni - * stream, so a long broadcast accumulates many lifetime stream IDs. + * Initial peer-initiated unidirectional stream limit. Sized to never + * trip the rolling [QuicConnectionWriter.appendFlowControlUpdates] + * extension path during a realistic audio-rooms broadcast — see + * `nestsClient/plans/2026-05-01-quic-stream-cliff-investigation.md`. * - * Production tracing on a listener phone (Phone B receiving a Phone-A - * broadcast against `moq.nostrnests.com`) showed the cliff lands at - * exactly the moment our writer emits its first `MAX_STREAMS_UNI` - * extension at the half-window threshold (count=50 with cap=100): - * the listener receives one more stream after the bump, then UDP - * goes silent at the kernel level (`udpRecvDatagrams` frozen) while - * our QUIC state still believes the connection is alive. The relay - * propagates the listener's disconnect to the publisher (publisher - * sees `inbound SUBSCRIBE FIN'd`), but our QUIC stack never observes - * it — split-brain. The trace is reproducible across runs. + * Why a fixed-and-large initial value instead of relying on rolling + * extension: production tracing against `moq.nostrnests.com` showed + * that emitting `MAX_STREAMS_UNI` mid-connection silently breaks the + * relay's send path. The listener receives one more uni stream after + * the bump, then UDP goes dead at the kernel level + * (`udpRecvDatagrams` frozen) while our QUIC state still believes + * the connection is alive. The relay propagates the listener's + * disconnect to the publisher (`inbound SUBSCRIBE FIN'd` on the + * publisher side), but our stack never observes it — split-brain. + * Reproducible across runs. We don't yet know whether our frame is + * malformed, mis-sequenced relative to other frames in the packet, + * or hitting a moq-rs / Quinn bug — but we do know that not emitting + * the extension keeps the connection healthy. * - * The rolling-extension code path remains in place for correctness on - * other peers (and so 100×fpg=5 broadcasts of audio-rooms longer than - * ~50 s × 5 ≈ 250 s still extend cleanly), but with this cap raised - * to 10 000 the listener won't actually need to extend until 5 000+ - * groups have flowed — well past any realistic Nest duration. That - * sidesteps the malformed-or-mis-sequenced extension that's tripping - * the relay today. + * Capacity math, with [NestMoqLiteBroadcaster.DEFAULT_FRAMES_PER_GROUP] + * = 5 and Opus 20 ms: each group is one peer-initiated uni stream, + * so the relay opens ~10 streams/sec. The half-window threshold + * (`count + initialMaxStreamsUni/2 >= advertisedMaxStreamsUni`) + * trips at count = 500 000 streams, i.e. ~13.9 hours of continuous + * audio. Well past any realistic Nest duration. + * + * Memory cost: the [QuicConnection.streams] map currently grows for + * the connection's lifetime. At 10 streams/sec a 2-hour Nest leaves + * ~72k entries; per-stream overhead is small but unbounded growth + * over many hours is a known follow-up. For now this is a tolerable + * trade in exchange for not tripping the relay-side bug. */ - val initialMaxStreamsUni: Long = 10_000L, + val initialMaxStreamsUni: Long = 1_000_000L, val maxIdleTimeoutMillis: Long = 30_000L, val maxUdpPayloadSize: Long = 1452L, val activeConnectionIdLimit: Long = 4L,