From 47436c061c07e10f6c48eecda5df3bf8ce490a8d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 5 Sep 2026 02:09:53 -0700 Subject: [PATCH 1/4] chore: claim quest 3001 (655d9920-7850-4f61-9d92-545a0456cebd) From b5f532c6a67b802aa945d49417c5c735ad720af1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 5 Sep 2026 09:55:33 -0700 Subject: [PATCH 2/4] fix(moq-net): map stream reset codes to the negotiated moq-transport draft Every RESET_STREAM and STOP_SENDING on an IETF session went out with a moq-lite code, and came back decoded against the moq-lite table, whichever draft was negotiated. The registries overlap but do not match: draft-18 took 0x4 for GOING_AWAY from UNKNOWN_OBJECT_STATUS, TOO_FAR_BEHIND arrived in draft-17 and MALFORMED_TRACK in draft-16, and moq-lite's provisional 32-63 placeholders and 64+ application codes have no home in this registry at all. `ietf::error` is now the per-draft mapping in both directions, and the new `coding::StreamCodes` trait picks a stream's registry from the version its Reader/Writer already carries, so encoding and decoding cannot drift apart. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CeKipxourMpoqgK36azULd --- drafts/draft-lcurley-moq-lite.md | 2 + quest/m0/README.md | 2 +- quest/m0/ietf-error-codes.md | 51 +++-- quest/m1/README.md | 1 + quest/m1/hls-cache-miss-codes.md | 53 +++++ rs/moq-net/src/coding/codes.rs | 181 +++++++++++++++ rs/moq-net/src/coding/mod.rs | 2 + rs/moq-net/src/coding/reader.rs | 49 ++-- rs/moq-net/src/coding/stream.rs | 4 +- rs/moq-net/src/coding/writer.rs | 37 +++- rs/moq-net/src/ietf/error.rs | 356 ++++++++++++++++++++++++++---- rs/moq-net/src/ietf/mod.rs | 2 +- rs/moq-net/src/ietf/session.rs | 10 +- rs/moq-net/src/ietf/subscriber.rs | 4 +- 14 files changed, 645 insertions(+), 109 deletions(-) create mode 100644 quest/m1/hls-cache-miss-codes.md create mode 100644 rs/moq-net/src/coding/codes.rs diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index bf386c6b38..1540a0d4a6 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -243,6 +243,8 @@ There are two independent error code spaces, one for terminating the session and The same numeric value means different things in each, so an endpoint MUST select the code from the space matching what it is terminating. Both spaces reuse the codes moq-transport assigns, unchanged and with the same meaning, so an endpoint that speaks both protocols has one vocabulary and a relay can forward a peer's code without translating it. +The assignments are those of moq-transport draft-18 and later; the earlier drafts differ, giving 0x4 to UNKNOWN_OBJECT_STATUS rather than GOING_AWAY and registering neither TOO_FAR_BEHIND (added in draft-17) nor MALFORMED_TRACK (added in draft-16). +An endpoint bridging moq-lite to one of those MUST map the code for the version it negotiated rather than forwarding it. The codes moq-lite uses are listed in full below; an endpoint MUST NOT assign a moq-lite specific meaning to any code below 32. Codes 64 and above are the application's, opaque to moq-lite. diff --git a/quest/m0/README.md b/quest/m0/README.md index 3bac764d3f..fdc5d3c7da 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -19,7 +19,7 @@ regression test per Root Cause First. - [#3360](/quest/m0/3360-js-watch-broadcast-is-undefined-at-initialization.md) - js/watch: a framework binding the element reads `broadcast` before the custom element is upgraded - [Adapter namespace map](/quest/m0/rs-adapter-namespace-map.md) - moq-net: a duplicate PUBLISH_NAMESPACE on draft-14/15 strands the first request, and the map never shrinks - [Playout clock](/quest/m0/playout-clock.md) - moq play presents on a clock it controls, with a `--delay` offset and forward re-anchoring -- [IETF error codes](/quest/m0/ietf-error-codes.md) - every code on a moq-transport wire is a registered value for the negotiated draft, requests and stream resets alike +- [IETF error codes](/quest/m0/ietf-error-codes.md) - every request error a moq-transport wire carries is a registered value for the negotiated draft, in Rust and js/net - [Resume info](/quest/m0/resume-info-newest.md) - moq-net: resume reports segment zero's track info, so a replaced broadcast rescales timestamps on the predecessor's timescale - [#3080](/quest/m0/3080-fix-watch-audio-ring-truncate-can-race-the-worklet-reader.md) - watch: an audio ring truncate can race the worklet reader for one quantum - [#3363](/quest/m0/3363-js-watch-a-broadcast-republished-on-one-session-keeps-resuming.md) - js/watch: a broadcast republished under its name on one session keeps resuming diff --git a/quest/m0/ietf-error-codes.md b/quest/m0/ietf-error-codes.md index f90385afa6..99ed98c435 100644 --- a/quest/m0/ietf-error-codes.md +++ b/quest/m0/ietf-error-codes.md @@ -7,8 +7,7 @@ registered value for the negotiated draft, in both directions: the SUBSCRIBE_ERROR, FETCH_ERROR, and REQUEST_ERROR payloads, and the RESET_STREAM and STOP_SENDING codes on data and request streams. A relay built on these crates passes the interop runner's subscribe-error and subscribe-before-announce -cases without the peer's compatibility branch, and a routine cancellation is -read as CANCELLED rather than INTERNAL_ERROR. +cases without the peer's compatibility branch. Boundaries: moq-lite's own code spaces are untouched, and js/net's inability to carry a code on a locally raised stream error stays with @@ -17,16 +16,22 @@ moq-lite space and the abstraction, not the registry. ## Plan -What the tree does today, identically on main and dev: +The Rust stream reset half is done: `rs/moq-net/src/ietf/error.rs` is the +per-draft `StreamError` <-> code mapping in both directions, and the +`coding::StreamCodes` trait picks a stream's registry from the version its +`Reader`/`Writer` already carries. What is left is the request errors in Rust +and the whole of js/net. + +What the tree does today: - `rs/moq-net/src/ietf/publisher.rs` `run_subscribe` rejects with the literal `404` at three sites and `run_fetch_stream` with `500`; `rs/moq-net/src/ietf/subscriber.rs` `write_error` uses `400`. `reject_subscribe`, `reject_fetch`, and `write_error` take a bare `u64`, and `ietf::SubscribeError`, `ietf::FetchError`, and `ietf::RequestError` carry a - bare `error_code`. The only named IETF code type is `TrackStatusCode` - (`rs/moq-net/src/ietf/track.rs`), for a different registry, plus a - function-local `NOT_SUPPORTED: u64 = 0x3` in the subscriber. + bare `error_code`. The only named request-error code type is + `TrackStatusCode` (`rs/moq-net/src/ietf/track.rs`), for a different registry, + plus a function-local `NOT_SUPPORTED: u64 = 0x3` in the subscriber. - `js/net/src/ietf/publisher.ts` `runSubscribe` writes `errorCode: 404` on both the draft-14 `SubscribeError` and the draft-15+ `RequestError` branch; `js/net/src/ietf/subscriber.ts` uses `400` and `409`; `errorCode` is a plain @@ -34,31 +39,26 @@ What the tree does today, identically on main and dev: - `js/net/src/ietf/subscriber.ts` `runPublish` writes a draft-14 `PublishError` with the function-local `NOT_SUPPORTED` and a `RequestError` on later drafts; the Rust subscriber's publish handling is the same shape. -- Stream resets on an IETF session send `Error::to_code()`, the moq-lite space - where `Cancel` is `0`; draft-19 section 3.3.4 assigns `0x0` to INTERNAL_ERROR - and `0x1` to CANCELLED. `Error::from_transport` maps only `0` back to - `Cancel`, so both directions have to move together. #2993 fixed one site - (`cancel_subscribe`, a named `STREAM_CANCELLED`). +- js/net resets streams with the moq-lite code space on IETF sessions, the + mistake `rs/moq-net` no longer makes. - `Version::Draft14` through `Draft20` are all negotiated, straddling the draft-19 consolidation into REQUEST_ERROR, so the values are per version. The work: -- One named code type per registry (request errors, stream errors), with - `Encode` and `Decode` in the `TrackStatusCode` style, that - maps `Error` variants to the registered value for the negotiated draft and - back. An incoming code with no named value decodes to the remote/opaque - variant, never to a named one. Cite the draft section each value comes from - in the type's docs. -- Use it at every construction site above, in Rust and JS, and in - `Reader::abort` / `Writer::abort` on IETF sessions. Delete the literals and - the function-local constant. `Error::NotFound`'s IETF wire value comes from - the same mapping. +- One named code type per registry (request errors, and the stream errors js/net + still sends from the wrong space), with `Encode` and + `Decode` in the `TrackStatusCode` style, that maps `Error` variants + to the registered value for the negotiated draft and back. An incoming code + with no named value decodes to the remote/opaque variant, never to a named + one. Cite the draft section each value comes from in the type's docs. +- Use it at every construction site above, in Rust and JS. Delete the literals + and the function-local constants. `Error::NotFound`'s IETF wire value comes + from the same mapping. `rs/moq-net/src/ietf/error.rs` is the shape to follow. - Tests: encode/decode round-trips per version; a draft-14 PublishError and a - draft-15+ RequestError round-trip for the publish path; a subscribe for a missing - broadcast rejects with the not-found value of each draft; a cancelled - subscription resets with CANCELLED and a moq-net peer reads it back as - `Cancel`; the two interop runner cases pass without their COMPAT branch. + draft-15+ RequestError round-trip for the publish path; a subscribe for a + missing broadcast rejects with the not-found value of each draft; the two + interop runner cases pass without their COMPAT branch. The reporter of #3359 offered a PR and asked whether to prefer a version-aware enum over a flat one: version-aware, per the above. @@ -66,7 +66,6 @@ enum over a flat one: version-aware, per the above. ## Closes - [#3359](https://github.com/moq-dev/moq/issues/3359) - close this issue when the quest finishes -- [#3001](https://github.com/moq-dev/moq/issues/3001) - close this issue when the quest finishes ## Related diff --git a/quest/m1/README.md b/quest/m1/README.md index 499cecc5f0..e021972567 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -42,6 +42,7 @@ with the current dev tree before starting. - [#2709](/quest/m1/2709-per-broadcast-bandwidth-estimates-and-reservation.md) - js/net mirrors the send-side bandwidth allocator so each publisher encodes against its own share - [#3000](/quest/m1/3000-track-teardown-on-poll-unused-is-not-atomic-against-a.md) - Track teardown on poll_unused is not atomic against a consumer reattaching - [JS stream codes](/quest/m1/js-net-stream-error-codes.md) - js/net: a locally raised group error reaches the wire as INTERNAL_ERROR +- [HLS cache misses](/quest/m1/hls-cache-miss-codes.md) - moq-hls: a segment the relay dropped is served as a 500, because the miss is matched against a table the wire stopped using - [#3002](/quest/m1/3002-no-test-drives-a-late-group-through-the-ietf-dispatch-loop.md) - No test drives a late group through the IETF dispatch loop - [#3187](/quest/m1/3187-preserve-structured-protocol-error-codes-across-ffi-and-c.md) - Preserve structured protocol error codes across FFI and C bindings - [#2318](/quest/m1/2318-js-net-remaining-capability-gaps-vs-rs-moq-net-setup-role.md) - js/net: remaining capability gaps vs rs/moq-net (SETUP role, finish_at and final sequence, range controls, typed errors) diff --git a/quest/m1/hls-cache-miss-codes.md b/quest/m1/hls-cache-miss-codes.md new file mode 100644 index 0000000000..168fe14bff --- /dev/null +++ b/quest/m1/hls-cache-miss-codes.md @@ -0,0 +1,53 @@ +# [S] moq-hls reads cache misses against a table the wire stopped using + +## Goal + +`moq-hls` answers 404 for a segment the relay no longer has, and 500 only for a +real failure, whichever shape the error arrives in. The classification is driven +by the registry the code actually came off the wire in, so it cannot drift from +`moq-net` again. + +## Plan + +`is_cache_miss` in `rs/moq-hls/src/export/rendition.rs` compares a wire code +against `moq_net::Error::to_code()`: + +```rust +let code = err.to_code(); +code == moq_net::Error::NotFound.to_code() // 13 + || code == moq_net::Error::Old.to_code() // 2 + || code == moq_net::Error::Evicted.to_code() // 31 +``` + +That table is the crate's own legacy numbering, and no stream reset has carried +it since #2620 replaced it with the `StreamError` registry. A remote miss now +arrives as `Error::Remote(0x20 | 0x22 | 0x23)` on a moq-lite wire, and as +`Error::Remote(0)` on a moq-transport one, since that registry has no value for +any of the three. None of those match, so every miss that crossed a session, +which in a relay is all of them, is served as a 500 instead of a 404. + +Worse, one value collides: `Error::Old.to_code()` is 2, which is DELIVERY_TIMEOUT +on both wires, so a peer's delivery timeout classifies as a cache miss. + +The tests do not catch it because they build the remote shape out of the same +stale table (`Error::Remote(local.to_code())`), so they agree with the code +rather than with the wire. + +The work: + +- Classify on the decoded error, not a hand-compared code. `StreamError` already + names `NotFound`, `Old`, and `Evicted`, so the fix is for `moq-net` to keep + them named through `Error` rather than flattening them into `Remote(code)`, + and for `moq-hls` to match variants. +- Decide what a moq-transport upstream can say at all: that registry has no + value for a cache miss on a stream reset, so a relay fetching over it cannot + distinguish one from a failure. Either the miss travels as a request error + rather than a stream reset, or the 500 is correct there and only the moq-lite + path is fixable. +- Rewrite the tests to build the remote shape from the wire registry + (`StreamError::to_code`, `ietf::error::to_stream_code`) so they fail when the + two drift again. + +## Related + +- [IETF error codes](/quest/m0/ietf-error-codes.md) - the registry work that this classification has to follow diff --git a/rs/moq-net/src/coding/codes.rs b/rs/moq-net/src/coding/codes.rs new file mode 100644 index 0000000000..aaf743f25e --- /dev/null +++ b/rs/moq-net/src/coding/codes.rs @@ -0,0 +1,181 @@ +//! Which stream reset registry a stream's codes are written and read with. + +use crate::{Error, StreamError, ietf, lite}; + +/// The stream reset registry of the protocol a stream belongs to. +/// +/// Every [`Reader`](super::Reader) and [`Writer`](super::Writer) carries the negotiated +/// version, which is what picks the registry here. The two wires draw on the same names, +/// but they do not agree on every value: moq-lite's are fixed by +/// [`StreamError::to_code`], while moq-transport's moved across the drafts we negotiate +/// (see the `ietf::error` module). Sending a code from the wrong table is silent, because the +/// number is valid in both. +/// +/// Encoding and decoding live on one trait so a peer cannot be told a code from one table +/// and read against another: implement both halves or neither. +pub trait StreamCodes { + /// The code to reset a stream, or send STOP_SENDING, with. + fn encode_stream_code(&self, err: &StreamError) -> u32; + + /// Read a code the peer reset a stream (or sent STOP_SENDING) with. + fn decode_stream_code(&self, code: u32) -> StreamError; + + /// Turn a transport failure into an [`Error`], reading a stream reset with this + /// registry. + /// + /// A session close is not stream-scoped, so it decodes through + /// [`SessionError`](crate::SessionError) exactly as [`Error::from_transport`] does. + fn transport_error(&self, err: E) -> Error { + if let Some((code, _reason)) = err.session_error() { + return crate::SessionError::from_code(code).into(); + } + + if let Some(code) = err.stream_error() { + return self.decode_stream_code(code).into(); + } + + Error::Transport(err.to_string()) + } +} + +/// moq-lite's registry, specified by draft-lcurley-moq-lite (Error Codes) and identical +/// across the versions we negotiate. +impl StreamCodes for lite::Version { + fn encode_stream_code(&self, err: &StreamError) -> u32 { + err.to_code() + } + + fn decode_stream_code(&self, code: u32) -> StreamError { + StreamError::from_code(code) + } +} + +/// moq-transport's registry, which is per draft. +impl StreamCodes for ietf::Version { + fn encode_stream_code(&self, err: &StreamError) -> u32 { + ietf::error::to_stream_code(err, *self) + } + + fn decode_stream_code(&self, code: u32) -> StreamError { + ietf::error::from_stream_code(code, *self) + } +} + +/// The negotiated version before it is narrowed to one protocol, e.g. the SETUP stream a +/// pre-lite-05 session opens. +impl StreamCodes for crate::Version { + fn encode_stream_code(&self, err: &StreamError) -> u32 { + match self { + Self::Lite(version) => version.encode_stream_code(err), + Self::Ietf(version) => version.encode_stream_code(err), + } + } + + fn decode_stream_code(&self, code: u32) -> StreamError { + match self { + Self::Lite(version) => version.decode_stream_code(code), + Self::Ietf(version) => version.decode_stream_code(code), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The registry follows the negotiated protocol, not the other way around: a group + /// dropped for being old is a moq-lite placeholder and an INTERNAL_ERROR on the IETF + /// wire, and the same number read back on each wire has to mean what that wire said. + #[test] + fn the_version_picks_the_registry() { + let lite = crate::Version::Lite(lite::Version::Lite05); + let ietf = crate::Version::Ietf(ietf::Version::Draft20); + + assert_eq!(lite.encode_stream_code(&StreamError::Old), StreamError::Old.to_code()); + assert_eq!(ietf.encode_stream_code(&StreamError::Old), ietf::error::INTERNAL_ERROR); + + // Both agree on a cancellation, which is the one code that has to be right. + assert_eq!(lite.encode_stream_code(&StreamError::Cancel), ietf::error::CANCELLED); + assert_eq!(ietf.encode_stream_code(&StreamError::Cancel), ietf::error::CANCELLED); + + // GOING_AWAY is moq-lite's 0x4 and draft-20's, but draft-17 gives 0x4 to + // UNKNOWN_OBJECT_STATUS, so it is not sent there and not read back there. + let draft17 = crate::Version::Ietf(ietf::Version::Draft17); + assert_eq!(lite.decode_stream_code(0x4), StreamError::GoingAway); + assert_eq!(ietf.decode_stream_code(0x4), StreamError::GoingAway); + assert_eq!(draft17.decode_stream_code(0x4), StreamError::Unknown(0x4)); + assert_eq!( + draft17.encode_stream_code(&StreamError::GoingAway), + ietf::error::INTERNAL_ERROR + ); + } + + /// A dropped [`Writer`](super::Writer) resets with a cancellation, and `Drop` cannot + /// carry the bound that would reach the negotiated registry. It does not need one only + /// while every registry we speak agrees on the value, so pin that. + #[test] + fn both_registries_agree_about_a_cancellation() { + // Every negotiable version, plus the work-in-progress one `Versions::all` holds back. + let versions = crate::Versions::all() + .iter() + .copied() + .chain([crate::Version::Lite(lite::Version::Lite06Wip)]) + .collect::>(); + + for version in versions { + assert_eq!( + version.encode_stream_code(&StreamError::Cancel), + StreamError::Cancel.to_code(), + "{version:?} cancels with a different code than the Writer's Drop sends" + ); + } + } + + /// A session close is not stream-scoped, so it keeps the session registry whichever + /// wire the stream is on. + #[test] + fn a_session_close_keeps_the_session_registry() { + #[derive(Debug, thiserror::Error)] + #[error("failed")] + struct Failed { + session: Option, + stream: Option, + } + + impl web_transport_trait::Error for Failed { + fn session_error(&self) -> Option<(u32, String)> { + self.session.map(|code| (code, "closed".to_string())) + } + fn stream_error(&self) -> Option { + self.stream + } + } + + let version = ietf::Version::Draft20; + + // 0x0 ends a session cleanly, but fails a stream. + assert!(matches!( + version.transport_error(Failed { + session: Some(0x0), + stream: None + }), + Error::Cancel + )); + assert!(matches!( + version.transport_error(Failed { + session: None, + stream: Some(0x0) + }), + Error::Remote(0) + )); + + // Neither: the transport itself failed. + assert!(matches!( + version.transport_error(Failed { + session: None, + stream: None + }), + Error::Transport(_) + )); + } +} diff --git a/rs/moq-net/src/coding/mod.rs b/rs/moq-net/src/coding/mod.rs index e04b66be32..5f427d0974 100644 --- a/rs/moq-net/src/coding/mod.rs +++ b/rs/moq-net/src/coding/mod.rs @@ -1,5 +1,6 @@ //! Contains encoding and decoding helpers. +mod codes; mod decode; mod encode; mod reader; @@ -9,6 +10,7 @@ mod varint; mod version; mod writer; +pub use codes::*; pub use decode::*; pub use encode::*; pub use reader::*; diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index f1cc99f503..79ec70d9fb 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -32,7 +32,7 @@ pub struct Reader { /// enough that an ordinary burst is still one wake. const WAKE_BUDGET: usize = 64 * 1024; -impl Reader { +impl Reader { pub fn new(stream: S, version: V) -> Self { Self { stream, @@ -146,7 +146,9 @@ impl Reader { let n = cmp::min(self.buffer.len(), max); return Poll::Ready(Ok(Some(self.buffer.split_to(n).freeze()))); } - self.stream.poll_read_chunk(cx, max).map_err(Error::from_transport) + self.stream + .poll_read_chunk(cx, max) + .map_err(|err| self.version.transport_error(err)) } /// Fill a frame's payload from the stream, returning `Pending` once it would block. @@ -240,24 +242,17 @@ impl Reader { match ready!(self.stream.poll_read_buf(cx, &mut self.buffer)) { Ok(Some(_)) => Poll::Ready(Ok(true)), Ok(None) => Poll::Ready(Ok(false)), - Err(e) => Poll::Ready(Err(Error::from_transport(e))), + Err(e) => Poll::Ready(Err(self.version.transport_error(e))), } } /// Abort the stream with the given error. pub fn abort(&mut self, err: &Error) { - // STOP_SENDING is a stream operation, so it carries a stream code. Sending the - // session code here would have the peer read it against the wrong registry. - self.stream.stop(StreamError::from(err).to_code()); - } - - /// Abort the stream with a raw application code. - /// - /// [`Self::abort`] encodes the moq-lite error space. A protocol with its own registry - /// of stream reset codes has to name one directly, since the two spaces do not agree - /// on what a given number means. - pub fn stop(&mut self, code: u32) { - self.stream.stop(code); + // STOP_SENDING is a stream operation, so it carries a stream code from the + // negotiated protocol's registry. A session code, or the other protocol's, would be + // read against the wrong table and mean something else. + self.stream + .stop(self.version.encode_stream_code(&StreamError::from(err))); } /// Cast the reader to a different version, used during version negotiation. @@ -310,6 +305,8 @@ mod tests { /// arrive as INTERNAL_ERROR, and `Unauthorized` as KEY_VALUE_FORMATTING_ERROR. #[test] fn abort_stops_with_a_stream_code() { + const VERSION: crate::lite::Version = crate::lite::Version::Lite05; + for (err, expected) in [ (Error::Cancel, StreamError::Cancel.to_code()), (Error::Lagged, StreamError::TooFarBehind.to_code()), @@ -318,7 +315,7 @@ mod tests { StreamError::Session(crate::SessionError::Unauthorized).to_code(), ), ] { - let mut reader = Reader::new(StopLog::default(), ()); + let mut reader = Reader::new(StopLog::default(), VERSION); reader.abort(&err); assert_eq!(reader.stream.stops, vec![expected], "{err:?} used the wrong registry"); } @@ -327,6 +324,22 @@ mod tests { assert_ne!(StreamError::Cancel.to_code(), crate::SessionError::Cancel.to_code()); } + /// And the registry follows the negotiated protocol, not just the stream direction. A + /// group dropped for being old is a moq-lite placeholder with no moq-transport value, so + /// the same abort has to leave a different number on each wire. + #[test] + fn abort_stops_with_the_negotiated_protocols_code() { + let mut lite = Reader::new(StopLog::default(), crate::lite::Version::Lite05); + lite.abort(&Error::Old); + + let mut ietf = Reader::new(StopLog::default(), crate::ietf::Version::Draft20); + ietf.abort(&Error::Old); + + assert_eq!(lite.stream.stops, vec![StreamError::Old.to_code()]); + assert_eq!(ietf.stream.stops, vec![crate::ietf::error::INTERNAL_ERROR]); + assert_ne!(lite.stream.stops, ietf.stream.stops); + } + /// Counts wakes delivered to a parked consumer. #[derive(Default)] struct CountWaker(std::sync::atomic::AtomicUsize); @@ -406,7 +419,7 @@ mod tests { let (mut frame, mut payload, waiter, wakes) = parked_frame(9); assert!(payload.poll_read_chunk(&waiter).is_pending()); - let mut reader = Reader::new(Chunks([b"foo".as_slice(), b"bar"].into()), ()); + let mut reader = Reader::new(Chunks([b"foo".as_slice(), b"bar"].into()), crate::lite::Version::Lite05); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_read_frame(&mut cx, &mut frame).is_pending()); @@ -477,7 +490,7 @@ mod tests { payload, waiter, }, - (), + crate::lite::Version::Lite05, ); let mut cx = Context::from_waker(std::task::Waker::noop()); assert!(reader.poll_read_frame(&mut cx, &mut frame).is_pending()); diff --git a/rs/moq-net/src/coding/stream.rs b/rs/moq-net/src/coding/stream.rs index 63f499db49..07e7b529ed 100644 --- a/rs/moq-net/src/coding/stream.rs +++ b/rs/moq-net/src/coding/stream.rs @@ -1,7 +1,7 @@ use std::task::{Context, Poll, ready}; use crate::Error; -use crate::coding::{Reader, Writer}; +use crate::coding::{Reader, StreamCodes, Writer}; /// The send order every control stream is opened at. /// @@ -25,7 +25,7 @@ pub struct Stream { pub reader: Reader, } -impl Stream { +impl Stream { /// Poll opening a new stream with the given version. pub fn poll_open(session: &mut S, version: V, cx: &mut Context<'_>) -> Poll> where diff --git a/rs/moq-net/src/coding/writer.rs b/rs/moq-net/src/coding/writer.rs index 8bd4bc1324..aa49889a11 100644 --- a/rs/moq-net/src/coding/writer.rs +++ b/rs/moq-net/src/coding/writer.rs @@ -18,7 +18,7 @@ pub struct Writer { version: V, } -impl Writer { +impl Writer { /// Create a new writer for the given stream and version. pub fn new(stream: S, version: V) -> Self { Self { @@ -53,7 +53,7 @@ impl Writer { pub fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll> { while !self.buffer.is_empty() { ready!(self.stream.as_mut().unwrap().poll_write_buf(cx, &mut self.buffer)) - .map_err(Error::from_transport)?; + .map_err(|err| self.version.transport_error(err))?; } Poll::Ready(Ok(())) } @@ -107,7 +107,11 @@ impl Writer { /// immediately after an `encode` are safe because `encode` flushes fully. pub fn finish(&mut self) -> Result<(), Error> { debug_assert!(self.buffer.is_empty(), "finish with unflushed bytes"); - self.stream.as_mut().unwrap().finish().map_err(Error::from_transport) + self.stream + .as_mut() + .unwrap() + .finish() + .map_err(|err| self.version.transport_error(err)) } /// Abort the stream with the given error. @@ -116,7 +120,9 @@ impl Writer { /// reset a second time and overwrite the reason with a plain [`Error::Cancel`]. pub fn abort(mut self, err: &Error) { if let Some(mut stream) = self.stream.take() { - stream.reset(StreamError::from(err).to_code()); + // The code comes from the negotiated protocol's registry: the same number means + // different things on the two wires, so the version is what picks the table. + stream.reset(self.version.encode_stream_code(&StreamError::from(err))); } } @@ -132,17 +138,19 @@ impl Writer { return Ok(()); }; - stream.finish().map_err(Error::from_transport)?; - std::future::poll_fn(|cx| stream.poll_closed(cx).map_err(Error::from_transport)).await + stream.finish().map_err(|err| self.version.transport_error(err))?; + let version = &self.version; + std::future::poll_fn(|cx| stream.poll_closed(cx).map_err(|err| version.transport_error(err))).await } /// Poll until the stream is closed, or the [Self::finish] is acknowledged by the peer. pub fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { + let version = &self.version; self.stream .as_mut() .unwrap() .poll_closed(cx) - .map_err(Error::from_transport) + .map_err(|err| version.transport_error(err)) } /// Poll-friendly [`Self::close`] for a finished writer: once the peer acknowledges @@ -192,6 +200,10 @@ impl Drop for Writer { fn drop(&mut self) { if let Some(mut stream) = self.stream.take() { // Unlike the Quinn default, we abort the stream on drop. + // + // A `Drop` impl cannot add the bound that would reach the version's registry, and + // it does not need one: CANCELLED is 0x1 in moq-lite and in every moq-transport + // draft we negotiate, which `both_registries_agree_about_a_cancellation` pins. stream.reset(StreamError::Cancel.to_code()); } } @@ -218,8 +230,8 @@ mod tests { #[derive(Debug)] struct Poison; - impl Encode<()> for Poison { - fn encode(&self, w: &mut W, _: ()) -> Result<(), EncodeError> { + impl Encode for Poison { + fn encode(&self, w: &mut W, _: crate::lite::Version) -> Result<(), EncodeError> { w.put_slice(b"junk"); Err(EncodeError::BoundsExceeded) } @@ -229,7 +241,7 @@ mod tests { /// partial bytes would desynchronize the stream for every later message. #[test] fn a_failed_encode_leaves_no_partial_bytes() { - let mut writer = Writer::new(SinkSend::new(Log::default()), ()); + let mut writer = Writer::new(SinkSend::new(Log::default()), crate::lite::Version::Lite05); writer.buffer(&5u8).unwrap(); writer.buffer(&Poison).unwrap_err(); @@ -246,7 +258,10 @@ mod tests { #[test] fn flush_resumes_after_pending() { let gate = kio::Producer::new(false); - let mut writer = Writer::new(SinkSend::gated(Log::default(), gate.consume()), ()); + let mut writer = Writer::new( + SinkSend::gated(Log::default(), gate.consume()), + crate::lite::Version::Lite05, + ); let log = writer.stream.as_ref().unwrap().log.clone(); writer.buffer(&5u8).unwrap(); diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 05b35bb751..61318af478 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -1,72 +1,342 @@ -//! Stream reset codes for the moq-transport wire. +//! The moq-transport stream reset code registry, per negotiated draft. +//! +//! Sent on RESET_STREAM and STOP_SENDING, and read back off both. The values are the +//! draft's, not [`StreamError::to_code`]'s: the two registries agree on most of what they +//! both assign, but not all of it, and this one grew (and moved a value) across the drafts +//! we negotiate, so a code only means something once you know which draft carried it. +//! +//! | Code | draft-14/15 | draft-16 | draft-17 | draft-18+ | +//! |------|-------------|----------|----------|-----------| +//! | 0x0 | INTERNAL_ERROR | INTERNAL_ERROR | INTERNAL_ERROR | INTERNAL_ERROR | +//! | 0x1 | CANCELLED | CANCELLED | CANCELLED | CANCELLED | +//! | 0x2 | DELIVERY_TIMEOUT | DELIVERY_TIMEOUT | DELIVERY_TIMEOUT | DELIVERY_TIMEOUT | +//! | 0x3 | SESSION_CLOSED | SESSION_CLOSED | SESSION_CLOSED | SESSION_CLOSED | +//! | 0x4 | - | UNKNOWN_OBJECT_STATUS | UNKNOWN_OBJECT_STATUS | GOING_AWAY | +//! | 0x5 | - | - | TOO_FAR_BEHIND | TOO_FAR_BEHIND | +//! | 0x12 | - | MALFORMED_TRACK | MALFORMED_TRACK | MALFORMED_TRACK | +//! +//! Encoding and decoding therefore move together and both take the negotiated version: +//! GOING_AWAY sent to a draft-17 peer reads as UNKNOWN_OBJECT_STATUS, and a draft-17 +//! peer's UNKNOWN_OBJECT_STATUS read as GOING_AWAY would retire a session that is not +//! going anywhere. -use crate::Error; +use super::Version; +use crate::{SessionError, StreamError}; -/// The stream died on our side, with no registry entry for why. -/// -/// Deliberately separate from [`Error::to_code`], which encodes moq-lite's space. Those -/// codes are our own and unstandardized; these are the registry every moq-transport peer -/// reads. The same number means different things on the two wires, so the mappings must -/// not be interchanged: moq-lite's cancel is 0, which here is this. +/// An implementation-specific error: the stream died on our side, with no registry entry +/// for why. Assigned by every draft we negotiate. pub const INTERNAL_ERROR: u32 = 0x0; -/// The stream was cancelled by either endpoint. +/// The stream was cancelled by either endpoint. A routine unsubscribe, not a failure. +/// Assigned by every draft we negotiate. pub const CANCELLED: u32 = 0x1; -// Draft-19 section 3.3.4 also defines DELIVERY_TIMEOUT (0x2) and SESSION_CLOSED (0x3). -// Neither is named here because nothing we can currently detect earns them: the first is -// the negotiated timeout of section 8 rather than any expiry, and the second asserts the -// whole session is going away. Add them alongside the path that can actually prove one. +/// The content missed its delivery deadline. +const DELIVERY_TIMEOUT: u32 = 0x2; + +/// The session is closing, taking this stream with it. +const SESSION_CLOSED: u32 = 0x3; + +/// A GOAWAY was sent or received. Draft-18 and later; draft-16 and 17 gave 0x4 to +/// UNKNOWN_OBJECT_STATUS instead. +const GOING_AWAY: u32 = 0x4; + +/// The subscription outran the publisher's resource limits. Draft-17 and later. +const TOO_FAR_BEHIND: u32 = 0x5; -/// The code to reset a stream or send STOP_SENDING with. +/// The track's content could not be parsed. Draft-16 and later. +const MALFORMED_TRACK: u32 = 0x12; + +/// Whether the draft assigns 0x4 to GOING_AWAY. /// -/// Only cancellation maps, because only cancellation has a meaning both spaces agree on. -/// The rest of the registry is narrower than it looks: DELIVERY_TIMEOUT is the negotiated -/// timeout of draft-19 section 8, not any expiry we happen to hit, and SESSION_CLOSED -/// asserts the session is going away rather than one handle or one piece of content. Our -/// generic errors do not establish either, so claiming them would tell a peer something -/// specific and untrue. +/// Draft-16 and 17 assign it to UNKNOWN_OBJECT_STATUS, which draft-18 moved to 0x6 when it +/// took 0x4 for this. Draft-14 and 15 assign it nothing. +fn has_going_away(version: Version) -> bool { + matches!(version, Version::Draft18 | Version::Draft19 | Version::Draft20) +} + +/// Whether the draft assigns TOO_FAR_BEHIND. Added in draft-17. +fn has_too_far_behind(version: Version) -> bool { + !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) +} + +/// Whether the draft assigns MALFORMED_TRACK. Added in draft-16. +fn has_malformed_track(version: Version) -> bool { + !matches!(version, Version::Draft14 | Version::Draft15) +} + +/// The code to reset a stream, or send STOP_SENDING, with on the negotiated draft. /// -/// Everything else is INTERNAL_ERROR, which is the honest answer: the stream died on our -/// side and we have no registry entry for why. -pub fn to_stream_code(err: &Error) -> u32 { +/// Only values the draft registers go out. Everything else is INTERNAL_ERROR, which is not +/// a loss: draft-20 section 14 makes a receiver treat any unregistered code as equivalent +/// to INTERNAL_ERROR, so an unregistered value would say the same thing less clearly. That +/// covers the conditions moq-lite carries in its provisional 32-63 range (a group dropped +/// as old or evicted, a malformed frame size) and application codes, which this registry +/// has no range for at all: 64 and above is ordinary registry space here, and part of it is +/// reserved for greasing. +pub fn to_stream_code(err: &StreamError, version: Version) -> u32 { match err { - Error::Cancel => CANCELLED, + StreamError::Internal => INTERNAL_ERROR, + StreamError::Cancel => CANCELLED, + // We never negotiate the section 8 DELIVERY_TIMEOUT parameter, so this is only ever + // our own delivery deadline. That is what the code describes, and the same claim + // moq-lite makes with it, so a relay can carry a peer's timeout across either wire. + StreamError::DeliveryTimeout => DELIVERY_TIMEOUT, + // Flattened, as on the moq-lite wire: the session registry is disjoint, so the + // specific reason travels on the session close instead. + StreamError::Session(_) => SESSION_CLOSED, + StreamError::GoingAway if has_going_away(version) => GOING_AWAY, + StreamError::TooFarBehind if has_too_far_behind(version) => TOO_FAR_BEHIND, + StreamError::MalformedTrack if has_malformed_track(version) => MALFORMED_TRACK, _ => INTERNAL_ERROR, } } +/// Read a stream reset (or STOP_SENDING) code received on the negotiated draft. +/// +/// A code the draft does not assign stays [`StreamError::Unknown`], which surfaces as +/// [`Error::Remote`](crate::Error::Remote): an error, but never one given a meaning it did +/// not carry. That includes the codes this crate has no local counterpart for +/// (UNKNOWN_OBJECT_STATUS, EXPIRED_AUTH_TOKEN, EXCESSIVE_LOAD) and every value a later +/// draft may add. +pub fn from_stream_code(code: u32, version: Version) -> StreamError { + match code { + INTERNAL_ERROR => StreamError::Internal, + CANCELLED => StreamError::Cancel, + DELIVERY_TIMEOUT => StreamError::DeliveryTimeout, + // The peer's session code is not on this stream, so the reason is unknown here. + SESSION_CLOSED => StreamError::Session(SessionError::Internal), + GOING_AWAY if has_going_away(version) => StreamError::GoingAway, + TOO_FAR_BEHIND if has_too_far_behind(version) => StreamError::TooFarBehind, + MALFORMED_TRACK if has_malformed_track(version) => StreamError::MalformedTrack, + code => StreamError::Unknown(code), + } +} + #[cfg(test)] mod tests { use super::*; + use crate::Error; - /// The two spaces disagree on every value that matters, which is the whole reason this - /// mapping exists rather than reusing `Error::to_code`. + const ALL: [Version; 7] = [ + Version::Draft14, + Version::Draft15, + Version::Draft16, + Version::Draft17, + Version::Draft18, + Version::Draft19, + Version::Draft20, + ]; + + /// A routine unsubscribe must not read as a fault on our side. moq-lite's own error + /// enum encodes a cancellation as 0, which is this wire's INTERNAL_ERROR, so the codes + /// come from here instead. #[test] - fn the_two_error_spaces_disagree() { - assert_eq!(to_stream_code(&Error::Cancel), CANCELLED); - assert_ne!(to_stream_code(&Error::Cancel), Error::Cancel.to_code()); + fn a_cancellation_is_not_an_internal_error() { + for version in ALL { + assert_eq!(to_stream_code(&StreamError::Cancel, version), CANCELLED); + assert_eq!(from_stream_code(CANCELLED, version), StreamError::Cancel); + assert!(matches!( + Error::from(from_stream_code(CANCELLED, version)), + Error::Cancel + )); + } + + assert_ne!(CANCELLED, Error::Cancel.to_code(), "the two spaces disagree about 0"); + assert_eq!(Error::Cancel.to_code(), INTERNAL_ERROR); + } + + /// Every code we send must decode back to what we meant on the same draft, or two + /// moq-net peers disagree about what a stream reset said. + #[test] + fn every_emitted_code_round_trips() { + let errors = [ + StreamError::Internal, + StreamError::Cancel, + StreamError::DeliveryTimeout, + StreamError::GoingAway, + StreamError::TooFarBehind, + StreamError::MalformedTrack, + StreamError::NotFound, + StreamError::Old, + StreamError::Evicted, + StreamError::App(7), + ]; + + for version in ALL { + for err in &errors { + let code = to_stream_code(err, version); + let decoded = from_stream_code(code, version); + assert_eq!( + to_stream_code(&decoded, version), + code, + "{err:?} on {version:?} did not survive a round trip" + ); + } + + // A session teardown flattens to SESSION_CLOSED and comes back as one, rather + // than as the session's own reason, which the stream never carried. + let code = to_stream_code(&StreamError::Session(SessionError::Unauthorized), version); + assert_eq!(code, SESSION_CLOSED); + assert_eq!( + from_stream_code(code, version), + StreamError::Session(SessionError::Internal) + ); + } + } + + /// Draft-18 took 0x4 for GOING_AWAY from UNKNOWN_OBJECT_STATUS, so the same integer + /// means different things on two drafts we both negotiate. Sending it to draft-17 would + /// claim the next object's status is unknowable; reading theirs as GOING_AWAY would + /// start draining a session that is not going anywhere. + #[test] + fn going_away_only_exists_from_draft_18() { + for version in [Version::Draft14, Version::Draft15, Version::Draft16, Version::Draft17] { + assert_eq!(to_stream_code(&StreamError::GoingAway, version), INTERNAL_ERROR); + assert_eq!(from_stream_code(GOING_AWAY, version), StreamError::Unknown(GOING_AWAY)); + } + + for version in [Version::Draft18, Version::Draft19, Version::Draft20] { + assert_eq!(to_stream_code(&StreamError::GoingAway, version), GOING_AWAY); + assert_eq!(from_stream_code(GOING_AWAY, version), StreamError::GoingAway); + } + } + + /// The rest of the per-draft registry: TOO_FAR_BEHIND arrived in draft-17 and + /// MALFORMED_TRACK in draft-16, so an older peer must be told neither. + #[test] + fn later_codes_are_not_sent_to_earlier_drafts() { + for version in ALL { + let too_far_behind = to_stream_code(&StreamError::TooFarBehind, version); + let malformed = to_stream_code(&StreamError::MalformedTrack, version); + + assert_eq!( + too_far_behind, + match has_too_far_behind(version) { + true => TOO_FAR_BEHIND, + false => INTERNAL_ERROR, + }, + "{version:?} disagrees about TOO_FAR_BEHIND" + ); + assert_eq!( + malformed, + match has_malformed_track(version) { + true => MALFORMED_TRACK, + false => INTERNAL_ERROR, + }, + "{version:?} disagrees about MALFORMED_TRACK" + ); + } + assert_eq!( - Error::Cancel.to_code(), - INTERNAL_ERROR, - "moq-lite's cancel is this wire's failure" + from_stream_code(TOO_FAR_BEHIND, Version::Draft16), + StreamError::Unknown(TOO_FAR_BEHIND) + ); + assert_eq!( + from_stream_code(MALFORMED_TRACK, Version::Draft15), + StreamError::Unknown(MALFORMED_TRACK) ); } - /// An error with no registry entry says "this died on our side" rather than inventing a - /// meaning a peer would read as something specific. That includes errors whose names - /// echo a registry entry: our timeouts are not the negotiated delivery timeout, and a - /// closed handle is not a closing session. + /// Conditions this registry has no value for say INTERNAL_ERROR rather than borrowing + /// moq-lite's provisional 32-63 range or its application offset. A peer treats an + /// unregistered code as INTERNAL_ERROR anyway (draft-20 section 14), so the placeholder + /// would carry no more meaning while looking like a registration. #[test] - fn unmapped_errors_are_internal() { + fn unregistered_conditions_are_internal() { for err in [ - Error::NotFound, - Error::Duplicate, - Error::Timeout, - Error::Closed, - Error::Dropped, + StreamError::NotFound, + StreamError::Unroutable, + StreamError::Old, + StreamError::Evicted, + StreamError::WrongSize, + StreamError::FrameTooLarge, + StreamError::TimestampMismatch, + StreamError::App(7), + StreamError::Unknown(0x1234), ] { - assert_eq!(to_stream_code(&err), INTERNAL_ERROR, "{err:?} has no registry meaning"); + assert_eq!( + to_stream_code(&err, Version::Draft20), + INTERNAL_ERROR, + "{err:?} has no value in this registry" + ); + } + + // And nothing decodes back into them: an unregistered code keeps its number and + // stays opaque instead of being read as a meaning the wire did not carry. + for code in [0x6, 0x7, 0x9, 0x20, 0x22, 64 + 7] { + assert_eq!(from_stream_code(code, Version::Draft20), StreamError::Unknown(code)); + assert!(matches!( + Error::from(from_stream_code(code, Version::Draft20)), + Error::Remote(remote) if remote == code + )); + } + } + + /// Every stream error this crate can hold, so the conformance check below covers the + /// whole space rather than the variants someone remembered. A new variant belongs here. + const EVERY_ERROR: [StreamError; 16] = [ + StreamError::Session(SessionError::Cancel), + StreamError::Internal, + StreamError::Cancel, + StreamError::DeliveryTimeout, + StreamError::GoingAway, + StreamError::TooFarBehind, + StreamError::MalformedTrack, + StreamError::NotFound, + StreamError::Unroutable, + StreamError::Old, + StreamError::Evicted, + StreamError::WrongSize, + StreamError::FrameTooLarge, + StreamError::TimestampMismatch, + StreamError::App(7), + StreamError::Unknown(0x22), + ]; + + /// Every code we can put on a moq-transport stream has to be one the negotiated draft + /// registers. moq-lite's own table is not: it emits provisional values in 32-63 and + /// offsets application codes past 64, neither of which this registry has a range for, + /// and it assigns 0x4 and 0x5 meanings the earlier drafts give to something else. + /// + /// The table is transcribed from the drafts (draft-14 section 13.1.8 through draft-20 + /// section 15.11.4), not derived from the mapping, so a mistake in the mapping cannot + /// talk the assertion into agreeing with it. + #[test] + fn only_registered_codes_reach_the_wire() { + fn registered(version: Version) -> &'static [u32] { + match version { + Version::Draft14 | Version::Draft15 => &[0x0, 0x1, 0x2, 0x3], + Version::Draft16 => &[0x0, 0x1, 0x2, 0x3, 0x4, 0x12], + Version::Draft17 => &[0x0, 0x1, 0x2, 0x3, 0x4, 0x5, 0x9, 0x12], + _ => &[0x0, 0x1, 0x2, 0x3, 0x4, 0x5, 0x6, 0x7, 0x9, 0x12], + } + } + + for version in ALL { + for err in EVERY_ERROR { + let code = to_stream_code(&err, version); + assert!( + registered(version).contains(&code), + "{err:?} sends {code:#x}, which {version} does not register" + ); + } + } + } + + /// A relay decodes a peer's code and re-encodes it onto the stream it tears down in + /// response. That hop must not change what the code says, on the same draft. + #[test] + fn relaying_a_code_does_not_change_its_meaning() { + for version in ALL { + for code in [INTERNAL_ERROR, CANCELLED, DELIVERY_TIMEOUT, SESSION_CLOSED] { + let relayed = StreamError::from(&Error::from(from_stream_code(code, version))); + assert_eq!( + to_stream_code(&relayed, version), + code, + "{code:#x} changed across a relay on {version:?}" + ); + } } } } diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 97a3a5ed48..8c73b2ea0b 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -9,7 +9,7 @@ mod parameters; mod adapter; pub mod cluster; mod control; -mod error; +pub(crate) mod error; mod fetch; mod filter; mod goaway; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index a17fe87722..874eb7538d 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -694,11 +694,11 @@ where let mut reader = reader.with_version(version); if let Err(err) = run_uni_group(&mut sub, &mut reader).await { tracing::debug!(%err, "uni stream error"); - // A moq-transport code, not `Error::to_code`'s moq-lite one. A group arriving - // for an alias we retired is the expected tail of our own cancellation, and - // moq-lite's cancel encodes to 0, which on this wire means the stream died of - // an internal failure on our side. - reader.stop(super::error::to_stream_code(&err)); + // The reader carries the negotiated draft, so this is a moq-transport code. A + // group arriving for an alias we retired is the expected tail of our own + // cancellation, and moq-lite's own cancel encodes to 0, which on this wire + // says the stream died of an internal failure on our side. + reader.abort(&err); } }); } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 31f3e9a505..5de90c4918 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1635,7 +1635,7 @@ where // STOP_SENDING needs no acknowledgement, so it goes first and the wait below covers // only what we still have to deliver. - reader.stop(super::error::CANCELLED); + reader.abort(&Error::Cancel); // Finishing alone would leave the writer's Drop free to RESET_STREAM, and a stream // that has sent its FIN is still retransmitting: the reset would discard the @@ -2578,7 +2578,7 @@ mod tests { .expect_err("a retired alias resolves to a cancellation"); assert_eq!( - crate::ietf::error::to_stream_code(&err), + crate::ietf::error::to_stream_code(&crate::StreamError::from(&err), Version::Draft20), crate::ietf::error::CANCELLED, "the code the dispatch loop maps this error onto", ); From 69dada1ea67040675be16c8693402b1a114c33c9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 5 Sep 2026 19:06:53 -0700 Subject: [PATCH 3/4] fix(moq-net): decode raw write failures with the negotiated registry Raw payload writes bypassed the version-aware stream code conversion, so draft-17 STOP_SENDING 0x4 became GoingAway. Route them through the shared mapping and add a regression that fails before the fix. Clarify the separate request/reset registries and the protocol bridging rule. Co-Authored-By: GPT-6 --- drafts/draft-lcurley-moq-lite.md | 8 +++-- rs/moq-net/src/coding/writer.rs | 55 +++++++++++++++++++++++++++++++- rs/moq-net/src/ietf/error.rs | 10 ++---- rs/moq-net/src/ietf/session.rs | 18 ++--------- 4 files changed, 64 insertions(+), 27 deletions(-) diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 1540a0d4a6..de97a2fe66 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -242,9 +242,9 @@ However, it is ultimately the other peer's responsibility to close their send di There are two independent error code spaces, one for terminating the session and one for resetting a stream. The same numeric value means different things in each, so an endpoint MUST select the code from the space matching what it is terminating. -Both spaces reuse the codes moq-transport assigns, unchanged and with the same meaning, so an endpoint that speaks both protocols has one vocabulary and a relay can forward a peer's code without translating it. -The assignments are those of moq-transport draft-18 and later; the earlier drafts differ, giving 0x4 to UNKNOWN_OBJECT_STATUS rather than GOING_AWAY and registering neither TOO_FAR_BEHIND (added in draft-17) nor MALFORMED_TRACK (added in draft-16). -An endpoint bridging moq-lite to one of those MUST map the code for the version it negotiated rather than forwarding it. +The shared codes match moq-transport draft-18 and later. +Earlier drafts differ: 0x4 denotes UNKNOWN_OBJECT_STATUS in draft-16/17 and is unassigned in draft-14/15; TOO_FAR_BEHIND was added in draft-17 and MALFORMED_TRACK in draft-16. +An endpoint bridging protocols MUST translate codes according to the negotiated version and error-code space; moq-lite's provisional and application ranges have no corresponding moq-transport ranges. The codes moq-lite uses are listed in full below; an endpoint MUST NOT assign a moq-lite specific meaning to any code below 32. Codes 64 and above are the application's, opaque to moq-lite. @@ -1256,6 +1256,8 @@ The `Message Length` describes the payload size on the wire. # Appendix A: Changelog ## moq-lite-06 + +- Require error-code translation when bridging protocols and draft versions. - Made a repeated non-zero Hop ID in one announcement's Hop ID list a PROTOCOL_VIOLATION, matching draft-lcurley-moq-cluster. Repeated 0 entries stay legal. - Moved the Qmux-over-WebSocket binding details to draft-lcurley-qmux-websocket; the binding itself is unchanged. - Extended the SETUP `Path` parameter to carry the URI query: a client appends `?` and the query component after the path, matching moq-transport's PATH option. The credential a deployment puts in the query was previously unrepresentable on a binding with no request URI. diff --git a/rs/moq-net/src/coding/writer.rs b/rs/moq-net/src/coding/writer.rs index aa49889a11..0b25ef84fc 100644 --- a/rs/moq-net/src/coding/writer.rs +++ b/rs/moq-net/src/coding/writer.rs @@ -81,7 +81,7 @@ impl Writer { .as_mut() .unwrap() .poll_write_buf(cx, buf) - .map_err(Error::from_transport) + .map_err(|err| self.version.transport_error(err)) } /// Poll until the entire `Buf` has been written to the stream. @@ -214,6 +214,59 @@ mod tests { use crate::lite::test_transport::{Log, SinkSend}; use std::task::Waker; + #[derive(Debug, Clone, thiserror::Error)] + #[error("stream stopped with {0}")] + struct Stopped(u32); + + impl web_transport_trait::Error for Stopped { + fn session_error(&self) -> Option<(u32, String)> { + None + } + + fn stream_error(&self) -> Option { + Some(self.0) + } + } + + impl web_transport_trait::poll::SendStream for Stopped { + type Error = Self; + + fn poll_write(&mut self, _: &mut Context<'_>, _: &[u8]) -> Poll> { + Poll::Ready(Err(self.clone())) + } + + fn set_priority(&mut self, _: u8) {} + + fn finish(&mut self) -> Result<(), Self::Error> { + Err(self.clone()) + } + + fn reset(&mut self, _: u32) {} + + fn poll_closed(&mut self, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(self.clone())) + } + } + + #[tokio::test] + async fn raw_payload_stop_uses_the_negotiated_registry() { + for (version, expected) in [ + (crate::Version::Ietf(crate::ietf::Version::Draft17), Error::Remote(0x4)), + (crate::Version::Ietf(crate::ietf::Version::Draft20), Error::GoingAway), + (crate::Version::Lite(crate::lite::Version::Lite05), Error::GoingAway), + ] { + let mut writer = Writer::new(Stopped(0x4), version); + let err = writer.write_all(&mut b"payload".as_slice()).await.unwrap_err(); + assert!( + matches!( + (err, expected), + (Error::Remote(4), Error::Remote(4)) | (Error::GoingAway, Error::GoingAway) + ), + "{version} decoded the STOP_SENDING with the wrong registry" + ); + } + } + #[test] fn set_priority_forwards_send_order() { let log = Log::default(); diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 61318af478..fd8f59af59 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -67,13 +67,9 @@ fn has_malformed_track(version: Version) -> bool { /// The code to reset a stream, or send STOP_SENDING, with on the negotiated draft. /// -/// Only values the draft registers go out. Everything else is INTERNAL_ERROR, which is not -/// a loss: draft-20 section 14 makes a receiver treat any unregistered code as equivalent -/// to INTERNAL_ERROR, so an unregistered value would say the same thing less clearly. That -/// covers the conditions moq-lite carries in its provisional 32-63 range (a group dropped -/// as old or evicted, a malformed frame size) and application codes, which this registry -/// has no range for at all: 64 and above is ordinary registry space here, and part of it is -/// reserved for greasing. +/// Conditions without a matching reset code use INTERNAL_ERROR. Request rejection has +/// its own registry: a missing track belongs in a request error response, not a reset. +/// moq-lite's provisional and application ranges have no corresponding ranges here. pub fn to_stream_code(err: &StreamError, version: Version) -> u32 { match err { StreamError::Internal => INTERNAL_ERROR, diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index caa26e1b40..473e46227f 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -694,10 +694,6 @@ where let mut reader = reader.with_version(version); if let Err(err) = run_uni_group(&mut sub, &mut reader).await { tracing::debug!(%err, "uni stream error"); - // The reader carries the negotiated draft, so this is a moq-transport code. A - // group arriving for an alias we retired is the expected tail of our own - // cancellation, and moq-lite's own cancel encodes to 0, which on this wire - // says the stream died of an internal failure on our side. reader.abort(&err); } }); @@ -1202,18 +1198,8 @@ mod tests { writes.clone() } - /// A group for an alias we retired is answered with STOP_SENDING carrying the - /// moq-transport CANCELLED code. - /// - /// Such a group is the expected tail of our own cancellation, still in flight when we - /// unsubscribed. moq-lite's cancel encodes to 0, which on this wire says the stream - /// died of an internal fault on our side, so sending that for a routine unsubscribe - /// distorts the publisher's error handling. - /// - /// Driven over a real receive stream through the loop that sends it, which is what - /// makes this the test that fails if the loop stops mapping the error or stops calling - /// `stop` at all. `subscriber::a_retired_alias_maps_to_the_cancelled_code` owns the - /// mapping itself and would pass either way. + /// A late group for a retired alias must reach the dispatch loop and stop with + /// CANCELLED. Testing only the alias lookup would miss a broken dispatch path. #[tokio::test(start_paused = true)] async fn a_group_for_a_retired_alias_is_stopped_with_cancelled() { const VERSION: Version = Version::Draft19; From a35b120d193992ce358c269fb9874517fb972dfb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 5 Sep 2026 20:21:52 -0700 Subject: [PATCH 4/4] fix(moq-net): keep uni stream rejection scoped to the stream Do not encode SESSION_CLOSED when the uni child handler leaves the session running. Preserve the INTERNAL_ERROR fallback and cover the dispatcher with a regression. Track padding and unknown-type session handling separately. Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 + quest/m1/hls-cache-miss-codes.md | 9 ++++--- quest/m1/ietf-uni-stream-types.md | 17 +++++++++++++ rs/moq-net/src/coding/reader.rs | 5 ++-- rs/moq-net/src/ietf/session.rs | 41 ++++++++++++++++++++++--------- 5 files changed, 54 insertions(+), 19 deletions(-) create mode 100644 quest/m1/ietf-uni-stream-types.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 3cbcd041bd..180312a8d8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -41,6 +41,7 @@ with the current dev tree before starting. - [Plan: binding rate control](/quest/m1/plan-binding-rate-control.md) - settle how a non-Rust publisher follows the send estimate before wiring five bindings - [#2709](/quest/m1/2709-per-broadcast-bandwidth-estimates-and-reservation.md) - js/net mirrors the send-side bandwidth allocator so each publisher encodes against its own share - [#3000](/quest/m1/3000-track-teardown-on-poll-unused-is-not-atomic-against-a.md) - Track teardown on poll_unused is not atomic against a consumer reattaching +- [IETF stream types](/quest/m1/ietf-uni-stream-types.md) - accept padding and close sessions for genuinely unknown uni-stream types - [HLS cache misses](/quest/m1/hls-cache-miss-codes.md) - moq-hls: a segment the relay dropped is served as a 500, because the miss is matched against a table the wire stopped using - [#3187](/quest/m1/3187-preserve-structured-protocol-error-codes-across-ffi-and-c.md) - Preserve structured protocol error codes across FFI and C bindings - [#2318](/quest/m1/2318-js-net-remaining-capability-gaps-vs-rs-moq-net-setup-role.md) - js/net: remaining capability gaps vs rs/moq-net (SETUP role, finish_at and final sequence, range controls, typed errors) diff --git a/quest/m1/hls-cache-miss-codes.md b/quest/m1/hls-cache-miss-codes.md index 168fe14bff..e70fd3d3da 100644 --- a/quest/m1/hls-cache-miss-codes.md +++ b/quest/m1/hls-cache-miss-codes.md @@ -2,10 +2,11 @@ ## Goal -`moq-hls` answers 404 for a segment the relay no longer has, and 500 only for a -real failure, whichever shape the error arrives in. The classification is driven -by the registry the code actually came off the wire in, so it cannot drift from -`moq-net` again. +`moq-hls` answers 404 when the decoded error identifies a cache miss, and 500 +for genuine failures or an IETF stream reset that cannot distinguish a miss. +Returning 404 over IETF depends on the request-error path preserving that +classification. Decode using the negotiated registry so classification cannot +drift from `moq-net` again. ## Plan diff --git a/quest/m1/ietf-uni-stream-types.md b/quest/m1/ietf-uni-stream-types.md new file mode 100644 index 0000000000..15cd0f762d --- /dev/null +++ b/quest/m1/ietf-uni-stream-types.md @@ -0,0 +1,17 @@ +# [S] Validate IETF unidirectional stream types + +## Goal + +Accept valid padding streams and close the session for genuinely unknown stream types according to the negotiated moq-transport draft. + +## Plan + +`rs/moq-net/src/ietf/session.rs` routes every non-SETUP uni stream to `run_uni_group`, which rejects padding and unknown types alike while leaving the session alive. The stream-only rejection uses INTERNAL_ERROR because SESSION_CLOSED would falsely claim a session shutdown. + +Classify stream types before spawning a group handler. Handle PADDING according to each supported draft, including draining it where required, and propagate genuinely unknown types to the session driver. Keep ordinary group failures scoped to their streams. Add regressions for padding, unknown types causing session shutdown, and group failures preserving the session. + +Consult [draft-19 section 3.4 and section 11.5.1](https://www.ietf.org/archive/id/draft-ietf-moq-transport-19.html) and [draft-20 section 11.5.1](https://www.ietf.org/archive/id/draft-ietf-moq-transport-20.html), which explicitly permits cancelling padding streams. Check the earlier supported drafts too. + +## Related + +- [IETF error codes](/quest/m0/ietf-error-codes.md) - request and session error registry follow-ups diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 79ec70d9fb..b109aec38b 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -247,12 +247,11 @@ impl Reader { } /// Abort the stream with the given error. - pub fn abort(&mut self, err: &Error) { + pub fn abort(&mut self, err: impl Into) { // STOP_SENDING is a stream operation, so it carries a stream code from the // negotiated protocol's registry. A session code, or the other protocol's, would be // read against the wrong table and mean something else. - self.stream - .stop(self.version.encode_stream_code(&StreamError::from(err))); + self.stream.stop(self.version.encode_stream_code(&err.into())); } /// Cast the reader to a different version, used during version negotiation. diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 473e46227f..034f7b6812 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -1,6 +1,6 @@ use crate::origin; use crate::{ - Error, Hop, SessionError, + Error, Hop, SessionError, StreamError, coding::{Decode, DecodeError, Encode, Reader, Stream, Writer}, ietf::{self, FetchHeader, RequestId}, setup, @@ -694,7 +694,12 @@ where let mut reader = reader.with_version(version); if let Err(err) = run_uni_group(&mut sub, &mut reader).await { tracing::debug!(%err, "uni stream error"); - reader.abort(&err); + // This handler stops only the stream, so it cannot claim the session closed. + let reset = match StreamError::from(&err) { + StreamError::Session(_) => StreamError::Internal, + reset => reset, + }; + reader.abort(reset); } }); } @@ -1198,18 +1203,13 @@ mod tests { writes.clone() } - /// A late group for a retired alias must reach the dispatch loop and stop with - /// CANCELLED. Testing only the alias lookup would miss a broken dispatch path. - #[tokio::test(start_paused = true)] - async fn a_group_for_a_retired_alias_is_stopped_with_cancelled() { + async fn dispatch_uni(payload: Vec, retired_alias: Option) -> crate::lite::test_transport::Log { const VERSION: Version = Version::Draft19; - const ALIAS: u64 = 7; let origin = crate::origin::Info::new(crate::Hop::new(1).unwrap()).produce(); - // The peer opens one uni stream carrying the group header and then goes quiet, so + // The peer opens one uni stream and then goes quiet, so // the loop is still running when the assertion is taken. - let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()) - .with_incoming_unis(vec![subgroup_header(VERSION, ALIAS).await]); + let session = crate::lite::test_transport::ScriptedSession::new(Vec::new()).with_incoming_unis(vec![payload]); let log = session.log.clone(); let (tasks, _task_set) = TaskSet::new(); @@ -1227,7 +1227,9 @@ mod tests { tasks, Default::default(), ); - subscriber.retire_alias(ALIAS); + if let Some(alias) = retired_alias { + subscriber.retire_alias(alias); + } // Held so the trigger side stays alive for as long as the loop runs. let (_goaway, goaway) = crate::goaway::Handle::new(false); @@ -1237,7 +1239,7 @@ mod tests { for _ in 0..100 { if let std::task::Poll::Ready(result) = futures::poll!(unis.as_mut()) { - panic!("the dispatch loop ended over one dropped group: {result:?}"); + panic!("the dispatch loop ended over one rejected stream: {result:?}"); } if !log.stops().is_empty() { break; @@ -1245,6 +1247,14 @@ mod tests { tokio::time::sleep(std::time::Duration::from_millis(1)).await; } + log + } + + /// A late group must reach the dispatch loop and stop with CANCELLED. + #[tokio::test(start_paused = true)] + async fn a_group_for_a_retired_alias_is_stopped_with_cancelled() { + let log = dispatch_uni(subgroup_header(Version::Draft19, 7).await, Some(7)).await; + assert_eq!( log.stops(), vec![crate::ietf::error::CANCELLED], @@ -1252,4 +1262,11 @@ mod tests { ); assert_eq!(log.closes(), vec![], "one dropped group may not close the session"); } + + #[tokio::test(start_paused = true)] + async fn unknown_uni_type_does_not_claim_the_session_closed() { + let log = dispatch_uni(vec![0], None).await; + assert_eq!(log.stops(), vec![crate::ietf::error::INTERNAL_ERROR]); + assert!(log.closes().is_empty()); + } }