From 1fc8dbceeebede46cfd133727f117063a6e7988f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 19:27:19 +0000 Subject: [PATCH] debug(nests): trace announce-flow chain to localise still-empty _announcedSpeakers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The replay=64 SharedFlow fix in d6517cf did NOT close the receiver's "announced=0" symptom — the next two-phone repro (15:20:08 onward) still shows the cliff detector ticking with `active=1 announced=0` indefinitely, even after the session-internal pumpAnnounceWatch clearly received an Active update. Either moq-rs sends Active to only the FIRST announce bidi (so the VM-level second bidi never gets it), OR the chain from that bidi to `_announcedSpeakers` is broken at a layer my last fix didn't visit. Adds focused logging at every step of the announce chain so the next repro pins down which one. All NestRx-tagged: `MoqLiteSession.announce` (per-bidi pump): - "session.announce(prefix='…') bidi opened, pump launching" - "session.announce(prefix='…') bidi pump emit #N status=… suffix='…' chunks=…" - "session.announce(prefix='…') bidi.incoming() ended naturally chunks=… emits=…" - per-emission so we can compare bidi#1 (session-internal) vs bidi#2 (VM-level) emit counts. If bidi#2 has 0 emits, the relay is sending Actives to only one bidi. `MoqLiteNestsListener.announces`: - "MoqLiteNestsListener.announces flow STARTING (state=…)" - "MoqLiteNestsListener.announces opened bidi#2 (VM-level)" - per-emission "bidi#2 #N status=… suffix='…'" - "bidi#2 collect ENDED naturally" / "finally emissions=N" `ReconnectingNestsListener.announces` (wrapper): - "wrapper.announces() flow starting collect on activeListener" - "wrapper.announces() iter=N activeListener=set/null" - "wrapper.announces() iter=N inner state=Connected/Closed/…" - "wrapper.announces() iter=N inner collect ended naturally/X fwd=N" `NestViewModel.observeAnnounces`: - "observeAnnounces launching collect on l.announces()" - per-emission "observeAnnounces #N active=… pubkey='…' → ADD/REMOVE" - "observeAnnounces collect ENDED naturally/X (emissions=N, _announcedSpeakers.size=…)" After the next repro, the receiver-side log will show one of: - bidi#2 never opens (terminalOrConnected != Connected somehow) - bidi#2 opens, no chunks/emits → relay-side issue - bidi#2 emits, but `MoqLiteNestsListener.announces` doesn't forward → bug in the flow body - forwarding works but observeAnnounces collect ends → bug in `ReconnectingNestsListener.announces` collectLatest behavior - everything works but `_announcedSpeakers.size=0` → mutation lost Tests: full suite still passes (CliffDetectorTest 12, NestPlayerTest 12, NestBroadcasterTest 6, MoqLiteSessionTest 11, …). --- .../commons/viewmodels/NestViewModel.kt | 26 ++++++++++++++----- .../nestsclient/MoqLiteNestsListener.kt | 13 ++++++++++ .../nestsclient/ReconnectingNestsListener.kt | 18 +++++++++++-- .../nestsclient/moq/lite/MoqLiteSession.kt | 16 ++++++++++-- 4 files changed, 62 insertions(+), 11 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 efd73e927..6776346a0 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 @@ -820,15 +820,27 @@ class NestViewModel( announcesJob?.cancel() announcesJob = viewModelScope.launch { - runCatching { - l.announces().collect { ann -> - if (closed) return@collect - if (ann.active) { - _announcedSpeakers.update { it + ann.pubkey } - } else { - _announcedSpeakers.update { it - ann.pubkey } + com.vitorpamplona.quartz.utils.Log + .d("NestRx") { "observeAnnounces launching collect on l.announces()" } + var emissionCount = 0 + val outcome = + runCatching { + l.announces().collect { ann -> + if (closed) return@collect + emissionCount += 1 + com.vitorpamplona.quartz.utils.Log.d("NestRx") { + "observeAnnounces #$emissionCount active=${ann.active} pubkey='${ann.pubkey.take(12)}' → ${if (ann.active) "ADD" else "REMOVE"}" + } + if (ann.active) { + _announcedSpeakers.update { it + ann.pubkey } + } else { + _announcedSpeakers.update { it - ann.pubkey } + } } } + com.vitorpamplona.quartz.utils.Log.w("NestRx") { + val why = outcome.exceptionOrNull()?.let { "${it::class.simpleName}: ${it.message}" } ?: "naturally" + "observeAnnounces collect ENDED $why (emissions=$emissionCount, _announcedSpeakers.size=${_announcedSpeakers.value.size})" } } } diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt index 5986ab11c..3ee7fcf97 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsListener.kt @@ -128,6 +128,8 @@ class MoqLiteNestsListener internal constructor( override fun announces(): Flow = flow { + com.vitorpamplona.quartz.utils.Log + .d("NestRx") { "MoqLiteNestsListener.announces flow STARTING (state=${state.value::class.simpleName})" } check(state.value is NestsListenerState.Connected) { "NestsListener.announces requires Connected state, was ${state.value}" } @@ -136,8 +138,15 @@ class MoqLiteNestsListener internal constructor( // suffix is the broadcast's path component within the // room — for nests this is the speaker pubkey hex. val handle = session.announce(prefix = "") + com.vitorpamplona.quartz.utils.Log + .d("NestRx") { "MoqLiteNestsListener.announces opened bidi#2 (VM-level)" } + var emissionCount = 0 try { handle.updates.collect { announce -> + emissionCount += 1 + com.vitorpamplona.quartz.utils.Log.d("NestRx") { + "MoqLiteNestsListener.announces bidi#2 #$emissionCount status=${announce.status} suffix='${announce.suffix.take(12)}'" + } emit( RoomAnnouncement( pubkey = announce.suffix, @@ -145,7 +154,11 @@ class MoqLiteNestsListener internal constructor( ), ) } + com.vitorpamplona.quartz.utils.Log + .w("NestRx") { "MoqLiteNestsListener.announces bidi#2 collect ENDED naturally after $emissionCount emissions" } } finally { + com.vitorpamplona.quartz.utils.Log + .w("NestRx") { "MoqLiteNestsListener.announces bidi#2 finally (emissions=$emissionCount)" } // The flow's `finally` runs on both normal completion // and cancellation — let cancel propagate after closing // the announce handle. diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 1b1b9b7a8..71bcd9b64 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -252,7 +252,11 @@ private class ReconnectingHandle( */ override fun announces(): Flow = flow { + Log.d("NestRx") { "wrapper.announces() flow starting collect on activeListener" } + var iter = 0 activeListener.collectLatest { listener -> + iter += 1 + Log.d("NestRx") { "wrapper.announces() iter=$iter activeListener=${if (listener == null) "null" else "set"}" } if (listener == null) return@collectLatest val terminalOrConnected = listener.state.first { state -> @@ -260,9 +264,19 @@ private class ReconnectingHandle( state is NestsListenerState.Closed || state is NestsListenerState.Failed } + Log.d("NestRx") { "wrapper.announces() iter=$iter inner state=${terminalOrConnected::class.simpleName}" } if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest - runCatching { - listener.announces().collect { emit(it) } + var fwd = 0 + val outcome = + runCatching { + listener.announces().collect { + fwd += 1 + emit(it) + } + } + Log.w("NestRx") { + val why = outcome.exceptionOrNull()?.let { "${it::class.simpleName}: ${it.message}" } ?: "naturally" + "wrapper.announces() iter=$iter inner collect ended $why fwd=$fwd" } } } 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 54c50b88b..903c7526e 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 @@ -136,21 +136,33 @@ class MoqLiteSession internal constructor( replay = 64, onBufferOverflow = BufferOverflow.DROP_OLDEST, ) + Log.d("NestRx") { "session.announce(prefix='$prefix') bidi opened, pump launching (replayCap=64)" } val pump = scope.launch { val buffer = MoqLiteFrameBuffer() + var chunkCount = 0 + var emitCount = 0 try { bidi.incoming().collect { chunk -> + chunkCount += 1 buffer.push(chunk) while (true) { val payload = buffer.readSizePrefixed() ?: break - updates.emit(MoqLiteCodec.decodeAnnounce(payload)) + val decoded = MoqLiteCodec.decodeAnnounce(payload) + emitCount += 1 + Log.d("NestRx") { + "session.announce(prefix='$prefix') bidi pump emit #$emitCount " + + "status=${decoded.status} suffix='${decoded.suffix.take(12)}' " + + "(chunks=$chunkCount)" + } + updates.emit(decoded) } } + Log.w("NestRx") { "session.announce(prefix='$prefix') bidi.incoming() ended naturally (chunks=$chunkCount, emits=$emitCount)" } } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { - Log.w("NestRx") { "announce(prefix='$prefix'): bidi.incoming() threw ${t::class.simpleName}: ${t.message}" } + Log.w("NestRx") { "announce(prefix='$prefix'): bidi.incoming() threw ${t::class.simpleName}: ${t.message} (chunks=$chunkCount, emits=$emitCount)" } // Flow terminated (peer FIN or transport close). // The Announce stream's emit-side just stops; consumers // see an end-of-flow.