nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(audio-rooms): proactive JWT refresh in the reconnecting wrapper (T4 #16)

Browse Source 
moq-auth issues bearer tokens with a 600 s lifetime. When a token
expires the relay tears down the WebTransport session and the
wrapper recovers via the regular Failed → Reconnecting → Connected
path with a brief audible dropout (smoothed by the SubscribeHandle
buffer pump but still user-visible as a Reconnecting chip).

Add `tokenRefreshAfterMs` to connectReconnectingNestsListener
(default 540 s — 1 min before the relay's 600 s expiry). The
orchestrator now waits for either a terminal listener state OR
the refresh deadline; whichever fires first wins. On refresh:

  * close the still-healthy listener,
  * skip the reconnect-schedule path (refresh is a planned
    cutover, not a backoff event),
  * loop straight to openOnce() which mints a fresh JWT and
    establishes a new session.

The SubscribeHandle re-issuance pump cuts subs over to the new
session, and the wrapper's outward state never enters Reconnecting
during the cutover — the user-facing UI stays Connected end-to-end.
Setting tokenRefreshAfterMs <= 0 disables the behavior.

Tests:
  * ReconnectingNestsListenerTest gains
    `proactive_token_refresh_recycles_listener_without_failure_state`
    — drives the refresh path with a 50 ms window, captures every
    state emission, asserts neither Reconnecting nor Failed appear
    during a clean recycle.
This commit is contained in:
Claude
2026-04-27 01:57:19 +00:00
parent c3ff82913b
commit efcd32f987
2 changed files with 122 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
+48 -3
View File
@@ -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
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListenerTest.kt
+74
View File
@@ -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<NestsListenerState>()
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()
}
}
}