From d6517cf346d54d91b841f46037e77e87e25dee52 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 19:11:51 +0000 Subject: [PATCH] fix(nests): announce SharedFlow late-attach race dropped Active updates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cliff-detector diagnostic logs from the user's two-phone repro (commit f2205dc, run 15:06) showed the smoking gun — every tick from the receiver-side cliff detector said `announced=0` even though an Active announce had clearly been received earlier (the session-level `pumpAnnounceWatch` logged it). The cliff detector only acts on speakers that are in `_announcedSpeakers`, so it never tripped, even when frames clearly stopped flowing for 6+ s. Root cause: `MoqLiteSession.announce` builds a `MutableSharedFlow` with `replay = 0` to bridge the bidi pump and the consumer's `collect`. With Kotlin's `MutableSharedFlow(replay = 0)`, an `emit` that happens BEFORE the first collector subscribes is silently dropped — there's no buffer for items emitted into a flow with no subscribers. The bidi pump is launched inside `announce()` and starts collecting `bidi.incoming()` immediately; the caller (the flow returned by `MoqLiteNestsListener.announces()`) attaches its collector AFTER `session.announce()` returns. In between, if the relay sends an Active update (which it does the moment the broadcaster's session is registered), the pump's emit hits a flow with no collectors and is lost. In the user's logs both an internal session-level `pumpAnnounceWatch` AND the VM-level `observeAnnounces` open separate announce bidis. The internal one started first and won the race for its bidi's Active. The VM-level one lost — the bidi pump emitted Active before the consumer subscribed, the SharedFlow dropped it, and `_announcedSpeakers` never populated. Fix: change the announce SharedFlow to `replay = 64` with `BufferOverflow.DROP_OLDEST`. Late-attaching collectors now see up to the last 64 announces (far more than nests rooms have speakers — typical stage size is ≤ 10). The bidi pump's emit still never suspends, so QUIC-level backpressure can't propagate. The replay cache evicts the oldest item if more than 64 announces arrive before any collector attaches; for the production workload this is unreachable. Tests: nestsClient jvmTest passes 57/57 unchanged (MoqLiteCodecTest, MoqLiteFrameBufferTest, MoqLitePathTest, MoqLiteSessionTest, NestBroadcasterTest, NestPlayerTest). The session-level pumpAnnounceWatch path used the same SharedFlow but its caller subscribes from `scope.launch { pumpAnnounceWatch(...) }` inside `ensureAnnounceWatchStarted` — close enough in time that it won the race in production. The fix closes the race for both callers regardless of timing. --- .../nestsclient/moq/lite/MoqLiteSession.kt | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) 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 6fc3a2755..54c50b88b 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 @@ -119,7 +119,23 @@ class MoqLiteSession internal constructor( bidi.write(Varint.encode(MoqLiteControlType.Announce.code)) bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix))) - val updates = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) + // replay=64, DROP_OLDEST: announces emitted before the + // caller's `collect` attaches MUST NOT be dropped — that's + // the late-attach race that left `_announcedSpeakers` empty + // in production receiver logs even after a clear Active + // arrived (the session-internal pumpAnnounceWatch beat the + // VM-level `observeAnnounces` to subscribing, then the + // VM-level bidi's pump emitted to a SharedFlow with no + // collector yet, and the prior `replay=0` silently dropped + // it). Replay buffer covers up to 64 distinct announces — + // far more than nests rooms have speakers — and DROP_OLDEST + // means the bidi pump never has to suspend on emit, so + // backpressure can't propagate up into the QUIC read loop. + val updates = + MutableSharedFlow( + replay = 64, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) val pump = scope.launch { val buffer = MoqLiteFrameBuffer()