laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-07-22 15:52:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(nests): listener subscriptions survive publisher recycle

Browse Source 
Two layered fixes for the gap captured in
nestsClient/plans/2026-04-28-listener-survives-publisher-recycle.md.

Session layer (MoqLiteSession.kt):
  - Lazy single shared announce-watch pump per session, opened on
    first subscribe. moq-lite Lite-03 has no explicit "publisher
    gone" message on the subscribe bidi (the relay keeps that
    bidi open across publisher cycles in case a fresh publisher
    takes over the suffix), so the announce stream's Ended event
    is the only reliable signal.
  - On Announce(Ended) for a broadcast suffix, close the
    matching ListenerSubscription's frames Channel and remove it
    from the map. The wrapper-level frames.consumeAsFlow() flow
    ends naturally — same shape as a user-driven
    handle.unsubscribe() — so the wrapper pump's collect-
    completion path drives the re-issue.

Wrapper layer (ReconnectingNestsListener.kt):
  - Inner re-subscribe `while (currentCoroutineContext().isActive)`
    loop in reissuingSubscribe. When the underlying frames flow
    completes (publisher cycled, signalled by the session layer
    above), re-issue subscribe against the same listener with a
    100 ms backoff. moq-lite supports subscribe-before-announce so
    a re-subscribe issued during the gap attaches cleanly when
    the next publisher comes up under the same suffix.

Verified against the real moq-rs relay (host build, external
mode): the new
NostrNestsReconnectingListenerInteropTest.subscribe_handle_survives_publisher_recycle
test passes — single SubscribeHandle keeps emitting frames across
multiple speaker JWT-refresh cycles. Speaker reconnect tests
still pass too.

In production, this closes the audio dropout that would have
fired every 9 minutes per speaker JWT refresh on long Nest calls
(see the plan doc for the original diagnosis).

https://claude.ai/code/session_01HXf3zG3F2ev2ASeQju7Y5S
This commit is contained in:
Claude
2026-04-28 15:06:07 +00:00
parent 1c8c961762
commit 851045c654
3 changed files with 351 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt
View File

@@ -28,6 +28,7 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
@@ -39,6 +40,7 @@ import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import java.util.concurrent.atomic.AtomicReference
@@ -263,9 +265,38 @@ private class ReconnectingHandle(
)
val liveHandleRef = AtomicReference<SubscribeHandle?>(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.
// Re-subscribe pump. Two re-issue triggers, layered:
//
// 1. Listener session swap (outer collectLatest) — fires
// when the orchestrator opens a fresh listener after
// the 540 s JWT-refresh window or a transport-loss
// reconnect. collectLatest cancels the prior pump
// iteration so the next iteration runs against the
// new listener.
//
// 2. Publisher session swap (inner while loop) — fires
// when the underlying SubscribeHandle.objects flow
// completes mid-stream because the *publisher*
// cycled. The moq-lite session layer detects publisher
// disconnect via the announce stream's Ended event
// and closes the underlying frames channel; that
// naturally ends `handle.objects.collect` here. We
// then loop into a fresh subscribe — moq-lite supports
// subscribe-before-announce, so the new subscribe
// attaches cleanly to whichever publisher serves the
// suffix next, including one that comes up AFTER us.
//
// Without the inner loop, a remote speaker's JWT refresh
// (every 9 min on the speaker side via
// [connectReconnectingNestsSpeaker]) would silently kill
// every listener's audio — the listener's own JWT refresh
// fires on a different cadence and can't be relied on to
// coincide.
//
// Bounded by:
// - listener swap → outer collectLatest cancels us
// - unsubscribeAction → pumpJob.cancel()
// - opener-throws → break + wait for next swap
val pumpJob =
scope.launch {
activeListener.collectLatest { listener ->
@@ -277,14 +308,23 @@ private class ReconnectingHandle(
state is NestsListenerState.Failed
}
if (terminalOrConnected !is NestsListenerState.Connected) return@collectLatest
val handle =
runCatching { opener(listener) }
.getOrNull() ?: return@collectLatest
liveHandleRef.set(handle)
try {
handle.objects.collect { frames.emit(it) }
} finally {
if (liveHandleRef.get() === handle) liveHandleRef.set(null)
while (currentCoroutineContext().isActive) {
val handle =
runCatching { opener(listener) }
.getOrNull() ?: break
liveHandleRef.set(handle)
try {
handle.objects.collect { frames.emit(it) }
} finally {
if (liveHandleRef.get() === handle) liveHandleRef.set(null)
}
// Brief backoff so a permanently-gone
// publisher doesn't tight-loop the relay
// with re-subscribes. 100 ms stays well
// under the SUBSCRIBE_BUFFER's 1.3 s of
// audio headroom.
delay(RESUBSCRIBE_BACKOFF_MS)
}
}
}
@@ -318,6 +358,12 @@ private class ReconnectingHandle(
// grow the queue unbounded.
private const val SUBSCRIBE_BUFFER = 64
// Inner-pump backoff between publisher-cycle re-subscribes.
// Short enough to stay well under the SUBSCRIBE_BUFFER's
// ~1.3 s of audio headroom; long enough that a permanently-
// gone publisher doesn't spin the relay with re-subscribes.
private const val RESUBSCRIBE_BACKOFF_MS = 100L
private val SYNTH_OK =
SubscribeOk(
subscribeId = -1L,

120
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
View File

@@ -77,6 +77,17 @@ class MoqLiteSession internal constructor(
/** Lazily-launched relay→us inbound bidi pump; only runs while a publisher is active. */
private var bidiPump: Job? = null
/**
* Single shared announce-watch pump that runs while we have any
* listener-side subscription. Closes the frames channel of any
* subscription whose broadcast suffix goes Ended on the relay's
* announce stream — see [pumpAnnounceWatch] for why this is the
* only reliable signal of publisher disconnect under moq-lite
* Lite-03. Lazily launched on first subscribe; lives until the
* session scope is cancelled.
*/
private var announceWatchJob: Job? = null
/** Single active publisher per session (moq-lite doesn't model multi-broadcast publishers). */
private var activePublisher: PublisherStateImpl? = null
@@ -173,6 +184,27 @@ class MoqLiteSession internal constructor(
bidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
bidi.write(MoqLiteCodec.encodeSubscribe(request))
// Single long-running collector pump for the bidi's response
// side. Reads the SubscribeResponse, then keeps collecting
// until the peer FINs (or scope is cancelled, or
// handle.unsubscribe() FINs our side and the relay echoes).
// The flow completion IS the moq-lite-03 signal that the
// publisher has disconnected mid-broadcast — Lite-03 has no
// explicit "publisher gone" message; bidi close is it.
// Without this watch, the frames Channel below would never
// close on remote disconnect, and any consumer collecting
// from the wrapper-level [MoqLiteSubscribeHandle.frames]
// flow would sit silent indefinitely after a publisher
// cycle even though the relay is happy to serve a fresh
// subscribe under the same broadcast suffix.
//
// Why a single pump (vs separate response read + death
// watch): the underlying QUIC stream's `incoming` is
// backed by `Channel<ByteArray>.consumeAsFlow()` — which
// CANCELS the channel when the first collect ends. A second
// collect on a fresh `bidi.incoming()` Flow would see an
// already-cancelled channel and fire prematurely. Keeping
// one collect alive sidesteps that entirely.
// moq-lite's subscribe-response is a single size-prefixed
// message on the response side of the bidi. Read incoming
// chunks into a buffer until the buffer holds a full payload,
@@ -202,6 +234,29 @@ class MoqLiteSession internal constructor(
state.withLock {
subscriptionsBySubscribeId[id] = sub
if (groupPump == null) groupPump = scope.launch { pumpUniStreams() }
// Lazy-launch the publisher-disconnect watcher
// — one shared announce bidi per session whose
// sole job is to close the frames channel on any
// ListenerSubscription whose broadcast path goes
// Ended. moq-lite Lite-03 has no explicit
// "publisher gone" message on the subscribe
// bidi (the relay keeps that bidi open across
// publisher cycles in case a fresh publisher
// takes over the suffix), so the announce
// stream IS the only signal.
//
// Without this, a wrapper-layer consumer
// collecting from [MoqLiteSubscribeHandle.frames]
// would sit silent indefinitely after a
// publisher cycle even though the relay is happy
// to serve a fresh subscribe under the same
// suffix. The wrapper-level
// [com.vitorpamplona.nestsclient.ReconnectingNestsListener]
// pump's natural collect-completion path then
// re-issues the subscribe.
if (announceWatchJob == null) {
announceWatchJob = scope.launch { pumpAnnounceWatch() }
}
}
return MoqLiteSubscribeHandle(
id = id,
@@ -213,6 +268,71 @@ class MoqLiteSession internal constructor(
}
}
/**
* Single shared announce-watch pump for ALL subscriptions on
* this session. Opens one announce bidi (prefix="") on first
* subscribe, then for each [MoqLiteAnnounceStatus.Ended] update
* iterates the subscription map and closes the frames channel of
* any subscription whose `broadcast` matches the announce
* suffix. The closed channel ends the consumer-facing
* `frames.consumeAsFlow()` flow naturally — same shape as a
* user-driven `handle.unsubscribe()` from the consumer's POV —
* which lets the wrapper's re-issuance pump drive a fresh
* subscribe against the same broadcast path. moq-lite supports
* subscribe-before-announce, so a subscribe issued during the
* gap (between Ended and the next Active under the same suffix)
* attaches cleanly when the new publisher comes up.
*
* This pump survives announce-bidi errors via best-effort
* silence — the session itself recovers via its own reconnect
* path. Cancelled when [scope] is cancelled (session close).
*/
private suspend fun pumpAnnounceWatch() {
val handle =
try {
announce(prefix = "")
} catch (ce: kotlinx.coroutines.CancellationException) {
throw ce
} catch (_: Throwable) {
// Couldn't open the announce bidi — best effort,
// bail. Subscriptions still work; we just lose
// automatic cycle detection.
return
}
try {
handle.updates.collect { update ->
if (update.status != MoqLiteAnnounceStatus.Ended) return@collect
val targets =
state.withLock {
subscriptionsBySubscribeId.values
.filter { it.request.broadcast == update.suffix }
.toList()
}
for (sub in targets) {
// Just close the frames channel — the
// wrapper-level collect of `frames.consumeAsFlow()`
// ends naturally and the wrapper pump re-issues.
// Don't fire `unsubscribe(id)` here: that'd FIN
// OUR side of the (still-alive) subscribe bidi,
// and the wrapper's re-issue would have to open
// a fresh bidi anyway. Keeping the subscribe
// bidi open lets a future subscribe-before-
// announce land cleanly.
sub.frames.close()
state.withLock { subscriptionsBySubscribeId.remove(sub.id) }
runCatching { sub.bidi.finish() }
}
}
} catch (ce: kotlinx.coroutines.CancellationException) {
throw ce
} catch (_: Throwable) {
// Announce bidi died — same best-effort fallback.
} finally {
runCatching { handle.close() }
state.withLock { announceWatchJob = null }
}
}
/**
* Drain inbound uni streams and route each one's group frames to
* the matching subscription. The relay opens a fresh uni stream

174
nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt
View File

@@ -24,12 +24,14 @@ import com.vitorpamplona.nestsclient.NestsListener
import com.vitorpamplona.nestsclient.NestsListenerState
import com.vitorpamplona.nestsclient.NestsReconnectPolicy
import com.vitorpamplona.nestsclient.NestsRoomConfig
import com.vitorpamplona.nestsclient.NestsSpeakerState
import com.vitorpamplona.nestsclient.OkHttpNestsClient
import com.vitorpamplona.nestsclient.audio.AudioCapture
import com.vitorpamplona.nestsclient.audio.OpusEncoder
import com.vitorpamplona.nestsclient.connectNestsListener
import com.vitorpamplona.nestsclient.connectNestsSpeaker
import com.vitorpamplona.nestsclient.connectReconnectingNestsListener
import com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker
import com.vitorpamplona.nestsclient.transport.QuicWebTransportFactory
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
@@ -408,4 +410,176 @@ class NostrNestsReconnectingListenerInteropTest {
harnessOrNull = null
}
}
/**
* Captures the listener-survives-publisher-recycle invariant —
* the gap discovered while validating
* [connectReconnectingNestsSpeaker]. The setup:
*
* - SUT (listener): a [connectReconnectingNestsListener]-backed
* handle, vanilla refresh disabled so the listener's own
* session never recycles during the test.
* - Driver (speaker): a [connectReconnectingNestsSpeaker] with
* a small `tokenRefreshAfterMs`, forcing the publisher's
* session to recycle mid-stream.
*
* The single [SubscribeHandle] returned from
* `subscribeSpeaker(pubkey)` MUST keep emitting frames after the
* publisher cycles. The session-layer death-watch in
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.subscribe]
* detects the publisher's bidi-FIN, closes the frames channel,
* the wrapper-level `reissuingSubscribe` pump's collect ends
* naturally, and the pump re-issues a fresh subscribe via the
* outer collectLatest's loop semantics.
*
* Skipped by default — set `-DnestsInterop=true` to enable.
*/
@Test
fun subscribe_handle_survives_publisher_recycle() =
runBlocking {
NostrNestsHarness.assumeNestsInterop()
val harness = harnessOrNull ?: return@runBlocking
val signer = NostrSignerInternal(KeyPair())
val pubkey = signer.pubKey
val room =
NestsRoomConfig(
authBaseUrl = harness.authBaseUrl,
endpoint = harness.moqEndpoint,
hostPubkey = pubkey,
roomId = "lst-pub-cycle-${System.currentTimeMillis()}",
)
val httpClient = OkHttpNestsClient()
val transport =
QuicWebTransportFactory(
certificateValidator = PermissiveCertificateValidator(),
)
val supervisor = SupervisorJob()
val pumpScope = CoroutineScope(supervisor + Dispatchers.IO)
val capturesLock = Any()
val captures = mutableListOf<DriverCapture>()
val captureFactory: () -> AudioCapture = {
val c = DriverCapture()
synchronized(capturesLock) { captures += c }
c
}
val encoder = StubEncoder(prefix = "LSP-".encodeToByteArray())
val speakerOpenCount = AtomicInteger(0)
val scope = "listener-survives-publisher-recycle"
try {
val speaker =
InteropDebug.stepSuspending(scope, "connectReconnectingNestsSpeaker (refresh=${PUBCYCLE_REFRESH_MS}ms)") {
connectReconnectingNestsSpeaker(
httpClient = httpClient,
transport = transport,
scope = pumpScope,
room = room,
signer = signer,
speakerPubkeyHex = pubkey,
captureFactory = captureFactory,
encoderFactory = { encoder },
policy = NestsReconnectPolicy(initialDelayMs = 250L),
tokenRefreshAfterMs = PUBCYCLE_REFRESH_MS,
connector = {
speakerOpenCount.incrementAndGet()
connectNestsSpeaker(
httpClient = httpClient,
transport = transport,
scope = pumpScope,
room = room,
signer = signer,
speakerPubkeyHex = pubkey,
captureFactory = captureFactory,
encoderFactory = { encoder },
)
},
)
}
val broadcast = speaker.startBroadcasting()
withTimeoutOrNull(BROADCAST_READY_MS) {
speaker.state.first { it is NestsSpeakerState.Broadcasting }
} ?: fail("[$scope] speaker never reached initial Broadcasting")
// SUT: reconnecting listener with refresh disabled.
val listener =
InteropDebug.stepSuspending(scope, "connectReconnectingNestsListener (refresh disabled)") {
connectReconnectingNestsListener(
httpClient = httpClient,
transport = transport,
scope = pumpScope,
room = room,
signer = signer,
policy = NestsReconnectPolicy(initialDelayMs = 250L),
tokenRefreshAfterMs = 0L,
)
}
withTimeoutOrNull(CONNECT_TIMEOUT_MS) {
listener.state.first { it is NestsListenerState.Connected }
} ?: fail("[$scope] listener never reached Connected")
// The single subscription that MUST survive every
// publisher recycle.
val subscription = listener.subscribeSpeaker(pubkey)
val received =
async(pumpScope.coroutineContext) {
withTimeoutOrNull(LISTENER_SURVIVAL_TIMEOUT_MS) {
subscription.objects.take(N_FRAMES_CYCLE).toList()
}
}
kotlinx.coroutines.delay(SUBSCRIBE_SETTLE_MS)
for (i in 0 until N_FRAMES_CYCLE) {
val current =
synchronized(capturesLock) { captures.lastOrNull() }
?: error("captureFactory was never invoked")
current.push(shortArrayOf(i.toShort()))
kotlinx.coroutines.delay(CYCLE_FRAME_SPACING_MS)
if (i == N_FRAMES_CYCLE / 2) {
InteropDebug.checkpoint(scope, "midpoint — waiting for speaker recycle")
withTimeoutOrNull(SWAP_TIMEOUT_MS) {
while (speakerOpenCount.get() < 2) kotlinx.coroutines.delay(50)
speaker.state.first { it is NestsSpeakerState.Broadcasting }
} ?: fail("[$scope] speaker did not recycle — openCount=${speakerOpenCount.get()}")
kotlinx.coroutines.delay(POST_RECYCLE_SETTLE_MS)
}
}
val datagrams = received.await()
if (datagrams == null) {
fail(
"[$scope] listener subscription went silent across publisher recycle — " +
"speakerOpenCount=${speakerOpenCount.get()}, " +
"listener=${InteropDebug.describe(listener.state.value)}",
)
}
assertEquals(N_FRAMES_CYCLE, datagrams.size, "all frames must arrive on the SAME subscribe handle across the publisher recycle")
val payloads = datagrams.map { it.payload.last().toInt() and 0xFF }.toSet()
assertEquals((0 until N_FRAMES_CYCLE).toSet(), payloads, "all frames pre- AND post-recycle must round-trip")
assertTrue(
speakerOpenCount.get() >= 2,
"expected ≥2 underlying speaker sessions across the burst (one before, one after recycle); got ${speakerOpenCount.get()}",
)
runCatching { subscription.unsubscribe() }
runCatching { listener.close() }
runCatching { broadcast.close() }
runCatching { speaker.close() }
} finally {
synchronized(capturesLock) { captures.forEach { runCatching { it.stop() } } }
supervisor.cancelAndJoin()
}
Unit
}
}
private const val PUBCYCLE_REFRESH_MS = 4_000L
private const val BROADCAST_READY_MS = 15_000L
private const val LISTENER_SURVIVAL_TIMEOUT_MS = 60_000L
private const val POST_RECYCLE_SETTLE_MS = 1_500L
private const val CYCLE_FRAME_SPACING_MS = 80L
private const val N_FRAMES_CYCLE = 10
private const val SWAP_TIMEOUT_MS = 60_000L