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()