fix(nests): wrapper.announces() ALSO needs channelFlow (collectLatest emits cross-coroutine)

The channelFlow conversion in f17e7ad fixed `MoqLiteNestsListener.announces`
but the SAME `IllegalStateException: Flow invariant is violated` kept
firing in the next two-phone repro (commit f17e7ad, run 15:46:51) — this
time the offending `flow{}` was one layer up, in
`ReconnectingNestsListener.announces`:

  override fun announces(): Flow<RoomAnnouncement> =
      flow {
          activeListener.collectLatest { listener ->
              ...
              runCatching {
                  listener.announces().collect { emit(it) }  // <-- HERE
              }
          }
      }

`Flow.collectLatest { lambda }` cancels and restarts a CHILD coroutine
for each upstream emission. The lambda body runs in that child, NOT in
the surrounding flow{} body's coroutine. So when the lambda invokes
`emit(it)`, it's emitting from a child coroutine. `flow{}`'s
SafeFlow guard rejects this with the same "Emission from another
coroutine is detected" error the inner listener was throwing before
my last fix.

Net effect on the user's repro: the inner channelFlow now correctly
sends RoomAnnouncement to the wrapper's `listener.announces().collect`
lambda, but that lambda's `emit(it)` to the wrapper's flow{} body
fails the same SafeFlow check, the wrapper's runCatching swallows the
exception (as `iter=2 inner collect ended IllegalStateException ...
fwd=1`), `_announcedSpeakers` stays empty, cliff detector never fires.

Fix: convert `ReconnectingNestsListener.announces` from `flow{} + emit`
to `channelFlow{} + send`, matching the inner listener's shape. The
combination of `channelFlow + collectLatest + send` is the canonical
pattern in kotlinx-coroutines for "switch-map" semantics with cross-
coroutine production. `awaitClose { }` is empty because `collectLatest`
on an infinite-StateFlow never completes naturally — cancellation
propagates through structured concurrency when the consumer cancels
the channelFlow.

Tests: every nestsClient + commons unit test still passes, including
the 12 CliffDetectorTest cases pinning the predicate's behaviour.

Together with f17e7ad (inner channelFlow), this should finally close
the chain: inner emits RoomAnnouncement on its channelFlow → wrapper's
collectLatest receives it inside its child coroutine → wrapper's
channelFlow.send forwards across the channel → consumer's
observeAnnounces collect receives → `_announcedSpeakers` populates →
cliff detector tick reports `announced=1` → on a real cliff event the
recycle fires.
This commit is contained in:
Claude
2026-05-05 19:50:23 +00:00
parent f17e7adfa7
commit 457e0f5997
@@ -29,6 +29,7 @@ import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.Flow
@@ -37,9 +38,9 @@ import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.isActive import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
@@ -251,8 +252,25 @@ private class ReconnectingHandle(
* swallowed so a future moq-lite session keeps emitting. * swallowed so a future moq-lite session keeps emitting.
*/ */
override fun announces(): Flow<RoomAnnouncement> = override fun announces(): Flow<RoomAnnouncement> =
flow { // `channelFlow` (NOT `flow`) is required here for the same
Log.d("NestRx") { "wrapper.announces() flow starting collect on activeListener" } // reason `MoqLiteNestsListener.announces` is channelFlow:
// we drive emissions from a `collectLatest` that internally
// launches a child coroutine per `activeListener` emission.
// `flow {}`'s SafeFlow guard rejects emissions from any
// coroutine other than the flow-builder's own — so on the
// very first listener emission the inner `emit(it)` lands
// in a child coroutine of `collectLatest`, throws
// IllegalStateException, the consumer's `runCatching`
// swallows it, and `_announcedSpeakers` never populates.
// Two-phone production logs at commit f17e7ad showed this
// exact failure even AFTER the inner listener was switched
// to channelFlow — the wrapper layer was the second
// un-fixed `flow {}`. `channelFlow` + `send(...)` instead
// of `emit(...)` allows cross-coroutine production, which
// is exactly what `collectLatest`'s per-emission child
// coroutines need.
channelFlow {
Log.d("NestRx") { "wrapper.announces() channelFlow starting collect on activeListener" }
var iter = 0 var iter = 0
activeListener.collectLatest { listener -> activeListener.collectLatest { listener ->
iter += 1 iter += 1
@@ -271,7 +289,7 @@ private class ReconnectingHandle(
runCatching { runCatching {
listener.announces().collect { listener.announces().collect {
fwd += 1 fwd += 1
emit(it) send(it)
} }
} }
Log.w("NestRx") { Log.w("NestRx") {
@@ -279,6 +297,12 @@ private class ReconnectingHandle(
"wrapper.announces() iter=$iter inner collect ended $why fwd=$fwd" "wrapper.announces() iter=$iter inner collect ended $why fwd=$fwd"
} }
} }
// collectLatest above never returns naturally — it
// collects activeListener forever. awaitClose fires when
// the consumer cancels the channelFlow, at which point
// the collectLatest is cancelled too via structured
// concurrency.
awaitClose { }
} }
/** /**