diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 0c6f9613a..2faa674a0 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -20,16 +20,24 @@ */ package com.vitorpamplona.nestsclient +import com.vitorpamplona.nestsclient.moq.MoqObject import com.vitorpamplona.nestsclient.moq.SubscribeHandle +import com.vitorpamplona.nestsclient.moq.SubscribeOk import com.vitorpamplona.nestsclient.transport.WebTransportFactory import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.collectLatest +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch +import java.util.concurrent.atomic.AtomicReference /** * `connectNestsListener` plus a transport-loss reconnect loop with @@ -44,13 +52,13 @@ import kotlinx.coroutines.launch * session. * * **Subscribe-handle re-issuance** — caller-owned [SubscribeHandle]s - * are NOT preserved across a reconnect. The listener's - * `subscribeSpeaker` returns a handle bound to the SESSION; once - * the session is replaced, that handle's flow stops. Callers can - * either re-subscribe in their own - * `state.collectLatest { if (Connected) sub() }` loop, or wait for - * the future `MutableSharedFlow`-buffered upgrade flagged in the - * Tier-4 plan. + * survive a reconnect. `subscribeSpeaker` returns a handle backed + * by a [MutableSharedFlow]; the orchestrator opens an underlying + * subscription against each fresh session and forwards its frames + * into the shared flow, so the consumer's collector keeps emitting + * even after a transport drop. The handle's [SubscribeHandle.unsubscribe] + * cancels the pump and best-effort cancels the live underlying + * subscription. * * Cancellation: cancelling [scope] (typically the room screen's * VM scope) cancels the reconnect loop and closes the active @@ -93,22 +101,24 @@ suspend fun connectReconnectingNestsListener( null } if (listener != null) { - // Forward state until the listener terminates. - listener.state.collect { s -> - state.value = s - if (s is NestsListenerState.Failed && !isUserCancelled(s)) { - // Transport-side failure → reconnect. - attempt++ - if (policy.isExhausted(attempt)) return@collect + // Mirror state until the listener terminates. + // `return@collect` does NOT break a StateFlow's + // collect (it just returns from the lambda for one + // emission), so we use `onEach { mirror } + first` + // to wait deterministically for a terminal state. + val terminal = + listener.state + .onEach { state.value = it } + .first { s -> + s is NestsListenerState.Failed || s is NestsListenerState.Closed + } + if (terminal is NestsListenerState.Failed && !isUserCancelled(terminal)) { + // Transport-side failure → schedule a reconnect. + attempt++ + if (!policy.isExhausted(attempt)) { val delayMs = policy.delayForAttempt(attempt) state.value = NestsListenerState.Reconnecting(attempt, delayMs) - return@collect } - if (s is NestsListenerState.Closed) { - // User-driven close — exit the loop entirely. - return@collect - } - // Connecting / Connected / Reconnecting (transient) — continue. } } val terminal = state.value @@ -125,7 +135,7 @@ suspend fun connectReconnectingNestsListener( } } - return ReconnectingHandle(state, activeListener, orchestrator) + return ReconnectingHandle(state, activeListener, orchestrator, scope) } private fun isUserCancelled(state: NestsListenerState.Failed): Boolean { @@ -142,14 +152,75 @@ private class ReconnectingHandle( private val mutableState: MutableStateFlow, private val activeListener: MutableStateFlow, private val orchestrator: Job, + private val scope: CoroutineScope, ) : NestsListener { override val state: StateFlow = mutableState.asStateFlow() override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle { - val live = - activeListener.value - ?: error("no live session — wait for state == Connected before subscribing") - return live.subscribeSpeaker(speakerPubkeyHex) + // Require a live (or just-connected) session at call time — + // matches the existing IETF / moq-lite listener contract so + // a caller can't accidentally subscribe in Idle and stall + // waiting for a session that may never arrive. + activeListener.value + ?: error("no live session — wait for state == Connected before subscribing") + + val frames = MutableSharedFlow(extraBufferCapacity = SUBSCRIBE_BUFFER) + val liveHandleRef = AtomicReference(null) + + // Re-subscribe pump: every time activeListener changes, drop + // the prior subscription (collectLatest cancels the inner + // body) and open a new one against the fresh session. + // Wait for the inner listener to reach Connected before + // calling subscribeSpeaker — IETF + moq-lite listeners both + // throw IllegalStateException if subscribed before Connected, + // and a fresh session may go through Connecting first. + val pumpJob = + scope.launch { + activeListener.collectLatest { listener -> + if (listener == null) return@collectLatest + val terminalOrConnected = + listener.state.first { state -> + state is NestsListenerState.Connected || + state is NestsListenerState.Closed || + state is NestsListenerState.Failed + } + if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest + val handle = + runCatching { listener.subscribeSpeaker(speakerPubkeyHex) } + .getOrNull() ?: return@collectLatest + liveHandleRef.set(handle) + try { + handle.objects.collect { frames.emit(it) } + } finally { + // Either the inner session ended, or we're + // being replaced because activeListener + // changed. Drop the dead reference; the + // inner unsubscribe is handled by the + // session's own teardown. + if (liveHandleRef.get() === handle) liveHandleRef.set(null) + } + } + } + + return SubscribeHandle( + // Synthetic ids — the consumer-facing handle is logical, + // not bound to any one session's wire ids. -1 is a + // reserved sentinel callers shouldn't compare against. + subscribeId = -1L, + trackAlias = -1L, + ok = SYNTH_OK, + objects = frames.asSharedFlow(), + unsubscribeAction = { + // Capture the live underlying handle BEFORE cancelling + // the pump — the pump's finally clears `liveHandleRef` + // on cancellation, and we need a stable reference to + // forward the user-initiated unsubscribe to the live + // session's wire SUBSCRIBE_DONE. + val live = liveHandleRef.getAndSet(null) + pumpJob.cancel() + live?.let { runCatching { it.unsubscribe() } } + }, + ) } override suspend fun close() { @@ -159,4 +230,21 @@ private class ReconnectingHandle( mutableState.value = NestsListenerState.Closed } } + + companion object { + // Buffer enough Opus frames to ride out a brief downstream + // stall during reconnect. Frames are ~20 ms each, so 64 ≈ 1.3 s + // of audio — long enough for a typical re-handshake without + // dropping speech, short enough that a slow consumer doesn't + // grow the queue unbounded. + private const val SUBSCRIBE_BUFFER = 64 + + private val SYNTH_OK = + SubscribeOk( + subscribeId = -1L, + expiresMs = 0L, + groupOrder = 0x01, + contentExists = false, + ) + } } diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt new file mode 100644 index 000000000..16e927b24 --- /dev/null +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt @@ -0,0 +1,269 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.nestsclient + +import com.vitorpamplona.nestsclient.moq.MoqObject +import com.vitorpamplona.nestsclient.moq.SubscribeHandle +import com.vitorpamplona.nestsclient.moq.SubscribeOk +import com.vitorpamplona.nestsclient.transport.WebTransportFactory +import com.vitorpamplona.nestsclient.transport.WebTransportSession +import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair +import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner +import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class ReconnectingNestsListenerTest { + private val room = + NestsRoomConfig( + authBaseUrl = "https://relay.example.com/api/v1/nests", + endpoint = "https://relay.example.com/moq", + hostPubkey = "0".repeat(64), + roomId = "abc", + ) + private val signer = NostrSignerInternal(KeyPair()) + + // The custom `connector` lambda below short-circuits the real + // mintToken / WebTransport handshake, so these two sentinels + // only satisfy the function signature — they're never invoked. + private val httpClient = + object : NestsClient { + override suspend fun mintToken( + room: NestsRoomConfig, + publish: Boolean, + signer: NostrSigner, + ): String = error("not invoked — connector overrides") + } + private val transport = + object : WebTransportFactory { + override suspend fun connect( + authority: String, + path: String, + bearerToken: String?, + ): WebTransportSession = error("not invoked — connector overrides") + } + + /** + * Scripted [NestsListener] that opens in [NestsListenerState.Connected] + * by default and exposes a [MutableSharedFlow] to push frames into + * the next [SubscribeHandle.objects] returned from [subscribeSpeaker]. + */ + private class ScriptedListener( + connectedRoom: NestsRoomConfig, + moqVersion: Long = 1, + ) : NestsListener { + private val mutableState = + MutableStateFlow( + NestsListenerState.Connected(connectedRoom, moqVersion), + ) + override val state: StateFlow = mutableState.asStateFlow() + val frames = MutableSharedFlow(extraBufferCapacity = 16) + + // Atomic counters — the wrapper's pump runs on Dispatchers.Default + // while the test reads the count from the runBlocking thread, so + // a plain `var subscribeCount` would risk stale reads. + private val _subscribeCount = + java.util.concurrent.atomic + .AtomicInteger(0) + private val _unsubscribeCount = + java.util.concurrent.atomic + .AtomicInteger(0) + val subscribeCount: Int get() = _subscribeCount.get() + val unsubscribeCount: Int get() = _unsubscribeCount.get() + + override suspend fun subscribeSpeaker(speakerPubkeyHex: String): SubscribeHandle { + val id = _subscribeCount.incrementAndGet().toLong() + return SubscribeHandle( + subscribeId = id, + trackAlias = id, + ok = + SubscribeOk( + subscribeId = id, + expiresMs = 0L, + groupOrder = 0x01, + contentExists = false, + ), + objects = frames, + unsubscribeAction = { _unsubscribeCount.incrementAndGet() }, + ) + } + + override suspend fun close() { + mutableState.value = NestsListenerState.Closed + } + + fun fail(reason: String) { + mutableState.value = NestsListenerState.Failed(reason) + } + } + + private fun frame(payload: ByteArray): MoqObject = + MoqObject( + trackAlias = 1L, + groupId = 0L, + objectId = 0L, + publisherPriority = 0, + payload = payload, + ) + + // The orchestrator + pump jobs are real coroutines on a real + // dispatcher — runTest's virtual clock interacts badly with the + // orchestrator's `delay(...)` reconnect loop (the test scheduler + // doesn't auto-advance through the StateFlow + delay chain + // under UnconfinedTestDispatcher). Use a dedicated supervisor + // scope and real time; tests stay fast (single-digit ms) because + // the policy's initialDelayMs is set to 1. + @Test + fun subscribeSpeaker_survives_session_swap() = + runBlocking { + // Plain SupervisorJob — NOT plus coroutineContext, otherwise + // `scope.cancel()` in the finally block would propagate up + // through the runBlocking parent and fail the test. + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedListener(room) + val second = ScriptedListener(room) + val listenersInOrder = mutableListOf(first, second) + + val reconnecting = + connectReconnectingNestsListener( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + policy = NestsReconnectPolicy(initialDelayMs = 1L), + connector = { listenersInOrder.removeAt(0) }, + ) + + withTimeout(5_000L) { + reconnecting.state.first { it is NestsListenerState.Connected } + } + + val handle = reconnecting.subscribeSpeaker("speaker-pubkey") + + val collected = + scope.async { + withTimeout(5_000L) { + handle.objects.take(2).toList() + } + } + + // Wait for the pump to actually subscribe to the first + // session before we emit / fail. With Dispatchers.Default + // the pumpJob's collectLatest doesn't run synchronously + // off the launch site — yielding here lets it advance. + withTimeout(5_000L) { + while (first.subscribeCount == 0) kotlinx.coroutines.delay(5) + } + + first.frames.emit(frame(byteArrayOf(0x01))) + + // Force a reconnect: fail the first listener, the + // orchestrator opens the next one. + first.fail("scripted-disconnect") + // Both ScriptedListeners report the same `room` and + // `negotiatedMoqVersion`, so we can't differentiate + // first.Connected from second.Connected via state alone. + // The actual postcondition we care about — the pump + // re-issued the subscription against the new session — + // is observable directly via second.subscribeCount. + withTimeout(5_000L) { + while (second.subscribeCount == 0) kotlinx.coroutines.delay(5) + } + + second.frames.emit(frame(byteArrayOf(0x02))) + + val result = collected.await() + assertEquals(2, result.size) + assertEquals(0x01.toByte(), result[0].payload[0]) + assertEquals(0x02.toByte(), result[1].payload[0]) + + // Both sessions saw exactly one subscribeSpeaker call — + // the wrapper re-issues against the new session, not the + // old one. + assertEquals(1, first.subscribeCount) + assertEquals(1, second.subscribeCount) + + handle.unsubscribe() + assertEquals(1, second.unsubscribeCount) + + reconnecting.close() + } finally { + scope.cancel() + } + } + + @Test + fun unsubscribe_before_session_swap_releases_handle() = + runBlocking { + // Plain SupervisorJob — NOT plus coroutineContext, otherwise + // `scope.cancel()` in the finally block would propagate up + // through the runBlocking parent and fail the test. + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedListener(room) + val reconnecting = + connectReconnectingNestsListener( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + connector = { first }, + ) + withTimeout(5_000L) { + reconnecting.state.first { it is NestsListenerState.Connected } + } + + val handle = reconnecting.subscribeSpeaker("speaker-pubkey") + + // Wait until the pump opened the underlying subscription — + // collectLatest schedules the inner block on a child of + // scope, so it runs after we yield. + withTimeout(5_000L) { + while (first.subscribeCount == 0) kotlinx.coroutines.yield() + } + assertTrue(first.subscribeCount >= 1, "pump should have opened the underlying sub") + + handle.unsubscribe() + assertEquals(1, first.unsubscribeCount) + + reconnecting.close() + } finally { + scope.cancel() + } + } +}