fix(nests): scope moq-lite publisher to a single track to stop catalog hijack

Listeners stuck on a spinning speaker avatar with no audio (against both Amethyst-hosted and nostrnests.com-hosted Nests). The trace captured on a working speaker phone showed every group keyed to subId=1 / track='catalog.json' even though the broadcaster was pumping Opus frames: 13:21:27.570 inbound SUBSCRIBE id=1 track='catalog.json' 13:21:27.574 inbound SUBSCRIBE id=0 track='audio/data' 13:21:27.609 openGroup seq=0 keyedOnSubId=1 track='catalog.json' 13:21:27.610 broadcaster: send accepted (subscriber attached) ... every subsequent group keyed=1, every audio frame on the catalog stream Root cause: `PublisherStateImpl.openNextGroupLocked` keyed each uni stream off `inboundSubs.first()` with no track filter. The kdoc asserted "Inbound subscription set is expected to be small (1 in nests's listener-per-room model)" — that was true when listeners only opened the audio sub. Once the listener side started opening a catalog sub alongside (commits ~Apr 2026), whichever SUBSCRIBE arrived first (typically the catalog one, by ~4 ms in this trace) became the routing target for all audio frames. The relay forwarded those frames to the listener with subscribeId=catalog, the listener routed them to its catalog handler, RoomSpeakerCatalog.parseOrNull returned null (it's not JSON), and the audio subscription's frames channel never received anything — so onSpeakerActivity never fired and the spinner stayed forever. Fix: per moq-lite Lite-03, a publisher is responsible for one (broadcast, track) tuple. Thread `track` through `MoqLiteSession.publish` and have `PublisherStateImpl.registerInboundSubscription` reject inbound SUBSCRIBEs whose track doesn't match — those still receive SUBSCRIBE_OK from the bidi handler (best-effort per Lite-03's optional-content semantics for catalog) but don't influence the audio routing. `MoqLiteNestsSpeaker.startBroadcasting` passes `track = MoqLiteNestsListener.AUDIO_TRACK` (= "audio/data"). Test callsites (MoqLiteSessionTest, NostrnestsProdAudioTransmissionTest, NostrNestsSustainedSendOutcomesInteropTest) updated to pass the new required parameter; all use track="audio/data" matching what they exercise. The diagnostic logs from the prior commits stay in — they're how we caught this and they'll catch the next one. They can be stripped in a follow-up once the fix is confirmed in the field. https://claude.ai/code/session_01PYYez8a6sjiakyjAxsfCEQ
2026-05-04 17:28:18 +00:00
parent 9b6e09838e
commit 43720dd14d
5 changed files with 55 additions and 14 deletions
@@ -75,8 +75,19 @@ class MoqLiteNestsSpeaker internal constructor(

            // Per the audio-rooms NIP draft + JS reference
            // (`@moq/publish/screen-B680RFft.js:5641`), publishers
-            // claim a broadcast suffix equal to their pubkey hex.
-            val publisher = session.publish(broadcastSuffix = speakerPubkeyHex)
+            // claim a broadcast suffix equal to their pubkey hex and
+            // the audio data sits on track "audio/data". The publisher
+            // must be track-scoped because listeners typically open
+            // BOTH a catalog subscribe AND an audio subscribe per
+            // speaker; without the track filter the publisher would
+            // route Opus frames onto whichever subscribe arrived first
+            // (in practice the catalog one) and the audio subscription
+            // would silently starve.
+            val publisher =
+                session.publish(
+                    broadcastSuffix = speakerPubkeyHex,
+                    track = MoqLiteNestsListener.AUDIO_TRACK,
+                )
            // From here on, the publisher is registered in the session
            // and has a live announce/subscribe path. Anything that
            // throws before we hand a working handle back to the caller
@@ -562,17 +562,20 @@ class MoqLiteSession internal constructor(
     * Only one [publish] is supported per session for now. Calling
     * [publish] twice on the same session is rejected with [IllegalStateException].
     */
-    suspend fun publish(broadcastSuffix: String): MoqLitePublisherHandle {
+    suspend fun publish(
+        broadcastSuffix: String,
+        track: String,
+    ): MoqLitePublisherHandle {
        ensureOpen()
        val normalised = MoqLitePath.normalize(broadcastSuffix)
-        Log.d("NestTx") { "publish suffix='$normalised'" }
+        Log.d("NestTx") { "publish suffix='$normalised' track='$track'" }
        val publisher: PublisherStateImpl
        state.withLock {
            check(!closed) { "session is closed" }
            check(activePublisher == null) {
                "MoqLiteSession.publish called twice — only one broadcast per session is supported"
            }
-            publisher = PublisherStateImpl(suffix = normalised)
+            publisher = PublisherStateImpl(suffix = normalised, track = track)
            activePublisher = publisher
            // Lazy launch — the inbound-bidi pump needs to keep running
            // for the lifetime of any active publisher.
@@ -783,13 +786,33 @@ class MoqLiteSession internal constructor(
    /**
     * Publisher state. Tracks the announce bidis the relay opened to us
     * + the inbound subscriptions a relay (or peer) opened against our
-     * broadcast, and owns the current group's uni stream.
+     * broadcast for [track], and owns the current group's uni stream.
+     *
+     * Per moq-lite Lite-03 a publisher is responsible for one
+     * `(broadcast, track)` tuple — the relay multiplexes multiple
+     * tracks per broadcast by routing each inbound SUBSCRIBE to the
+     * publisher whose track field matches. Subs whose `sub.track`
+     * doesn't match this publisher's [track] are intentionally
+     * ignored so a listener subscribing to e.g. `catalog.json` while
+     * we're only publishing `audio/data` doesn't accidentally hijack
+     * the audio routing — see [registerInboundSubscription] for the
+     * filter and the bug history below.
+     *
+     * Bug history (`nestsClient/plans/2026-05-04-publisher-track-routing.md`):
+     * before this filter, [openNextGroupLocked] keyed each group
+     * stream off `inboundSubs.first()`. When a listener opened both a
+     * `catalog.json` subscribe and an `audio/data` subscribe, whichever
+     * arrived first won the routing race — and because the catalog
+     * SUBSCRIBE typically races ahead of the audio one by a few ms,
+     * every Opus frame ended up on the catalog stream. Listeners saw
+     * a perpetually-spinning speaker avatar with no audio.
     *
     * `gate` serialises access to per-group state so concurrent
     * `send` / `startGroup` / `endGroup` / `close` can't race.
     */
    private inner class PublisherStateImpl(
        override val suffix: String,
+        private val track: String,
    ) : MoqLitePublisherHandle {
        private val gate = Mutex()
        private val announceBidis = mutableListOf<AnnounceBidiEntry>()
@@ -815,6 +838,13 @@ class MoqLiteSession internal constructor(
        suspend fun registerInboundSubscription(sub: MoqLiteSubscribe) {
            gate.withLock {
                if (publisherClosed) return
+                if (sub.track != track) {
+                    Log.d("NestTx") {
+                        "ignoring inbound SUBSCRIBE id=${sub.id} track='${sub.track}' " +
+                            "(publisher serves track='$track' only)"
+                    }
+                    return
+                }
                val wasEmpty = inboundSubs.isEmpty()
                inboundSubs += sub
                if (wasEmpty) {
@@ -258,7 +258,7 @@ class MoqLiteSessionTest {
            val (clientSide, serverSide) = FakeWebTransport.pair()
            val session = MoqLiteSession.client(clientSide, pumpScope)

-            val publisher = session.publish(broadcastSuffix = "speakerPubkey")
+            val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")

            // Relay (serverSide) opens an Announce bidi to us with
            // AnnouncePlease(prefix="").
@@ -283,7 +283,7 @@ class MoqLiteSessionTest {
            val (clientSide, serverSide) = FakeWebTransport.pair()
            val session = MoqLiteSession.client(clientSide, pumpScope)

-            val publisher = session.publish(broadcastSuffix = "speakerPubkey")
+            val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")

            // Step 1: relay opens Subscribe bidi.
            val subBidi = serverSide.openBidiStream()
@@ -346,7 +346,7 @@ class MoqLiteSessionTest {
            val (clientSide, _) = FakeWebTransport.pair()
            val session = MoqLiteSession.client(clientSide, pumpScope)

-            val publisher = session.publish(broadcastSuffix = "speakerPubkey")
+            val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")
            // No relay-opened Subscribe bidi → no subscribers → send is
            // a silent no-op (returns false), matching the listener
            // semantics where the speaker keeps capturing audio even
@@ -363,7 +363,7 @@ class MoqLiteSessionTest {
            val (clientSide, serverSide) = FakeWebTransport.pair()
            val session = MoqLiteSession.client(clientSide, pumpScope)

-            val publisher = session.publish(broadcastSuffix = "speakerPubkey")
+            val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")

            // Relay opens an announce bidi.
            val relayBidi = serverSide.openBidiStream()
@@ -121,7 +121,7 @@ class NostrNestsSustainedSendOutcomesInteropTest {
            val speakerSession = MoqLiteSession.client(speakerWt, pumpScope)
            val publisher =
                InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") {
-                    speakerSession.publish(broadcastSuffix = hostSigner.pubKey)
+                    speakerSession.publish(broadcastSuffix = hostSigner.pubKey, track = "audio/data")
                }

            // ---- listener side: production code path, unchanged.
@@ -452,7 +452,7 @@ class NostrNestsSustainedSendOutcomesInteropTest {
        val speakerSession = MoqLiteSession.client(speakerWt, pumpScope)
        val publisher =
            InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") {
-                speakerSession.publish(broadcastSuffix = hostSigner.pubKey)
+                speakerSession.publish(broadcastSuffix = hostSigner.pubKey, track = "audio/data")
            }

        val listeners = mutableListOf<com.vitorpamplona.nestsclient.NestsListener>()
@@ -644,7 +644,7 @@ class NostrnestsProdAudioTransmissionTest {
                    .client(speakerWt, pumpScope)
            val publisher =
                InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") {
-                    speakerSession.publish(broadcastSuffix = hostSigner.pubKey)
+                    speakerSession.publish(broadcastSuffix = hostSigner.pubKey, track = "audio/data")
                }

            // ---- listener side: production code path, unchanged.
@@ -1013,7 +1013,7 @@ class NostrnestsProdAudioTransmissionTest {
                .client(speakerWt, pumpScope)
        val publisher =
            InteropDebug.stepSuspending(scope, "host: session.publish(broadcastSuffix=hostPub)") {
-                speakerSession.publish(broadcastSuffix = hostSigner.pubKey)
+                speakerSession.publish(broadcastSuffix = hostSigner.pubKey, track = "audio/data")
            }

        val listeners = mutableListOf<com.vitorpamplona.nestsclient.NestsListener>()