nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(quic): raise stream-id cap to 1M to support multi-hour Nests; strip diagnostic logs

Browse Source 
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
This commit is contained in:
Claude
2026-05-04 21:55:49 +00:00
parent f32131d073
commit c3d6cadffb
8 changed files with 46 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt
+3 -14
View File
@@ -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,
),
)
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
-1
View File
@@ -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)
}
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt
+1 -21
View File
@@ -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,
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+9 -74
View File
@@ -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<MoqLiteAnnounce>(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)
}
}
nestsClient/src/jvmAndroid/kotlin/com/vitorpamplona/nestsclient/transport/QuicWebTransportFactory.kt
+1 -50
View File
@@ -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(