From cdf930938d554d378804c4ee3101b36e9a046aa9 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 26 Apr 2026 23:38:09 +0000 Subject: [PATCH] feat(audio-rooms): connectReconnectingNestsListener (T4 #2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wraps `connectNestsListener` with a transport-loss reconnect loop that consumes [NestsReconnectPolicy] for exponential backoff. Mirrors the JS reference's `Connection.Reload` behaviour. connectReconnectingNestsListener(httpClient, transport, scope, room, signer, policy = default, connector = real) Behaviour: * Returned [NestsListener] forwards the underlying listener's state while a session is alive. * On Failed (transport / handshake error), the orchestrator increments the attempt counter, surfaces [NestsListenerState.Reconnecting(attempt, delayMs)] for that delay, then opens a fresh session via the injected connector. * On Closed (user-driven disconnect), the loop exits — Closed is terminal. * `policy.isExhausted(attempt+1)` halts the loop when [NestsReconnectPolicy.maxAttempts] hits. Default policy is Int.MAX_VALUE so a long-running room keeps trying. Subscribe-handle preservation across a reconnect is intentionally NOT in this commit. Caller-owned [SubscribeHandle]s bind to the SESSION; once the session is replaced, the handle's flow stops emitting. The KDoc spells this out so callers can either: 1. Re-subscribe inside their own `state.collectLatest { if (Connected) sub() }` loop 2. Wait for the future MutableSharedFlow-buffered upgrade flagged in the Tier-4 plan ("MutableSharedFlow per handle that the per-session pump emits into"). Test seam: the `connector: suspend () -> NestsListener` parameter defaults to the real connect call. Tests can pass a scripted fake that returns a sequence of (Failed, Connected, Failed, Connected) listeners and verify the orchestrator walks the policy correctly — production path is unchanged. --- .../nestsclient/ReconnectingNestsListener.kt | 162 ++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt new file mode 100644 index 000000000..0c6f9613a --- /dev/null +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -0,0 +1,162 @@ +/* + * 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.SubscribeHandle +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.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.launch + +/** + * `connectNestsListener` plus a transport-loss reconnect loop with + * exponential backoff (the JS reference's + * `Connection.Reload`-style behaviour). + * + * The returned [NestsListener] surfaces the underlying listener's + * state directly while a session is alive, but flips to + * [NestsListenerState.Reconnecting] between attempts. The listener + * is auto-redirected to the freshly-opened session under the hood + * — `subscribeSpeaker(...)` calls always go to the latest live + * 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. + * + * Cancellation: cancelling [scope] (typically the room screen's + * VM scope) cancels the reconnect loop and closes the active + * session. [NestsListener.close] is idempotent. + */ +suspend fun connectReconnectingNestsListener( + httpClient: NestsClient, + transport: WebTransportFactory, + scope: CoroutineScope, + room: NestsRoomConfig, + signer: NostrSigner, + policy: NestsReconnectPolicy = NestsReconnectPolicy(), + /** + * Test seam — defaults to the production + * [connectNestsListener]. Tests pass a fake that returns a + * scripted [NestsListener] so the reconnect state machine can + * be exercised without a real WebTransport stack. + */ + connector: suspend () -> NestsListener = { + connectNestsListener(httpClient, transport, scope, room, signer) + }, +): NestsListener { + val state = MutableStateFlow(NestsListenerState.Idle) + val activeListener = MutableStateFlow(null) + + suspend fun openOnce(): NestsListener { + val listener = connector() + activeListener.value = listener + state.value = listener.state.value + return listener + } + + val orchestrator = + scope.launch { + var attempt = 0 + while (true) { + val listener = + runCatching { openOnce() }.getOrElse { + state.value = NestsListenerState.Failed("connect failed: ${it.message}", it) + 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 + 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 + if (terminal is NestsListenerState.Closed) break + if (policy.isExhausted(attempt + 1)) break + val delayMs = + if (terminal is NestsListenerState.Reconnecting) { + terminal.delayMs + } else { + policy.delayForAttempt(++attempt) + } + state.value = NestsListenerState.Reconnecting(attempt.coerceAtLeast(1), delayMs) + delay(delayMs) + } + } + + return ReconnectingHandle(state, activeListener, orchestrator) +} + +private fun isUserCancelled(state: NestsListenerState.Failed): Boolean { + val msg = state.reason + // The user-driven close path comes through Closed, not Failed, + // so a Failed state is always a transport / handshake error. + // This hook is a forward-compat seam in case a future + // listener emits Failed on user dispose — extending the + // pattern is local. + return msg.contains("user cancelled", ignoreCase = true) +} + +private class ReconnectingHandle( + private val mutableState: MutableStateFlow, + private val activeListener: MutableStateFlow, + private val orchestrator: Job, +) : 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) + } + + override suspend fun close() { + orchestrator.cancel() + runCatching { activeListener.value?.close() } + if (mutableState.value !is NestsListenerState.Closed) { + mutableState.value = NestsListenerState.Closed + } + } +}