fix(nestsclient): keep stream priority pack non-negative — Lite-03 priority bit-layout
Pre-fix, the M1 priority pack `((trackPriority and 0xFF) shl 24)`
would set bit 31 whenever `trackPriority >= 0x80`, producing a
negative `Int`. `QuicConnectionWriter.sortedByDescending { it.priority }`
sorts via signed `Int.compareTo`, so negative-signed values land
BELOW every positive priority — exactly inverting the intended
"higher trackPriority drains first" ordering.
The default `DEFAULT_TRACK_PRIORITY = 0x80` sits exactly on this
boundary, so this would have hit production the moment a hand-
tuned `trackPriority=0xFF` (highest intended) appeared in any
multi-track scenario.
Reshape the layout to keep bit 31 clear:
bit 31 : 0 (reserved as the sign bit)
bits 30..23 : trackPriority u8 (0..255)
bits 22..0 : sequence low 23 bits (~97 days at 1 grp/sec)
The trackPriority byte still occupies an 8-bit slot, just shifted
one bit lower. Sequence saturation moves from 24 bits (≈ 6 days)
to 23 bits (≈ 97 days) — still ample for the 9-minute JWT refresh
cycle.
Two regression tests:
- The existing `publisher_packs_trackPriority_and_sequence_into_setPriority_value`
is updated to assert the new bit layout AND that the encoded
value is non-negative.
- New `publisher_priority_pack_keeps_top_trackPriority_above_lower_trackPriority`
is the direct guard against the bug: `trackPriority=0xFF`
must encode HIGHER than `trackPriority=0x7F`. Pre-fix this
test fails with `0xFF` encoding to a negative Int below the
positive `0x7F` encoding; post-fix both are positive and
correctly ordered.
Surfaced by the merge-prep security review of the moq-lite
Lite-03/04 audit branch — agent flagged it as "QoS/correctness
not security, excluded by DoS rule" but it's a real bug worth
fixing before merge.
Audit doc updated: bit layout in `nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md`
(M1 + L4 entries).
https://claude.ai/code/session_012TGfo99Ugz7fcCPv85a8Tq
This commit is contained in:
@@ -122,7 +122,7 @@ Severity legend (matches the prior QUIC audit):
|
|||||||
| L1 | Lite-04 ALPN constant defined but codec is Lite-03 only | 🟦 → ✅ | Pre-fix `MoqLiteAlpn.LITE_04` was a documented constant but the codec was hard-wired to Lite-03 — advertising Lite-04 would have caused the very first Announce to desync. Now full Lite-04 support: codec is version-aware via a new `MoqLiteVersion` enum, all three differing fields (Announce.hops as `OriginList`, AnnouncePlease.excludeHop, Probe.rtt) are encoded / decoded correctly, ALPN negotiation surfaces the server's pick via `WebTransportSession.negotiatedSubProtocol`, and `connectNestsListener` / `connectNestsSpeaker` advertise both versions and use whichever the relay echoes. | `MoqLiteCodec.kt`, `MoqLiteSession.kt`, `MoqLiteAlpn.kt`, `QuicWebTransportFactory.kt`, `WebTransportSession.kt`, `NestsConnect.kt` | **fixed in this audit** — see `Fix #8` |
|
| L1 | Lite-04 ALPN constant defined but codec is Lite-03 only | 🟦 → ✅ | Pre-fix `MoqLiteAlpn.LITE_04` was a documented constant but the codec was hard-wired to Lite-03 — advertising Lite-04 would have caused the very first Announce to desync. Now full Lite-04 support: codec is version-aware via a new `MoqLiteVersion` enum, all three differing fields (Announce.hops as `OriginList`, AnnouncePlease.excludeHop, Probe.rtt) are encoded / decoded correctly, ALPN negotiation surfaces the server's pick via `WebTransportSession.negotiatedSubProtocol`, and `connectNestsListener` / `connectNestsSpeaker` advertise both versions and use whichever the relay echoes. | `MoqLiteCodec.kt`, `MoqLiteSession.kt`, `MoqLiteAlpn.kt`, `QuicWebTransportFactory.kt`, `WebTransportSession.kt`, `NestsConnect.kt` | **fixed in this audit** — see `Fix #8` |
|
||||||
| L2 | SubscribeOk always echoes `null/null` for startGroup/endGroup | 🟦 → ✅ | Per spec the publisher MAY narrow the subscriber's requested group bounds. Pre-fix we echoed `null/null` and lost the diagnostic "which group am I about to start sending?" signal. Now the publisher narrows `startGroup` to its `nextSequence` — useful for hot-swap continuations where the seeded `startSequence` is non-zero. `endGroup` stays null (live broadcast, no end). | `MoqLiteSession.kt` Subscribe accept arm | **fixed in this audit** — see `Fix #6` |
|
| L2 | SubscribeOk always echoes `null/null` for startGroup/endGroup | 🟦 → ✅ | Per spec the publisher MAY narrow the subscriber's requested group bounds. Pre-fix we echoed `null/null` and lost the diagnostic "which group am I about to start sending?" signal. Now the publisher narrows `startGroup` to its `nextSequence` — useful for hot-swap continuations where the seeded `startSequence` is non-zero. `endGroup` stays null (live broadcast, no end). | `MoqLiteSession.kt` Subscribe accept arm | **fixed in this audit** — see `Fix #6` |
|
||||||
| L3 | No subscriber-driven Probe API | 🟦 → ✅ | Pre-fix the publisher-side Probe handler existed but no subscriber-side counterpart did, leaving the protocol surface incomplete. New `MoqLiteSession.probe()` opens a Probe bidi (writes `ControlType=4`) and returns a `MoqLiteProbeHandle` whose `updates` flow yields each `MoqLiteProbe` the publisher pushes. Mirrors kixelated's `Subscriber::run_probe_stream`. No production consumer (audio rooms are fixed-rate Opus, no ABR), but the API completes the surface for diagnostic tools. | `MoqLiteSession.kt` `probe()` (new), `MoqLiteHandles.kt` `MoqLiteProbeHandle` (new) | **fixed in this audit** — see `Fix #7` |
|
| L3 | No subscriber-driven Probe API | 🟦 → ✅ | Pre-fix the publisher-side Probe handler existed but no subscriber-side counterpart did, leaving the protocol surface incomplete. New `MoqLiteSession.probe()` opens a Probe bidi (writes `ControlType=4`) and returns a `MoqLiteProbeHandle` whose `updates` flow yields each `MoqLiteProbe` the publisher pushes. Mirrors kixelated's `Subscriber::run_probe_stream`. No production consumer (audio rooms are fixed-rate Opus, no ABR), but the API completes the surface for diagnostic tools. | `MoqLiteSession.kt` `probe()` (new), `MoqLiteHandles.kt` `MoqLiteProbeHandle` (new) | **fixed in this audit** — see `Fix #7` |
|
||||||
| L4 | Stream-priority overflow guard is a saturating cast | 🟦 | Pre-fix, `setPriority(sequence.coerceAtMost(Int.MAX_VALUE).toInt())` saturated at `Int.MAX_VALUE` (≈ 71 yrs at 1 grp/sec). After the M1 fix, sequence saturates at the 24-bit boundary (≈ 6 days within a single track) but the per-track high byte keeps cross-track ordering intact. Defensive only; production sessions cycle on JWT refresh every 9 min. | `MoqLiteSession.kt:openGroupStream` (post-M1) | closed (subsumed by M1 fix) |
|
| L4 | Stream-priority overflow guard is a saturating cast | 🟦 | Pre-fix, `setPriority(sequence.coerceAtMost(Int.MAX_VALUE).toInt())` saturated at `Int.MAX_VALUE` (≈ 71 yrs at 1 grp/sec). After the M1 fix, sequence saturates at the 23-bit boundary (≈ 97 days within a single track) but the per-track high byte keeps cross-track ordering intact. Defensive only; production sessions cycle on JWT refresh every 9 min. | `MoqLiteSession.kt:openGroupStream` (post-M1) | closed (subsumed by M1 fix) |
|
||||||
| L5 | Inbound bidi pump captures `publishersSnapshot` at bidi-arrival time | 🟦 → ✅ | If a new publisher was added to the session (`session.publish(track="…")`) AFTER an inbound bidi opened, the dispatcher couldn't see it. In practice both nests publishers (`audio/data` and `catalog.json`) register before any subscriber arrives, so this never bit. Tightened anyway: snapshot is now read at first-byte dispatch time. | `MoqLiteSession.kt:781-790` (pre-fix) | **fixed in this audit** — see `Fix #4` |
|
| L5 | Inbound bidi pump captures `publishersSnapshot` at bidi-arrival time | 🟦 → ✅ | If a new publisher was added to the session (`session.publish(track="…")`) AFTER an inbound bidi opened, the dispatcher couldn't see it. In practice both nests publishers (`audio/data` and `catalog.json`) register before any subscriber arrives, so this never bit. Tightened anyway: snapshot is now read at first-byte dispatch time. | `MoqLiteSession.kt:781-790` (pre-fix) | **fixed in this audit** — see `Fix #4` |
|
||||||
|
|
||||||
## Specifically checked items (per audit-prompt request)
|
## Specifically checked items (per audit-prompt request)
|
||||||
@@ -274,14 +274,21 @@ lower-rate track the moment audio's outbound queue got congested.
|
|||||||
The fix: add `trackPriority: Int = DEFAULT_TRACK_PRIORITY` parameter
|
The fix: add `trackPriority: Int = DEFAULT_TRACK_PRIORITY` parameter
|
||||||
to `MoqLiteSession.publish()`; store on `PublisherStateImpl`; bit-pack
|
to `MoqLiteSession.publish()`; store on `PublisherStateImpl`; bit-pack
|
||||||
in `openGroupStream` as
|
in `openGroupStream` as
|
||||||
`((trackPriority and 0xFF) shl 24) or (sequence.toInt() and 0x00FF_FFFF)`.
|
`((trackPriority and 0xFF) shl 23) or (sequence.toInt() and 0x007F_FFFF)`.
|
||||||
|
|
||||||
|
Wire layout:
|
||||||
|
- bit 31 : 0 (reserved as the sign bit — required because
|
||||||
|
`QuicConnectionWriter.sortedByDescending { priority }` uses
|
||||||
|
signed `Int.compareTo`; a negative value would sort BELOW
|
||||||
|
every positive priority and invert the intent).
|
||||||
|
- bits 30..23 : trackPriority u8 (0..255).
|
||||||
|
- bits 22..0 : sequence low 23 bits.
|
||||||
|
|
||||||
`DEFAULT_TRACK_PRIORITY = 0x80` matches the existing subscriber-side
|
`DEFAULT_TRACK_PRIORITY = 0x80` matches the existing subscriber-side
|
||||||
`DEFAULT_PRIORITY` midpoint, so all existing call sites keep their
|
`DEFAULT_PRIORITY` midpoint. The 23-bit sequence window is ample
|
||||||
prior behavior. The 24-bit sequence window is ample (≈ 6 days at
|
(≈ 97 days at 1 grp/sec, beyond which all newer groups within a
|
||||||
1 grp/sec, beyond which all newer groups within a single track tie
|
single track tie — but still beat older groups of any LOWER-priority
|
||||||
— but still beat older groups of any LOWER-priority track via the
|
track via the high byte).
|
||||||
top byte).
|
|
||||||
|
|
||||||
Test seam: `FakeWebTransport.openUniStream` now records the most-
|
Test seam: `FakeWebTransport.openUniStream` now records the most-
|
||||||
recent `setPriority` value via a shared `AtomicInteger` cell that the
|
recent `setPriority` value via a shared `AtomicInteger` cell that the
|
||||||
|
|||||||
+22
-10
@@ -1264,18 +1264,30 @@ class MoqLiteSession internal constructor(
|
|||||||
// queue gets congested.
|
// queue gets congested.
|
||||||
//
|
//
|
||||||
// Wire layout into `Int`:
|
// Wire layout into `Int`:
|
||||||
// bits 31..24 trackPriority u8 (0..255) — top byte
|
// bit 31 0 (always — keeps the value non-negative)
|
||||||
// bits 23..0 sequence low 24 bits — wraps every
|
// bits 30..23 trackPriority u8 (0..255)
|
||||||
// ~16M groups within a single track
|
// bits 22..0 sequence low 23 bits (~8.4M groups)
|
||||||
|
//
|
||||||
|
// Bit 31 is reserved as the sign bit because
|
||||||
|
// [com.vitorpamplona.quic.connection.QuicConnectionWriter] sorts
|
||||||
|
// streams via signed `sortedByDescending { it.priority }`.
|
||||||
|
// Setting bit 31 (which would happen for `trackPriority >= 0x80`
|
||||||
|
// shifted by 24) produces a negative `Int` that sorts BELOW
|
||||||
|
// positive priorities, inverting the intent. The audit's
|
||||||
|
// priority-byte default `DEFAULT_TRACK_PRIORITY = 0x80` sits
|
||||||
|
// exactly on this boundary — without the 23-bit shift the
|
||||||
|
// default-vs-default case would be fine but a hand-tuned 0xFF
|
||||||
|
// (highest intended) would sort behind 0x7F (lower intended).
|
||||||
//
|
//
|
||||||
// Saturating cast on `sequence` guards a theoretical broadcast
|
// Saturating cast on `sequence` guards a theoretical broadcast
|
||||||
// long enough to outgrow 24 bits (≈ 6 days at the production
|
// long enough to outgrow 23 bits (≈ 97 days at the production
|
||||||
// 1-group/sec cadence; the priority degrades to "all newer
|
// 1-group/sec cadence). Past that the priority degrades to "all
|
||||||
// groups tie" past that, but they still beat older groups of
|
// newer groups within a track tie", but they still beat older
|
||||||
// any LOWER-priority track via the top byte). Defensive only;
|
// groups of any LOWER-priority track via the high byte.
|
||||||
// production sessions cycle on JWT refresh every 9 min.
|
// Production sessions cycle on JWT refresh every 9 min, so the
|
||||||
val seq24 = sequence.coerceAtMost(0xFF_FFFFL).toInt() and 0x00FF_FFFF
|
// wrap is defensive only.
|
||||||
val packedPriority = ((trackPriority and 0xFF) shl 24) or seq24
|
val seq23 = sequence.coerceAtMost(0x7F_FFFFL).toInt() and 0x007F_FFFF
|
||||||
|
val packedPriority = ((trackPriority and 0xFF) shl 23) or seq23
|
||||||
uni.setPriority(packedPriority)
|
uni.setPriority(packedPriority)
|
||||||
uni.write(Varint.encode(MoqLiteDataType.Group.code))
|
uni.write(Varint.encode(MoqLiteDataType.Group.code))
|
||||||
uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence)))
|
uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence)))
|
||||||
|
|||||||
+95
-15
@@ -507,20 +507,23 @@ class MoqLiteSessionTest {
|
|||||||
fun publisher_packs_trackPriority_and_sequence_into_setPriority_value() =
|
fun publisher_packs_trackPriority_and_sequence_into_setPriority_value() =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
// Lite-03 audit M1: openGroupStream packs the publisher's
|
// Lite-03 audit M1: openGroupStream packs the publisher's
|
||||||
// trackPriority byte into bits 31..24 and the group sequence
|
// trackPriority byte into bits 30..23 and the group sequence
|
||||||
// into bits 23..0 of the value passed to
|
// into bits 22..0 of the value passed to
|
||||||
// WebTransportWriteStream.setPriority. This mirrors the
|
// WebTransportWriteStream.setPriority. Bit 31 stays 0 so
|
||||||
// (track.priority, sequence) ordering kixelated's
|
// the resulting Int is always non-negative — required
|
||||||
// PriorityHandle uses at `rs/moq-lite/src/lite/priority.rs`.
|
// because [QuicConnectionWriter] sorts via signed
|
||||||
// Verified on the *peer* side via FakeReadStream.lastSetPriority,
|
// `sortedByDescending { it.priority }`. Verified on the
|
||||||
// which the in-memory transport uses as a side-channel for
|
// *peer* side via FakeReadStream.lastSetPriority, which the
|
||||||
// priority introspection.
|
// in-memory transport uses as a side-channel for priority
|
||||||
|
// introspection.
|
||||||
val (clientSide, serverSide) = FakeWebTransport.pair()
|
val (clientSide, serverSide) = FakeWebTransport.pair()
|
||||||
val session = MoqLiteSession.client(clientSide, pumpScope)
|
val session = MoqLiteSession.client(clientSide, pumpScope)
|
||||||
|
|
||||||
// Use a non-default trackPriority so a regression that
|
// Use a non-default trackPriority so a regression that
|
||||||
// accidentally drops it back to DEFAULT_TRACK_PRIORITY shows
|
// accidentally drops it back to DEFAULT_TRACK_PRIORITY shows
|
||||||
// up in the assertion.
|
// up in the assertion. 0xC0 sits in the upper half of the
|
||||||
|
// byte range — pre-fix this would have produced a negative
|
||||||
|
// Int (bit 31 set) that mis-sorts.
|
||||||
val publisher =
|
val publisher =
|
||||||
session.publish(
|
session.publish(
|
||||||
broadcastSuffix = "speakerPubkey",
|
broadcastSuffix = "speakerPubkey",
|
||||||
@@ -552,12 +555,18 @@ class MoqLiteSessionTest {
|
|||||||
|
|
||||||
val relayUni = withTimeout(2_000) { serverSide.incomingUniStreams().first() }
|
val relayUni = withTimeout(2_000) { serverSide.incomingUniStreams().first() }
|
||||||
val fakeRead = relayUni as FakeReadStream
|
val fakeRead = relayUni as FakeReadStream
|
||||||
// Expected: (0xC0 shl 24) or (7 and 0x00FF_FFFF) =
|
// Expected: (0xC0 shl 23) or (7 and 0x7FFFFF). Compute via
|
||||||
// 0xC000_0007 in unsigned terms = -1073741817 as Int.
|
// the same formula so the test asserts the CONTRACT, not
|
||||||
// Compute via the same formula so the test asserts the
|
// a hex literal.
|
||||||
// CONTRACT, not a hex literal.
|
val expected = ((0xC0 and 0xFF) shl 23) or (7 and 0x007F_FFFF)
|
||||||
val expected = ((0xC0 and 0xFF) shl 24) or (7 and 0x00FF_FFFF)
|
assertEquals(expected, fakeRead.lastSetPriority, "trackPriority occupies bits 30..23; sequence fills the low 23 bits")
|
||||||
assertEquals(expected, fakeRead.lastSetPriority, "trackPriority dominates the high byte; sequence fills the low 24 bits")
|
// Critical guard: the encoded value MUST be non-negative so
|
||||||
|
// the QUIC writer's signed `sortedByDescending` orders it
|
||||||
|
// correctly relative to lower-trackPriority streams.
|
||||||
|
kotlin.test.assertTrue(
|
||||||
|
(fakeRead.lastSetPriority ?: -1) >= 0,
|
||||||
|
"encoded priority must be non-negative — bit 31 is reserved as the sign bit",
|
||||||
|
)
|
||||||
|
|
||||||
// Drain the stream so the cleanup path doesn't leak.
|
// Drain the stream so the cleanup path doesn't leak.
|
||||||
relayUni.incoming().toList()
|
relayUni.incoming().toList()
|
||||||
@@ -565,6 +574,77 @@ class MoqLiteSessionTest {
|
|||||||
session.close()
|
session.close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun publisher_priority_pack_keeps_top_trackPriority_above_lower_trackPriority() =
|
||||||
|
runBlocking {
|
||||||
|
// Direct guard against the pre-fix sign-extension bug: a
|
||||||
|
// higher trackPriority MUST produce a higher encoded
|
||||||
|
// priority value. Pre-fix `trackPriority=0xFF` with
|
||||||
|
// sequence=0 produced a negative Int that sorted BELOW
|
||||||
|
// `trackPriority=0x7F` with the same sequence — exactly
|
||||||
|
// the inversion the audit M1 fix was supposed to prevent.
|
||||||
|
// We exercise the path by opening a uni stream via a
|
||||||
|
// publisher of each priority and comparing.
|
||||||
|
val (clientSide1, serverSide1) = FakeWebTransport.pair()
|
||||||
|
val (clientSide2, serverSide2) = FakeWebTransport.pair()
|
||||||
|
val sessionHigh = MoqLiteSession.client(clientSide1, pumpScope)
|
||||||
|
val sessionLow = MoqLiteSession.client(clientSide2, pumpScope)
|
||||||
|
|
||||||
|
val pubHigh =
|
||||||
|
sessionHigh.publish(
|
||||||
|
broadcastSuffix = "spk",
|
||||||
|
track = "audio/data",
|
||||||
|
trackPriority = 0xFF,
|
||||||
|
)
|
||||||
|
val pubLow =
|
||||||
|
sessionLow.publish(
|
||||||
|
broadcastSuffix = "spk",
|
||||||
|
track = "audio/data",
|
||||||
|
trackPriority = 0x7F,
|
||||||
|
)
|
||||||
|
|
||||||
|
// Wire each up with a subscriber so send() doesn't drop.
|
||||||
|
for ((server, _) in listOf(serverSide1 to "high", serverSide2 to "low")) {
|
||||||
|
val sub = server.openBidiStream()
|
||||||
|
sub.write(Varint.encode(MoqLiteControlType.Subscribe.code))
|
||||||
|
sub.write(
|
||||||
|
MoqLiteCodec.encodeSubscribe(
|
||||||
|
MoqLiteSubscribe(
|
||||||
|
id = 1L,
|
||||||
|
broadcast = "spk",
|
||||||
|
track = "audio/data",
|
||||||
|
priority = 0x80,
|
||||||
|
ordered = true,
|
||||||
|
maxLatencyMillis = 0L,
|
||||||
|
startGroup = null,
|
||||||
|
endGroup = null,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
withTimeout(2_000) { sub.incoming().first() }
|
||||||
|
}
|
||||||
|
|
||||||
|
pubHigh.send("h".encodeToByteArray())
|
||||||
|
pubHigh.endGroup()
|
||||||
|
pubLow.send("l".encodeToByteArray())
|
||||||
|
pubLow.endGroup()
|
||||||
|
|
||||||
|
val highUni = withTimeout(2_000) { serverSide1.incomingUniStreams().first() } as FakeReadStream
|
||||||
|
val lowUni = withTimeout(2_000) { serverSide2.incomingUniStreams().first() } as FakeReadStream
|
||||||
|
val highPriority = highUni.lastSetPriority ?: error("high priority not set")
|
||||||
|
val lowPriority = lowUni.lastSetPriority ?: error("low priority not set")
|
||||||
|
|
||||||
|
kotlin.test.assertTrue(highPriority > lowPriority, "0xFF trackPriority must encode higher than 0x7F (was $highPriority vs $lowPriority)")
|
||||||
|
kotlin.test.assertTrue(highPriority >= 0, "0xFF trackPriority encoding must stay non-negative (was $highPriority)")
|
||||||
|
|
||||||
|
highUni.incoming().toList()
|
||||||
|
lowUni.incoming().toList()
|
||||||
|
pubHigh.close()
|
||||||
|
pubLow.close()
|
||||||
|
sessionHigh.close()
|
||||||
|
sessionLow.close()
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun publisher_send_returns_false_when_no_inbound_subscriber() =
|
fun publisher_send_returns_false_when_no_inbound_subscriber() =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
|
|||||||
Reference in New Issue
Block a user