nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(nestsclient): pack trackPriority + sequence into stream priority — moq-lite Lite-03 priority.rs

Browse Source 
Per moq-lite Lite-03 (kixelated/moq `rs/moq-lite/src/lite/priority.rs`),
`Publisher::serve_group` calls `priority.insert(track.priority,
sequence)` and feeds the resulting position into `stream.set_priority`.
Priority sorts first by `track.priority u8` (higher track = drains
ahead under congestion), then by group `sequence` within a track
(newer = drains ahead).

Pre-fix we passed raw `sequence.toInt()` directly to setPriority,
ignoring the per-track byte. For our single-Opus-track production case
this was unobservable — newer-first ordering held by sequence
monotonicity — but a future multi-track broadcast (audio + companion
catalog / status track) would have starved the lower-rate track the
moment audio's outbound queue got congested.

Fix: add `trackPriority: Int = DEFAULT_TRACK_PRIORITY` parameter to
`MoqLiteSession.publish()`, store on PublisherStateImpl, and bit-pack
in `openGroupStream`:
  bits 31..24  trackPriority u8 (0..255)
  bits 23..0   sequence low 24 bits

The 24-bit sequence window is ample (≈ 6 days at 1 group/sec, beyond
which all newer groups within a single track tie — but they still
beat older groups of any LOWER-priority track via the top byte).
Production sessions cycle on JWT refresh every 9 min, so the wrap is
defensive only.

`DEFAULT_TRACK_PRIORITY = 0x80` matches the existing subscriber-side
DEFAULT_PRIORITY midpoint, so all existing call sites keep their
prior behavior.

Test seam: `FakeWebTransport.openUniStream` now records the most-
recent `setPriority` value via a shared `AtomicInteger` cell that the
peer-side `FakeReadStream` exposes as `lastSetPriority`. Lets the new
regression test verify the bit-pack formula on the actual peer-side
read stream rather than peeking into private state.

Regression test:
`MoqLiteSessionTest.publisher_packs_trackPriority_and_sequence_into_setPriority_value`.

Audit doc: nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md (M1).

https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
This commit is contained in:
Claude
2026-05-09 14:16:36 +00:00
parent 626521a6f2
commit 38df70504a
3 changed files with 159 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSession.kt
+61 -9
View File
@@ -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<AnnounceBidiEntry>()
@@ -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
nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/transport/FakeWebTransport.kt
+35 -3
View File
@@ -78,8 +78,17 @@ class FakeWebTransport private constructor(
// simulate in the in-memory channel.
stateLock.withLock { check(open) { "session closed" } }
val pipe = Channel<ByteArray>(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<ByteArray>,
/**
* 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<ByteArray> = 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<ByteArray>,
/**
* 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)
}
}
nestsClient/src/commonTest/kotlin/com/vitorpamplona/nestsclient/moq/lite/MoqLiteSessionTest.kt
+63
View File
@@ -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 {