From 851045c65471765fb6eef10c40afb2077f201f22 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Apr 2026 15:06:07 +0000 Subject: [PATCH] fix(nests): listener subscriptions survive publisher recycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../nestsclient/ReconnectingNestsListener.kt | 68 +++++-- .../nestsclient/moq/lite/MoqLiteSession.kt | 120 ++++++++++++ ...strNestsReconnectingListenerInteropTest.kt | 174 ++++++++++++++++++ 3 files changed, 351 insertions(+), 11 deletions(-) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt index e3686e4ec..d2e738b8d 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsListener.kt @@ -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(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, diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt index b61ac96cc..1697cc566 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt @@ -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.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 diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt index 54bdedb5f..724482d72 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsReconnectingListenerInteropTest.kt @@ -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() + 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