diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index 27674720a..96eb98893 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -37,6 +37,7 @@ import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeoutOrNull import java.util.concurrent.atomic.AtomicReference /** @@ -71,6 +72,18 @@ suspend fun connectReconnectingNestsListener( room: NestsRoomConfig, signer: NostrSigner, policy: NestsReconnectPolicy = NestsReconnectPolicy(), + /** + * Proactive JWT refresh window. moq-auth issues bearer tokens + * with a 600 s lifetime; once the token expires the relay + * tears down the WebTransport session and we'd otherwise + * recover via the regular reconnect path with a brief audible + * dropout. By recycling the session a minute before expiry we + * stay ahead of the relay's tear-down: the new session opens, + * the [SubscribeHandle] re-issuance pump cuts subs over to + * it, and the listener never enters the user-visible + * Reconnecting state. Set to 0 or negative to disable. + */ + tokenRefreshAfterMs: Long = 540_000L, /** * Test seam — defaults to the production * [connectNestsListener]. Tests pass a fake that returns a @@ -100,19 +113,43 @@ suspend fun connectReconnectingNestsListener( state.value = NestsListenerState.Failed("connect failed: ${it.message}", it) null } + var refreshTriggered = false if (listener != null) { - // Mirror state until the listener terminates. + // Wait for either a terminal state OR the proactive + // JWT-refresh deadline. withTimeoutOrNull returns + // null when the timer fires first; we then close + // the (still-healthy) listener and loop to mint a + // fresh JWT via openOnce(). The SubscribeHandle + // re-issuance pump cuts subs over to the new + // session without surfacing Reconnecting to the UI. + // // `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 = + val terminalAwait: suspend () -> NestsListenerState = { listener.state .onEach { state.value = it } .first { s -> s is NestsListenerState.Failed || s is NestsListenerState.Closed } - if (terminal is NestsListenerState.Failed && !isUserCancelled(terminal)) { + } + val terminal = + if (tokenRefreshAfterMs > 0L) { + withTimeoutOrNull(tokenRefreshAfterMs) { terminalAwait() } + } else { + terminalAwait() + } + if (terminal == null) { + // Refresh deadline hit before any terminal state — + // planned recycle, not a failure. Close the old + // listener; don't bump `attempt` (it's not a + // backoff event) so the next openOnce() runs + // immediately. + runCatching { listener.close() } + attempt = 0 + refreshTriggered = true + } else if (terminal is NestsListenerState.Failed && !isUserCancelled(terminal)) { // Transport-side failure → schedule a reconnect. attempt++ if (!policy.isExhausted(attempt)) { @@ -121,6 +158,14 @@ suspend fun connectReconnectingNestsListener( } } } + if (refreshTriggered) { + // Skip the reconnect-schedule path entirely — a + // refresh is a planned cutover, not a backoff + // event. The next iteration's openOnce() runs + // immediately and the wrapper's outward state + // never enters Reconnecting. + continue + } val terminal = state.value if (terminal is NestsListenerState.Closed) break if (policy.isExhausted(attempt + 1)) break diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt index 16e927b24..e5cbec6e2 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt @@ -39,6 +39,7 @@ import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout import kotlin.test.Test @@ -266,4 +267,77 @@ class ReconnectingNestsListenerTest { scope.cancel() } } + + @Test + fun proactive_token_refresh_recycles_listener_without_failure_state() = + runBlocking { + val scope = CoroutineScope(SupervisorJob()) + try { + val first = ScriptedListener(room) + val second = ScriptedListener(room) + val third = ScriptedListener(room) + val listenersInOrder = mutableListOf(first, second, third) + + // Track every state the wrapper surfaces. The contract + // for proactive refresh is that we recycle WITHOUT + // ever flipping to Reconnecting / Failed — the new + // session is up before the old one's token would have + // expired. + val seenStates = mutableListOf() + + val reconnecting = + connectReconnectingNestsListener( + httpClient = httpClient, + transport = transport, + scope = scope, + room = room, + signer = signer, + // 50 ms refresh window — small enough for the + // test to fire it immediately, large enough + // that the orchestrator's first Connected + // observation lands before the timeout. + tokenRefreshAfterMs = 50L, + connector = { listenersInOrder.removeAt(0) }, + ) + + // Watch state transitions — must NOT see Reconnecting + // or Failed at any point during a clean refresh. + val watcher = + scope.launch { + reconnecting.state.collect { seenStates += it } + } + + withTimeout(5_000L) { + reconnecting.state.first { it is NestsListenerState.Connected } + } + + // Wait for at least 2 underlying listeners (one for + // initial, one after first refresh). + withTimeout(5_000L) { + while (listenersInOrder.size > 1) kotlinx.coroutines.delay(10) + } + + // Verify: the wrapper's outward state never went to + // Reconnecting or Failed during the refresh — the + // user-visible UI stays Connected throughout. + val sawReconnecting = seenStates.any { it is NestsListenerState.Reconnecting } + val sawFailed = seenStates.any { it is NestsListenerState.Failed } + assertTrue( + !sawReconnecting && !sawFailed, + "proactive refresh must not surface Reconnecting/Failed; saw=$seenStates", + ) + + // Verify: the previous listener was actually closed + // (state.value should reach Closed for first/second + // before the test ends). + withTimeout(5_000L) { + first.state.first { it is NestsListenerState.Closed } + } + + watcher.cancel() + reconnecting.close() + } finally { + scope.cancel() + } + } }