diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 972d4c7c20..2f098b4bfa 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -251,7 +251,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 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. @@ -1384,6 +1386,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/quest/m0/README.md b/quest/m0/README.md index 188d4996f4..285b1ee68d 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 bb15b24cf7..d751954043 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. js/net already carries a code on a locally raised stream error, via the `toStreamCode` / `fromTransport` @@ -17,6 +16,12 @@ registry. ## Plan +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 both +languages and the richer per-draft JS stream mappings. + Current implementation paths (verify against the target branch before starting): - `rs/moq-net/src/ietf/publisher.rs` `run_subscribe` rejects with the literal @@ -24,9 +29,9 @@ Current implementation paths (verify against the target branch before starting): `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,30 +39,28 @@ Current implementation paths (verify against the target branch before starting): - `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. -- Rust's `ietf/error.rs::to_stream_code` and JS's `toStreamCode` with an IETF - version send CANCELLED (1) for cancellation and INTERNAL_ERROR (0) otherwise. - JS keeps received IETF resets opaque rather than decoding them as moq-lite - local error classes. Per-draft typed decoding and richer mappings remain here. +- JS's `toStreamCode` with an IETF version sends CANCELLED (1) for cancellation + and INTERNAL_ERROR (0) otherwise. JS keeps received IETF resets opaque rather + than decoding them as moq-lite local error classes. Per-draft typed decoding + and richer mappings remain here. - `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 richer JS stream + error mappings), 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. @@ -65,7 +68,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 6f65e8ff09..180312a8d8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -41,6 +41,8 @@ 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) - [#2774](/quest/m1/2774-collapse-reload-and-shared-into-one-connection-class.md) - Collapse Reload and Shared into one Connection class diff --git a/quest/m1/hls-cache-miss-codes.md b/quest/m1/hls-cache-miss-codes.md new file mode 100644 index 0000000000..e70fd3d3da --- /dev/null +++ b/quest/m1/hls-cache-miss-codes.md @@ -0,0 +1,54 @@ +# [S] moq-hls reads cache misses against a table the wire stopped using + +## Goal + +`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 + +`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/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/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..b109aec38b 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,16 @@ 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); + 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(&err.into())); } /// Cast the reader to a different version, used during version negotiation. @@ -310,6 +304,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 +314,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 +323,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 +418,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 +489,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..0b25ef84fc 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(())) } @@ -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. @@ -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()); } } @@ -202,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(); @@ -218,8 +283,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 +294,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 +311,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..fd8f59af59 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -1,72 +1,338 @@ -//! 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 { +/// 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 { - 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 67ffc3c08e..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,11 +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"); - // 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)); + // 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); } }); } @@ -1202,28 +1203,13 @@ 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. - #[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(); @@ -1241,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); @@ -1251,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; @@ -1259,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], @@ -1266,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()); + } } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index b08e1ed085..18e66a5190 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1652,7 +1652,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 @@ -2595,7 +2595,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", );