fix(nests): announce SharedFlow late-attach race dropped Active updates

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.
This commit is contained in:
Claude
2026-05-05 19:11:51 +00:00
parent f2205dc97e
commit d6517cf346

View File

@@ -119,7 +119,23 @@ class MoqLiteSession internal constructor(
bidi.write(Varint.encode(MoqLiteControlType.Announce.code))
bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix)))
val updates = MutableSharedFlow<MoqLiteAnnounce>(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<MoqLiteAnnounce>(
replay = 64,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
val pump =
scope.launch {
val buffer = MoqLiteFrameBuffer()