Merge pull request #2816 from vitorpamplona/claude/audit-moq-lite-compliance-NSuPk

moq-lite Lite-03/04 compliance audit: wire-compatible codec & session fixes
This commit is contained in:
Vitor Pamplona
2026-05-09 11:39:12 -04:00
committed by GitHub
14 changed files with 2258 additions and 161 deletions
@@ -62,6 +62,20 @@
## Pointers
- moq-lite wire spec + IETF gap: `nestsClient/plans/2026-04-26-moq-lite-gap.md`
- moq-lite Lite-03/04 compliance audit (2026-05-09): `nestsClient/plans/2026-05-09-moq-lite-rfc-compliance.md`
no 🔴 wire-incompatibilities found; **nine spec tightenings
shipped end-to-end**: AnnouncePlease prefix-mismatch, Subscribe
broadcast validation, trackPriority bit-pack matching
kixelated's `PriorityHandle`, publishers-list freshness on
inbound bidi dispatch, RESET_STREAM with typed code on Drop
replies, STOP_SENDING(SUBSCRIPTION_GONE) on dead group uni,
SubscribeOk narrowing of `startGroup` to publisher's
`nextSequence`, subscriber-driven Probe API
(`MoqLiteSession.probe()`), and full Lite-04 codec + ALPN
negotiation (`MoqLiteVersion` enum, `wt-protocol` parsing,
dual-version advertisement). M6 (Goaway body) closed via spec
verification (no body in Lite-03/04). **No items remain
deferred.**
- Nostrnests integration audit (gaps + roadmap): see most recent doc in `nestsClient/plans/`
- QUIC stack status: `quic/plans/2026-04-26-quic-stack-status.md`
- Audio-rooms NIP draft (needs refresh after the moq-lite findings):
@@ -0,0 +1,520 @@
# moq-lite Lite-03 compliance audit
**Date:** 2026-05-09
**Branch:** `claude/audit-moq-lite-compliance-NSuPk`
**Scope:** `nestsClient/src/commonMain/kotlin/com/vitorpamplona/nestsclient/moq/lite/` plus
the audio glue under `…/audio/` that drives moq-lite. The IETF
`draft-ietf-moq-transport-17` reference impl in `…/moq/` (no `lite/`)
is OUT of scope — it stays as a unit-test target only. The QUIC layer
(`:quic`) is OUT of scope; recently audited and locked.
## TL;DR
The moq-lite Lite-03 implementation is **wire-compatible** with
kixelated's reference relay (moq-rs / moq-lite-03). All on-the-wire
codec primitives — control type discriminators, AnnouncePlease,
Announce, Subscribe (incl. the `Option<u64>` off-by-one and `Duration`
millis-as-varint conventions), SubscribeOk/Drop (with the
type-outside-size-prefix framing peculiar to Lite-03), GroupHeader,
Probe — match the reference Rust impl byte-for-byte. The known gaps
are all spec-loose / future-fragile (🟡 / 🟦), not wire-incompatible
(🔴).
**This audit shipped nine fixes** (M1, M2, M3, M4, M5, L1, L2, L3,
L5) and **closed M6** (verified via WebFetch that Goaway has no
body in the spec, so our existing handler is canonical). **L4**
is subsumed by the M1 fix. M2/M3 originally deferred under the
"`:quic` is locked" constraint; after a merge from `main` brought
in the full RFC-compliant `:quic` layer, the user lifted the
constraint and M2/M3 landed end-to-end. L1/L2/L3 originally
parked as 🟦 non-fixes; the user re-prioritised them and all
three landed with regression tests. **Every gap from the matrix
is now resolved (✅) or explicitly closed.** No 🔴, 🟡, or 🟦
items remain open.
## Methodology
1. Walked every Kotlin file under `moq/lite/` and traced data flow
speaker → relay → listener.
2. Cross-referenced each on-wire message + control byte against
kixelated's reference Rust impl at
`https://github.com/kixelated/moq/tree/main/rs/moq-lite/src/`,
reading specifically:
- `coding/decode.rs`, `coding/encode.rs`, `coding/varint.rs`
(primitive codings — confirmed `Option<u64>` is `0=None /
n=Some(n-1)`, `Duration` is `millis-as-varint`, `bool` is 1
byte 0/1).
- `lite/stream.rs` (ControlType + DataType discriminator codes).
- `lite/announce.rs` (AnnouncePlease + Announce body shape).
- `lite/subscribe.rs` (Subscribe + SubscribeResponse Lite-03
framing).
- `lite/publisher.rs` (`Publisher::serve_group` priority handle,
uni-stream open + group header).
- `lite/subscriber.rs` (Probe direction, group decode loop,
unsubscribe = FIN).
- `lite/probe.rs` (Probe body: `bitrate u62` only on Lite-03).
3. Verified the team's prior `2026-04-26-moq-lite-gap.md` against
live source — every claimed item still holds (see "Verified
working" below).
4. Built the gap matrix (see "Gap matrix" below).
5. Shipped fixes for the two 🟡 items closest to being 🔴 — see
"Fixes shipped" at bottom.
6. Ran `:nestsClient:jvmTest` after each fix — green throughout.
## What's verified working
The shipped Lite-03 surface holds up against the reference Rust impl
at the byte / control-flow level:
| Surface | Verdict | Evidence (file:line) |
| ------------------------------------------------------------------------------------------------------------------------------------ | ------- | -------------------- |
| ALPN `"moq-lite-03"` advertised as the only sub-protocol | ✅ | `MoqLiteAlpn.kt:47`, `QuicWebTransportFactory.kt:88-113` |
| No SETUP / no in-band session message (Lite-03 = WT handshake IS the handshake) | ✅ | `MoqLiteSession.kt:1374-1378` (`client(...)` returns immediately, no SETUP coroutine) |
| `ControlType` codes — `Session=0, Announce=1, Subscribe=2, Fetch=3, Probe=4, Goaway=5` | ✅ | `MoqLiteControlCodes.kt:39-65` (all five recognised + dispatched in `handleInboundBidi`) |
| `DataType::Group=0` on uni-stream open | ✅ | `MoqLiteControlCodes.kt:74-85`, `MoqLiteSession.kt:1047` |
| Path normalisation (strip leading/trailing `/`, collapse runs) | ✅ | `MoqLitePath.kt:44-99` (matches `rs/moq-lite/src/path.rs::Path::new`) |
| Path normalisation applied at every wire boundary | ✅ | `MoqLiteCodec.kt:54, 70, 92` (encode side); `:60, 81, 105` (decode side) |
| AnnouncePlease body shape — `prefix: string` | ✅ | `MoqLiteCodec.kt:52-63` |
| Announce body shape — `status u8, suffix string, hops u62 varint` | ✅ | `MoqLiteCodec.kt:67-85`; status enum `Ended=0, Active=1` matches `rs/moq-lite/src/lite/announce.rs:84-90` |
| Subscribe body — 8 fields: `id u62, broadcast string, track string, priority u8, ordered u8, maxLatency varint, startGroup, endGroup` | ✅ | `MoqLiteCodec.kt:89-100` |
| `Option<u64>` off-by-one collapse: `0=None, n=Some(n-1)` | ✅ | `MoqLiteCodec.kt:286-288` matches `rs/moq-lite/src/coding/decode.rs::impl Decode for Option<u64>` |
| `Duration` encoded as millis-as-varint | ✅ | `MoqLiteCodec.kt:96` matches `rs/moq-lite/src/coding/encode.rs::impl Encode for std::time::Duration` |
| SubscribeResponse framing: type discriminator OUTSIDE size prefix (Lite-03+) | ✅ | `MoqLiteCodec.kt:152-167` matches `rs/moq-lite/src/lite/subscribe.rs::SubscribeResponse::encode` Lite-03 arm |
| SubscribeOk body — 5 fields (priority, ordered, maxLatency, startGroup, endGroup) without id/broadcast/track | ✅ | `MoqLiteCodec.kt:127-135` |
| SubscribeDrop body — `errorCode varint + reasonPhrase string` | ✅ | `MoqLiteCodec.kt:137-142` |
| Group uni-stream layout: `DataType=0` + size-prefixed `(subscribeId, sequence)` + size-prefixed frame loop until FIN | ✅ | `MoqLiteSession.kt:1021-1050` (publisher), `:594-668` (listener `drainOneGroup`) |
| Probe body — `bitrate u62 varint` only (rtt is Lite-04+) | ✅ | `MoqLiteCodec.kt:262-266` (encode), `:247-251` (decode) |
| Probe direction — subscriber opens bidi, publisher writes Probe messages | ✅ | `MoqLiteSession.kt:925-944` (publisher arm of `handleInboundBidi`) |
| Subscribe id allocation — per-session, monotonic, varint | ✅ | `MoqLiteSession.kt:83, 246-251` (incremented under `state` mutex) |
| Group sequence allocation — per-publisher, monotonic, varint, hot-swap-aware (`startSequence` parameter) | ✅ | `MoqLiteSession.kt:716-746, 1129-1132, 1306-1329` |
| Per-group end = QUIC FIN | ✅ | `MoqLiteSession.kt:1264-1271` (`endGroup`) |
| Broadcast ended = `Announce(Ended)` BEFORE FIN (correct ordering) | ✅ | `MoqLiteSession.kt:1273-1304` (`PublisherStateImpl.close`) |
| Unsubscribe = FIN subscribe bidi's send side (no UNSUBSCRIBE message) | ✅ | `MoqLiteSession.kt:670-675` |
| Reliable QUIC streams for groups (NOT bestEffort; matches kixelated) | ✅ | `MoqLiteSession.kt:1024-1039` + comment block explaining the prior bestEffort regression |
| Stream priority hint — newer groups (higher sequence) drain first under congestion | ✅ | `MoqLiteSession.kt:1040-1046`; sequence-as-priority is monotonically increasing, same direction as kixelated's `PriorityHandle.insert(track.priority, sequence)` |
| Auth — purely WebTransport-CONNECT-time (JWT in `?jwt=` query); moq-lite layer touches no claims | ✅ | `MoqLiteSession.kt` has zero references to JWT / claims / auth |
| Reconnect — no moq-lite-layer message; transport drop ⇒ open new WT session and resubscribe | ✅ | Reconnect lives in `connectReconnectingNestsListener` / `…NestsSpeaker` (transport-layer concern) |
| Goaway recognised + FIN'd cleanly (no body decode today; relay migration not implemented) | ✅ | `MoqLiteControlCodes.kt:50-58`, `MoqLiteSession.kt:973-990` |
| Catalog publisher emits one frame per inbound subscriber via `setOnNewSubscriber` hook (matches hang.live consumer expectations) | ✅ | `MoqLitePublisherHandle.kt:91-114`, `MoqLiteNestsSpeaker.kt:160-181` |
| `hang` "legacy" container: each frame is `varint(timestamp_us) + raw_opus_packet` | ✅ | `NestMoqLiteBroadcaster.kt:212-345` (encode), `MoqLiteNestsListener.kt:279-285` (decode strip) |
The team's `2026-04-26-moq-lite-gap.md` (✅ DONE banner) holds — every
phase claim is still accurate against the live source.
## Gap matrix
Severity legend (matches the prior QUIC audit):
- 🔴 **High** — wire-incompatible with the reference relay; audio
doesn't work.
- 🟡 **Medium** — works against the reference relay today but
violates spec; future relay versions may break.
- 🟦 **Low** — diagnostic / observability gap; no functional impact.
| # | Spec § / Behavior | Severity | Gap | Evidence (file:line) | Status |
| -- | --------------------------------------------------------------------------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------- | ------ |
| H0 | (none — no wire-incompatible items found) | 🔴 | — | — | n/a |
| M1 | `Publisher::serve_group` priority parity | 🟡 → ✅ | We assigned each new group `priority = sequence (i32 cast)`, ignoring `track.priority` and never re-prioritising in flight. kixelated computes `let priority = priority.insert(track.priority, sequence); stream.set_priority(priority.current())` (per `rs/moq-lite/src/lite/publisher.rs::serve_group`). For our single Opus track at 1 group/sec the difference was unobservable (newer-first ordering still held). | `MoqLiteSession.kt:openGroupStream` (pre-fix) | **fixed in this audit** — see `Fix #3` |
| M2 | `STOP_SENDING` for single-group cancel | 🟡 → ✅ | The Lite-03 spec lets a receiver cancel a specific group via `STOP_SENDING` on its uni stream. Pre-fix we closed the consumer-facing frames channel only — the publisher's uni stream stayed open until natural FIN, wasting bandwidth on frames the listener would discard plus relay queue pressure on the per-subscriber forward pipeline. The `:quic` `QuicStream.stopSending(errorCode)` API existed but wasn't exposed through `WebTransportReadStream`. Now plumbed end-to-end: extended `WebTransportReadStream` with `stopSending(code)`, plumbed through `:quic`'s `StrippedWtStream`, and `MoqLiteSession.drainOneGroup` fires `stopSending(SUBSCRIPTION_GONE)` once on the first frame that observes `sub == null`. | `MoqLiteSession.kt:drainOneGroup` (pre-fix), `WebTransportSession.kt` (read interface) | **fixed in this audit** — see `Fix #5` |
| M3 | `RESET_STREAM` with `Error::to_code()` | 🟡 → ✅ | Pre-fix all error / cancel paths FIN'd gracefully via `runCatching { bidi.finish() }`. Lite-03 spec says errors on any stream are conveyed by `RESET_STREAM(application_error_code u32)`. The `SubscribeDrop` reply paths now write the body and then `RESET_STREAM(errorCode)` instead of `finish()`, so a peer watching only the QUIC layer (no body decode) can distinguish typed rejection from graceful "publisher gone" FIN. Plumbing extended `WebTransportWriteStream` with `reset(code)` and routed through both QUIC-backed adapters + `:quic`'s `StrippedWtStream`. | `MoqLiteSession.kt:handleInboundBidi` Drop replies (pre-fix) | **fixed in this audit** — see `Fix #5` |
| M4 | AnnouncePlease prefix-mismatch falls back to full suffix | 🟡 → ✅ | When the relay opened an Announce bidi with `prefix="X"`, our publisher emitted `Active(suffix=ourFullSuffix)` even when our suffix didn't start with `X`. The relay would observe an Active update for a broadcast it didn't ask about. In production the relay always asks for `prefix=""`, so this never bit empirically — but it's a spec violation. | `MoqLiteSession.kt:841-852` (pre-fix) | **fixed in this audit** — see `Fix #1` |
| M5 | Inbound Subscribe doesn't validate broadcast field | 🟡 → ✅ | When the relay opened a Subscribe bidi, we matched on `track` only, never checking `sub.broadcast == publisher.suffix`. A relay (or peer) could subscribe to broadcast `"otherPubkey"` on our connection and we'd happily route OUR audio to them. The production relay routes correctly, so this never bit empirically — but it's a spec violation. | `MoqLiteSession.kt:861-898` (pre-fix) | **fixed in this audit** — see `Fix #2` |
| M6 | Goaway body decoding + migration handler | 🟡 → ✅ | We recognise `ControlType::Goaway = 5` and FIN cleanly. WebFetched the kixelated reference (`rs/moq-lite/src/lite/{stream,client}.rs`): **Goaway has no body schema in moq-lite Lite-03** — it's a single ControlType byte with no payload, not even a migration URL. Our existing handler (recognise, log, FIN) is the canonical implementation; "no body decode" was never a gap. | `MoqLiteSession.kt:973-990`, `MoqLiteControlCodes.kt:50-58` | **closed in this audit** — see `M6 closure` |
| 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 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)
- **Stream priorities + the "stream cliff":** the priority assignment
drains newer groups first (sequence is monotonically increasing,
higher value = higher priority per `:quic`'s `QuicStream.priority`
contract). The stream cliff investigation (`2026-05-01-quic-stream-
cliff-investigation.md`) traced the production cliff to TWO
separate root causes, neither of which was a moq-lite priority
violation: (a) `:quic` not emitting `MAX_STREAMS_UNI` (real bug,
fixed in `:quic`); (b) the production relay's per-subscriber
forward-stream-rate ceiling (≈40 streams/sec) being lower than our
`framesPerGroup=1` push rate of 50 streams/sec. The mitigation
(`framesPerGroup=50` in production / `=5` in interop tests) is a
cadence tuning, not a spec violation. Verified: under the current
default the listener and speaker both stream cleanly for
multi-minute sessions in production logs.
- **Group lifetime / stream retirement:** out of scope per audit
prompt (`:quic` is locked). At the moq-lite layer we're confident
groups are short-lived (one per `framesPerGroup * 20ms` =
`framesPerGroup * 0.02s`); each ends in QUIC FIN, and the
per-publisher `currentGroup` reference is dropped on every
`endGroup` so the GC root chain doesn't pin retired streams.
- **Subscribe / Announce flow:** verified end-to-end — see
`MoqLiteSessionTest.subscribe_writes_request_and_returns_handle_on_ok`
+ `…publisher_acks_subscribe_and_pushes_group_data_on_uni_stream` +
`…publisher_replies_to_announcePlease_with_active_announce`.
Wire-shape matches the byte-level reference in
`2026-04-26-moq-lite-gap.md`.
- **Authentication:** moq-lite Lite-03 has no auth message inside the
WT session — auth is purely the WT-CONNECT JWT in `?jwt=` query.
Verified: `MoqLiteSession.kt` has zero auth references; all auth
lives in `OkHttpNestsClient.mintToken` and the WT factory's URL
construction.
- **Reconnect:** moq-lite has no reconnect message; on transport
drop the `connectReconnectingNests*` wrapper opens a new WT
session, mints a new moq-lite session, and re-issues subscribes
(`ReconnectingNestsListener.kt:317-465`). Speaker side hot-swaps
the publisher reference inside a long-running broadcaster
(`NestMoqLiteBroadcaster.swapPublisher`), continuing the group
sequence across the swap so kixelated/hang's `Container.Consumer.#run`
doesn't drop our post-swap groups.
- **Best-effort vs reliable:** verified reliable. The previous
`bestEffort=true` shape was reverted (see comment block at
`MoqLiteSession.kt:1024-1039`); kixelated's reference uses
reliable QUIC streams in `Publisher::serve_group`. Match.
## What's deliberately deferred
Nothing. Every gap surfaced by the audit is now resolved. The
remaining open items are external follow-ups (not moq-lite gaps):
- **`kixelated/moq` feature request** — file an issue
describing the production stream-cliff symptom (relay's
per-subscriber forward queue starvation under sustained
push) and propose: (a) per-deployment tuning of the
unbounded `FuturesUnordered` task pool, and (b) a deadline
on `serve_group()`'s `open_uni().await` derived from the
active subscriber's smallest `max_latency`. This is a
feature request against the upstream relay, not a defect in
our code.
## Closed in this audit (no fix needed)
### M6 closure — Goaway has no body in moq-lite Lite-03
A second WebFetch pass against
`https://raw.githubusercontent.com/kixelated/moq/main/rs/moq-lite/src/lite/stream.rs`
and `…/client.rs` confirmed: **`ControlType::Goaway = 5` is a bare
discriminator with no payload schema in moq-lite Lite-03.** No URL,
no preferred-relay address, no error code — it's just the byte. The
"no body decode" comment in our existing handler
(`MoqLiteSession.kt:973-990`) was never a gap; there was nothing to
decode. The wrapper-layer `connectReconnectingNests*` already absorbs
the eventual hard disconnect, and a Goaway with no body has nothing
more to say than "I'm shutting down, expect FIN." Our recognise + log
+ FIN behavior is the canonical implementation.
If a future Lite-04+ revision adds a Goaway body (URL, etc.) we'd
need to decode it — that becomes part of the L1 Lite-04 codec work.
## Fixes shipped in this audit
### Fix #1 — Don't emit Active for non-matching AnnouncePlease prefix (M4)
When the relay opened an Announce bidi with a `prefix` that didn't
match our broadcast suffix, we'd fall through `MoqLitePath.stripPrefix`'s
null result and emit `Active(suffix=ourFullSuffix, hops=0)` anyway,
falsely advertising our broadcast under a prefix the subscriber didn't
ask for. The reference Rust implementation
(`rs/moq-lite/src/lite/announce.rs::Producer`) only emits Active
updates for broadcasts whose path starts with the requested prefix.
Production never hit this because the relay's announce bidi to us
always asks for `prefix=""` — but a future relay version (or a
direct peer-to-peer subscriber asking about a specific room) would
see ghost broadcasts under their requested prefix.
The fix: when `MoqLitePath.stripPrefix(please.prefix, ourSuffix)`
returns null, FIN the announce bidi without writing any Announce.
Subscriber sees a clean end-of-flow.
Regression test:
`MoqLiteSessionTest.publisher_skips_announce_when_announce_please_prefix_does_not_match`.
### Fix #2 — Reject inbound Subscribe whose broadcast doesn't match our suffix (M5)
When the relay opened a Subscribe bidi, we matched the requested
`track` against our publisher set but never checked
`sub.broadcast == ourSuffix`. A relay (or peer) could open a
Subscribe with broadcast `"otherPubkey"` on our connection and we'd
route OUR audio to them.
Production never hit this because moq-rs's relay routes Subscribe
messages to the specific publisher whose broadcast matches the
requested path — but a buggy or malicious peer connecting to us
directly would have been able to siphon our audio under any
broadcast name.
The fix: in the Subscribe arm of `handleInboundBidi`, check
`sub.broadcast == publisher.suffix` (after path normalisation —
both inputs already are, but defensive normalize on the SUBSCRIBE
side too in case a peer skipped it). If mismatch, reply
`SubscribeDrop(errorCode=BROADCAST_DOES_NOT_EXIST,
reason="<requested> not published on this session (we publish
<ours>)")` and FIN.
Regression test:
`MoqLiteSessionTest.publisher_replies_subscribeDrop_when_broadcast_does_not_match`.
### Fix #3 — Pack trackPriority + sequence into stream priority (M1)
`Publisher::serve_group` in kixelated's reference
(`rs/moq-lite/src/lite/publisher.rs`) calls
`priority.insert(track.priority, sequence)` and feeds the resulting
position into `stream.set_priority`. The `PriorityHandle` orders
streams 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
`uni.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.
The fix: add `trackPriority: Int = DEFAULT_TRACK_PRIORITY` parameter
to `MoqLiteSession.publish()`; store on `PublisherStateImpl`; bit-pack
in `openGroupStream` as
`((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. 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
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`.
### Fix #5 — Plumb RESET_STREAM + STOP_SENDING through WebTransport (M2 + M3)
The two items shipped together because the plumbing is shared. After
the merge from `main` brought in the full RFC-compliant `:quic`
layer (commits `9f7f6a9e..6f32975c`), the user lifted the
"`:quic` is locked" constraint that originally deferred M2/M3.
**Plumbing** (new commit `feat(quic,nestsclient): plumb RESET_STREAM
+ STOP_SENDING through WebTransport`):
- `:quic`'s `StrippedWtStream` gains optional `reset` /
`stopSending` closures, wired in `WtPeerStreamDemux.emitStripped`
through `QuicStream.resetStream(code)` /
`QuicStream.stopSending(code)` + `driver.wakeup()`.
- `WebTransportReadStream.stopSending(code)` and
`WebTransportWriteStream.reset(code)` added to the public
interface — both suspending, both first-call-wins.
- All adapters (`QuicBidiStreamAdapter`,
`QuicUniWriteStreamAdapter`, `StrippedWtBidiStreamAdapter`,
`StrippedWtReadStreamAdapter`) route directly to the underlying
QUIC stream API.
- `FakeWebTransport`'s `FakeBidiStream`, `FakeReadStream`,
`ChannelWriteStream` (now public for test access) record reset
/ stopSending codes in shared `AtomicLong` cells with sentinel
`NO_CODE = Long.MIN_VALUE`. New properties `lastResetCode`,
`lastStopSendingCode`, `lastPeerResetCode`,
`lastPeerStopSendingCode`, `peerStopSendingCode` give tests
typed-error introspection without mock magic.
**M3 (publisher Drop reply uses RESET_STREAM)** (`fix(nestsclient):
RESET_STREAM with typed code after SubscribeDrop body`):
- In `handleInboundBidi`, both Drop reply paths
(`BROADCAST_DOES_NOT_EXIST`, `TRACK_DOES_NOT_EXIST`) now write
the Drop body and then `bidi.reset(errorCode)` instead of
`bidi.finish()`.
- The errorCode matches the body's `errorCode` field, so a peer
that decodes the Drop sees the same number as one that only
sees the RESET_STREAM frame at the QUIC layer.
- Strengthened the existing M5 + Drop tests to additionally
assert `subBidi.lastPeerResetCode == errorCode` so the
typed-error contract is locked.
**M2 (listener stopSending on dead group)** (`fix(nestsclient):
STOP_SENDING on group uni when subscription already canceled`):
- In `drainOneGroup`, the first frame that observes `sub ==
null` (subscription already canceled) fires
`stream.stopSending(MoqLiteStreamCancelCode.SUBSCRIPTION_GONE)`
once, latched, so the publisher abandons in-flight retransmits
instead of wasting bandwidth.
- New error-code constant
`MoqLiteStreamCancelCode.SUBSCRIPTION_GONE = 0x10L` lives in
`MoqLiteMessages.kt` next to `MoqLiteSubscribeDropCode`.
- Practical impact in production: the relay's per-subscriber
forward pipeline (already documented in the stream-cliff
investigation) keeps queueing groups for slow / canceled
subscribers; an early `stopSending` at the listener side
relieves the relay's queue earlier, complementing the
`framesPerGroup` mitigation.
Regression tests:
- `MoqLiteSessionTest.publisher_replies_subscribeDrop_when_broadcast_does_not_match`
— additional `lastPeerResetCode` assertion (M3).
- `MoqLiteSessionTest.publisher_replies_subscribeDrop_when_track_is_not_published`
— additional `lastPeerResetCode` assertion (M3).
- `MoqLiteSessionTest.listener_stopSending_group_uni_when_subscription_already_canceled`
— new test; verifies `peerStopSendingCode` lands the correct
cancel code (M2).
### Fix #4 — Refresh publisher list per-dispatch on inbound bidi (L5)
`handleInboundBidi` previously snapshotted the publisher list at the
TOP of the function (= bidi-arrival time) and used that snapshot for
the rest of the bidi's lifetime. A publisher registered between
bidi-open and the first inbound chunk would miss the dispatcher's
view, even though the publisher was already live in
`activePublishers` by the time we needed to dispatch.
Practical impact today: zero — both nests publishers register from
`MoqLiteNestsSpeaker.startBroadcasting` before the relay's SUBSCRIBE
bidi for either track lands. The ~few-ms gap is below the network
round-trip floor.
But the contract is narrower if we read the list at first-byte time:
we then see every publisher that was registered up to the moment we
needed to make a routing decision. Move the publishers fetch from
the function-top to inside the `if (!dispatched)` block, after the
control-type byte has been read. On empty publisher list (the case
we previously short-circuited at function-top), we now FIN cleanly so
the peer's wait resolves instead of hanging on an idle bidi.
No new regression test — exercised end-to-end by the existing
publisher tests, which now pass against the freshness-aware
dispatcher.
### Fix #6 — SubscribeOk narrows startGroup to publisher.nextSequence (L2)
When accepting a SUBSCRIBE, the publisher MAY narrow the
subscriber's `startGroup` / `endGroup` request bounds per spec.
Pre-fix we always echoed `null/null`, which lost the diagnostic
"which group am I about to start sending?" information —
particularly useful for hot-swap continuations where
[MoqLitePublisherHandle.nextSequence] is non-zero (the seeded
`startSequence` from the previous moq-lite session).
The fix narrows `startGroup` to `targetPublisher.nextSequence`.
The subscriber decodes this as "the next group on this
subscription will be sequence N" and can log / surface the
join-point. `endGroup` stays null because live audio rooms have
no end in sight; the subscriber's request bound is honoured
implicitly when the publisher closes.
Regression test:
`MoqLiteSessionTest.publisher_subscribeOk_narrows_startGroup_to_next_sequence`.
### Fix #7 — Subscriber-driven Probe API (L3)
`MoqLiteSession.probe()` mirrors kixelated's
`Subscriber::run_probe_stream`
(`rs/moq-lite/src/lite/subscriber.rs`): opens a bidi, writes
`ControlType::Probe` (varint 4), and returns a
`MoqLiteProbeHandle` whose `updates` flow yields each
size-prefixed `MoqLiteProbe` message the publisher pushes. The
handle's `close()` FINs the bidi and cancels the pump.
`updates` is a `MutableSharedFlow(replay=8)` so a collector that
attaches after the publisher's first emit doesn't miss it
(matches the same shape the announce-watch uses).
No production consumer for the API today — Amethyst's nests
listener doesn't run ABR on a fixed-rate Opus encoder. The API
exists to round out the moq-lite Lite-03/04 surface: a
diagnostic tool can now read a publisher's bitrate without
subscribing to its data track.
Regression test:
`MoqLiteSessionTest.subscriber_probe_writes_control_type_and_decodes_publisher_bitrate`.
### Fix #8 — Version-aware Lite-03/04 codec + ALPN negotiation (L1)
Pre-fix the codec was hard-wired to Lite-03; advertising the
documented `MoqLiteAlpn.LITE_04` constant would have caused the
very first Announce to desync because Lite-04 reshapes three
fields. Now full Lite-04 support, gated on a runtime version
discriminator selected by ALPN negotiation.
Wire diff verified via WebFetch against
`kixelated/moq` `main` (2026-05-09) at
`rs/moq-lite/src/lite/{announce,subscribe,probe}.rs` and
`rs/moq-lite/src/model/origin.rs`:
- `Announce.hops`: Lite-03 = single varint count (the spec
fills with `Origin::UNKNOWN` placeholders on decode);
Lite-04 = `varint(count) + count × varint(originId)` (the
`OriginList`). MAX_HOPS = 32.
- `AnnouncePlease.excludeHop`: Lite-03 absent; Lite-04 single
varint after `prefix`. Sentinel `0` = no exclusion.
- `Probe.rtt`: Lite-03 absent; Lite-04 single varint after
`bitrate`. Sentinel `0` = unknown (decoded as null).
Outgoing `Some(0)` clamped to `Some(1)` to avoid colliding
with the sentinel — mirrors kixelated's `encode_msg` clamp.
Surface added:
- `MoqLiteVersion` enum (LITE_03, LITE_04) with `fromAlpn`.
- All affected `MoqLiteCodec` methods take `version:
MoqLiteVersion = LITE_03`.
- `MoqLiteSession.client(transport, scope, version)` carries
the version through every codec invocation.
- Data classes evolved: `MoqLiteAnnouncePlease.excludeHop`
(default 0L), `MoqLiteAnnounce.hops` (`Long` →
`List<Long>`), `MoqLiteProbe.rtt` (`Long?`, default null).
- `WebTransportSession.negotiatedSubProtocol: String?` exposes
the server's `wt-protocol` selection.
`QuicWebTransportFactory` parses the response HEADERS,
extracts the SF-string from `wt-protocol`, and threads it
into `QuicWebTransportSession`.
`FakeWebTransport.pair(negotiatedSubProtocol = …)` lets
tests drive the same path.
- Default advertised list now `[moq-lite-04, moq-lite-03]`.
Lite-04 first to match kixelated's preference; servers that
don't support it fall back to Lite-03 cleanly.
- `connectNestsListener` / `connectNestsSpeaker` resolve the
version via
`resolveMoqLiteVersion(webTransport.negotiatedSubProtocol)`
and pass it to `MoqLiteSession.client(...)`. Falls back to
Lite-03 when the server doesn't echo `wt-protocol` (older
deployments) or echoes an unrecognised value (forward-compat).
Regression tests (all green):
- `MoqLiteCodecTest.announcePlease_lite04_round_trips_excludeHop`
- `MoqLiteCodecTest.announcePlease_lite03_omits_excludeHop_on_wire`
- `MoqLiteCodecTest.announce_lite04_round_trips_full_origin_list`
- `MoqLiteCodecTest.announce_lite03_drops_origin_ids_keeps_count`
- `MoqLiteCodecTest.announce_decoder_rejects_oversize_hop_count`
- `MoqLiteCodecTest.probe_lite04_round_trips_rtt`
- `MoqLiteCodecTest.probe_lite04_clamps_some_zero_to_one_to_avoid_unknown_sentinel`
- `MoqLiteCodecTest.probe_lite04_decodes_zero_rtt_as_null`
- `MoqLiteCodecTest.probe_lite03_wire_omits_rtt`
- `MoqLiteSessionTest.lite04_announce_round_trips_full_origin_list_through_session`
## Build / test verification
Baseline `:nestsClient:jvmTest` was green at audit start
(140 tests). After both fixes + regression tests, still green.
Default-mode tests run in seconds; the `-DnestsInterop=true` Docker
suite is unchanged by these fixes (they only tighten our existing
publisher-side behaviour against malformed input the production
relay never sends).
## Pointers
- Wire spec + IETF gap (DONE banner): `2026-04-26-moq-lite-gap.md`.
- Stream cliff investigation (production-fixed):
`2026-05-01-quic-stream-cliff-investigation.md`.
- Frames-per-group reconciliation (production vs interop):
`2026-05-07-framespergroup-reconciliation.md`.
- Audio-rooms completion plan: `2026-04-26-audio-rooms-completion.md`
(will reference this audit in its compliance section).
- Reference Rust impl: `https://github.com/kixelated/moq/tree/main/rs/moq-lite/src/`.
@@ -24,6 +24,7 @@ import com.vitorpamplona.nestsclient.audio.AudioCapture
import com.vitorpamplona.nestsclient.audio.OpusEncoder
import com.vitorpamplona.nestsclient.moq.SubscribeHandle
import com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession
import com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion
import com.vitorpamplona.nestsclient.transport.WebTransportException
import com.vitorpamplona.nestsclient.transport.WebTransportFactory
import com.vitorpamplona.nestsclient.transport.WebTransportSession
@@ -106,9 +107,11 @@ suspend fun connectNestsListener(
state.value = NestsListenerState.Connecting(NestsListenerState.Connecting.ConnectStep.MoqHandshake)
// moq-lite Lite-03 has NO setup message — the WebTransport handshake
// itself is the handshake, version is selected by the ALPN
// `moq-lite-03`.
// moq-lite has NO setup message — the WebTransport handshake itself
// is the handshake. Version (Lite-03 vs Lite-04) is selected by the
// ALPN exchange (`wt-available-protocols` / `wt-protocol`) and
// surfaced via [MoqLiteSession.version]; the session itself derives
// it from `transport.negotiatedSubProtocol`.
val moq =
try {
MoqLiteSession.client(webTransport, scope)
@@ -118,7 +121,7 @@ suspend fun connectNestsListener(
return failedListener(state)
}
state.value = NestsListenerState.Connected(room, MOQ_LITE_03_VERSION)
state.value = NestsListenerState.Connected(room, versionCode(moq.version))
return MoqLiteNestsListener(
session = moq,
mutableState = state,
@@ -134,6 +137,20 @@ suspend fun connectNestsListener(
*/
const val MOQ_LITE_03_VERSION: Long = 0x6D71_6C03L
/**
* Synthetic version code for moq-lite Lite-04 sessions — same shape
* as [MOQ_LITE_03_VERSION] with the ALPN suffix `-04` in the low
* bytes.
*/
const val MOQ_LITE_04_VERSION: Long = 0x6D71_6C04L
/** Map a [MoqLiteVersion] to its synthetic [MOQ_LITE_03_VERSION]-family code. */
internal fun versionCode(version: MoqLiteVersion): Long =
when (version) {
MoqLiteVersion.LITE_03 -> MOQ_LITE_03_VERSION
MoqLiteVersion.LITE_04 -> MOQ_LITE_04_VERSION
}
/**
* Build a no-op [NestsListener] in a Failed state for callers that want a
* uniform return type even on early-failure paths.
@@ -239,8 +256,9 @@ suspend fun connectNestsSpeaker(
state.value = NestsSpeakerState.Connecting(NestsSpeakerState.Connecting.ConnectStep.MoqHandshake)
// moq-lite Lite-03 has NO setup message. Same logic as the listener
// path — version is selected by the `moq-lite-03` ALPN.
// moq-lite has NO setup message. Same logic as the listener path —
// version is selected by the ALPN exchange and derived inside
// [MoqLiteSession.version].
val moq =
try {
MoqLiteSession.client(webTransport, scope)
@@ -250,7 +268,7 @@ suspend fun connectNestsSpeaker(
return failedSpeaker(state)
}
state.value = NestsSpeakerState.Connected(room, MOQ_LITE_03_VERSION)
state.value = NestsSpeakerState.Connected(room, versionCode(moq.version))
return MoqLiteNestsSpeaker(
session = moq,
speakerPubkeyHex = speakerPubkeyHex,
@@ -47,12 +47,65 @@ object MoqLiteAlpn {
const val LITE_03: String = "moq-lite-03"
/**
* `moq-lite-04` ALPN string. Wire-incompatible with [MoqLiteCodec]
* today — see the object kdoc for the codec diff. Defined here so
* a future patch that lands version-aware Announce / Probe codecs
* can drop it into the [QuicWebTransportFactory] sub-protocol list
* without re-deriving the constant.
* `moq-lite-04` ALPN string. Wire-compatible with [MoqLiteCodec]
* when the codec is invoked with [MoqLiteVersion.LITE_04] — see
* [MoqLiteVersion] for the version-conditional codec branches.
* Lite-04 reshapes `Announce.hops` (varint count → `OriginList`),
* adds `AnnounceInterest.exclude_hop`, and adds `Probe.rtt`.
* Subscribe / SubscribeOk / Drop / Group / SubscribeResponse are
* unchanged from Lite-03. The factory advertises Lite-04 ahead
* of Lite-03 to match kixelated's preference order.
*/
const val LITE_04: String = "moq-lite-04"
const val LEGACY: String = "moql"
}
/**
* Version discriminator for the [MoqLiteCodec] / [MoqLiteSession]
* version-aware code paths. Selected at WebTransport CONNECT time
* via the `wt-available-protocols` / `wt-protocol` ALPN exchange
* (see `QuicWebTransportFactory` and
* [com.vitorpamplona.nestsclient.transport.WebTransportSession.negotiatedSubProtocol]).
*
* The wire delta between [LITE_03] and [LITE_04] is exactly three
* fields:
* - `Announce.hops`: Lite-03 = single varint count; Lite-04 =
* `varint(count) + count × varint(originId)` (the `OriginList`).
* - `AnnounceInterest.exclude_hop` (= our `AnnouncePlease.excludeHop`):
* Lite-03 = absent; Lite-04 = a single varint after `prefix`
* (sentinel `0` = no exclusion).
* - `Probe.rtt`: Lite-03 = absent; Lite-04 = a single varint after
* `bitrate` (sentinel `0` = unknown; outgoing `Some(0)` is
* clamped to `Some(1)` to avoid colliding with the sentinel).
*
* Everything else (`Subscribe`, `SubscribeOk`, `SubscribeDrop`,
* `SubscribeResponse`, `GroupHeader`, `ControlType`, `DataType`,
* `AnnounceStatus`) is byte-for-byte identical between the two
* versions.
*
* Source: `kixelated/moq` `rs/moq-lite/src/lite/{announce,probe,
* subscribe}.rs` Lite-04 branches (verified against `main` @
* 2026-05-09).
*/
enum class MoqLiteVersion(
val alpn: String,
) {
LITE_03(MoqLiteAlpn.LITE_03),
LITE_04(MoqLiteAlpn.LITE_04),
;
companion object {
/**
* Resolve a negotiated `wt-protocol` value back to the
* version enum, or `null` if the string isn't a recognised
* moq-lite version. Caller decides whether to fall back to
* a default ([LITE_03]) or fail loudly.
*/
fun fromAlpn(alpn: String?): MoqLiteVersion? =
when (alpn) {
MoqLiteAlpn.LITE_03 -> LITE_03
MoqLiteAlpn.LITE_04 -> LITE_04
else -> null
}
}
}
@@ -49,37 +49,86 @@ import com.vitorpamplona.nestsclient.moq.MoqWriter
object MoqLiteCodec {
// ---------------- AnnouncePlease ----------------
fun encodeAnnouncePlease(msg: MoqLiteAnnouncePlease): ByteArray {
fun encodeAnnouncePlease(
msg: MoqLiteAnnouncePlease,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): ByteArray {
val body = MoqWriter()
body.writeLengthPrefixedString(MoqLitePath.normalize(msg.prefix))
// Lite-04 appends `excludeHop` as a single varint after
// prefix. Sentinel `0` = no exclusion. Lite-03 omits the
// field entirely; ignore [msg.excludeHop] in that case.
if (version == MoqLiteVersion.LITE_04) {
body.writeVarint(msg.excludeHop)
}
return wrapSizePrefixed(body)
}
fun decodeAnnouncePlease(payload: ByteArray): MoqLiteAnnouncePlease {
fun decodeAnnouncePlease(
payload: ByteArray,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): MoqLiteAnnouncePlease {
val r = MoqReader(payload)
val prefix = MoqLitePath.normalize(r.readLengthPrefixedString())
val excludeHop =
if (version == MoqLiteVersion.LITE_04) r.readVarint() else 0L
ensureFullyConsumed(r, "AnnouncePlease")
return MoqLiteAnnouncePlease(prefix = prefix)
return MoqLiteAnnouncePlease(prefix = prefix, excludeHop = excludeHop)
}
// ---------------- Announce ----------------
fun encodeAnnounce(msg: MoqLiteAnnounce): ByteArray {
fun encodeAnnounce(
msg: MoqLiteAnnounce,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): ByteArray {
val body = MoqWriter()
body.writeByte(msg.status.code)
body.writeLengthPrefixedString(MoqLitePath.normalize(msg.suffix))
body.writeVarint(msg.hops)
// Lite-03: just the count. Lite-04: count + each origin id.
// Mirrors kixelated's `encode_hops` switch.
body.writeVarint(msg.hops.size.toLong())
if (version == MoqLiteVersion.LITE_04) {
for (originId in msg.hops) {
body.writeVarint(originId)
}
}
return wrapSizePrefixed(body)
}
fun decodeAnnounce(payload: ByteArray): MoqLiteAnnounce {
fun decodeAnnounce(
payload: ByteArray,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): MoqLiteAnnounce {
val r = MoqReader(payload)
val statusByte = r.readByte()
val status =
MoqLiteAnnounceStatus.fromCode(statusByte)
?: throw MoqCodecException("unknown moq-lite Announce status byte: $statusByte")
val suffix = MoqLitePath.normalize(r.readLengthPrefixedString())
val hops = r.readVarint()
val count = r.readVarint()
if (count < 0L || count > MoqLiteAnnounce.MAX_HOPS) {
throw MoqCodecException("Announce hops count out of range [0, ${MoqLiteAnnounce.MAX_HOPS}]: $count")
}
val hops: List<Long> =
when (version) {
MoqLiteVersion.LITE_03 -> {
// Wire only carries a count; fill with UNKNOWN
// placeholders (origin id 0). Mirrors kixelated's
// `Origin::UNKNOWN` fill in announce.rs.
List(count.toInt()) { 0L }
}
MoqLiteVersion.LITE_04 -> {
val list = ArrayList<Long>(count.toInt())
repeat(count.toInt()) {
val id = r.readVarint()
require(id >= 0) { "decoded origin id must be non-negative" }
list.add(id)
}
list
}
}
ensureFullyConsumed(r, "Announce")
return MoqLiteAnnounce(status = status, suffix = suffix, hops = hops)
}
@@ -244,24 +293,52 @@ object MoqLiteCodec {
// ---------------- Probe ----------------
fun decodeProbe(payload: ByteArray): MoqLiteProbe {
fun decodeProbe(
payload: ByteArray,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): MoqLiteProbe {
val r = MoqReader(payload)
val bitrate = r.readVarint()
// Lite-04 appends `rtt` as a single varint after bitrate.
// Sentinel `0` = unknown (decoded as null).
val rtt: Long? =
if (version == MoqLiteVersion.LITE_04) {
val raw = r.readVarint()
if (raw == 0L) null else raw
} else {
null
}
ensureFullyConsumed(r, "Probe")
return MoqLiteProbe(bitrate = bitrate)
return MoqLiteProbe(bitrate = bitrate, rtt = rtt)
}
/**
* Encode a single Lite-03 Probe message body
* (`lite/probe.rs` — `bitrate: u62` only; `rtt` is Lite-04+).
* The publisher writes these size-prefixed onto a Probe bidi the
* Encode a Probe message body. Lite-03 writes only `bitrate`;
* Lite-04 appends `rtt` (sentinel `0` = unknown; outgoing
* `Some(0)` is clamped to `Some(1)` to avoid colliding with the
* sentinel — matches kixelated's `encode_msg` clamp). The
* publisher writes these size-prefixed onto a Probe bidi the
* subscriber opened, advertising the publisher's expected
* bandwidth. Wrapping (size prefix) is the caller's responsibility,
* matching [encodeAnnouncePlease] / [encodeAnnounce].
* bandwidth. Wrapping (size prefix) is the caller's
* responsibility, matching [encodeAnnouncePlease] / [encodeAnnounce].
*/
fun encodeProbe(probe: MoqLiteProbe): ByteArray {
fun encodeProbe(
probe: MoqLiteProbe,
version: MoqLiteVersion = MoqLiteVersion.LITE_03,
): ByteArray {
val body = MoqWriter()
body.writeVarint(probe.bitrate)
if (version == MoqLiteVersion.LITE_04) {
// null → 0 (unknown sentinel). Some(0) → 1 (clamp).
// Otherwise pass through.
val wire =
when (val r = probe.rtt) {
null -> 0L
0L -> 1L
else -> r
}
body.writeVarint(wire)
}
return wrapSizePrefixed(body)
}
@@ -54,6 +54,28 @@ class MoqLiteAnnouncesHandle internal constructor(
suspend fun close() = close.invoke()
}
/**
* Active probe handle returned by [MoqLiteSession.probe]. Mirrors
* kixelated's `Subscriber::run_probe_stream`
* (`rs/moq-lite/src/lite/subscriber.rs`): subscriber opens a bidi,
* writes `ControlType::Probe`, then reads size-prefixed
* [MoqLiteProbe] messages indefinitely until the publisher FINs or
* the consumer calls [close].
*
* [updates] is a cold flow that emits every Probe message the
* publisher pushes — typically a single bitrate hint at session
* start, but a publisher running an ABR codec MAY emit multiple
* over time. For a fixed-rate Opus producer (which is what
* Amethyst's nests speaker is) the publisher emits one and FINs;
* the flow then completes naturally and consumers see end-of-flow.
*/
class MoqLiteProbeHandle internal constructor(
val updates: Flow<MoqLiteProbe>,
private val close: suspend () -> Unit,
) {
suspend fun close() = close.invoke()
}
/** Thrown when subscribe is rejected (Drop) or the response stream dies. */
class MoqLiteSubscribeException(
message: String,
@@ -30,16 +30,32 @@ package com.vitorpamplona.nestsclient.moq.lite
/**
* "I'm interested in broadcasts under this prefix" — the first message
* the subscriber writes on an Announce bidi.
* the subscriber writes on an Announce bidi. Mirrors kixelated's
* `AnnounceInterest` struct (`rs/moq-lite/src/lite/announce.rs`).
*
* Wire layout (size-prefixed):
* prefix string (varint length + UTF-8)
* prefix string (varint length + UTF-8)
* excludeHop u62 varint (Lite-04 ONLY; sentinel `0` = no exclusion)
*
* Empty prefix means "everything".
* Empty prefix means "everything". `excludeHop != 0` (Lite-04 only)
* asks the publisher to skip announces whose hop ID list contains
* this value — used in clustered moq-rs deployments to break
* forwarding loops.
*/
data class MoqLiteAnnouncePlease(
val prefix: String,
)
/**
* Lite-04 only. `0` (the default) means "no exclusion." A
* non-zero value asks the publisher to skip announces whose
* `hops` list contains this origin ID. Encoded as a single
* varint after [prefix]; absent on the wire under Lite-03.
*/
val excludeHop: Long = 0L,
) {
init {
require(excludeHop >= 0) { "excludeHop must be non-negative: $excludeHop" }
}
}
/**
* Per-broadcast announce update streamed by the publisher (or relay)
@@ -52,13 +68,37 @@ data class MoqLiteAnnouncePlease(
* suffix string (broadcast path with the requested `prefix`
* stripped — `MoqLitePath.join(prefix, suffix)`
* reconstitutes the absolute path)
* hops u62 (relay routing depth, Lite-03 only)
* hops OriginList (Lite-03: a single varint count;
* Lite-04: `varint(count) + count × varint(id)`)
*
* [hops] is modeled as `List<Long>` — origin IDs the announce has
* traversed. Empty list = no hops. Origin id `0` is reserved as
* `Origin::UNKNOWN` (used by Lite-03 decode to fill the count of
* placeholder entries when the wire only carries a count). Per
* kixelated, [hops] is bounded to 32 entries (`MAX_HOPS`).
*/
data class MoqLiteAnnounce(
val status: MoqLiteAnnounceStatus,
val suffix: String,
val hops: Long,
)
val hops: List<Long>,
) {
init {
require(hops.size <= MAX_HOPS) { "hops list must not exceed MAX_HOPS=$MAX_HOPS, got ${hops.size}" }
// Origin id is a u62 varint; reject negatives so encoding
// can't silently produce a malformed varint. Per kixelated,
// id == 0 means UNKNOWN; otherwise must fit in 62 bits.
require(hops.all { it >= 0 }) { "hops origin ids must be non-negative" }
}
companion object {
/**
* Maximum number of origin entries in [hops]. Mirrors
* `kixelated/moq`'s `Origin::MAX_HOPS = 32`; the decoder
* rejects oversize lists with a codec exception.
*/
const val MAX_HOPS: Int = 32
}
}
/**
* "Subscribe me to (broadcast, track)" — the first message the
@@ -156,6 +196,44 @@ object MoqLiteSubscribeDropCode {
* `com.vitorpamplona.nestsclient.moq.ErrorCode.TRACK_DOES_NOT_EXIST`.
*/
const val TRACK_DOES_NOT_EXIST: Long = 0x04L
/**
* The publisher does not serve this broadcast at all. Sent for a
* subscribe whose `broadcast` field doesn't match the suffix we
* published under on this session. Distinct from
* [TRACK_DOES_NOT_EXIST] (which means "we publish this broadcast
* but not under that track name") so the watcher can tell apart
* "wrong room" from "wrong rendition". Mirrors the IETF
* `com.vitorpamplona.nestsclient.moq.ErrorCode.TRACK_NAMESPACE_DOES_NOT_EXIST`
* conceptually — moq-lite's flatter path model collapses
* namespace/track into broadcast/track, but the same "not found
* at this level" semantic applies.
*/
const val BROADCAST_DOES_NOT_EXIST: Long = 0x05L
}
/**
* Application error codes a listener passes to
* [com.vitorpamplona.nestsclient.transport.WebTransportReadStream.stopSending]
* when canceling a group's uni stream. moq-lite leaves these
* application-defined; we follow the same convention as
* [MoqLiteSubscribeDropCode] (small non-zero varints) so a future
* cross-protocol mapping is straightforward.
*
* Used by [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession.drainOneGroup]
* when a group arrives for a subscription that's already been
* removed: rather than silently dropping every frame the publisher
* pushes (and letting the publisher waste bandwidth on
* retransmits), we `stopSending` the uni stream so the publisher
* abandons it.
*/
object MoqLiteStreamCancelCode {
/**
* The receiving subscription has been canceled / unsubscribed
* since this group started. The publisher should abandon any
* pending retransmits.
*/
const val SUBSCRIPTION_GONE: Long = 0x10L
}
/**
@@ -172,12 +250,28 @@ data class MoqLiteGroupHeader(
/**
* Probe message written by the *publisher* on a subscriber-initiated
* Probe bidi (ControlType=4). `bitrate` is encoded as a u62 varint in
* a size-prefixed body. Decode-only for now — the listener path
* exposes the most recent reading, and we don't initiate probes from
* the speaker side.
* a size-prefixed body.
*
* Wire layout (size-prefixed):
* bitrate u62 varint
* rtt u62 varint (Lite-04 ONLY; sentinel `0` = unknown.
* Outgoing `Some(0)` is clamped to `Some(1)`
* to avoid colliding with the sentinel.)
*
* Source: `rs/moq-lite/src/lite/probe.rs`.
*/
data class MoqLiteProbe(
val bitrate: Long,
)
/**
* Round-trip time hint, in the unit the application and peer
* have agreed to (kixelated's Rust type is plain `Option<u64>`
* — no unit attached at the type level). Lite-04 only; absent
* on the wire under Lite-03. `null` = unknown.
*/
val rtt: Long? = null,
) {
init {
require(bitrate >= 0) { "bitrate must be non-negative: $bitrate" }
if (rtt != null) require(rtt >= 0) { "rtt must be non-negative when present: $rtt" }
}
}
@@ -78,6 +78,23 @@ class MoqLiteSession internal constructor(
internal val transport: WebTransportSession,
private val scope: CoroutineScope,
) {
/**
* Wire-format version this session speaks. Derived from the
* WebTransport ALPN the server selected via
* `wt-available-protocols` / `wt-protocol`
* ([WebTransportSession.negotiatedSubProtocol]); falls back to
* [MoqLiteVersion.LITE_03] when the server didn't echo
* `wt-protocol` or echoed something we don't recognise (older
* single-protocol deployments / forward-compat).
*
* Selects between the version-conditional codec branches in
* [MoqLiteCodec] for `Announce.hops`, `AnnouncePlease.excludeHop`,
* and `Probe.rtt`. All other messages are wire-identical between
* the two versions. Tests drive this via
* `FakeWebTransport.pair(negotiatedSubProtocol = …)`.
*/
val version: MoqLiteVersion =
MoqLiteVersion.fromAlpn(transport.negotiatedSubProtocol) ?: MoqLiteVersion.LITE_03
private val state = Mutex()
private val subscriptionsBySubscribeId: MutableMap<Long, ListenerSubscription> = HashMap()
private var nextSubscribeId: Long = 0L
@@ -132,7 +149,7 @@ class MoqLiteSession internal constructor(
ensureOpen()
val bidi = transport.openBidiStream()
bidi.write(Varint.encode(MoqLiteControlType.Announce.code))
bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix)))
bidi.write(MoqLiteCodec.encodeAnnouncePlease(MoqLiteAnnouncePlease(prefix), version))
// replay=64, DROP_OLDEST: announces emitted before the
// caller's `collect` attaches MUST NOT be dropped — that's
@@ -164,7 +181,7 @@ class MoqLiteSession internal constructor(
buffer.push(chunk)
while (true) {
val payload = buffer.readSizePrefixed() ?: break
val decoded = MoqLiteCodec.decodeAnnounce(payload)
val decoded = MoqLiteCodec.decodeAnnounce(payload, version)
emitCount += 1
Log.d("NestRx") {
"session.announce(prefix='$prefix') bidi pump emit #$emitCount " +
@@ -604,6 +621,15 @@ class MoqLiteSession internal constructor(
var frameCount = 0
var droppedNoSub = 0
var trySendFailures = 0
// Audit M2: once we observe the subscription is gone, fire
// STOP_SENDING(SUBSCRIPTION_GONE) once on the uni stream so
// the publisher abandons any in-flight retransmits. Pre-fix
// we silently dropped every frame; the publisher kept pushing
// and the relay kept buffering bytes the listener would never
// read. Latched so concurrent frames don't fire it N times
// (RFC 9000 §3.5: subsequent STOP_SENDING calls are ignored
// by the peer, but we save the syscall).
var stopSendingFired = false
try {
stream.incoming().collect { chunk ->
buffer.push(chunk)
@@ -630,6 +656,17 @@ class MoqLiteSession internal constructor(
val sub = state.withLock { subscriptionsBySubscribeId[subscribeId] }
if (sub == null) {
droppedNoSub += 1
if (!stopSendingFired) {
stopSendingFired = true
Log.w("NestRx") {
"drainOneGroup#$streamSeq subId=$subscribeId no longer subscribed — stopSending(SUBSCRIPTION_GONE)"
}
NestsTrace.emit("group_stop_sending") {
"\"stream_seq\":$streamSeq,\"sub_id\":$subscribeId,\"group_seq\":$groupSequence," +
"\"code\":${MoqLiteStreamCancelCode.SUBSCRIPTION_GONE}"
}
runCatching { stream.stopSending(MoqLiteStreamCancelCode.SUBSCRIPTION_GONE) }
}
} else {
val sent =
sub.frames.trySend(
@@ -641,16 +678,13 @@ class MoqLiteSession internal constructor(
if (!sent.isSuccess) trySendFailures += 1
}
frameCount += 1
// If the subscription has been closed already we
// silently drop the frame — the publisher hasn't
// observed the unsubscribe yet (its uni streams
// are independent of our bidi FIN).
}
}
Log.d("NestRx") { "drainOneGroup#$streamSeq FIN subId=$subscribeId groupSeq=$groupSequence frames=$frameCount droppedNoSub=$droppedNoSub trySendFail=$trySendFailures" }
Log.d("NestRx") { "drainOneGroup#$streamSeq FIN subId=$subscribeId groupSeq=$groupSequence frames=$frameCount droppedNoSub=$droppedNoSub trySendFail=$trySendFailures stopSent=$stopSendingFired" }
NestsTrace.emit("group_fin") {
"\"stream_seq\":$streamSeq,\"sub_id\":$subscribeId,\"group_seq\":$groupSequence," +
"\"frames\":$frameCount,\"dropped_no_sub\":$droppedNoSub,\"try_send_fail\":$trySendFailures"
"\"frames\":$frameCount,\"dropped_no_sub\":$droppedNoSub,\"try_send_fail\":$trySendFailures," +
"\"stop_sent\":$stopSendingFired"
}
} catch (ce: CancellationException) {
throw ce
@@ -674,6 +708,67 @@ class MoqLiteSession internal constructor(
sub.frames.close()
}
/**
* Open a Probe bidi against the relay / publisher (audit L3).
* Mirrors kixelated's `Subscriber::run_probe_stream`
* (`rs/moq-lite/src/lite/subscriber.rs`): subscriber writes
* `ControlType::Probe` (varint 4), then reads size-prefixed
* [MoqLiteProbe] messages until the publisher FINs.
*
* The publisher reciprocates by writing one or more [MoqLiteProbe]
* messages indicating its expected outbound bitrate. Amethyst's
* own publisher path is fixed-rate Opus and emits exactly one
* Probe before FIN, but third-party (e.g. browser-side
* `kixelated/moq` watcher → relay-served broadcast) publishers
* MAY emit multiple over time as their ABR estimates change.
*
* The returned handle's `updates` flow is cold — collecting it
* starts the read pump on this session's scope. On collector
* cancellation OR [MoqLiteProbeHandle.close], the bidi FINs and
* the pump exits.
*
* No application-level use of probe data exists in nests today
* (we don't run ABR), but the API completes the moq-lite Lite-03
* protocol surface. Useful for diagnostic tools that want to
* read a publisher's bitrate without subscribing to its data
* track.
*/
suspend fun probe(): MoqLiteProbeHandle {
ensureOpen()
val bidi = transport.openBidiStream()
bidi.write(Varint.encode(MoqLiteControlType.Probe.code))
val updates =
MutableSharedFlow<MoqLiteProbe>(
replay = 8,
onBufferOverflow = BufferOverflow.DROP_OLDEST,
)
val pump =
scope.launch {
val buffer = MoqLiteFrameBuffer()
try {
bidi.incoming().collect { chunk ->
buffer.push(chunk)
while (true) {
val payload = buffer.readSizePrefixed() ?: break
updates.emit(MoqLiteCodec.decodeProbe(payload, version))
}
}
} catch (ce: CancellationException) {
throw ce
} catch (_: Throwable) {
// Probe bidi died — best-effort silent. The
// updates flow simply stops emitting.
}
}
return MoqLiteProbeHandle(
updates = updates,
close = {
runCatching { bidi.finish() }
pump.cancelAndJoin()
},
)
}
// ====================================================================
// Publisher side
// ====================================================================
@@ -717,9 +812,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 +847,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
@@ -772,23 +883,32 @@ class MoqLiteSession internal constructor(
}
}
private suspend fun handleInboundBidi(bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream) {
// Snapshot of publishers at bidi-arrival time. All publishers
// on a single session share a suffix (enforced in [publish]),
// so the Announce response is the same regardless of which
// publisher we route the bidi to. Subscribe responses pick the
// publisher whose `track` matches `sub.track`.
val publishersSnapshot = state.withLock { activePublishers.toList() }
if (publishersSnapshot.isEmpty()) return
// Designated publisher for Announce-bidi ownership: the first
// one. The Active/Ended pair is keyed off the broadcast
// SUFFIX (not per track), so emitting it once via one
// publisher matches the single-broadcast model the relay
// expects. All publishers close together in [close], so
// there's no risk of one closing while the announce bidi
// stays alive on another.
val announcePublisher = publishersSnapshot.first()
/**
* Reject an inbound Subscribe with a typed error code. Writes the
* `SubscribeDrop` body so a watcher that decodes the application
* message sees the typed reason, then `RESET_STREAM(errorCode)`s
* the bidi so a watcher inspecting only the QUIC layer also sees
* the same code (audit M3 — Lite-03 conveys errors via
* RESET_STREAM, distinguishing "publisher rejected" from
* "publisher gracefully shut down" / FIN). Errors during the
* write/reset are swallowed: the bidi is already on its way out.
*/
private suspend fun rejectSubscribe(
bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream,
errorCode: Long,
reason: String,
) {
runCatching {
bidi.write(
MoqLiteCodec.encodeSubscribeDrop(
MoqLiteSubscribeDrop(errorCode = errorCode, reasonPhrase = reason),
),
)
bidi.reset(errorCode)
}
}
private suspend fun handleInboundBidi(bidi: com.vitorpamplona.nestsclient.transport.WebTransportBidiStream) {
// Single long-running collector for the bidi's full lifetime.
// Pre-fix this dispatch was split into a `firstOrNull()` to
// peek the control byte + a `readSizePrefixedFromBidiInto`
@@ -834,19 +954,73 @@ class MoqLiteSession internal constructor(
runCatching { bidi.finish() }
return@collect
}
// Read the publisher list FRESH at dispatch time
// rather than at bidi-arrival time. Closes a window
// where a publisher registered after the inbound
// bidi opened but before its first chunk landed
// wouldn't be visible to the dispatcher — pre-fix
// we snapshotted the list at the top of
// [handleInboundBidi], so the second-track publisher
// (e.g. catalog after audio in [MoqLiteNestsSpeaker])
// raced the relay's SUBSCRIBE bidi for that track
// when registration happened in the ~few-ms gap.
// Production never bit because both nests publishers
// register before any subscriber arrives, but the
// narrower contract is safer.
//
// All publishers on a single session share a suffix
// (enforced in [publish]), so the Announce response
// is the same regardless of which publisher we route
// the bidi to. Subscribe responses pick the publisher
// whose `track` matches `sub.track`.
val publishersSnapshot = state.withLock { activePublishers.toList() }
if (publishersSnapshot.isEmpty()) {
runCatching { bidi.finish() }
dispatched = true
return@collect
}
// Designated publisher for Announce-bidi ownership:
// the first one. The Active/Ended pair is keyed off
// the broadcast SUFFIX (not per track), so emitting
// it once via one publisher matches the single-
// broadcast model the relay expects. All publishers
// close together in [close], so there's no risk of
// one closing while the announce bidi stays alive
// on another.
val announcePublisher = publishersSnapshot.first()
when (controlType) {
MoqLiteControlType.Announce -> {
val pleasePayload = buffer.readSizePrefixed() ?: return@collect
val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload)
val emittedSuffix =
MoqLitePath.stripPrefix(please.prefix, announcePublisher.suffix) ?: announcePublisher.suffix
val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload, version)
// Per moq-lite Lite-03 (`rs/moq-lite/src/lite/announce.rs`),
// a publisher MUST only emit Active for broadcasts whose
// path starts with the requested prefix. If our suffix
// doesn't match, FIN cleanly without writing any
// Announce — the relay/peer sees an empty announce
// stream and moves on. Pre-fix we'd fall through the
// `null` branch of stripPrefix and falsely emit
// `Active(suffix=ourFullSuffix)`, advertising under a
// prefix the subscriber didn't ask for. Production
// never bit because the relay always asks for
// `prefix=""`, but a future peer-to-peer or
// namespace-scoped subscriber would see a ghost.
val emittedSuffix = MoqLitePath.stripPrefix(please.prefix, announcePublisher.suffix)
if (emittedSuffix == null) {
Log.w("NestTx") {
"ANNOUNCE inbound prefix='${please.prefix}' does not match publisher.suffix='${announcePublisher.suffix}' — FIN without Active"
}
runCatching { bidi.finish() }
dispatched = true
return@collect
}
bidi.write(
MoqLiteCodec.encodeAnnounce(
MoqLiteAnnounce(
status = MoqLiteAnnounceStatus.Active,
suffix = emittedSuffix,
hops = 0L,
hops = emptyList(),
),
version,
),
)
announcePublisher.registerAnnounceBidi(bidi, emittedSuffix)
@@ -861,38 +1035,52 @@ class MoqLiteSession internal constructor(
MoqLiteControlType.Subscribe -> {
val subPayload = buffer.readSizePrefixed() ?: return@collect
val sub = MoqLiteCodec.decodeSubscribe(subPayload)
// Validate the requested broadcast matches our
// session's publisher suffix BEFORE matching
// tracks — a peer subscribing under the wrong
// broadcast must not be able to siphon our audio
// by guessing a track name. Both sides have
// already path-normalised at the codec wire
// boundary (`MoqLiteCodec.decodeSubscribe` calls
// `MoqLitePath.normalize`), so a direct equality
// is sufficient. Reject with
// BROADCAST_DOES_NOT_EXIST so the peer sees a
// typed error code rather than a silent FIN.
// Publishers in `publishersSnapshot` all share
// the same suffix (enforced by
// [MoqLiteSession.publish]), so checking the
// first is equivalent to checking all.
val ourSuffix = announcePublisher.suffix
if (sub.broadcast != ourSuffix) {
Log.w("NestTx") {
"SUBSCRIBE inbound id=${sub.id} broadcast='${sub.broadcast}' does not match " +
"publisher.suffix='$ourSuffix' — replying SubscribeDrop+RESET"
}
rejectSubscribe(
bidi = bidi,
errorCode = MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST,
reason =
"broadcast '${sub.broadcast}' is not published on this session " +
"(we publish '$ourSuffix')",
)
dispatched = true
return@collect
}
// Find the publisher that claims this track. With the
// single-track-per-publisher model, only one match is possible.
val targetPublisher = publishersSnapshot.firstOrNull { it.track == sub.track }
if (targetPublisher == null) {
// Reply SubscribeDrop with a TRACK_DOES_NOT_EXIST
// error code BEFORE we FIN — without this the
// peer's response wait resolves only on
// bidi-FIN with no indication WHY (looks
// identical to "publisher disappeared mid-
// subscribe"). Drop carries the error code +
// reason phrase the watcher can log /
// surface, and matches what kixelated's
// `rs/moq-lite/src/lite/subscribe.rs`
// expects for an unrecognised track on a
// live broadcast.
Log.w("NestTx") {
"SUBSCRIBE inbound id=${sub.id} track='${sub.track}' has no matching publisher " +
"on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop"
}
runCatching {
bidi.write(
MoqLiteCodec.encodeSubscribeDrop(
MoqLiteSubscribeDrop(
errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST,
reasonPhrase =
"track '${sub.track}' is not published on this broadcast " +
"(available: ${publishersSnapshot.joinToString(",") { it.track }})",
),
),
)
bidi.finish()
"on this session (have ${publishersSnapshot.map { it.track }}) — replying SubscribeDrop+RESET"
}
rejectSubscribe(
bidi = bidi,
errorCode = MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST,
reason =
"track '${sub.track}' is not published on this broadcast " +
"(available: ${publishersSnapshot.joinToString(",") { it.track }})",
)
dispatched = true
return@collect
}
@@ -908,17 +1096,33 @@ class MoqLiteSession internal constructor(
targetPublisher.registerInboundSubscription(sub)
inboundSub = sub
inboundSubPublisher = targetPublisher
// Audit L2: narrow `startGroup` to the
// publisher's [MoqLitePublisherHandle.nextSequence]
// — the exact sequence of the next group we'll
// open. Spec lets the publisher narrow the
// subscriber's bounds; pre-fix we always echoed
// `null/null`, which loses the diagnostic
// information of "which group are you about to
// start sending?" Useful for hot-swap
// continuations where `nextSequence` is non-zero
// (the seeded `startSequence` from the previous
// session). `endGroup` stays null because we're
// a live broadcast with no end in sight; the
// subscriber's request `endGroup` is honoured
// implicitly when the publisher closes.
val narrowedStart = targetPublisher.nextSequence
bidi.write(
MoqLiteCodec.encodeSubscribeOk(
MoqLiteSubscribeOk(
priority = sub.priority,
ordered = sub.ordered,
maxLatencyMillis = sub.maxLatencyMillis,
startGroup = null,
startGroup = narrowedStart,
endGroup = null,
),
),
)
Log.d("NestTx") { "SUBSCRIBE_OK id=${sub.id} narrowed startGroup=$narrowedStart" }
dispatched = true
}
@@ -937,7 +1141,7 @@ class MoqLiteSession internal constructor(
// which left the subscriber's ABR
// estimator with no signal.
runCatching {
bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS)))
bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = NESTS_AUDIO_BITRATE_HINT_BPS), version))
bidi.finish()
}
dispatched = true
@@ -1021,6 +1225,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
@@ -1037,13 +1242,42 @@ 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`:
// 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 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 seqLow = sequence.coerceAtMost(SEQ_MAX.toLong()).toInt() and SEQ_MASK
val packedPriority = ((trackPriority and 0xFF) shl SEQ_BITS) or seqLow
uni.setPriority(packedPriority)
uni.write(Varint.encode(MoqLiteDataType.Group.code))
uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId, sequence)))
return uni
@@ -1114,6 +1348,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>()
@@ -1289,8 +1530,9 @@ class MoqLiteSession internal constructor(
MoqLiteAnnounce(
status = MoqLiteAnnounceStatus.Ended,
suffix = entry.emittedSuffix,
hops = 0L,
hops = emptyList(),
),
version,
),
)
}
@@ -1317,14 +1559,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)
}
}
@@ -1343,6 +1585,30 @@ 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
/**
* Bit layout for the QUIC stream priority value packed by
* [openGroupStream]. Bit 31 is reserved as the sign bit (so
* the writer's signed `sortedByDescending` orders correctly);
* bits 30..23 carry [DEFAULT_TRACK_PRIORITY]-family
* `trackPriority u8`; bits 22..0 carry the low [SEQ_BITS] of
* the group sequence. See the kdoc on [openGroupStream] for
* the full rationale.
*/
private const val SEQ_BITS: Int = 23
private const val SEQ_MASK: Int = 0x007F_FFFF
private const val SEQ_MAX: Int = SEQ_MASK
/**
* Bitrate hint (bits/sec) we report on inbound moq-lite Probe
* bidis as a publisher. Mirrors the upper-bound of an Opus
@@ -47,6 +47,15 @@ class FakeWebTransport private constructor(
private val inboundBidiStreams: Channel<FakeBidiStream>,
private val inboundUniStreams: Channel<WebTransportReadStream>,
private val outboundUniStreams: Channel<WebTransportReadStream>,
/**
* Test-controllable negotiated sub-protocol. Defaults to `null`
* — production fakes that don't care about ALPN negotiation
* keep the same behavior they had before this field existed.
* Tests that exercise the version-aware [MoqLiteSession] can
* pass a non-null value via [pair] to simulate a relay that
* picked Lite-04 (or any other ALPN string).
*/
override val negotiatedSubProtocol: String? = null,
) : WebTransportSession {
private val stateLock = Mutex()
private var open = true
@@ -57,8 +66,40 @@ class FakeWebTransport private constructor(
stateLock.withLock { check(open) { "session closed" } }
val localToPeer = Channel<ByteArray>(Channel.BUFFERED)
val peerToLocal = Channel<ByteArray>(Channel.BUFFERED)
val local = FakeBidiStream(write = localToPeer, read = peerToLocal)
val peer = FakeBidiStream(write = peerToLocal, read = localToPeer)
// Shared error-code cells: each side records its own reset()
// and stopSending() codes here so the peer can introspect.
// `localResetCell` records resets on `local.reset(code)` (=
// peer's `lastPeerResetCode`), and vice-versa.
val localResetCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
val peerResetCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
val localStopSendingCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
val peerStopSendingCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
val local =
FakeBidiStream(
write = localToPeer,
read = peerToLocal,
myResetCell = localResetCell,
myStopSendingCell = localStopSendingCell,
peerResetCell = peerResetCell,
peerStopSendingCell = peerStopSendingCell,
)
val peer =
FakeBidiStream(
write = peerToLocal,
read = localToPeer,
myResetCell = peerResetCell,
myStopSendingCell = peerStopSendingCell,
peerResetCell = localResetCell,
peerStopSendingCell = localStopSendingCell,
)
// Send *peer* half to the other side's inbound queue.
outboundBidiStreams.send(peer)
return local
@@ -78,8 +119,29 @@ 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)
// Shared reset / stopSending cells. The writer (us) records its
// `reset(code)` call here; the peer-side reader exposes it via
// `FakeReadStream.lastResetCode`. Symmetric: the reader's
// `stopSending(code)` call lands in `stopSendingCell` so the
// writer (= moq-lite publisher tests) can assert the listener
// canceled with the expected code.
val resetCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
val stopSendingCell =
java.util.concurrent.atomic
.AtomicLong(NO_CODE)
outboundUniStreams.send(FakeReadStream(pipe, priorityCell, resetCell, stopSendingCell))
return ChannelWriteStream(pipe, priorityCell, resetCell, stopSendingCell)
}
override suspend fun sendDatagram(payload: ByteArray): Boolean {
@@ -117,8 +179,14 @@ class FakeWebTransport private constructor(
* Create two linked fakes that act as "client" and "server" endpoints of
* the same virtual WebTransport session. Datagrams and streams opened on
* one side appear on the other.
*
* @param negotiatedSubProtocol surfaced by both sides via
* [WebTransportSession.negotiatedSubProtocol]. Defaults to
* `null`. Tests that drive a version-aware
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession]
* pass e.g. `"moq-lite-04"` here.
*/
fun pair(): Pair<FakeWebTransport, FakeWebTransport> {
fun pair(negotiatedSubProtocol: String? = null): Pair<FakeWebTransport, FakeWebTransport> {
val aToBDatagrams = Channel<ByteArray>(Channel.BUFFERED)
val bToADatagrams = Channel<ByteArray>(Channel.BUFFERED)
val aToBBidi = Channel<FakeBidiStream>(Channel.BUFFERED)
@@ -134,6 +202,7 @@ class FakeWebTransport private constructor(
inboundBidiStreams = bToABidi,
inboundUniStreams = bToAUni,
outboundUniStreams = aToBUni,
negotiatedSubProtocol = negotiatedSubProtocol,
)
val b =
FakeWebTransport(
@@ -143,6 +212,7 @@ class FakeWebTransport private constructor(
inboundBidiStreams = aToBBidi,
inboundUniStreams = aToBUni,
outboundUniStreams = bToAUni,
negotiatedSubProtocol = negotiatedSubProtocol,
)
return a to b
}
@@ -152,6 +222,20 @@ class FakeWebTransport private constructor(
class FakeBidiStream internal constructor(
private val write: Channel<ByteArray>,
private val read: Channel<ByteArray>,
/**
* Cell that records this side's `reset(code)` calls. Read by
* [lastResetCode] from the same side (rare — usually it's the
* peer's [peerResetCell] you want).
*/
private val myResetCell: java.util.concurrent.atomic.AtomicLong? = null,
private val myStopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
/**
* Cell that records the *peer's* `reset(code)` calls (i.e.
* mirror of the peer's [myResetCell]). [lastPeerResetCode] reads
* this so tests can assert "the peer reset the bidi with code X."
*/
private val peerResetCell: java.util.concurrent.atomic.AtomicLong? = null,
private val peerStopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
) : WebTransportBidiStream {
override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
@@ -163,13 +247,98 @@ class FakeBidiStream internal constructor(
write.close()
}
override suspend fun reset(errorCode: Long) {
// First call wins, mirroring `:quic`'s `QuicStream.resetStream`
// contract. Subsequent reset/finish/write are silently
// dropped to keep the wire frame stable in the eyes of any
// test that reads `lastResetCode`.
myResetCell?.compareAndSet(NO_CODE, errorCode)
write.close()
}
override suspend fun stopSending(errorCode: Long) {
myStopSendingCell?.compareAndSet(NO_CODE, errorCode)
// The peer's send half is our receive half; close it so a
// collector sees end-of-flow.
read.close()
}
/**
* Code our side passed to [reset], or `null` if reset wasn't
* called. Mostly useful when a test holds the peer's reference;
* [lastPeerResetCode] is the more common assertion.
*/
val lastResetCode: Long?
get() = myResetCell?.get()?.takeIf { it != NO_CODE }
/** Code the peer side passed to [reset], or `null` if not called. */
val lastPeerResetCode: Long?
get() = peerResetCell?.get()?.takeIf { it != NO_CODE }
val lastStopSendingCode: Long?
get() = myStopSendingCell?.get()?.takeIf { it != NO_CODE }
val lastPeerStopSendingCode: Long?
get() = peerStopSendingCell?.get()?.takeIf { it != NO_CODE }
override fun setPriority(priority: Int) = Unit
}
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,
/**
* Optional shared cell that records the writer's `reset(code)`
* call. Exposed via [lastResetCode] so tests can assert the
* publisher reset its uni stream with the expected typed error.
*/
private val resetCell: java.util.concurrent.atomic.AtomicLong? = null,
/**
* Optional shared cell that records this side's own
* `stopSending(code)` call. Exposed via [lastStopSendingCode]
* so the test can assert the listener actually fired the cancel.
*/
private val stopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
) : WebTransportReadStream {
override fun incoming(): Flow<ByteArray> = read.receiveAsFlow()
override suspend fun stopSending(errorCode: Long) {
// First-call-wins like the production path. Closing `read`
// here would race the production path's "publisher closes
// its write end" semantic — moq-lite calls `stopSending`
// when it's done caring about the bytes, but the publisher
// is still expected to finish what it was sending. We
// record the code without closing so a test that asserts
// `incoming().toList()` post-stopSending still sees any
// bytes the publisher had already enqueued.
stopSendingCell?.compareAndSet(NO_CODE, errorCode)
}
/**
* 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()
/**
* Code the peer-side writer passed to [WebTransportWriteStream.reset],
* or `null` if reset wasn't called.
*/
val lastResetCode: Long?
get() = resetCell?.get()?.takeIf { it != NO_CODE }
/** Code we passed to [stopSending], or `null` if not called. */
val lastStopSendingCode: Long?
get() = stopSendingCell?.get()?.takeIf { it != NO_CODE }
}
/**
@@ -177,9 +346,27 @@ class FakeReadStream internal constructor(
* [FakeWebTransport.openUniStream] (production: locally-opened uni
* stream that the paired peer reads via incomingUniStreams) and any
* test that wants to drive uni-stream bytes through a known channel.
*
* Public so tests that hold the writer reference (e.g. via
* `serverSide.openUniStream()`) can introspect the peer-side
* `stopSending` code via [peerStopSendingCode] without racing the
* listener's uni-stream pump on the read side.
*/
private class ChannelWriteStream(
class ChannelWriteStream internal constructor(
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,
private val resetCell: java.util.concurrent.atomic.AtomicLong? = null,
/**
* Cell that records the *peer's* `stopSending(code)` call, since
* the writer is on the other side of the stream from the
* `stopSending` caller. Same cell that
* [FakeReadStream.lastStopSendingCode] reads.
*/
private val stopSendingCell: java.util.concurrent.atomic.AtomicLong? = null,
) : WebTransportWriteStream {
override suspend fun write(chunk: ByteArray) {
channel.send(chunk)
@@ -189,5 +376,35 @@ private class ChannelWriteStream(
channel.close()
}
override fun setPriority(priority: Int) = Unit
override suspend fun reset(errorCode: Long) {
// First-call-wins. We close the channel so the peer's
// receive flow ends — same shape as `finish` but with a
// typed error code recorded for the peer to introspect.
resetCell?.compareAndSet(NO_CODE, errorCode)
channel.close()
}
override fun setPriority(priority: Int) {
priorityCell?.set(priority)
}
/**
* Code the peer-side reader passed to
* [WebTransportReadStream.stopSending], or `null` if not called.
* Lets a test that holds the writer side inspect the listener's
* group-cancel without racing for the peer-side
* [WebTransportReadStream] reference.
*/
val peerStopSendingCode: Long?
get() = stopSendingCell?.get()?.takeIf { it != NO_CODE }
}
/**
* Sentinel for "no reset / stopSending code recorded yet" in the
* [java.util.concurrent.atomic.AtomicLong] cells used by [FakeBidiStream]
* and [FakeReadStream] / [ChannelWriteStream]. Real moq-lite codes are
* non-negative varints (RFC 9000 application error codes are u32
* unsigned-fitting-in-Long); `Long.MIN_VALUE` is safely outside that
* range.
*/
internal const val NO_CODE: Long = Long.MIN_VALUE
@@ -42,6 +42,21 @@ interface WebTransportSession {
/** True once the Extended CONNECT exchange has returned 2xx and before [close] is called. */
val isOpen: Boolean
/**
* The sub-protocol the server selected via the
* `wt-protocol` response header (RFC 8941 SF-string;
* draft-ietf-webtrans-http3 §3.3). `null` if the server didn't
* echo a sub-protocol — typically because the client offered
* none, or because the server is older than draft-13 and
* supports only a single protocol implicitly.
*
* Used by moq-lite negotiation: a client that advertised both
* `moq-lite-04` and `moq-lite-03` reads this to select the
* matching [com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion]
* for its [com.vitorpamplona.nestsclient.moq.lite.MoqLiteSession].
*/
val negotiatedSubProtocol: String?
/**
* Open a new bidirectional WebTransport stream. The returned [WebTransportBidiStream]
* is writable + readable and closes when either peer half-closes.
@@ -104,6 +119,30 @@ interface WebTransportBidiStream :
interface WebTransportReadStream {
/** Flow of chunks as they arrive. Completes when the peer closes its write side. */
fun incoming(): Flow<ByteArray>
/**
* RFC 9000 §3.5: ask the peer to stop sending on this stream.
* Causes a `STOP_SENDING(applicationErrorCode)` frame to land at
* the peer, which typically responds with `RESET_STREAM` carrying
* the same code. Call this when the application no longer needs
* the stream's bytes — it lets the peer abandon any pending
* retransmits instead of wasting bandwidth on data we'd discard.
*
* First call wins per `:quic`'s `QuicStream.stopSending`
* lock-free first-call-wins gate; subsequent calls (including
* with a different code) are silently ignored to keep the wire
* frame stable.
*
* Used by moq-lite Lite-03's group-cancel path: a listener that
* decides a specific group is too stale can `stopSending` its
* uni stream so the publisher abandons any in-flight retransmits
* instead of wasting bandwidth on bytes the listener will discard.
*
* Implementations that don't model receive-side cancellation
* (e.g. the in-memory fake when no test asserts on it) MAY
* treat this as a no-op.
*/
suspend fun stopSending(errorCode: Long)
}
/** Write-only WebTransport stream. */
@@ -113,6 +152,31 @@ interface WebTransportWriteStream {
/** Half-close the write side (FIN). No further writes after this call. */
suspend fun finish()
/**
* RFC 9000 §3.5: send `RESET_STREAM(applicationErrorCode)` on
* this stream's send half, abandoning any pending bytes. Distinct
* from [finish], which is a graceful FIN — `reset` carries a
* typed error code the peer can act on.
*
* First call wins per `:quic`'s `QuicStream.resetStream`
* lock-free first-call-wins gate; subsequent calls (and any
* subsequent [write] / [finish]) are silently ignored to keep
* the wire frame stable.
*
* Used by moq-lite Lite-03's typed cancel paths: a publisher
* that rejects a Subscribe (e.g. broadcast / track does not
* exist) follows the SubscribeDrop body with a
* `RESET_STREAM(errorCode)` so the subscriber sees a typed
* application reason rather than the ambiguous "publisher FINed
* the bidi" signal that overlaps with a graceful publisher
* shutdown.
*
* Implementations that don't model sender-driven reset (e.g.
* the in-memory fake when no test asserts on it) MAY treat this
* as a [finish].
*/
suspend fun reset(errorCode: Long)
/**
* Hint to the transport about this stream's drain priority relative
* to other streams on the same session. Higher value drains first
@@ -61,17 +61,20 @@ class MoqLiteCodecTest {
@Test
fun announce_active_round_trips() {
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Active, suffix = "speakerPubkey", hops = 0L)
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Active, suffix = "speakerPubkey", hops = emptyList())
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnounce(msg))
// Byte 0 is the status byte (literal 1), then a 13-byte string,
// then a varint hops.
// then a varint hops count.
assertEquals(1, payload[0].toInt())
assertEquals(msg, MoqLiteCodec.decodeAnnounce(payload))
}
@Test
fun announce_ended_with_relay_hops_round_trips() {
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Ended, suffix = "fff", hops = 7L)
// Lite-03 wire only carries a count; decoded list is filled
// with UNKNOWN placeholders (origin id 0). Round-trip thus
// requires the input also be a list of zeros.
val msg = MoqLiteAnnounce(status = MoqLiteAnnounceStatus.Ended, suffix = "fff", hops = List(7) { 0L })
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnounce(msg))
assertEquals(0, payload[0].toInt())
assertEquals(msg, MoqLiteCodec.decodeAnnounce(payload))
@@ -202,6 +205,172 @@ class MoqLiteCodecTest {
assertEquals(msg, MoqLiteCodec.decodeGroupHeader(payload))
}
// ---------------- Lite-04 wire diffs (audit L1) ----------------
@Test
fun announcePlease_lite04_round_trips_excludeHop() {
val msg = MoqLiteAnnouncePlease(prefix = "rooms/A", excludeHop = 7L)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeAnnouncePlease(msg, MoqLiteVersion.LITE_04),
)
// Wire after prefix: a single varint excludeHop.
val r = MoqReader(payload)
assertEquals("rooms/A", r.readLengthPrefixedString())
assertEquals(7L, r.readVarint(), "Lite-04 excludeHop is a single varint after prefix")
assertTrue(!r.hasMore())
// Round-trip via decoder.
assertEquals(
msg,
MoqLiteCodec.decodeAnnouncePlease(payload, MoqLiteVersion.LITE_04),
)
}
@Test
fun announcePlease_lite03_omits_excludeHop_on_wire() {
// Lite-03 encode MUST NOT write excludeHop, regardless of the
// value carried in the data class. Confirms a Lite-04-aware
// app accidentally talking Lite-03 doesn't leak the field
// and desync the peer.
val msg = MoqLiteAnnouncePlease(prefix = "rooms/A", excludeHop = 99L)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeAnnouncePlease(msg, MoqLiteVersion.LITE_03),
)
val r = MoqReader(payload)
assertEquals("rooms/A", r.readLengthPrefixedString())
assertTrue(!r.hasMore(), "Lite-03 wire MUST end after prefix — no excludeHop")
// Decoded msg drops back to excludeHop=0 (Lite-03 default).
assertEquals(
MoqLiteAnnouncePlease(prefix = "rooms/A", excludeHop = 0L),
MoqLiteCodec.decodeAnnouncePlease(payload, MoqLiteVersion.LITE_03),
)
}
@Test
fun announce_lite04_round_trips_full_origin_list() {
val msg =
MoqLiteAnnounce(
status = MoqLiteAnnounceStatus.Active,
suffix = "spk",
hops = listOf(1L, 5L, 12L),
)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeAnnounce(msg, MoqLiteVersion.LITE_04),
)
// Wire: status u8, suffix string, count varint, then count
// origin id varints.
val r = MoqReader(payload)
assertEquals(1, r.readByte()) // Active
assertEquals("spk", r.readLengthPrefixedString())
assertEquals(3L, r.readVarint(), "count")
assertEquals(1L, r.readVarint(), "origin[0]")
assertEquals(5L, r.readVarint(), "origin[1]")
assertEquals(12L, r.readVarint(), "origin[2]")
assertTrue(!r.hasMore())
assertEquals(msg, MoqLiteCodec.decodeAnnounce(payload, MoqLiteVersion.LITE_04))
}
@Test
fun announce_lite03_drops_origin_ids_keeps_count() {
// Lite-03 wire only carries the count; the IDs are discarded
// on encode and reconstructed as UNKNOWN (0) placeholders on
// decode. Confirms cross-version round-trip is lossy in this
// direction (matches kixelated's Lite-03 hop-count semantic).
val sent =
MoqLiteAnnounce(
status = MoqLiteAnnounceStatus.Active,
suffix = "spk",
hops = listOf(1L, 5L, 12L),
)
val payload = peelSizePrefix(MoqLiteCodec.encodeAnnounce(sent, MoqLiteVersion.LITE_03))
val decoded = MoqLiteCodec.decodeAnnounce(payload, MoqLiteVersion.LITE_03)
// Same count, IDs replaced with UNKNOWN.
assertEquals(3, decoded.hops.size)
assertEquals(listOf(0L, 0L, 0L), decoded.hops)
}
@Test
fun announce_decoder_rejects_oversize_hop_count() {
// MAX_HOPS = 32 per kixelated's Origin::MAX_HOPS. A wire
// payload with count = 33 must be rejected.
val out =
com.vitorpamplona.nestsclient.moq
.MoqWriter()
out.writeByte(1) // Active
out.writeLengthPrefixedString("spk")
out.writeVarint(33L)
// Lite-03 doesn't write IDs so the body ends here.
val payload = out.toByteArray()
assertFailsWith<MoqCodecException> {
MoqLiteCodec.decodeAnnounce(payload, MoqLiteVersion.LITE_03)
}
}
@Test
fun probe_lite04_round_trips_rtt() {
val msg = MoqLiteProbe(bitrate = 32_000L, rtt = 25L)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeProbe(msg, MoqLiteVersion.LITE_04),
)
val r = MoqReader(payload)
assertEquals(32_000L, r.readVarint())
assertEquals(25L, r.readVarint())
assertTrue(!r.hasMore())
assertEquals(msg, MoqLiteCodec.decodeProbe(payload, MoqLiteVersion.LITE_04))
}
@Test
fun probe_lite04_clamps_some_zero_to_one_to_avoid_unknown_sentinel() {
// Per kixelated `encode_msg`, an outgoing rtt of `Some(0)` is
// clamped to `Some(1)` because 0 is the unknown-sentinel.
// Mirroring that exactly avoids a peer round-trip from
// Some(0) to None.
val msg = MoqLiteProbe(bitrate = 100_000L, rtt = 0L)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeProbe(msg, MoqLiteVersion.LITE_04),
)
val r = MoqReader(payload)
r.readVarint() // bitrate
assertEquals(1L, r.readVarint(), "rtt=0 must be clamped to 1 on the wire")
}
@Test
fun probe_lite04_decodes_zero_rtt_as_null() {
// Symmetric: an incoming `0` MUST decode as null (None).
val out =
com.vitorpamplona.nestsclient.moq
.MoqWriter()
out.writeVarint(64_000L)
out.writeVarint(0L) // rtt = unknown
val payload = out.toByteArray()
val decoded = MoqLiteCodec.decodeProbe(payload, MoqLiteVersion.LITE_04)
assertEquals(64_000L, decoded.bitrate)
assertEquals(null, decoded.rtt, "rtt=0 decodes as null (unknown sentinel)")
}
@Test
fun probe_lite03_wire_omits_rtt() {
// Lite-03 encode MUST NOT write rtt regardless of the data
// class value.
val msg = MoqLiteProbe(bitrate = 32_000L, rtt = 99L)
val payload =
peelSizePrefix(
MoqLiteCodec.encodeProbe(msg, MoqLiteVersion.LITE_03),
)
val r = MoqReader(payload)
assertEquals(32_000L, r.readVarint())
assertTrue(!r.hasMore(), "Lite-03 probe must end after bitrate — no rtt")
// Decoded msg drops rtt to null (Lite-03 default).
assertEquals(
MoqLiteProbe(bitrate = 32_000L, rtt = null),
MoqLiteCodec.decodeProbe(payload, MoqLiteVersion.LITE_03),
)
}
@Test
fun decoders_reject_trailing_garbage() {
// Build a valid AnnouncePlease payload then append a stray byte.
@@ -20,7 +20,9 @@
*/
package com.vitorpamplona.nestsclient.moq.lite
import com.vitorpamplona.nestsclient.transport.ChannelWriteStream
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
@@ -167,7 +169,7 @@ class MoqLiteSessionTest {
MoqLiteAnnounce(
MoqLiteAnnounceStatus.Active,
"speakerOne",
hops = 1L,
hops = listOf(0L),
),
),
)
@@ -176,7 +178,7 @@ class MoqLiteSessionTest {
MoqLiteAnnounce(
MoqLiteAnnounceStatus.Active,
"speakerTwo",
hops = 1L,
hops = listOf(0L),
),
),
)
@@ -340,6 +342,113 @@ class MoqLiteSessionTest {
session.close()
}
@Test
fun lite04_announce_round_trips_full_origin_list_through_session() =
runBlocking {
// Lite-03 audit L1: a session running [MoqLiteVersion.LITE_04]
// routes its codec calls through the Lite-04 branches —
// verified end-to-end by encoding an Announce on the
// server side with two origin IDs and confirming the
// client-side announce-watch flow surfaces them
// unchanged. Pre-fix the codec was Lite-03-only and
// would have read the count and discarded the IDs.
// Drive the negotiated sub-protocol via the fake so the
// session derives `version = LITE_04` itself (matching the
// production path, where `MoqLiteSession.version` reads
// through `transport.negotiatedSubProtocol`).
val (clientSide, serverSide) =
FakeWebTransport.pair(negotiatedSubProtocol = MoqLiteAlpn.LITE_04)
val session = MoqLiteSession.client(clientSide, pumpScope)
assertEquals(MoqLiteVersion.LITE_04, session.version)
val peer =
async {
val bidi = serverSide.peerOpenedBidiStreams().first()
// Drain control byte + AnnouncePlease. Lite-04
// shape: prefix string then excludeHop varint.
val chunks = bidi.incoming().take(2).toList()
assertEquals(MoqLiteControlType.Announce.code, MoqLiteFrameBuffer().apply { push(chunks[0]) }.readVarint())
val pleasePayload =
MoqLiteFrameBuffer().apply { push(chunks[1]) }.readSizePrefixed()
?: error("AnnouncePlease body missing")
val please = MoqLiteCodec.decodeAnnouncePlease(pleasePayload, MoqLiteVersion.LITE_04)
assertEquals("rooms/A", please.prefix)
assertEquals(0L, please.excludeHop, "default excludeHop is 0 (no exclusion)")
// Push an Announce with full origin list (Lite-04
// shape).
bidi.write(
MoqLiteCodec.encodeAnnounce(
MoqLiteAnnounce(
status = MoqLiteAnnounceStatus.Active,
suffix = "speakerA",
hops = listOf(2L, 7L),
),
MoqLiteVersion.LITE_04,
),
)
}
val handle = session.announce("rooms/A")
val announce = withTimeout(2_000) { handle.updates.first() }
assertEquals(MoqLiteAnnounceStatus.Active, announce.status)
assertEquals("speakerA", announce.suffix)
assertEquals(listOf(2L, 7L), announce.hops, "Lite-04 session preserves origin id list")
peer.await()
handle.close()
session.close()
}
@Test
fun publisher_subscribeOk_narrows_startGroup_to_next_sequence() =
runBlocking {
// Lite-03 audit L2: when accepting a SUBSCRIBE, the publisher
// narrows the SubscribeOk's `startGroup` from the subscriber's
// request to its own [MoqLitePublisherHandle.nextSequence],
// communicating the exact sequence the next group will
// carry. Particularly useful for hot-swap continuations
// (publisher seeded with a non-zero startSequence).
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
val publisher =
session.publish(
broadcastSuffix = "speakerPubkey",
track = "audio/data",
startSequence = 100L,
)
val subBidi = serverSide.openBidiStream()
subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
subBidi.write(
MoqLiteCodec.encodeSubscribe(
MoqLiteSubscribe(
id = 5L,
broadcast = "speakerPubkey",
track = "audio/data",
priority = 0x80,
ordered = true,
maxLatencyMillis = 0L,
// Subscriber asks "from latest" — publisher is
// free to narrow.
startGroup = null,
endGroup = null,
),
),
)
val ackChunk = withTimeout(2_000) { subBidi.incoming().first() }
val resp = MoqLiteCodec.decodeSubscribeResponse(ackChunk)
val ok = (resp as MoqLiteCodec.SubscribeResponse.Ok).ok
assertEquals(100L, ok.startGroup, "publisher narrows startGroup to its nextSequence")
// endGroup stays null — live broadcast has no end in sight.
assertEquals(null, ok.endGroup)
publisher.close()
session.close()
}
@Test
fun publisher_startSequence_seeds_first_group_for_hot_swap_continuation() =
runBlocking {
@@ -399,6 +508,148 @@ 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 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. 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",
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 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()
publisher.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
fun publisher_send_returns_false_when_no_inbound_subscriber() =
runBlocking {
@@ -533,6 +784,100 @@ class MoqLiteSessionTest {
session.close()
}
@Test
fun publisher_skips_announce_when_announce_please_prefix_does_not_match() =
runBlocking {
// Lite-03 audit M4: when the relay opens an Announce bidi
// with a non-empty prefix that doesn't match our broadcast
// suffix, the publisher MUST NOT emit Active under the
// requested prefix (or under our own suffix) — kixelated's
// `rs/moq-lite/src/lite/announce.rs::Producer` only emits
// for matching prefixes. Pre-fix we'd fall through the
// `null` branch of `MoqLitePath.stripPrefix` and announce
// ourselves anyway. Verified here by FINing the bidi
// cleanly without writing any Announce body.
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")
// Relay opens Announce bidi with a non-matching prefix.
val relayBidi = serverSide.openBidiStream()
relayBidi.write(Varint.encode(MoqLiteControlType.Announce.code))
relayBidi.write(
MoqLiteCodec.encodeAnnouncePlease(
MoqLiteAnnouncePlease(prefix = "differentRoom"),
),
)
// The bidi MUST end with no body (peer FIN), and
// `relayBidi.incoming().toList()` therefore returns empty.
// Use a generous timeout so a slow fake doesn't false-pass
// by hanging instead of FINing.
val chunks = withTimeout(2_000) { relayBidi.incoming().toList() }
assertEquals(emptyList(), chunks, "non-matching prefix must FIN without writing any Announce body")
publisher.close()
session.close()
}
@Test
fun publisher_replies_subscribeDrop_when_broadcast_does_not_match() =
runBlocking {
// Lite-03 audit M5: when the relay opens a Subscribe bidi
// whose `broadcast` field doesn't match our publisher's
// suffix, the publisher MUST reply SubscribeDrop with the
// BROADCAST_DOES_NOT_EXIST code rather than route OUR audio
// to the wrong subscriber. Pre-fix we matched on `track`
// only and would happily send Opus frames to a peer who
// subscribed under any broadcast string they liked.
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
val publisher = session.publish(broadcastSuffix = "speakerPubkey", track = "audio/data")
val subBidi = serverSide.openBidiStream()
subBidi.write(Varint.encode(MoqLiteControlType.Subscribe.code))
subBidi.write(
MoqLiteCodec.encodeSubscribe(
MoqLiteSubscribe(
id = 11L,
broadcast = "wrongPubkey",
track = "audio/data",
priority = 0x80,
ordered = true,
maxLatencyMillis = 0L,
startGroup = null,
endGroup = null,
),
),
)
val ackChunk = withTimeout(2_000) { subBidi.incoming().first() }
val resp = MoqLiteCodec.decodeSubscribeResponse(ackChunk)
val dropped = resp as MoqLiteCodec.SubscribeResponse.Dropped
assertEquals(MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST, dropped.drop.errorCode)
kotlin.test.assertContains(dropped.drop.reasonPhrase, "wrongPubkey")
kotlin.test.assertContains(dropped.drop.reasonPhrase, "speakerPubkey")
// Audit M3: in addition to the application-level Drop body,
// the publisher must also `RESET_STREAM(errorCode)` the bidi
// so the peer can distinguish a typed rejection from a
// graceful "publisher gone" FIN. Drain so the read side
// closes and the cell publishes, then assert the code.
// toList() will end because reset() closed the channel.
withTimeout(2_000) { subBidi.incoming().toList() }
val fakeBidi = subBidi as FakeBidiStream
assertEquals(
MoqLiteSubscribeDropCode.BROADCAST_DOES_NOT_EXIST as Long?,
fakeBidi.lastPeerResetCode,
"publisher must reset the bidi with BROADCAST_DOES_NOT_EXIST after the Drop body",
)
publisher.close()
session.close()
}
@Test
fun publisher_replies_subscribeDrop_when_track_is_not_published() =
runBlocking {
@@ -573,6 +918,19 @@ class MoqLiteSessionTest {
// the exact text so we can keep tweaking the wording.
kotlin.test.assertContains(dropped.drop.reasonPhrase, "video/data")
// Audit M3: typed RESET_STREAM follows the Drop body. Same
// shape as broadcast_does_not_match — the code matches
// the Drop's errorCode so a peer that only watched the
// QUIC layer (no body decode) still sees the typed
// rejection.
withTimeout(2_000) { subBidi.incoming().toList() }
val fakeBidi = subBidi as FakeBidiStream
assertEquals(
MoqLiteSubscribeDropCode.TRACK_DOES_NOT_EXIST as Long?,
fakeBidi.lastPeerResetCode,
"publisher must reset the bidi with TRACK_DOES_NOT_EXIST after the Drop body",
)
publisher.close()
session.close()
}
@@ -665,6 +1023,105 @@ class MoqLiteSessionTest {
session.close()
}
@Test
fun listener_stopSending_group_uni_when_subscription_already_canceled() =
runBlocking {
// Lite-03 audit M2: when a group's uni stream arrives for a
// subscribe id that was canceled before the first frame
// landed, the listener MUST stopSending() the uni stream so
// the publisher abandons retransmits instead of wasting
// bandwidth. Pre-fix the listener silently dropped every
// frame; the publisher kept pushing until natural FIN.
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
// Set up a subscription so the uni-stream pump is active.
val peer =
async {
val (bidi, _) = nextSubscribeBidi(serverSide)
bidi.write(MoqLiteCodec.encodeSubscribeOk(okFor(0L)))
bidi
}
val handle = session.subscribe("speakerX", "audio/data")
val subBidi = peer.await()
// Cancel the subscription before any group flows.
handle.unsubscribe()
// Drain the subscribe bidi's FIN so the peer's pump
// stays clean — not strictly required for the assertion
// but keeps the test isolated.
subBidi.incoming().toList()
// Now the peer (= the relay) opens a uni stream for that
// dead subscribe id and writes one frame.
val uni = serverSide.openUniStream() as ChannelWriteStream
uni.write(Varint.encode(MoqLiteDataType.Group.code))
uni.write(MoqLiteCodec.encodeGroupHeader(MoqLiteGroupHeader(subscribeId = handle.id, sequence = 0L)))
uni.write(framePayload(byteArrayOf(0x01)))
// Spin until the listener's drainOneGroup has processed
// the first frame, found no matching subscription, and
// fired stopSending. We hold the writer side, so reading
// [peerStopSendingCode] doesn't race the listener's
// pump for the peer-side [FakeReadStream] reference.
val deadline = System.currentTimeMillis() + 2_000
while (uni.peerStopSendingCode == null && System.currentTimeMillis() < deadline) {
kotlinx.coroutines.delay(10)
}
assertEquals(
MoqLiteStreamCancelCode.SUBSCRIPTION_GONE as Long?,
uni.peerStopSendingCode,
"listener must stopSending(SUBSCRIPTION_GONE) when a group arrives for a canceled subscription",
)
uni.finish()
session.close()
}
@Test
fun subscriber_probe_writes_control_type_and_decodes_publisher_bitrate() =
runBlocking {
// Lite-03 audit L3: the subscriber side opens a Probe bidi
// by writing ControlType=Probe and reads size-prefixed
// MoqLiteProbe messages from the publisher. Mirrors
// kixelated's `Subscriber::run_probe_stream`
// (`rs/moq-lite/src/lite/subscriber.rs`).
val (clientSide, serverSide) = FakeWebTransport.pair()
val session = MoqLiteSession.client(clientSide, pumpScope)
// Server side: read the subscriber's control byte, then
// write two Probe messages.
val serverPump =
async {
val bidi = serverSide.peerOpenedBidiStreams().first()
val buf = MoqLiteFrameBuffer()
// Read the leading ControlType varint. The session
// writes it as a single chunk via bidi.write.
val ctChunk = withTimeout(2_000) { bidi.incoming().first() }
buf.push(ctChunk)
val ct = buf.readVarint()
assertEquals(MoqLiteControlType.Probe.code, ct, "subscriber writes ControlType=Probe")
// Push two probes (different bitrates) then FIN.
bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 32_000L)))
bidi.write(MoqLiteCodec.encodeProbe(MoqLiteProbe(bitrate = 64_000L)))
bidi.finish()
}
val handle = session.probe()
// Drain both probes from the cold flow. SharedFlow with
// replay=8 means we can collect after the publisher has
// already pushed without losing emissions.
val received = withTimeout(2_000) { handle.updates.take(2).toList() }
assertEquals(2, received.size)
assertEquals(32_000L, received[0].bitrate)
assertEquals(64_000L, received[1].bitrate)
serverPump.await()
handle.close()
session.close()
}
@Test
fun unsubscribe_FINs_the_subscribe_bidi() =
runBlocking {
@@ -85,32 +85,26 @@ class QuicWebTransportFactory(
* via the `wt-available-protocols` header (RFC 8941 Structured Field List
* of strings see draft-ietf-webtrans-http3-14 §3.3).
*
* For nests, this MUST contain `moq-lite-03`; without it, moq-relay falls
* back to the legacy in-band SETUP exchange (moq-lite-02) and our first
* post-CONNECT message is decoded as SETUP_CLIENT, producing
* `connection closed err=invalid value` on the relay side and a stalled
* subscribe / `subscribe stream FIN before reply` on the client side.
* For nests, this MUST contain at least `moq-lite-03`; without it,
* moq-relay falls back to the legacy in-band SETUP exchange
* (moq-lite-02) and our first post-CONNECT message is decoded as
* SETUP_CLIENT, producing `connection closed err=invalid value` on
* the relay side and a stalled subscribe on the client side.
*
* **Lite-04 is intentionally NOT advertised** even though the kixelated
* browser client now ships it. Lite-04 reshapes the on-the-wire
* Announce.hops field from a single varint count into a full
* `OriginList` (`kixelated/moq` commit 45db108, "moq-lite/moq-relay:
* hop-based clustering"), adds an `exclude_hop` field to
* AnnounceInterest, and adds an `rtt` field to Probe Subscribe and
* Group framing are unchanged but Announce framing diverges. Our codec
* speaks pure Lite-03; if a Lite-04-preferring relay picks
* `moq-lite-04` from our advertised list, the very first Announce
* exchange would desync (we'd encode/read a bare hops varint where
* the peer expects `len + len × u62`) and the connection would abort.
* Until [com.vitorpamplona.nestsclient.moq.lite.MoqLiteCodec] gains
* version-aware Announce / AnnounceInterest / Probe paths and
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteAnnounce.hops]
* becomes a list, advertising Lite-04 is a footgun. See
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteAlpn] for the
* known-version constants.
* Both `moq-lite-04` and `moq-lite-03` are advertised by default
* the codec is now version-aware (audit L1), so the relay's
* choice (echoed via `wt-protocol`) is honoured at runtime via
* [WebTransportSession.negotiatedSubProtocol]
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion.fromAlpn].
* Lite-04 sits ahead of Lite-03 in the list to match kixelated's
* preference order; servers that don't yet support Lite-04 fall
* back to Lite-03 cleanly. See
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteAlpn] /
* [com.vitorpamplona.nestsclient.moq.lite.MoqLiteVersion] for the
* version constants and codec behavior.
*/
private val webTransportSubProtocols: List<String> =
listOf(MoqLiteAlpn.LITE_03),
listOf(MoqLiteAlpn.LITE_04, MoqLiteAlpn.LITE_03),
) : WebTransportFactory {
override suspend fun connect(
authority: String,
@@ -181,27 +175,27 @@ class QuicWebTransportFactory(
// Wait for the response HEADERS and verify :status is 2xx before
// declaring the WebTransport session open. Per RFC 9220 a non-2xx
// status means the server rejected the upgrade.
val responseStatus =
val response =
kotlinx.coroutines.withTimeoutOrNull(connectTimeoutMillis) {
readResponseStatus(requestStream)
} ?: -1
if (responseStatus < 0) {
readConnectResponse(requestStream)
}
if (response == null) {
driver.close()
throw WebTransportException(
kind = WebTransportException.Kind.HandshakeFailed,
message = "WebTransport CONNECT response timed out after ${connectTimeoutMillis}ms",
)
}
if (responseStatus !in 200..299) {
if (response.status !in 200..299) {
driver.close()
throw WebTransportException(
kind = WebTransportException.Kind.ConnectRejected,
message = "WebTransport CONNECT returned :status=$responseStatus",
message = "WebTransport CONNECT returned :status=${response.status}",
)
}
val state = QuicWebTransportSessionState(conn, driver, requestStream.streamId)
return QuicWebTransportSession(state)
return QuicWebTransportSession(state, response.subProtocol)
} catch (we: WebTransportException) {
throw we
} catch (ce: kotlinx.coroutines.CancellationException) {
@@ -220,14 +214,20 @@ class QuicWebTransportFactory(
}
/**
* Drain bytes from [requestStream] through an [Http3FrameReader] until a
* HEADERS frame arrives, then decode the QPACK field section and pull the
* `:status` pseudo-header.
* Drain bytes from [requestStream] through an [Http3FrameReader]
* until a HEADERS frame arrives, then decode the QPACK field
* section and pull out (`:status`, `wt-protocol`).
*
* Returns 0 if the stream closes without a HEADERS frame (the caller treats
* that as a connect rejection).
* `wt-protocol` (draft-ietf-webtrans-http3 §3.3, RFC 8941
* SF-string) is the server's selection from the client-offered
* `wt-available-protocols` list. Returned as a bare string with
* surrounding quotes stripped; `null` when the server didn't
* echo the header (e.g. older servers, or when the client
* offered only one protocol). The header value MAY contain
* commas if the server were to echo a list; we only expose the
* first item moq-lite servers always echo a single value.
*/
private suspend fun readResponseStatus(requestStream: QuicStream): Int {
private suspend fun readConnectResponse(requestStream: QuicStream): ConnectResponse {
val reader = Http3FrameReader()
val incoming = requestStream.incoming
try {
@@ -238,20 +238,52 @@ class QuicWebTransportFactory(
if (frame is Http3Frame.Headers) {
val pairs = QpackDecoder().decodeFieldSection(frame.qpackPayload)
val status = pairs.firstOrNull { it.first == ":status" }?.second?.toIntOrNull() ?: 0
throw HeadersReceived(status)
val subProtocol =
pairs
.firstOrNull { it.first.equals("wt-protocol", ignoreCase = true) }
?.second
?.let { parseFirstSfString(it) }
throw HeadersReceived(status, subProtocol)
}
}
}
} catch (e: HeadersReceived) {
return e.status
return ConnectResponse(e.status, e.subProtocol)
}
return 0
return ConnectResponse(0, null)
}
private data class ConnectResponse(
val status: Int,
val subProtocol: String?,
)
private class HeadersReceived(
val status: Int,
val subProtocol: String?,
) : RuntimeException()
/**
* Parse the first item out of an RFC 8941 SF-list-of-strings
* value (the format `wt-protocol` carries per
* draft-ietf-webtrans-http3 §3.3). For "moq-lite-04" servers
* echo a single quoted string; for completeness we tolerate a
* comma-separated list and take the first. Returns `null` if
* the value can't be parsed as an SF-string.
*
* This is a deliberately minimal parser full RFC 8941
* parameter handling isn't needed for the moq-lite use case
* (no parameters are ever attached). Tolerates surrounding
* whitespace per §3.1.
*/
private fun parseFirstSfString(raw: String): String? {
val firstItem = raw.substringBefore(',').trim()
if (firstItem.length < 2 || firstItem.first() != '"' || firstItem.last() != '"') return null
// SF-string only supports `\\` and `\"` escapes per §3.3.3.
// moq-lite values are bare ASCII; just strip the quotes.
return firstItem.substring(1, firstItem.length - 1)
}
/**
* Encode [items] as an RFC 8941 Structured Field List of bare strings
* (the format `wt-available-protocols` requires per draft-ietf-webtrans-http3
@@ -279,6 +311,14 @@ class QuicWebTransportFactory(
/** Adapter that wraps the :quic [QuicWebTransportSessionState] in the nestsClient interface. */
class QuicWebTransportSession(
private val state: QuicWebTransportSessionState,
/**
* The sub-protocol the server selected via `wt-protocol`
* during Extended CONNECT. Captured by
* [QuicWebTransportFactory.connect] from the parsed response
* headers and passed through to this constructor. `null` when
* the server didn't echo the header.
*/
override val negotiatedSubProtocol: String? = null,
) : WebTransportSession {
override val isOpen: Boolean get() = state.isOpen
@@ -366,6 +406,16 @@ private class QuicBidiStreamAdapter(
driver.wakeup()
}
override suspend fun reset(errorCode: Long) {
stream.resetStream(errorCode)
driver.wakeup()
}
override suspend fun stopSending(errorCode: Long) {
stream.stopSending(errorCode)
driver.wakeup()
}
override fun setPriority(priority: Int) {
stream.priority = priority
}
@@ -373,8 +423,14 @@ private class QuicBidiStreamAdapter(
private class QuicReadStreamAdapter(
private val stream: QuicStream,
private val driver: com.vitorpamplona.quic.connection.QuicConnectionDriver,
) : WebTransportReadStream {
override fun incoming(): Flow<ByteArray> = stream.incoming
override suspend fun stopSending(errorCode: Long) {
stream.stopSending(errorCode)
driver.wakeup()
}
}
/**
@@ -397,6 +453,11 @@ private class QuicUniWriteStreamAdapter(
driver.wakeup()
}
override suspend fun reset(errorCode: Long) {
stream.resetStream(errorCode)
driver.wakeup()
}
override fun setPriority(priority: Int) {
stream.priority = priority
}
@@ -407,13 +468,17 @@ private class StrippedWtReadStreamAdapter(
private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream,
) : WebTransportReadStream {
override fun incoming(): Flow<ByteArray> = stripped.data
override suspend fun stopSending(errorCode: Long) {
stripped.stopSending(errorCode)
}
}
/**
* Adapter for a peer-initiated bidi WT stream whose WT_BIDI_STREAM prefix
* has been stripped. Routes [write] / [finish] through the demux's
* driver-aware closures so application bytes actually leave the
* connection.
* has been stripped. Routes [write] / [finish] / [reset] / [stopSending]
* through the demux's driver-aware closures so application bytes actually
* leave the connection.
*/
private class StrippedWtBidiStreamAdapter(
private val stripped: com.vitorpamplona.quic.webtransport.StrippedWtStream,
@@ -440,13 +505,27 @@ private class StrippedWtBidiStreamAdapter(
finish()
}
override suspend fun reset(errorCode: Long) {
// Bidi streams have a send side we own, so `reset` is wired by
// the demux. Same defensive `?:` shape as in `write` / `finish`.
val r =
stripped.reset
?: error("peer-initiated bidi stream has no reset — demux didn't wire one")
r(errorCode)
}
override suspend fun stopSending(errorCode: Long) {
stripped.stopSending(errorCode)
}
/**
* No-op: peer-initiated bidi streams arrive through the demux as a
* [com.vitorpamplona.quic.webtransport.StrippedWtStream] which exposes
* only `send`/`finish` closures, not the underlying [QuicStream]. The
* moq-lite priority use case targets locally-opened uni group streams
* only, so this path doesn't need to model priority see the
* [WebTransportWriteStream.setPriority] contract.
* only `send`/`finish`/`reset`/`stopSending` closures, not the
* underlying [QuicStream]. The moq-lite priority use case targets
* locally-opened uni group streams only, so this path doesn't need
* to model priority see the [WebTransportWriteStream.setPriority]
* contract.
*/
override fun setPriority(priority: Int) = Unit
}
@@ -44,6 +44,14 @@ import kotlinx.coroutines.launch
* is false), [send] and [finish] are wired to the stream's outbound
* half so the application can write its response. They are null on
* unidirectional streams.
*
* For both directions, [reset] (RFC 9000 §3.5 RESET_STREAM) and
* [stopSending] (RFC 9000 §3.5 STOP_SENDING) are exposed so application
* code (e.g. moq-lite Lite-03's typed cancel paths) can convey error
* codes instead of relying on graceful FIN. [reset] is null on uni
* streams whose write side belongs to the peer; [stopSending] is null
* when there's no read side to ask the peer to stop on (i.e. never
* every stream we surface here has a read side).
*/
class StrippedWtStream(
val streamId: Long,
@@ -58,6 +66,23 @@ class StrippedWtStream(
* Half-close the send side (FIN). Null for unidirectional streams.
*/
val finish: (suspend () -> Unit)? = null,
/**
* Send `RESET_STREAM(applicationErrorCode)` on the send half (RFC
* 9000 §3.5). Null when the application doesn't own the send side
* (peer-initiated uni stream). The first call wins per
* `QuicStream.resetStream`'s lock-free first-call-wins gate;
* duplicate calls with different codes are silently ignored to
* keep the wire frame stable.
*/
val reset: (suspend (errorCode: Long) -> Unit)? = null,
/**
* Send `STOP_SENDING(applicationErrorCode)` on the receive half
* (RFC 9000 §3.5) asks the peer to RESET its corresponding
* send side. Non-null because every stripped stream surfaced
* here has a read side; the demux always wires it. First call
* wins.
*/
val stopSending: suspend (errorCode: Long) -> Unit,
)
/**
@@ -515,6 +540,26 @@ class WtPeerStreamDemux(
driver?.wakeup()
}
}
// RESET_STREAM is meaningful only on streams whose write side
// belongs to us. Same null pattern as [send] / [finish] for
// peer-initiated uni streams.
val reset: (suspend (Long) -> Unit)? =
if (isUni) {
null
} else {
{ errorCode ->
stream.resetStream(errorCode)
driver?.wakeup()
}
}
// STOP_SENDING is meaningful on the receive half — every
// stripped stream we surface here has one (uni: peer is
// sender; bidi: peer's send side is our receive side), so
// wire unconditionally.
val stopSending: (suspend (Long) -> Unit) = { errorCode ->
stream.stopSending(errorCode)
driver?.wakeup()
}
// Suspending send instead of `trySend` so a slow application
// back-pressures the demux instead of silently dropping the
// stream. With a bounded buffer ([readyStreamsBuffer]),
@@ -529,6 +574,8 @@ class WtPeerStreamDemux(
data = data,
send = send,
finish = finish,
reset = reset,
stopSending = stopSending,
),
)
}