From 43720dd14dc130b91bad1bd3af7a5a3993365677 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 4 May 2026 17:28:18 +0000 Subject: [PATCH] fix(nests): scope moq-lite publisher to a single track to stop catalog hijack MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../nestsclient/MoqLiteNestsSpeaker.kt | 15 +++++++- .../nestsclient/moq/lite/MoqLiteSession.kt | 38 +++++++++++++++++-- .../moq/lite/MoqLiteSessionTest.kt | 8 ++-- ...trNestsSustainedSendOutcomesInteropTest.kt | 4 +- .../NostrnestsProdAudioTransmissionTest.kt | 4 +- 5 files changed, 55 insertions(+), 14 deletions(-) diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 63057173f..b1e9b3dac 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -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 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 e0f8180ce..f2319e357 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 @@ -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() @@ -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) { 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 a258b8527..a7cad504d 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 @@ -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() diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt index 3ae5197f7..4a81cc3f8 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrNestsSustainedSendOutcomesInteropTest.kt @@ -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() diff --git a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt index 94b3880f0..67feda78e 100644 --- a/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt +++ b/nestsClient/src/jvmTest/kotlin/com/vitorpamplona/nestsclient/interop/NostrnestsProdAudioTransmissionTest.kt @@ -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()