The diagnostic logging in 1fc8dbc surfaced the real root cause —
not the SharedFlow replay race I fixed in d6517cf, but a strict
flow{}-builder concurrency-confinement violation.
The receiver-side log (15:34:40, repro after the diagnostic patch)
showed:
session.announce(prefix='') bidi pump emit #1 status=Active suffix='fe52579aa30e' (chunks=1)
MoqLiteNestsListener.announces bidi#2 #1 status=Active suffix='fe52579aa30e'
MoqLiteNestsListener.announces bidi#2 finally (emissions=1)
wrapper.announces() iter=2 inner collect ended IllegalStateException:
Flow invariant is violated:
Emission from another coroutine is detected.
Child of StandaloneCoroutine{Active}@e9b5981, expected child of
StandaloneCoroutine{Active}@8309a26.
FlowCollector is not thread-safe and concurrent emissions are
prohibited.
To mitigate this restriction please use 'channelFlow' builder
instead of 'flow'
The session's bidi pump (`scope.launch { bidi.incoming().collect …
updates.emit(decoded) }`) emits to the announce SharedFlow from the
PUMP's coroutine. SharedFlow's `collect { … }` resumes the lambda
inline on the emitter's coroutine when there's no dispatcher hop,
so `MoqLiteNestsListener.announces`' `handle.updates.collect {
emit(RoomAnnouncement(…)) }` lambda fires on the pump's coroutine,
not the flow{} body's coroutine. `flow {}`'s SafeFlow guard throws
on the cross-coroutine emit. The wrapper's
`runCatching { listener.announces().collect { emit(it) } }` swallows
the exception, the wrapper's collect ends with `fwd=1`, and
`_announcedSpeakers` stays empty for the lifetime of the session.
That, in turn, kept the cliff detector permanently gated:
cliff-detector tick=N active=1 announced=0 lastFrameAges=[fe52579a=…ms]
even though the receiver was clearly subscribed and clearly seeing
frames. The cliff detector requires the speaker to be in
`announcedSpeakers` to recycle, so the relay-forward cliff at ~135
streams went undetected — the original two-phone receiver-side
silence symptom.
Fix: switch `MoqLiteNestsListener.announces()` from `flow { … }` to
`channelFlow { … }`. `channelFlow` is the documented escape hatch
for cross-coroutine emission (the error message itself recommends
it). The new shape:
channelFlow {
val handle = session.announce(prefix = "")
val pump = launch {
try {
handle.updates.collect { send(RoomAnnouncement(…)) }
} finally {
withContext(NonCancellable) {
runCatching { handle.close() }
}
}
}
awaitClose { pump.cancel() }
}
Notes:
- `send` (channel) replaces `emit` (flow) — channelFlow buffers
via a Channel, so cross-coroutine producers are explicitly
supported.
- The inner `collect` is wrapped in a launched child of the
channelFlow scope so we can use `awaitClose` for the producer-
side teardown contract channelFlow requires.
- `handle.close` is suspend (FINs the bidi + joins the pump);
putting it in the launched pump's `finally` under
`NonCancellable` keeps the FIN reliable when the consumer
cancels mid-stream.
Diagnostic logging from 1fc8dbc preserved — the next two-phone
repro should now show:
observeAnnounces #1 active=true pubkey='fe52579aa30e' → ADD
cliff-detector tick=N active=1 announced=1 lastFrameAges=[…]
…and on a real cliff event, the recycle should fire.
Tests: full suite still passes
- CliffDetectorTest: 12/12
- NestPlayerTest: 12/12
- NestBroadcasterTest: 6/6
- MoqLiteSessionTest: 11/11
- All other nestsClient jvmTest classes (~240 tests across 37
classes) green.
Claude
f17e7adfa7