diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt index 3c0769d55..d324bc753 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/MoqLiteNestsSpeaker.kt @@ -190,7 +190,15 @@ class MoqLiteNestsSpeaker internal constructor( * reconnect wrapper's hot-swap path; not called from the * non-reconnecting path which goes through [startBroadcasting]. */ - override suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle = session.publish(broadcastSuffix = speakerPubkeyHex, track = track) + override suspend fun openPublisherForHotSwap( + track: String, + startSequence: Long, + ): MoqLitePublisherHandle = + session.publish( + broadcastSuffix = speakerPubkeyHex, + track = track, + startSequence = startSequence, + ) /** * Compare-and-clear that runs from inside [close] (already holds @@ -348,8 +356,20 @@ internal interface HotSwappablePublisherSource { * session. Caller owns the returned handle's lifetime (typically * via [com.vitorpamplona.nestsclient.audio.NestMoqLiteBroadcaster.swapPublisher]'s * close-the-old contract). + * + * @param startSequence first group sequence the new publisher will + * assign. Used by the hot-swap path to seed the new session's + * audio track with the previous session's + * [MoqLitePublisherHandle.nextSequence] so kixelated/hang's + * `Container.Consumer.#run` doesn't drop every post-recycle + * group as `sequence < #active`. Pass `0L` for fresh (non- + * continuation) publishers — the catalog track is one such + * case, since its `#active` semantics are different from audio. */ - suspend fun openPublisherForHotSwap(track: String): MoqLitePublisherHandle + suspend fun openPublisherForHotSwap( + track: String, + startSequence: Long = 0L, + ): MoqLitePublisherHandle /** * Surface a broadcast-pipeline terminal failure (e.g. sustained diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt index e532e28a0..79cf850d7 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/ReconnectingNestsSpeaker.kt @@ -496,9 +496,25 @@ private class ReissuingBroadcastHandle( * when the wrapper itself closes. */ private suspend fun runHotSwapIteration(hotSwap: HotSwappablePublisherSource) { + // Carry the previous session's audio-track group sequence + // forward so kixelated/hang's `Container.Consumer.#run` + // doesn't drop every post-recycle group as `sequence < + // #active`. Read BEFORE opening the new publisher — there is + // a tiny race window where the broadcaster could send one + // more group on the old publisher between our read and the + // swap-snapshot below; in practice that window is microseconds + // (the broadcaster's swap is a single volatile write) and the + // broadcaster's group cadence is 1 group/sec at the production + // [NestMoqLiteBroadcaster.framesPerGroup], so the chance of a + // duplicate sequence is negligible. Worst case the watcher + // briefly stalls on one duplicate then catches up. + val startSequence: Long = hotSwapBroadcaster?.currentPublisher?.nextSequence ?: 0L val newPublisher = try { - hotSwap.openPublisherForHotSwap(MoqLiteNestsListener.AUDIO_TRACK) + hotSwap.openPublisherForHotSwap( + track = MoqLiteNestsListener.AUDIO_TRACK, + startSequence = startSequence, + ) } catch (ce: kotlinx.coroutines.CancellationException) { throw ce } catch (_: Throwable) { diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt index 1535d690a..f83344db1 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/audio/NestMoqLiteBroadcaster.kt @@ -119,6 +119,18 @@ class NestMoqLiteBroadcaster( */ @Volatile private var publisher: MoqLitePublisherHandle = initialPublisher + /** + * Volatile read of the broadcaster's current publisher reference, + * for callers (typically the hot-swap pump in + * [com.vitorpamplona.nestsclient.connectReconnectingNestsSpeaker]) + * that want to read its [MoqLitePublisherHandle.nextSequence] + * before swapping in a fresh publisher. Don't use this to send + * — that contract belongs to the broadcaster's send loop and + * the caller would race the loop's swap-snapshot. + */ + val currentPublisher: MoqLitePublisherHandle + get() = publisher + /** * Start capturing + encoding + publishing in the background. * Returns immediately. Calling twice is an error. If 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 4b7f3cbe1..bb9989eca 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 @@ -703,12 +703,24 @@ class MoqLiteSession internal constructor( * distinct `track`. Re-publishing the same `(suffix, track)` pair * or mixing different suffixes is rejected with * [IllegalStateException]. + * + * @param startSequence first group sequence the publisher will + * assign. Defaults to 0 for fresh broadcasts. The hot-swap path + * in [com.vitorpamplona.nestsclient.MoqLiteNestsSpeaker.openPublisherForHotSwap] + * passes the previous publisher's + * [MoqLitePublisherHandle.nextSequence] so the new session's + * group lineage continues monotonically across JWT refreshes — + * kixelated/hang's `Container.Consumer.#run` drops any group + * with `sequence < #active`, so a reset to 0 after a recycle + * silences the watcher until `#active` rolls over. */ suspend fun publish( broadcastSuffix: String, track: String, + startSequence: Long = 0L, ): MoqLitePublisherHandle { ensureOpen() + require(startSequence >= 0L) { "startSequence must be >= 0, got $startSequence" } val normalised = MoqLitePath.normalize(broadcastSuffix) val publisher: PublisherStateImpl state.withLock { @@ -721,7 +733,12 @@ class MoqLiteSession internal constructor( check(activePublishers.none { it.track == track }) { "MoqLiteSession.publish called twice for the same track '$track' on suffix '$normalised'." } - publisher = PublisherStateImpl(suffix = normalised, track = track) + publisher = + PublisherStateImpl( + suffix = normalised, + track = track, + startSequence = startSequence, + ) activePublishers += publisher // Lazy launch — the inbound-bidi pump needs to keep running // for the lifetime of any active publisher. @@ -1083,12 +1100,23 @@ class MoqLiteSession internal constructor( private inner class PublisherStateImpl( override val suffix: String, internal val track: String, + startSequence: Long, ) : MoqLitePublisherHandle { private val gate = Mutex() private val announceBidis = mutableListOf() private val inboundSubs = mutableListOf() private var currentGroup: GroupOutbound? = null - private var nextSequence: Long = 0L + + // `@Volatile` so the hot-swap caller can read this from outside + // the publisher's gate (see [MoqLitePublisherHandle.nextSequence] + // kdoc). Mutation happens only inside [openNextGroupLocked] + // (which holds [gate]); the volatile guarantees a cross-thread + // read sees the latest write without contending the gate. + @Volatile + private var nextSequenceField: Long = startSequence + + override val nextSequence: Long + get() = nextSequenceField // Diagnostic: throttled counter for "send returned false" logs so a // long no-subscriber window doesn't flood logcat at 50 Hz. @@ -1272,7 +1300,8 @@ class MoqLiteSession internal constructor( // expected to be small (1 in nests's listener-per-room // model), so this is fine. val sub = inboundSubs.first() - val sequence = nextSequence++ + val sequence = nextSequenceField + nextSequenceField = sequence + 1L val uni = try { openGroupStream(subscribeId = sub.id, sequence = sequence) @@ -1408,6 +1437,31 @@ interface MoqLitePublisherHandle { */ val suffix: String + /** + * The next group sequence number that will be assigned by [send] / + * [startGroup]. Snapshot-only — read AFTER the broadcaster has + * stopped sending into this publisher (typically just before the + * caller closes the publisher in a hot-swap), so the value is the + * highest-already-used sequence + 1. + * + * Used by [com.vitorpamplona.nestsclient.MoqLiteNestsSpeaker]'s + * hot-swap path to seed the new session's publisher with a + * monotonically-continuing sequence — without this, every JWT + * refresh restarts at sequence 0 and kixelated/hang's + * `Container.Consumer.#run` drops every group whose sequence is + * less than its current `#active` high-water mark, killing audio + * for the watcher until either `#active` rolls over or the + * watcher re-subscribes. + * + * `@Volatile` on the implementation; safe to read from any + * coroutine. The accept-tiny-race window between read and a + * concurrent `send` is closed in practice because the broadcaster + * is responsible for swapping its publisher reference BEFORE the + * caller reads this value (so no further sends land on this + * publisher). + */ + val nextSequence: Long + /** * Start a new group. Allocates a fresh sequence id and opens a new * uni stream pre-loaded with `DataType=Group + GroupHeader`. Idempotent 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 f823438ce..37eaa430c 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 @@ -340,6 +340,65 @@ class MoqLiteSessionTest { session.close() } + @Test + fun publisher_startSequence_seeds_first_group_for_hot_swap_continuation() = + runBlocking { + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Mint a publisher with a non-zero startSequence — simulates + // the hot-swap path's "carry forward old publisher's + // nextSequence" contract. + val publisher = + session.publish( + broadcastSuffix = "speakerPubkey", + track = "audio/data", + startSequence = 42L, + ) + assertEquals(42L, publisher.nextSequence, "fresh publisher reports startSequence as next") + + // Wire up a subscriber so send() doesn't short-circuit. + val subBidi = serverSide.openBidiStream() + subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 11L, + broadcast = "speakerPubkey", + track = "audio/data", + priority = 0x80, + ordered = true, + maxLatencyMillis = 0L, + startGroup = null, + endGroup = null, + ), + ), + ) + withTimeout(2_000) { subBidi.incoming().first() } + + assertEquals(true, publisher.send("opus-1".encodeToByteArray())) + // FIN before draining so toList() terminates rather than + // blocking indefinitely on the still-open uni stream. + publisher.endGroup() + // After the first send, nextSequence should advance to 43. + assertEquals(43L, publisher.nextSequence) + + // First uni stream's GroupHeader.sequence MUST be 42, not 0. + val relayUni = withTimeout(2_000) { serverSide.incomingUniStreams().first() } + val uniChunks = relayUni.incoming().toList() + val buf = MoqLiteFrameBuffer() + uniChunks.forEach { buf.push(it) } + assertEquals(MoqLiteDataType.Group.code, buf.readVarint()) + val header = + MoqLiteCodec.decodeGroupHeader( + buf.readSizePrefixed() ?: error("group header missing"), + ) + assertEquals(42L, header.sequence, "first group's sequence is the seeded startSequence") + + publisher.close() + session.close() + } + @Test fun publisher_send_returns_false_when_no_inbound_subscriber() = runBlocking {