diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 2fdd8e44b..3c0769d55 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -28,13 +28,9 @@ import com.vitorpamplona.nestsclient.moq.lite.MoqLitePublisherHandle import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Job -import kotlinx.coroutines.cancelAndJoin -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow -import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -150,39 +146,25 @@ class MoqLiteNestsSpeaker internal constructor( runCatching { publisher.close() } throw t } - // Push catalog once before launching the periodic republish - // pump so the manifest is on the wire before any subscriber - // attaches; the republisher then re-emits a fresh group - // every CATALOG_REPUBLISH_INTERVAL_MS so late-joining - // watchers don't wait an arbitrary amount of time for the - // next refresh. + // Catalog emit-on-subscribe: every time the relay opens a + // SUBSCRIBE bidi for catalog.json, fire the hook to write + // one group + FIN. moq-lite serves new listeners from the + // relay's per-track latest-group cache, so emitting once + // per relay-side subscribe is enough — late-joining + // watchers behind the same relay get the cached blob + // without us having to maintain a periodic re-emit loop. + // Set BEFORE the relay can race a SUBSCRIBE in; in + // practice the relay's SUBSCRIBE bidi takes a network + // round-trip after our ANNOUNCE Active, so this is safe + // even though the setter is non-suspending. val catalogJson = MoqLiteHangCatalog.opusMono48k(MoqLiteNestsListener.AUDIO_TRACK).encodeJsonBytes() - val republishJob = - try { + catalogPublisher.setOnNewSubscriber { + runCatching { catalogPublisher.send(catalogJson) catalogPublisher.endGroup() - scope.launch { - try { - while (true) { - delay(CATALOG_REPUBLISH_INTERVAL_MS) - catalogPublisher.send(catalogJson) - catalogPublisher.endGroup() - } - } catch (ce: CancellationException) { - throw ce - } catch (_: Throwable) { - // Best-effort: a transient catalog send - // failure is non-fatal — the audio path - // owns terminal-failure detection. - } - } - } catch (t: Throwable) { - runCatching { catalogPublisher.close() } - runCatching { broadcaster.stop() } - runCatching { publisher.close() } - throw t } + } mutableState.value = NestsSpeakerState.Broadcasting( room = current.room, @@ -194,7 +176,6 @@ class MoqLiteNestsSpeaker internal constructor( broadcaster = broadcaster, publisher = publisher, catalogPublisher = catalogPublisher, - catalogRepublishJob = republishJob, parent = this, ) activeHandle = handle @@ -202,17 +183,6 @@ class MoqLiteNestsSpeaker internal constructor( } } - companion object { - /** - * How often the catalog group is re-emitted. The relay drops - * the catalog group when its last subscriber drops, so a - * watcher that attaches mid-broadcast needs a freshly-published - * group to receive the manifest. 2 s keeps late-attach worst - * case bounded without spamming the relay. - */ - const val CATALOG_REPUBLISH_INTERVAL_MS: Long = 2_000L - } - /** * [HotSwappablePublisherSource] implementation. See the interface * kdoc — this method mints a fresh publisher on the session @@ -302,7 +272,6 @@ internal class MoqLiteBroadcastHandle( private val broadcaster: NestMoqLiteBroadcaster, private val publisher: MoqLitePublisherHandle, private val catalogPublisher: MoqLitePublisherHandle, - private val catalogRepublishJob: Job, private val parent: MoqLiteNestsSpeaker, ) : BroadcastHandle { @Volatile private var muted: Boolean = false @@ -321,20 +290,8 @@ internal class MoqLiteBroadcastHandle( override suspend fun close() { if (closed) return closed = true - // Stop the catalog republisher first so it doesn't race a - // concurrent send against the publisher.close below. - try { - catalogRepublishJob.cancelAndJoin() - } catch (ce: kotlinx.coroutines.CancellationException) { - // The parent scope is being cancelled. Continue cleanup of - // the audio + publisher resources, then rethrow. - runCatching { catalogPublisher.close() } - runCatching { publisher.close() } - parent.broadcastClosed(this) - throw ce - } catch (_: Throwable) { - // Best-effort. - } + // Stop the broadcaster first so the audio capture + encoder + // don't keep producing into a closing publisher. try { broadcaster.stop() } catch (ce: kotlinx.coroutines.CancellationException) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index 85d3b0d8d..e532e28a0 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -30,7 +30,6 @@ import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.awaitCancellation -import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow @@ -442,8 +441,6 @@ private class ReissuingBroadcastHandle( */ @Volatile private var hotSwapCatalogPublisher: MoqLitePublisherHandle? = null - @Volatile private var hotSwapCatalogJob: Job? = null - /** Legacy path's per-session handle. Cleared when the session swaps. */ private val liveHandle = AtomicReference(null) private var pumpJob: Job? = null @@ -572,11 +569,10 @@ private class ReissuingBroadcastHandle( // this the catalog goes silent the moment the session recycles // and any watcher that attaches AFTER the recycle sees nothing // to subscribe to. Mirror of [MoqLiteNestsSpeaker.startBroadcasting]'s - // catalog setup; same JSON, same republish cadence. + // catalog setup; same JSON, same emit-on-subscribe pattern. val catalogPayload = MoqLiteHangCatalog.opusMono48k(MoqLiteNestsListener.AUDIO_TRACK).encodeJsonBytes() val priorCatalogPublisher = hotSwapCatalogPublisher - val priorCatalogJob = hotSwapCatalogJob val newCatalogPublisher = try { hotSwap.openPublisherForHotSwap(MoqLiteNestsListener.CATALOG_TRACK) @@ -590,45 +586,23 @@ private class ReissuingBroadcastHandle( null } if (newCatalogPublisher != null) { - val newCatalogJob = - try { + // Set the emit-on-subscribe hook BEFORE installing the + // publisher reference, so the relay can't race a SUBSCRIBE + // in between. + newCatalogPublisher.setOnNewSubscriber { + runCatching { newCatalogPublisher.send(catalogPayload) newCatalogPublisher.endGroup() - scope.launch { - try { - while (true) { - delay(MoqLiteNestsSpeaker.CATALOG_REPUBLISH_INTERVAL_MS) - newCatalogPublisher.send(catalogPayload) - newCatalogPublisher.endGroup() - } - } catch (ce: kotlinx.coroutines.CancellationException) { - throw ce - } catch (_: Throwable) { - // Best-effort: a transient catalog send - // failure on this session is non-fatal — - // the audio path owns terminal-failure - // detection. - } - } - } catch (ce: kotlinx.coroutines.CancellationException) { - runCatching { newCatalogPublisher.close() } - throw ce - } catch (_: Throwable) { - runCatching { newCatalogPublisher.close() } - null } - if (newCatalogJob != null) { - hotSwapCatalogPublisher = newCatalogPublisher - hotSwapCatalogJob = newCatalogJob - // Tear down the prior session's catalog state AFTER the - // new one is installed so the watcher only sees a brief - // overlap rather than a gap. The prior publisher's - // session is about to be torn down by the orchestrator - // anyway, but graceful Announce(Ended) keeps the relay - // book-keeping clean. - if (priorCatalogJob != null) runCatching { priorCatalogJob.cancelAndJoin() } - if (priorCatalogPublisher != null) runCatching { priorCatalogPublisher.close() } } + hotSwapCatalogPublisher = newCatalogPublisher + // Tear down the prior session's catalog publisher AFTER the + // new one is installed so the watcher only sees a brief + // overlap rather than a gap. The prior publisher's session + // is about to be torn down by the orchestrator anyway, but + // graceful Announce(Ended) keeps the relay book-keeping + // clean. + if (priorCatalogPublisher != null) runCatching { priorCatalogPublisher.close() } } try { @@ -703,10 +677,10 @@ private class ReissuingBroadcastHandle( // // Catalog gets explicit teardown because — unlike the audio // publisher which broadcaster.stop() closes for us — the - // catalog publisher + republisher have no broadcaster wrapper. - // Cancel the republish loop FIRST so it can't race the close. - hotSwapCatalogJob?.let { runCatching { it.cancelAndJoin() } } - hotSwapCatalogJob = null + // catalog publisher has no broadcaster wrapper. The + // emit-on-subscribe hook itself doesn't need shutdown: it + // only fires inside [PublisherStateImpl.registerInboundSubscription], + // and closing the publisher prevents any further hook fires. hotSwapCatalogPublisher?.let { runCatching { it.close() } } hotSwapCatalogPublisher = null hotSwapBroadcaster?.let { runCatching { it.stop() } } 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 bf85c60e4..961f9cfe9 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 @@ -1079,6 +1079,18 @@ class MoqLiteSession internal constructor( @Volatile private var publisherClosed = false + /** + * Caller-installed hook fired once per accepted inbound + * SUBSCRIBE — see [MoqLitePublisherHandle.setOnNewSubscriber]. + * Read-and-fire happens OUTSIDE [gate] to avoid deadlocking on + * the hook's own calls to [send] / [endGroup]. + */ + @Volatile private var onNewSubscriberHook: (suspend () -> Unit)? = null + + override fun setOnNewSubscriber(hook: (suspend () -> Unit)?) { + onNewSubscriberHook = hook + } + suspend fun registerAnnounceBidi( bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream, emittedSuffix: String, @@ -1093,6 +1105,11 @@ class MoqLiteSession internal constructor( } suspend fun registerInboundSubscription(sub: MoqLiteSubscribe) { + // Capture the hook INSIDE the lock — guarantees the hook + // observes a fully-registered subscriber when it fires — + // but invoke it OUTSIDE so a hook that calls [send] / + // [endGroup] doesn't deadlock on the same gate. + var hookToFire: (suspend () -> Unit)? = null gate.withLock { if (publisherClosed) { Log.w("NestTx") { "SUBSCRIBE inbound rejected (publisher closed) id=${sub.id} track='${sub.track}'" } @@ -1103,8 +1120,15 @@ class MoqLiteSession internal constructor( return } inboundSubs += sub + hookToFire = onNewSubscriberHook Log.d("NestTx") { "SUBSCRIBE registered id=${sub.id} broadcast='${sub.broadcast}' track='${sub.track}' inboundSubs.size=${inboundSubs.size}" } } + // Launch on the session's scope so the hook outlives the + // bidi pump's per-bidi coroutine if it's slow (e.g. a + // catalog send blocked on transport backpressure). + hookToFire?.let { hook -> + scope.launch { runCatching { hook.invoke() } } + } } /** @@ -1388,6 +1412,31 @@ interface MoqLitePublisherHandle { /** FIN the current group's uni stream. The next [send] starts a fresh group. */ suspend fun endGroup() + /** + * Register a callback that fires once each time a new inbound + * subscriber is registered against this publisher's track (i.e. + * each track-matching SUBSCRIBE bidi the relay opens to us). Used + * to push a "track-latest" payload — the canonical example is the + * broadcast catalog manifest, which a watcher needs to receive on + * subscribe but doesn't change between subscribers — without + * forcing the publisher to maintain a periodic re-emit loop. + * + * Called once per accepted SUBSCRIBE (track filter passed). Fires + * OUTSIDE the publisher's serialisation lock, so the hook can + * safely call [send] / [endGroup] without deadlocking. + * + * Caller MUST set the hook before any subscriber attaches (typically + * immediately after [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.publish] + * returns) — there's no "fire-on-set for existing subscribers" + * replay. For the typical catalog use case the publisher is fresh + * when the hook is set, and the relay's SUBSCRIBE bidi takes a + * round-trip to arrive, so this is safe in practice. + * + * Pass `null` to clear the hook. Calling twice with non-null + * replaces the previous hook (no de-duplication). + */ + fun setOnNewSubscriber(hook: (suspend () -> Unit)?) + /** * Stop publishing. Sends `Announce(Ended)` on every active announce * bidi, FINs the current group, and releases all per-publisher diff --git a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt index 70b1d5eb4..f823438ce 100644 --- a/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt +++ b/nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt @@ -357,6 +357,123 @@ class MoqLiteSessionTest { session.close() } + @Test + fun publisher_setOnNewSubscriber_hook_fires_per_inbound_subscribe() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data") + val hookFireCount = + java.util.concurrent.atomic + .AtomicInteger(0) + publisher.setOnNewSubscriber { + hookFireCount.incrementAndGet() + } + + // First inbound SUBSCRIBE → hook fires once. + val subBidi1 = serverSide.openBidiStream() + subBidi1.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi1.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 0L, + broadcast = "speakerPubkey", + track = "audio/data", + priority = 0x80, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ), + ), + ) + // Wait for SubscribeOk to drain so we know registerInboundSubscription + // ran (and therefore the hook had its chance to launch). + withTimeout(2_000) { subBidi1.incoming().first() } + // Hook fires asynchronously on the session scope; give it a + // moment to land. Use a bounded retry rather than a flat + // delay so the test is fast on the happy path. + withTimeout(2_000) { + while (hookFireCount.get() < 1) kotlinx.coroutines.yield() + } + assertEquals(1, hookFireCount.get()) + + // Second inbound SUBSCRIBE → hook fires again. + val subBidi2 = serverSide.openBidiStream() + subBidi2.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi2.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 1L, + broadcast = "speakerPubkey", + track = "audio/data", + priority = 0x80, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ), + ), + ) + withTimeout(2_000) { subBidi2.incoming().first() } + withTimeout(2_000) { + while (hookFireCount.get() < 2) kotlinx.coroutines.yield() + } + assertEquals(2, hookFireCount.get()) + + publisher.close() + session.close() + } + + @Test + fun publisher_setOnNewSubscriber_hook_does_not_fire_on_track_mismatch() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Publisher serves audio/data only. + val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data") + val hookFireCount = + java.util.concurrent.atomic + .AtomicInteger(0) + publisher.setOnNewSubscriber { + hookFireCount.incrementAndGet() + } + + // Inbound SUBSCRIBE for a different track. Replies + // SubscribeDrop (covered in another test); hook MUST NOT + // fire because no subscriber was actually registered on + // this publisher. + val subBidi = serverSide.openBidiStream() + subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 7L, + broadcast = "speakerPubkey", + track = "video/data", + priority = 0x80, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ), + ), + ) + // Drain the Drop reply. + withTimeout(2_000) { subBidi.incoming().first() } + // No way to wait deterministically for "the hook didn't + // fire"; sleep briefly to let any racing launch surface, + // then assert. Short delay because the hook would launch + // on the same scope as registerInboundSubscription's caller. + kotlinx.coroutines.delay(100) + assertEquals(0, hookFireCount.get()) + + publisher.close() + session.close() + } + @Test fun publisher_replies_subscribeDrop_when_track_is_not_published() = runBlocking {