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 f4a302d36..51d8f2a18 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 @@ -717,9 +717,24 @@ class MoqLiteSession internal constructor( broadcastSuffix: String, track: String, startSequence: Long = 0L, + /** + * Per-track priority byte, 0..255. Defaults to + * [DEFAULT_TRACK_PRIORITY] (the moq-lite midpoint). Mirrors the + * `track.priority u8` field that kixelated's + * `Publisher::serve_group` mixes with `sequence` via its + * `PriorityHandle` (see `rs/moq-lite/src/lite/priority.rs`). + * Higher values drain ahead of lower ones under congestion; + * within a single track, newer groups (higher sequence) drain + * ahead of older ones. For audio rooms the default is fine — + * we only run one track at the same priority — but a future + * multi-track broadcast (e.g. mixing audio with a low-rate + * status track) can lift audio above its peer. + */ + trackPriority: Int = DEFAULT_TRACK_PRIORITY, ): MoqLitePublisherHandle { ensureOpen() require(startSequence >= 0L) { "startSequence must be >= 0, got $startSequence" } + require(trackPriority in 0..255) { "trackPriority must fit in a byte: $trackPriority" } val normalised = MoqLitePath.normalize(broadcastSuffix) val publisher: PublisherStateImpl state.withLock { @@ -737,6 +752,7 @@ class MoqLiteSession internal constructor( suffix = normalised, track = track, startSequence = startSequence, + trackPriority = trackPriority, ) activePublishers += publisher // Lazy launch — the inbound-bidi pump needs to keep running @@ -1077,6 +1093,7 @@ class MoqLiteSession internal constructor( internal suspend fun openGroupStream( subscribeId: Long, sequence: Long, + trackPriority: Int = DEFAULT_TRACK_PRIORITY, ): com.vitorpamplona.nestsclient.transport.WebTransportWriteStream { // Group streams use reliable QUIC delivery to match the // moq-lite reference (kixelated/moq-rs `serve_group` writes @@ -1093,13 +1110,30 @@ class MoqLiteSession internal constructor( // 150 ms late still falls inside hang's default ~200 ms // jitter buffer) and avoids the dropout entirely. val uni = transport.openUniStream() - // Mirror moq-rs `Publisher::serve_group` - // (`rs/moq-lite/src/lite/publisher.rs`): newer groups get higher - // priority so the QUIC writer drains fresh audio first when - // retransmits queue up under loss. Saturating cast guards a - // theoretical broadcast that runs long enough for `sequence` to - // exceed Int.MAX_VALUE (≈ 71 years at 1 group/sec); defensive only. - uni.setPriority(sequence.coerceAtMost(Int.MAX_VALUE.toLong()).toInt()) + // Mirror moq-rs `Publisher::serve_group` (`rs/moq-lite/src/lite/ + // publisher.rs`) and `PriorityHandle.insert(track.priority, + // sequence)`: priority sorts first by track (higher track = + // drains ahead under congestion), then by group sequence within + // a track (newer = drains ahead). Pre-fix we passed raw + // `sequence` and ignored the per-track byte entirely, which + // worked for our single-track Opus case but starves a low-rate + // companion track (e.g. catalog) the moment audio's outbound + // queue gets congested. + // + // Wire layout into `Int`: + // bits 31..24 trackPriority u8 (0..255) — top byte + // bits 23..0 sequence low 24 bits — wraps every + // ~16M groups within a single track + // + // Saturating cast on `sequence` guards a theoretical broadcast + // long enough to outgrow 24 bits (≈ 6 days at the production + // 1-group/sec cadence; the priority degrades to "all newer + // groups tie" past that, but they still beat older groups of + // any LOWER-priority track via the top byte). Defensive only; + // production sessions cycle on JWT refresh every 9 min. + val seq24 = sequence.coerceAtMost(0xFF_FFFFL).toInt() and 0x00FF_FFFF + val packedPriority = ((trackPriority and 0xFF) shl 24) or seq24 + uni.setPriority(packedPriority) uni.write(Varint.encode(MoqLiteDataType.Group.code)) uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) return uni @@ -1170,6 +1204,13 @@ class MoqLiteSession internal constructor( override val suffix: String, internal val track: String, startSequence: Long, + /** + * Per-track priority byte (0..255) — see [publish] kdoc. + * Mixed with each group's [GroupOutbound.sequence] in + * [openGroupStream] to mirror kixelated's `(track.priority, + * sequence)` priority ordering. + */ + internal val trackPriority: Int, ) : MoqLitePublisherHandle { private val gate = Mutex() private val announceBidis = mutableListOf() @@ -1373,14 +1414,14 @@ class MoqLiteSession internal constructor( nextSequenceField = sequence + 1L val uni = try { - openGroupStream(subscribeId = sub.id, sequence = sequence) + openGroupStream(subscribeId = sub.id, sequence = sequence, trackPriority = trackPriority) } catch (ce: CancellationException) { throw ce } catch (t: Throwable) { Log.w("NestTx") { "openGroupStream threw subId=${sub.id} seq=$sequence: ${t::class.simpleName}: ${t.message}" } throw t } - Log.d("NestTx") { "openGroupStream subId=${sub.id} seq=$sequence" } + Log.d("NestTx") { "openGroupStream subId=${sub.id} seq=$sequence trackPriority=$trackPriority" } return GroupOutbound(sequence = sequence, uni = uni) } } @@ -1399,6 +1440,17 @@ class MoqLiteSession internal constructor( /** moq-lite priority byte midpoint — neutral default. */ const val DEFAULT_PRIORITY: Int = 0x80 + /** + * Default per-track priority byte applied when [publish] is + * called without an explicit `trackPriority`. Same midpoint + * as the subscriber-side [DEFAULT_PRIORITY] — kept distinct + * because the two values are conceptually independent (one + * is the subscriber's priority hint in the SUBSCRIBE wire + * field, the other is the publisher's per-stream drain + * priority in `kixelated/moq`'s `PriorityHandle`). + */ + const val DEFAULT_TRACK_PRIORITY: Int = 0x80 + /** * Bitrate hint (bits/sec) we report on inbound moq-lite Probe * bidis as a publisher. Mirrors the upper-bound of an Opus diff --git a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt index 875bcd568..cb1ef868d 100644 --- a/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt +++ b/nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt @@ -78,8 +78,17 @@ class FakeWebTransport private constructor( // simulate in the in-memory channel. stateLock.withLock { check(open) { "session closed" } } val pipe = Channel(Channel.BUFFERED) - outboundUniStreams.send(FakeReadStream(pipe)) - return ChannelWriteStream(pipe) + // Shared priority cell: the writer (us) stores its most recent + // setPriority call here; the peer-side reader exposes it via + // [FakeReadStream.lastSetPriority] so tests can verify the + // priority value the moq-lite publisher computed without + // peeking into private state. Defaults to 0 (the + // [WebTransportWriteStream.setPriority] kdoc default). + val priorityCell = + java.util.concurrent.atomic + .AtomicInteger(0) + outboundUniStreams.send(FakeReadStream(pipe, priorityCell)) + return ChannelWriteStream(pipe, priorityCell) } override suspend fun sendDatagram(payload: ByteArray): Boolean { @@ -168,8 +177,24 @@ class FakeBidiStream internal constructor( class FakeReadStream internal constructor( private val read: Channel, + /** + * Optional shared cell with the writer side. When present, exposes + * the most recent [WebTransportWriteStream.setPriority] value the + * writer applied. `null` for read streams not paired with a fake + * uni-stream writer (e.g. tests that hand-roll a [Channel]). + */ + private val priorityCell: java.util.concurrent.atomic.AtomicInteger? = null, ) : WebTransportReadStream { override fun incoming(): Flow = read.receiveAsFlow() + + /** + * Last priority value the paired write side applied via + * [WebTransportWriteStream.setPriority], or `0` if no priority was + * ever set. Returns `null` when this read stream isn't paired with + * a fake uni-stream writer. + */ + val lastSetPriority: Int? + get() = priorityCell?.get() } /** @@ -180,6 +205,11 @@ class FakeReadStream internal constructor( */ private class ChannelWriteStream( private val channel: Channel, + /** + * Optional shared cell with the peer-side reader. When present, + * stores each [setPriority] call so the peer can introspect. + */ + private val priorityCell: java.util.concurrent.atomic.AtomicInteger? = null, ) : WebTransportWriteStream { override suspend fun write(chunk: ByteArray) { channel.send(chunk) @@ -189,5 +219,7 @@ private class ChannelWriteStream( channel.close() } - override fun setPriority(priority: Int) = Unit + override fun setPriority(priority: Int) { + priorityCell?.set(priority) + } } 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 47092eeb4..df333ffa7 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 @@ -21,6 +21,7 @@ package com.vitorpamplona.nestsclient.moq.lite import com.vitorpamplona.nestsclient.transport.FakeBidiStream +import com.vitorpamplona.nestsclient.transport.FakeReadStream import com.vitorpamplona.nestsclient.transport.FakeWebTransport import com.vitorpamplona.quic.Varint import kotlinx.coroutines.CoroutineScope @@ -399,6 +400,68 @@ class MoqLiteSessionTest { session.close() } + @Test + fun publisher_packs_trackPriority_and_sequence_into_setPriority_value() = + runBlocking { + // Lite-03 audit M1: openGroupStream packs the publisher's + // trackPriority byte into bits 31..24 and the group sequence + // into bits 23..0 of the value passed to + // WebTransportWriteStream.setPriority. This mirrors the + // (track.priority, sequence) ordering kixelated's + // PriorityHandle uses at `rs/moq-lite/src/lite/priority.rs`. + // Verified on the *peer* side via FakeReadStream.lastSetPriority, + // which the in-memory transport uses as a side-channel for + // priority introspection. + val (clientSide, serverSide) = FakeWebTransport.pair() + val session = MoqLiteSession.client(clientSide, pumpScope) + + // Use a non-default trackPriority so a regression that + // accidentally drops it back to DEFAULT_TRACK_PRIORITY shows + // up in the assertion. + val publisher = + session.publish( + broadcastSuffix = "speakerPubkey", + track = "audio/data", + startSequence = 7L, + trackPriority = 0xC0, + ) + + val subBidi = serverSide.openBidiStream() + subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code)) + subBidi.write( + MoqLiteCodec.encodeSubscribe( + MoqLiteSubscribe( + id = 99L, + 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-frame".encodeToByteArray())) + publisher.endGroup() + + val relayUni = withTimeout(2_000) { serverSide.incomingUniStreams().first() } + val fakeRead = relayUni as FakeReadStream + // Expected: (0xC0 shl 24) or (7 and 0x00FF_FFFF) = + // 0xC000_0007 in unsigned terms = -1073741817 as Int. + // Compute via the same formula so the test asserts the + // CONTRACT, not a hex literal. + val expected = ((0xC0 and 0xFF) shl 24) or (7 and 0x00FF_FFFF) + assertEquals(expected, fakeRead.lastSetPriority, "trackPriority dominates the high byte; sequence fills the low 24 bits") + + // Drain the stream so the cleanup path doesn't leak. + relayUni.incoming().toList() + publisher.close() + session.close() + } + @Test fun publisher_send_returns_false_when_no_inbound_subscriber() = runBlocking {