mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-07-22 15:52:22 +00:00
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:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user