diff --git a/nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md b/nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md index 00538054d..73578bad6 100644 --- a/nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md +++ b/nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md @@ -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` | | 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` | -| 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` | ## 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 to `MoqLiteSession.publish()`; store on `PublisherStateImpl`; bit-pack 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_PRIORITY` midpoint, so all existing call sites keep their -prior behavior. The 24-bit sequence window is ample (≈ 6 days at -1 grp/sec, beyond which all newer groups within a single track tie -— but still beat older groups of any LOWER-priority track via the -top byte). +`DEFAULT_PRIORITY` midpoint. The 23-bit sequence window is ample +(≈ 97 days at 1 grp/sec, beyond which all newer groups within a +single track tie — but still beat older groups of any LOWER-priority +track via the high byte). Test seam: `FakeWebTransport.openUniStream` now records the most- recent `setPriority` value via a shared `AtomicInteger` cell that the 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 d6ba0d80d..d16d9aed6 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 @@ -1264,18 +1264,30 @@ class MoqLiteSession internal constructor( // 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 + // bit 31 0 (always — keeps the value non-negative) + // bits 30..23 trackPriority u8 (0..255) + // 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 - // 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 + // long enough to outgrow 23 bits (≈ 97 days at the production + // 1-group/sec cadence). Past that the priority degrades to "all + // newer groups within a track tie", but they still beat older + // groups of any LOWER-priority track via the high byte. + // Production sessions cycle on JWT refresh every 9 min, so the + // wrap is defensive only. + val seq23 = sequence.coerceAtMost(0x7F_FFFFL).toInt() and 0x007F_FFFF + val packedPriority = ((trackPriority and 0xFF) shl 23) or seq23 uni.setPriority(packedPriority) uni.write(Varint.encode(MoqLiteDataType.Group.code)) uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence))) 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 566be7e86..046234f22 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 @@ -507,20 +507,23 @@ class MoqLiteSessionTest { 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. + // trackPriority byte into bits 30..23 and the group sequence + // into bits 22..0 of the value passed to + // WebTransportWriteStream.setPriority. Bit 31 stays 0 so + // the resulting Int is always non-negative — required + // because [QuicConnectionWriter] sorts via signed + // `sortedByDescending { it.priority }`. 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. + // 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 = session.publish( broadcastSuffix = "speakerPubkey", @@ -552,12 +555,18 @@ class MoqLiteSessionTest { 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") + // Expected: (0xC0 shl 23) or (7 and 0x7FFFFF). Compute via + // the same formula so the test asserts the CONTRACT, not + // a hex literal. + val expected = ((0xC0 and 0xFF) shl 23) or (7 and 0x007F_FFFF) + assertEquals(expected, fakeRead.lastSetPriority, "trackPriority occupies bits 30..23; sequence fills the low 23 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. relayUni.incoming().toList() @@ -565,6 +574,77 @@ class MoqLiteSessionTest { 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 fun publisher_send_returns_false_when_no_inbound_subscriber() = runBlocking {