diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 939945e36a..8b323cf4a5 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -385,6 +385,12 @@ datagram: unreliable, unordered, under about 1200 bytes, and never retransmitted. It suits real-time audio and sensor data. There is no stream fallback, so a datagram that doesn't fit isn't delivered that way. +Nothing caches a datagram. A subscription gets the ones inside its group +range, and a new one may get the few still in the publisher's short send +buffer, but `FETCH` never returns one. A publisher that knows the sequence was +a datagram refuses with `NOT_FETCHABLE` (moq-lite-07), otherwise with +`NOT_FOUND`. Use a group for anything a late joiner needs. + ## What moq-lite leaves out Compared with moq-transport: no request IDs (a stream per request instead), no diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 4757aaa99e..980b846ec3 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -86,7 +86,10 @@ tracks waiting for a group that never arrives. A moq-lite datagram is a single-frame group, so on moq-transport it travels as an `OBJECT_DATAGRAM` at object 0 whose Group ID is the sequence, and a relay forwards it without renumbering. A datagram carrying any other Object ID, or a -status other than Normal, is dropped. JavaScript does not yet carry datagrams +status other than Normal, is dropped. Datagrams are never fetchable: a fetch +object flagged as a datagram fails only that fetch, as `NotFetchable`, and a +relay answers its own downstream FETCH with `DOES_NOT_EXIST`, or `NOT_FETCHABLE` +to a moq-lite-07 peer. JavaScript does not yet carry datagrams on moq-transport. A client may present one credential in its `SETUP` with the `AUTHORIZATION diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 85e4a33a2d..04ec11d00f 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -330,6 +330,8 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST | ------- | ------------- | ----------- | | 0x39 | TIMESTAMP_MISMATCH | A frame's timestamp does not match its track's timescale. | | ------- | ------------- | ----------- | +| 0x3A | NOT_FETCHABLE | The FETCH named a group delivered only as a datagram, which is never cached (see [Datagrams](#datagrams)). It has no moq-transport value; a bridge refuses the FETCH with DOES_NOT_EXIST. | +| ------- | ------------- | ----------- | Note that CANCELLED is 0x1, not 0x0: a stream reset with 0x0 is an INTERNAL_ERROR, not a routine cancellation. An endpoint terminating a stream because the session is ending SHOULD use SESSION_CLOSED rather than the session's own code, since the two spaces are disjoint. @@ -658,7 +660,11 @@ The subscriber MAY cache the error and potentially retry later. QUIC datagrams provide unreliable, unordered delivery for latency-sensitive content that does not need retransmission. A publisher MAY transmit a Group consisting of exactly one Frame as a single QUIC datagram, in addition to (or instead of) opening a Group Stream, based on application hints, group size, and network conditions; a multi-frame Group is delivered via a Group Stream only. -A datagram-delivered group is not cached or retransmitted; a publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. +A publisher sends a datagram to a subscription only when its Group Sequence is inside the subscription's range, where a non-zero `Frame Start` excludes the start group (see [Positions](#positions)). +A datagram is never cached or retransmitted, and FETCH never returns one. +A publisher MAY send a new subscription the datagrams still in a short send buffer, so a subscriber cannot assume a datagram was sent after it subscribed. +A publisher that knows a FETCH names a group delivered only as a datagram resets the stream with NOT_FETCHABLE; otherwise it answers as for a group that does not exist, with NOT_FOUND. +A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence. Each datagram body has the following encoding (note: there is no message length prefix; the QUIC datagram boundary delimits the payload): @@ -1382,6 +1388,8 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 +- A subscription's range bounds datagrams like groups, and FETCH never returns a datagram. +- Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram. - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. - A refusal is not retried at another route of the same prefix either. - Made TRACK_INFO Publisher Max Age optional, encoded as milliseconds plus one with zero meaning no limit. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 9697149830..9b615d3871 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -156,6 +156,22 @@ test("appendDatagram delivers to a subscriber", async () => { expect(got && dec.decode(got.payload)).toBe("hello"); }); +// The in-range datagrams, sent between the dropped ones, show the reader is live, so the +// range is what dropped the others. +test("the group range bounds datagrams", async () => { + const producer = new TrackProducer("test"); + const track = producer.subscribe({ groups: { start: { included: 5 } } }); + track.setGroups({ end: { excluded: 7 } }); + for (const sequence of [4, 5, 7, 6]) { + producer.insertDatagram(sequence, Timestamp.fromMillis(0), enc.encode("x")); + } + producer.close(); + + expect((await track.recvDatagram())?.sequence).toBe(5); + expect((await track.recvDatagram())?.sequence).toBe(6); + expect(await track.recvDatagram()).toBeUndefined(); +}); + test("datagram buffers drop oldest at capacity without delaying active subscribers", async () => { const producer = new TrackProducer("test"); const slow = producer.subscribe(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 54c42587b9..0fd7b73855 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -1430,8 +1430,9 @@ export class Subscriber { * Receive the next datagram in arrival order. * * Datagrams are a separate best-effort channel from groups (see - * {@link Producer.appendDatagram}); they share only the sequence namespace. A consumer - * that falls too far behind silently loses the oldest datagrams. Read this alongside + * {@link Producer.appendDatagram}); they share only the sequence namespace. Those outside + * the group range are skipped. A consumer that falls too far behind silently loses the + * oldest datagrams. Read this alongside * {@link recvGroup} (e.g. in a separate loop) to receive both channels concurrently. * The two cursors are independent: a datagram never moves the group cursor. */ @@ -1446,8 +1447,11 @@ export class Subscriber { for (;;) { const datagrams = this.#state.datagrams.peek(); - if (datagrams.length > 0) { - return datagrams.shift(); + // The group range bounds datagrams too, judged when each is read. One outside it is + // gone for good: nothing holds it for a later raise, unlike a group past the cap. + const { start, end } = this.#cursor.peek(); + for (let datagram = datagrams.shift(); datagram; datagram = datagrams.shift()) { + if (datagram.sequence >= start && (end === undefined || datagram.sequence < end)) return datagram; } const closed = this.#state.closed.peek(); diff --git a/quest/m1/README.md b/quest/m1/README.md index 76699d10e7..55d5882370 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -80,7 +80,6 @@ blocks. The quests that gated m0 lines moved under them. - [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - a relay fetches from an IETF upstream without subscribing, finished tracks included, with End of Track always reported - [IETF object gaps](/quest/m1/ietf-object-gaps.md) - a gapped object ID is refused loudly like a subgroup in Rust and JS, and an overflowing one closes the session in Rust - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust -- [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - no FETCH, replay to a new subscriber, or cache fill ever returns a datagram group - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface - [Bounded HLS playlists](/quest/m1/hls-bounded.md) - `moq-hls` renders a capped sliding window for a fresh viewer, with a stable media sequence, gap slots, and no video before a sync point - [JS request window](/quest/m1/js-request-window.md) - @moq/net grants MAX_REQUEST_ID as requests close, so a peer never stalls after ~21k requests diff --git a/quest/m1/datagram-unfetchable.md b/quest/m1/datagram-unfetchable.md deleted file mode 100644 index 76fa052c69..0000000000 --- a/quest/m1/datagram-unfetchable.md +++ /dev/null @@ -1,55 +0,0 @@ -# [S] Datagrams are live-only - -## Goal - -A datagram group is delivered only to subscribers present when it is sent. -FETCH never returns one, a new subscription never replays buffered datagrams, -and a relay fills no cache miss with one. Anything that needs reliable or -late-joiner delivery should use a stream group instead. - -## Plan - -Decided in planning (09-29), after the moxygen fetch-fill agent found that the -relay's group-fetch decoder refuses a draft-16+ fetch object that carries the -datagram flag, and reports it as "unsupported": - -- Keep the existing 64-datagram send buffer per track - (`rs/moq-net/src/model/track.rs`, `MAX_DATAGRAMS`). An age bound was - considered and dropped. -- A new subscription starts at the next datagram sent, not at the buffered - backlog. This covers Rust and `@moq/net`, lite and IETF. -- FETCH never serves a datagram group. A publisher answers a FETCH for one the - way it answers any group it won't deliver (dropped or does not exist), - never with its payload. -- A received fetch object with the datagram flag is refused as not fetchable. - It fails only that fetch or fill, not the session, because draft-16+ allows - the flag (maintainer, 09-29, on Codex's review). That covers the relay's group - fill (`recv_group_fetch_objects`) and the joining-fetch - fill (`run_fill_objects`). -- Update `drafts/draft-lcurley-moq-lite.md` to say datagrams are neither - cached nor fetchable, and `doc/concept/`. Run `just drafts check` and - `just test interop --all`. - -Not planned: reliable or cached datagrams. If they are ever wanted, the way to -get them is back-pressure in the QUIC library and treating each datagram like -a one-shot stream, not a cache. - -The subscription's range bounds datagrams the way it bounds groups, fixed in -the model rather than filtered per session. Decided in the 2026-09-30 audit: -the moxygen line's datagram-range quest merged here, since it duplicated this -late-join rule. Its edge cases: - -- a datagram at the start group when a frame offset skips object 0; -- SUBSCRIBE_UPDATE moving the range while datagrams are in flight; -- a datagram that lands before the subscription's alias or id is known. - -Tests: a subscriber joining after datagrams were sent receives only later -ones. A FETCH covering a datagram group gets no payload. A fetch stream -carrying a datagram-flagged object fails that fetch and leaves the session up. -Each edge case above gets one, and a test tells a datagram filtered by the -range apart from one dropped for any other reason. - -## Related - -- [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - JS datagram send and receive over moq-transport -- [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - datagram groups stay best effort there too diff --git a/quest/m1/js-ietf-datagram.md b/quest/m1/js-ietf-datagram.md index 7ecfb8faab..0326b949ca 100644 --- a/quest/m1/js-ietf-datagram.md +++ b/quest/m1/js-ietf-datagram.md @@ -15,7 +15,3 @@ same way Rust does, and close the session on a malformed Type. The integration test `ietf does not deliver datagrams` flips to delivery on every supported draft, and `just test interop --all` covers both directions. - -## Related - -- [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - the subscribe range for datagrams, settled on both protocols diff --git a/rs/moq-c/src/error.rs b/rs/moq-c/src/error.rs index ddea6193fa..c0c0f6b10c 100644 --- a/rs/moq-c/src/error.rs +++ b/rs/moq-c/src/error.rs @@ -393,7 +393,8 @@ fn stream_kind(err: &moq_net::StreamError) -> moq_protocol_kind { moq_net::StreamError::GoingAway => MOQ_PROTOCOL_KIND_GOING_AWAY, moq_net::StreamError::TooFarBehind => MOQ_PROTOCOL_KIND_TOO_FAR_BEHIND, moq_net::StreamError::MalformedTrack => MOQ_PROTOCOL_KIND_MALFORMED_TRACK, - moq_net::StreamError::NotFound => MOQ_PROTOCOL_KIND_NOT_FOUND, + // A datagram reached by a FETCH is, to a binding, a miss like any other. + moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable => MOQ_PROTOCOL_KIND_NOT_FOUND, moq_net::StreamError::Unroutable => MOQ_PROTOCOL_KIND_UNROUTABLE, moq_net::StreamError::Old => MOQ_PROTOCOL_KIND_OLD, moq_net::StreamError::Evicted => MOQ_PROTOCOL_KIND_EVICTED, diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 148a18fae3..2061b1fb14 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -302,7 +302,7 @@ impl MoqBroadcastConsumer { fn map_fetch_error(err: moq_net::Error) -> MoqError { match err { - moq_net::Error::NotFound => MoqError::NotFound, + moq_net::Error::NotFound | moq_net::Error::NotFetchable => MoqError::NotFound, moq_net::Error::Unsupported | moq_net::Error::Version => MoqError::Unsupported, err => err.into(), } diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index 30e370dbbf..1f5ef3cdec 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -136,7 +136,8 @@ fn stream_kind(err: &moq_net::StreamError) -> MoqProtocolKind { moq_net::StreamError::GoingAway => MoqProtocolKind::GoingAway, moq_net::StreamError::TooFarBehind => MoqProtocolKind::TooFarBehind, moq_net::StreamError::MalformedTrack => MoqProtocolKind::MalformedTrack, - moq_net::StreamError::NotFound => MoqProtocolKind::NotFound, + // A datagram reached by a FETCH is, to a binding, a miss like any other. + moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable => MoqProtocolKind::NotFound, moq_net::StreamError::Unroutable => MoqProtocolKind::Unroutable, moq_net::StreamError::Old => MoqProtocolKind::Old, moq_net::StreamError::Evicted => MoqProtocolKind::Evicted, @@ -280,7 +281,7 @@ impl From for MoqError { fn from(err: moq_net::Error) -> Self { match err { moq_net::Error::Transport(message) => Self::Transport(message), - moq_net::Error::NotFound => Self::NotFound, + moq_net::Error::NotFound | moq_net::Error::NotFetchable => Self::NotFound, moq_net::Error::Closed | moq_net::Error::GoingAway | moq_net::Error::SessionClosed => Self::Closed, moq_net::Error::Cancel => Self::Cancelled, moq_net::Error::Unauthorized => Self::Unauthorized, diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index e90623e7db..2db0dca4d1 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -1047,10 +1047,14 @@ fn is_cache_miss(err: &moq_net::Error) -> bool { matches!( err, moq_net::Error::NotFound + | moq_net::Error::NotFetchable | moq_net::Error::Old | moq_net::Error::Evicted | moq_net::Error::Stream( - moq_net::StreamError::NotFound | moq_net::StreamError::Old | moq_net::StreamError::Evicted + moq_net::StreamError::NotFound + | moq_net::StreamError::NotFetchable + | moq_net::StreamError::Old + | moq_net::StreamError::Evicted ) ) } @@ -1100,7 +1104,12 @@ mod tests { #[test] fn a_cache_miss_that_crossed_a_session_is_still_a_cache_miss() { - for local in [moq_net::Error::NotFound, moq_net::Error::Old, moq_net::Error::Evicted] { + for local in [ + moq_net::Error::NotFound, + moq_net::Error::NotFetchable, + moq_net::Error::Old, + moq_net::Error::Evicted, + ] { assert!(is_cache_miss(&local), "{local:?} locally"); let lite = over_lite(&local); assert!(is_cache_miss(&lite), "{local:?} over lite ({lite:?})"); diff --git a/rs/moq-net/src/coding/codes.rs b/rs/moq-net/src/coding/codes.rs index 723b4ebe36..409ae657bb 100644 --- a/rs/moq-net/src/coding/codes.rs +++ b/rs/moq-net/src/coding/codes.rs @@ -7,8 +7,8 @@ use crate::{Error, StreamError, ietf, lite}; /// 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 +/// [`StreamError::to_code`] (a later version adds codes but never moves one), 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 @@ -38,15 +38,21 @@ pub trait StreamCodes { } } -/// moq-lite's registry, specified by draft-lcurley-moq-lite (Error Codes) and identical -/// across the versions we negotiate. +/// moq-lite's registry, specified by draft-lcurley-moq-lite (Error Codes). A code a version +/// lacks is sent there as the closest one it has, and not read back there. impl StreamCodes for lite::Version { fn encode_stream_code(&self, err: &StreamError) -> u32 { - err.to_code() + match err { + StreamError::NotFetchable if !self.has_not_fetchable() => StreamError::NotFound.to_code(), + err => err.to_code(), + } } fn decode_stream_code(&self, code: u32) -> StreamError { - StreamError::from_code(code) + match StreamError::from_code(code) { + StreamError::NotFetchable if !self.has_not_fetchable() => StreamError::Unknown(code), + err => err, + } } } @@ -110,6 +116,32 @@ mod tests { ); } + /// NOT_FETCHABLE is lite-07's: an earlier peer is told NOT_FOUND, the miss it would + /// have been told before, and is not credited with a code its version never assigned. + #[test] + fn not_fetchable_is_lite_07s() { + for version in [lite::Version::Lite05, lite::Version::Lite06] { + assert_eq!( + version.encode_stream_code(&StreamError::NotFetchable), + StreamError::NotFound.to_code(), + "{version}" + ); + assert_eq!( + version.decode_stream_code(0x3a), + StreamError::Unknown(0x3a), + "{version}" + ); + } + + let version = lite::Version::Lite07; + assert_eq!(version.encode_stream_code(&StreamError::NotFetchable), 0x3a); + assert_eq!(version.decode_stream_code(0x3a), StreamError::NotFetchable); + assert_eq!( + version.decode_stream_code(StreamError::NotFound.to_code()), + StreamError::NotFound + ); + } + /// 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. diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index af188dcb0e..a606cdedcc 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -146,6 +146,11 @@ pub enum StreamError { #[error("not found")] NotFound, + /// A FETCH reached a datagram, which is never cached. Sent on moq-lite-07 + /// and later; an earlier version sends [`NotFound`](Self::NotFound) instead. + #[error("not fetchable")] + NotFetchable, + /// The broadcast is neither announced nor served, so there is no route to it. #[error("unroutable")] Unroutable, @@ -207,6 +212,7 @@ impl StreamError { Self::WrongSize => 0x37, Self::FrameTooLarge => 0x38, Self::TimestampMismatch => 0x39, + Self::NotFetchable => 0x3a, Self::App(app) => *app as u32 + 64, Self::Unknown(code) => *code, } @@ -238,6 +244,7 @@ impl StreamError { 0x37 => Self::WrongSize, 0x38 => Self::FrameTooLarge, 0x39 => Self::TimestampMismatch, + 0x3a => Self::NotFetchable, code @ 64.. => match u16::try_from(code - 64) { Ok(app) => Self::App(app), Err(_) => Self::Unknown(code), @@ -323,6 +330,10 @@ pub enum Error { #[error("not found")] NotFound, + /// A FETCH reached a datagram, which is never cached. + #[error("not fetchable")] + NotFetchable, + /// A joining FETCH named a request that is not an active subscription. #[error("invalid joining request ID")] InvalidJoiningRequestId, @@ -567,6 +578,7 @@ impl From<&Error> for StreamError { Error::Evicted => Self::Evicted, Error::Lagged => Self::TooFarBehind, Error::NotFound => Self::NotFound, + Error::NotFetchable => Self::NotFetchable, Error::Unroutable => Self::Unroutable, Error::WrongSize => Self::WrongSize, Error::FrameTooLarge => Self::FrameTooLarge, @@ -675,6 +687,7 @@ mod tests { StreamError::WrongSize, StreamError::FrameTooLarge, StreamError::TimestampMismatch, + StreamError::NotFetchable, StreamError::App(7), ]; for err in registered { @@ -692,6 +705,7 @@ mod tests { (StreamError::WrongSize, 0x37), (StreamError::FrameTooLarge, 0x38), (StreamError::TimestampMismatch, 0x39), + (StreamError::NotFetchable, 0x3a), ] { assert_eq!(err.to_code(), code, "{err:?} moved off its assigned code"); } @@ -726,9 +740,12 @@ mod tests { let relayed = StreamError::from(&Error::from(StreamError::from_code(0x3))); assert_eq!(relayed.to_code(), 0x3); + // A datagram reached by a FETCH keeps its own code rather than reading as a plain miss. + assert_eq!(StreamError::from(&Error::NotFetchable), StreamError::NotFetchable); + // Registered stream codes survive the hop unchanged. for code in [ - 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, + 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, ] { let relayed = StreamError::from(&Error::from(StreamError::from_code(code))); assert_eq!(relayed.to_code(), code, "stream {code:#x} changed across a relay"); diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 66f1163408..44b1010059 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -289,7 +289,12 @@ pub(crate) mod request { return TIMEOUT; } Error::Unsupported | Error::Version | Error::Session(SessionError::Version) => return NOT_SUPPORTED, - Error::NotFound | Error::Stream(StreamError::NotFound) => does_not_exist(kind, version), + // moq-transport has no code for a FETCH that reached a datagram, which is never + // cached, so it is a group that is not there. NOT_SUPPORTED would claim the peer + // does not implement FETCH at all. + Error::NotFound + | Error::NotFetchable + | Error::Stream(StreamError::NotFound | StreamError::NotFetchable) => does_not_exist(kind, version), Error::InvalidRange => invalid_range(kind, version), Error::InvalidJoiningRequestId => invalid_joining_request_id(kind, version), // A path with no route is one we will not carry. A subscriber that asked for it @@ -361,12 +366,13 @@ pub(crate) mod request { /// Every error a rejection distinguishes, plus one it does not, so the checks below cover /// the whole registry rather than the variants someone remembered. A new arm in /// [`to_code`] belongs here. - const EVERY_ERROR: [Error; 11] = [ + const EVERY_ERROR: [Error; 12] = [ Error::Duplicate, Error::Unauthorized, Error::Timeout, Error::Unsupported, Error::NotFound, + Error::NotFetchable, Error::InvalidRange, Error::InvalidJoiningRequestId, Error::Unroutable, @@ -395,6 +401,16 @@ pub(crate) mod request { // A broadcast with no route is, to the peer that asked for it, not here either. assert_eq!(to_code(&Error::Unroutable, Kind::Subscribe, Version::Draft20), 0x10); + + // So is a datagram group: never cached, so a FETCH reaching one finds nothing, and + // NOT_SUPPORTED would say the endpoint has no FETCH at all. + assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft14), 0x4); + assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft20), 0x10); + // Including a lite-07 upstream's NOT_FETCHABLE, bridged by a relay. + assert_eq!( + to_code(&Error::Stream(StreamError::NotFetchable), Kind::Fetch, Version::Draft20), + 0x10 + ); } /// Draft-14 gives 0x4 to UNINTERESTED on the requests that offer content and to @@ -695,6 +711,7 @@ mod tests { fn unregistered_conditions_are_internal() { for err in [ StreamError::NotFound, + StreamError::NotFetchable, StreamError::Unroutable, StreamError::Old, StreamError::Evicted, @@ -728,6 +745,7 @@ mod tests { 0x37, 0x38, 0x39, + 0x3a, 64 + 7, ] { assert_eq!(from_stream_code(code, Version::Draft20), StreamError::Unknown(code)); @@ -740,7 +758,7 @@ mod tests { /// 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; 18] = [ + const EVERY_ERROR: [StreamError; 19] = [ StreamError::Session(SessionError::Cancel), StreamError::Internal, StreamError::Cancel, @@ -750,6 +768,7 @@ mod tests { StreamError::TooFarBehind, StreamError::MalformedTrack, StreamError::NotFound, + StreamError::NotFetchable, StreamError::Unroutable, StreamError::Old, StreamError::Evicted, diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index dd81ec4063..43ba5b95a4 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4872,6 +4872,34 @@ mod serve_tests { } } + /// A datagram group is never fetchable: a FETCH for one is refused like a group that + /// does not exist, newest or not, and opens no fetch stream for its payload. + #[moq_net_sim::test] + async fn a_standalone_fetch_of_a_datagram_group_is_refused() { + for version in FETCH_DRAFTS { + for newest in [true, false] { + let mut h = serve(version); + publish_pairs(&mut h, 2, None); + let sequence = h.track.append_datagram(timestamp(), b"d".as_slice()).unwrap(); + if !newest { + h.track.append_group().unwrap().finish().unwrap(); + } + settle().await; + + let location = Location { + group: sequence, + object: 0, + }; + let buf = standalone_fetch(&h, location, location, GroupOrder::Ascending).await; + assert_eq!( + fetch_refusal(buf, version), + does_not_exist(version), + "{version}: newest={newest}" + ); + } + } + } + /// A joining FETCH reaching back before the subscription's group is refused, like any /// FETCH touching several groups. #[moq_net_sim::test] diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 8f325a52e3..e3994e3f75 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3259,7 +3259,7 @@ where let mut prior_group = None; let mut first = true; while let Some(object) = decode_fetch_object(stream, self.version, std::mem::take(&mut first)).await? { - if !object.subgroup_ok { + if !object.datagram && !object.subgroup_ok { tracing::warn!("subgroup ID is not supported, dropping fill"); return Err(Error::Unsupported); } @@ -3270,6 +3270,28 @@ where prior_group = Some(sequence); } + if object.datagram { + // An absolute join spans whole groups, so a datagram among them is a group of + // its own: the head before it is complete, and the groups after it still fill. + // Any other fill is the one group, which a datagram cannot be. The group is the + // resolved one, since a later object of a datagram group inherits its Group ID. + let sequence = prior_group.filter(|_| matches!(joining, Some(JoiningFetch::Absolute { .. }))); + let Some(sequence) = sequence else { + tracing::debug!(?group, "a datagram group is not fetchable"); + return Err(Error::NotFetchable); + }; + // A datagram inside the head's own group mixes the two, which no group can hold. + if head.as_ref().is_some_and(|(head, _, _)| *head == sequence) { + tracing::debug!(sequence, "a fetched group mixes stream and datagram objects"); + return Err(Error::NotFetchable); + } + end_fill_group(head, largest)?; + // Draft-16 on, where a datagram can appear, a fetch object has no status field. + let mut remaining = usize::try_from(stream.varint().await?).map_err(|_| Error::FrameTooLarge)?; + std::future::poll_fn(|cx| stream.poll_skip(cx, &mut remaining)).await?; + continue; + } + match head.as_ref().map(|(sequence, next, _)| (*sequence, *next)) { None => { let (Some(sequence), Some(0)) = (group, object.object) else { @@ -3718,6 +3740,10 @@ where tracing::warn!(sequence, "a group fetch continued past its end marker"); return Err(Error::ProtocolViolation); } + if object.datagram { + tracing::debug!(sequence, "a datagram group is not fetchable"); + return Err(Error::NotFetchable); + } if !object.subgroup_ok { tracing::warn!("subgroup ID is not supported, dropping group fetch"); return Err(Error::Unsupported); @@ -3776,6 +3802,9 @@ where struct FetchedObject { group: Option, object: Option, + /// Draft-16 on lets a fetch carry an Object published as a datagram. A datagram group is + /// never cached, so it is never filled from a fetch. + datagram: bool, subgroup_ok: bool, properties: Option>, } @@ -3802,6 +3831,7 @@ async fn decode_fetch_object( return Ok(Some(FetchedObject { group: Some(group), object: Some(object), + datagram: false, subgroup_ok: subgroup == 0, properties: Some(properties), })); @@ -3831,6 +3861,7 @@ async fn decode_fetch_object( Some(FetchedObject { group, object, + datagram: subgroup == ietf::FetchSubgroup::Datagram, subgroup_ok: matches!( subgroup, ietf::FetchSubgroup::Zero | ietf::FetchSubgroup::Prior | ietf::FetchSubgroup::Explicit(0) @@ -3886,8 +3917,17 @@ fn advance_fill_group( return Err(Error::Unsupported); } + end_fill_group(head, largest)?; + open_fill_group(track, head, sequence) +} + +/// Finish the group an absolute joining FETCH was writing, since a later group started. +fn end_fill_group( + head: &mut Option<(u64, u64, crate::recv::Group)>, + largest: Option, +) -> Result<(), Error> { let Some((prev, _, producer)) = head.take() else { - return open_fill_group(track, head, sequence); + return Ok(()); }; if largest.is_some_and(|largest| prev >= largest.group) { @@ -3897,7 +3937,7 @@ fn advance_fill_group( } producer.finish()?; - open_fill_group(track, head, sequence) + Ok(()) } /// Ready once a finished subscription's data streams are accounted for: Stream Count of @@ -4659,6 +4699,55 @@ mod tests { assert!(is_protocol_violation(&subscriber.recv_datagram(malformed).unwrap_err())); } + /// A datagram that lands before SUBSCRIBE_OK binds its alias is dropped, not held for the + /// subscription to replay once it is bound. The reader was already subscribed, so only the + /// missing alias can have dropped it. + #[moq_net_sim::test] + async fn a_datagram_before_its_alias_is_dropped() { + use crate::coding::Encode as _; + use futures::FutureExt as _; + + let subscriber = subscriber_with_tracks(&[(RequestId(11), "cam", "audio")]); + let mut consumer = { + let mut state = subscriber.state.lock(); + let track = state.subscribes.get_mut(&RequestId(11)).unwrap(); + track.timescale = Some(Timescale::default()); + track.producer.as_ref().unwrap().subscribe(None) + }; + let datagram = |group_id: u64| { + let mut properties = Vec::new(); + ietf::encode_object_time( + &mut crate::coding::Encoder::new(&mut properties, Version::Draft19.into()), + crate::Timestamp::new(96_000, Timescale::default()).unwrap(), + Timescale::default(), + Version::Draft19, + ) + .unwrap(); + ietf::ObjectDatagram { + track_alias: 7, + group_id, + object_id: None, + publisher_priority: None, + end_of_group: true, + properties: Some(properties), + body: ietf::DatagramBody::Payload(bytes::Bytes::from_static(b"d")), + } + .encode_bytes(Version::Draft19) + .unwrap() + }; + + subscriber.recv_datagram(datagram(4)).unwrap(); + subscriber.register_alias(RequestId(11), 7).unwrap(); + subscriber.recv_datagram(datagram(5)).unwrap(); + + let received = consumer.recv_datagram().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(received.sequence, 5); + assert!( + consumer.recv_datagram().now_or_never().is_none(), + "group 4 never arrives" + ); + } + /// One alias naming two different tracks is the collision section 11.1 makes fatal. #[test] fn an_alias_reused_for_another_track_is_fatal() { @@ -8169,6 +8258,181 @@ mod stitch_tests { } } + /// A group's first fetch Object, with `group` as the wire's Group ID field. + fn first_object(subgroup: ietf::FetchSubgroup, group: u64, payload: &[u8]) -> Vec { + fetch_object(subgroup, Some(group), payload) + } + + /// The next fetch Object of the prior one's group, inheriting its Group ID. + fn next_object(subgroup: ietf::FetchSubgroup, payload: &[u8]) -> Vec { + fetch_object(subgroup, None, payload) + } + + /// A stamped fetch Object: a group's first when `group` names it, else the next one. + fn fetch_object(subgroup: ietf::FetchSubgroup, group: Option, payload: &[u8]) -> Vec { + let mut properties = Vec::new(); + let w = &mut Encoder::new(&mut properties, VERSION.into()); + ietf::encode_object_time(w, timestamp(0), Timescale::MICRO, VERSION).unwrap(); + let mut buf = Vec::new(); + ietf::FetchObject::Object { + subgroup, + group, + object: group.map(|_| 0), + priority: group.map(|_| 0), + properties: Some(properties), + } + .encode(&mut crate::coding::Encoder::new(&mut buf, VERSION.into()), VERSION) + .unwrap(); + crate::coding::Encoder::new(&mut buf, VERSION.into()) + .varint(payload.len() as u64) + .unwrap(); + buf.extend_from_slice(payload); + buf + } + + /// One Object published as a datagram, as draft-16 on lets a fetch stream carry it. + fn datagram_object(sequence: u64) -> Vec { + first_object(ietf::FetchSubgroup::Datagram, sequence, b"d") + } + + /// An absolute join spans whole groups, so a datagram among them is skipped: the group + /// before it is complete, and the groups after it still fill and stitch into the tail. + #[moq_net_sim::test] + async fn an_absolute_join_skips_a_datagram_group() { + const START: u64 = 7; + const LIVE_GROUP: u64 = 9; + let largest = ietf::Location { + group: LIVE_GROUP, + object: 0, + }; + // Group 8 went out as two datagram objects, the second inheriting its Group ID. From + // draft-18 on a later Group ID is an ascending delta, so 7, 8, 9 is 7, 0, 0. + let mut fill = fill_stream_for(FETCH, &[(START, &[b"g7-0"])]); + fill.extend(first_object(ietf::FetchSubgroup::Datagram, 0, b"g8-0")); + fill.extend(next_object(ietf::FetchSubgroup::Datagram, b"g8-1")); + fill.extend(first_object(ietf::FetchSubgroup::Zero, 0, b"g9-0")); + + let h = Harness::new( + Fill::Serving(Some(Timescale::MICRO)), + vec![fill, tail_stream(LIVE_GROUP, 1, &[b"g9-1"])], + ) + .with_joining(JoiningFetch::Absolute { group_id: START }, FETCH, largest); + let mut consumer = h + .track + .subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(60))); + + let mut fill = h.stream().await; + let mut tail = h.stream().await; + h.subscriber.clone().recv_fill(&mut fill).await.expect("absolute fetch"); + h.subscriber.clone().recv_group(&mut tail).await.expect("tail"); + + let mut groups = Vec::new(); + for _ in 0..2 { + let (sequence, frames) = read_group(&mut consumer).await; + groups.push(( + sequence, + frames.into_iter().map(|(_, payload)| payload).collect::>(), + )); + } + assert_eq!( + groups, + vec![ + (START, vec![b"g7-0".to_vec()]), + (LIVE_GROUP, vec![b"g9-0".to_vec(), b"g9-1".to_vec()]), + ] + ); + } + + /// A datagram Object inside the head's own group mixes stream and datagram objects, which + /// no group can hold. The join fails as not fetchable and drops that group rather than + /// finishing it partway. + #[moq_net_sim::test] + async fn an_absolute_join_refuses_a_datagram_in_the_head() { + const START: u64 = 7; + let largest = ietf::Location { group: 9, object: 0 }; + let mut fill = fill_stream_for(FETCH, &[(START, &[b"g7-0"])]); + fill.extend(next_object(ietf::FetchSubgroup::Datagram, b"g7-1")); + + let h = Harness::new(Fill::Serving(Some(Timescale::MICRO)), vec![fill]).with_joining( + JoiningFetch::Absolute { group_id: START }, + FETCH, + largest, + ); + let mut consumer = h + .track + .subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(60))); + + let mut fill = h.stream().await; + assert!(matches!( + h.subscriber.clone().recv_fill(&mut fill).await, + Err(Error::NotFetchable) + )); + + assert!( + futures::FutureExt::now_or_never(consumer.recv_group()).is_none(), + "the partial head must not be published as a complete group" + ); + } + + /// A datagram group is never fetchable, so a group fetch answered with one fails as + /// not fetchable, and nothing is cached. + #[moq_net_sim::test] + async fn a_group_fetch_refuses_a_datagram_object() { + let mut run = GroupFetchRun::new(VERSION, datagram_object(SEQUENCE)).await; + let group = run.track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); + let mut consumer = group.consume(); + let slot = kio::Producer::new(GroupFetch::Ready { + producer: group, + timescale: None, + start: 0, + end: None, + }); + + let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; + assert!(matches!(res, Err(Error::NotFetchable)), "{res:?}"); + assert!( + matches!(consumer.read_frame().await, Err(Error::NotFetchable)), + "the group fails without the payload" + ); + } + + /// A fill answered with a datagram Object is refused the same way. Only the fill goes: + /// the subscription keeps delivering the groups after it. + #[moq_net_sim::test] + async fn a_datagram_fill_fails_only_the_fill() { + let mut fill = Vec::new(); + crate::coding::Encoder::new(&mut fill, VERSION.into()) + .varint(ietf::FetchHeader::TYPE) + .unwrap(); + ietf::FetchHeader { request_id: REQUEST } + .encode(&mut crate::coding::Encoder::new(&mut fill, VERSION.into()), VERSION) + .unwrap(); + fill.extend(datagram_object(SEQUENCE)); + + let h = Harness::new( + Fill::Serving(Some(Timescale::MICRO)), + vec![fill, tail_stream(SEQUENCE + 1, 0, &[b"next"])], + ); + let mut consumer = h.track.subscribe(None); + let mut fill = h.stream().await; + let mut tail = h.stream().await; + + assert!(matches!( + h.subscriber.clone().recv_fill(&mut fill).await, + Err(Error::NotFetchable) + )); + h.subscriber + .clone() + .recv_group(&mut tail) + .await + .expect("the subscription carries on"); + + let (sequence, frames) = read_group(&mut consumer).await; + assert_eq!(sequence, SEQUENCE + 1); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].1, b"next"); + } + /// A group fetch's objects after the FETCH_HEADER: the first one names the group and /// `start`, and every later one is the next object. fn group_fetch_objects(sequence: u64, start: u64, payloads: &[&[u8]]) -> Vec { diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index 727035b7a1..fe1c15b04a 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -227,6 +227,17 @@ impl Version { } } + /// Whether the stream error table has NOT_FETCHABLE, for a FETCH that reached a + /// datagram. Added in lite-07; earlier versions say NOT_FOUND instead. + #[allow(clippy::match_like_matches_macro)] + pub(crate) fn has_not_fetchable(self) -> bool { + // Match form so future versions default forward (AGENTS.md convention). + match self { + Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 | Self::Lite05 | Self::Lite06 => false, + _ => true, + } + } + /// Whether announcements and track requests carry the publisher epoch, the /// identity that says two routes serve the same bytes. Added in lite-07. /// Older versions carry nothing, so a received route has no epoch and is diff --git a/rs/moq-net/src/model/datagram.rs b/rs/moq-net/src/model/datagram.rs index 7f92e0db3d..04fca1e78a 100644 --- a/rs/moq-net/src/model/datagram.rs +++ b/rs/moq-net/src/model/datagram.rs @@ -10,6 +10,9 @@ //! body exceeds the transport's datagram size, and sessions that can't carry datagrams at all //! (moq-lite before 05, or stream-only transports like WebSocket) never deliver them. //! +//! A datagram is never cached or served by a fetch. Only the short per-track send buffer holds +//! it, which a new subscription may still read. +//! //! Wire counterparts: [`crate::lite::Datagram`], and on moq-transport an OBJECT_DATAGRAM at //! object 0 whose Group ID is the sequence ([`crate::ietf::ObjectDatagram`]). diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 77a7465e5f..017dc99d92 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -513,6 +513,11 @@ impl TrackState { } } + /// Whether `sequence` was sent as a datagram still in the send buffer. + fn holds_datagram(&self, sequence: u64) -> bool { + self.datagrams.iter().any(|datagram| datagram.sequence == sequence) + } + /// Push a datagram, dropping the oldest when the send buffer is full. fn push_datagram(&mut self, datagram: Datagram) { if self.datagrams.len() == MAX_DATAGRAMS { @@ -1546,9 +1551,10 @@ impl Producer { /// [`Self::append_group`] never reuses a number). There is no group fallback: each /// session drops (with a debug log) any datagram whose encoded body exceeds the /// transport's datagram size, and sessions that can't carry datagrams at all (moq-lite - /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. An origin - /// publisher uses this; a relay preserving upstream numbering uses - /// [`Self::insert_datagram`]. + /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. + /// A datagram is never cached or served by a fetch, so use a group for anything a late + /// joiner needs. An origin publisher uses this; a relay preserving upstream numbering + /// uses [`Self::insert_datagram`]. /// /// Pass `None` on an untimed track; a datagram that doesn't match its track is /// [`Error::TimestampMismatch`]. @@ -2753,8 +2759,9 @@ impl Consumer { /// or `group::Fetch::default()`. /// /// The returned future resolves to [`Error::NotFound`] when the group can never be served - /// (past the final sequence, or no [`Dynamic`] on the track), the handler's rejection - /// (a relay's upstream miss is [`StreamError::NotFound`](crate::StreamError::NotFound)), + /// (past the final sequence, or no [`Dynamic`] on the track), [`Error::NotFetchable`] + /// instead when this track sent the sequence as a datagram still buffered, the handler's + /// rejection (a relay's upstream miss is [`StreamError::NotFound`](crate::StreamError::NotFound)), /// or the track's abort error if it's already closed. Concurrent fetches for the same sequence coalesce onto one /// handler request. pub fn fetch_group(&self, sequence: u64, options: impl Into>) -> kio::Pending { @@ -3221,11 +3228,15 @@ impl kio::Task for Fetching { let Some(result) = result else { // Never queued: no handler existed when the fetch was made. Fail fast while // that's still true; a handler that appeared since may yet fill the cache. + // A sequence this track sent as a datagram says so, rather than a plain miss. return match fetch.poll(waiter, |fetch| match fetch.has_handlers() { false => Poll::Ready(()), true => Poll::Pending, }) { - Poll::Ready(_guard) => Poll::Ready(Err(Error::NotFound)), + Poll::Ready(_guard) => Poll::Ready(Err(match state.read().holds_datagram(sequence) { + true => Error::NotFetchable, + false => Error::NotFound, + })), Poll::Pending => Poll::Pending, }; }; @@ -3638,15 +3649,34 @@ impl Cursor { } } + /// The next datagram inside this subscriber's range, dropping the ones outside it. + /// + /// The range is the one groups get: the floor, the cap, and a start frame past 0, which + /// excludes the start group's only frame. A datagram is judged when the cursor reaches + /// it, against the range in force then, and one outside it is gone for good: unlike a + /// group past the cap, nothing holds it for a later raise. fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { - let Some((datagram, found_index)) = - ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?) - else { - return Poll::Ready(Ok(None)); - }; - - self.datagram_index = found_index + 1; - Poll::Ready(Ok(Some(datagram))) + loop { + let Some((datagram, found_index)) = + ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?) + else { + return Poll::Ready(Ok(None)); + }; + self.datagram_index = found_index + 1; + + let sequence = datagram.sequence; + let mid_group = self + .subscription + .read() + .start + .is_some_and(|start| start.group == sequence && start.frame > 0); + if sequence >= self.min_sequence + && super::subscription::before_end(sequence, self.end_sequence) + && !mid_group + { + return Poll::Ready(Ok(Some(datagram))); + } + } } /// The lowest servable group past the last one returned, walking off everything the @@ -3867,8 +3897,9 @@ impl Subscriber { /// /// Datagrams are a separate best-effort channel from groups (see /// [`Producer::append_datagram`]); they share only the sequence namespace, and - /// neither cursor moves the other. A consumer that falls too far behind silently - /// loses the oldest datagrams. + /// neither cursor moves the other. A new subscriber may get the few still in the send + /// buffer, and any outside its group range are skipped. A consumer that falls too far + /// behind silently loses the oldest datagrams. /// /// Returns `Poll::Ready(Ok(Some(datagram)))` when one is available, /// `Poll::Ready(Ok(None))` when the track is finished, `Poll::Ready(Err(e))` when the track @@ -4810,6 +4841,94 @@ mod test { assert_eq!(&recv_datagram(&mut dg).payload[..], b"go"); } + /// A fetch never serves a datagram group, and says so rather than reporting a plain + /// miss, whether or not a group followed it. + #[test] + fn a_datagram_group_is_not_fetchable() { + let mut producer = track_producer("test", None); + let consumer = producer.consume(); + let _subscriber = producer.subscribe(None); + let sequence = producer.append_datagram(Timestamp::ZERO, &b"x"[..]).unwrap(); + let fetch = |sequence| consumer.fetch_group(sequence, None).now_or_never(); + assert!( + matches!(fetch(sequence), Some(Err(Error::NotFetchable))), + "the newest sequence" + ); + + producer.append_group().unwrap(); + assert!( + matches!(fetch(sequence), Some(Err(Error::NotFetchable))), + "below a newer group" + ); + // A sequence that was never a datagram is a plain miss. + assert!( + matches!(fetch(sequence + 2), Some(Err(Error::NotFound))), + "a missing group" + ); + } + + /// The subscription's floor and cap bound datagrams as they bound groups. The datagrams + /// in range, sent between the dropped ones, show the reader is live, so the range is + /// what dropped the others. + #[test] + fn the_group_range_bounds_datagrams() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(5))); + subscriber.set_groups(..7); + for sequence in [4, 5, 7, 6] { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + assert_eq!(recv_datagram(&mut subscriber).sequence, 5); + assert_eq!(recv_datagram(&mut subscriber).sequence, 6); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + } + + /// A start past the start group's first frame excludes that group's datagram, its only + /// frame; a start at frame 0 keeps it. + #[test] + fn a_mid_group_start_drops_the_start_groups_datagram() { + let mut producer = track_producer("test", None); + let mut mid = producer.subscribe(Subscription::default().with_start(Position { group: 5, frame: 1 })); + let mut whole = producer.subscribe(Subscription::default().with_start(Position::group(5))); + for sequence in [5, 6] { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + assert_eq!(recv_datagram(&mut mid).sequence, 6); + assert_eq!(recv_datagram(&mut whole).sequence, 5); + assert_eq!(recv_datagram(&mut whole).sequence, 6); + } + + /// A range update judges the datagrams still buffered when they are read, and one it + /// dropped stays dropped once the cap rises again: nothing holds it like a parked group. + #[test] + fn a_range_update_applies_to_buffered_datagrams() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(None); + for sequence in 1..=4 { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + subscriber.set_groups(2..4); + assert_eq!(recv_datagram(&mut subscriber).sequence, 2); + assert_eq!(recv_datagram(&mut subscriber).sequence, 3); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + + subscriber.set_groups(..); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + producer + .insert_datagram(5, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + assert_eq!(recv_datagram(&mut subscriber).sequence, 5); + } + /// Exercises the full producer -> publisher-encode -> subscriber-decode -> producer seam /// (everything but the QUIC datagram send/recv), catching any field-order mismatch between /// the wire codec and the model. diff --git a/rs/moq-net/tests/datagram.rs b/rs/moq-net/tests/datagram.rs index 72ad4001a2..4f7c3e45b4 100644 --- a/rs/moq-net/tests/datagram.rs +++ b/rs/moq-net/tests/datagram.rs @@ -10,7 +10,7 @@ mod support; use std::time::Duration; use moq_net::{Hop, Timestamp, Version}; -use support::harness::{MockConnectOptions, MockPair, connect_mock}; +use support::harness::{MockConnectOptions, MockPair, connect_mock, peer}; /// Maximum time any single test may run before being treated as a deadlock. const TEST_TIMEOUT: Duration = Duration::from_secs(10); @@ -226,3 +226,53 @@ async fn inserted_sequences_survive_the_lite_wire() { .await .expect("timed out"); } + +/// A FETCH for a sequence the publisher sent as a datagram is refused with NOT_FETCHABLE on +/// moq-lite-07, and with the NOT_FOUND an earlier version has for it. A relay carries the +/// answer across in the downstream version's terms. The moq-transport side (DOES_NOT_EXIST) +/// is covered by the IETF publisher's own tests. +#[moq_net_sim::test] +async fn a_fetched_datagram_is_refused_per_version() { + const NOT_FETCHABLE: &str = "not fetchable"; + const NOT_FOUND: &str = "not found"; + + /// Fetch a datagram's sequence over one hop per version, publisher first. + async fn fetch_a_datagram(versions: &[&str]) -> &'static str { + let nodes: Vec<_> = (1..=versions.len() as u64 + 1).map(produce_origin).collect(); + let mut _pairs = Vec::new(); + for (pair, version) in nodes.windows(2).zip(versions) { + _pairs.push(peer(version.parse().unwrap(), &pair[0], &pair[1]).await); + } + + let broadcast = nodes[0].create_broadcast("room").unwrap(); + let mut producer = broadcast.create_track("datagrams", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let sequence = producer.append_datagram(Timestamp::ZERO, &b"x"[..]).unwrap(); + + let consumer = nodes.last().unwrap().consume(); + consumer.routed("room").await.unwrap(); + let remote = consumer.request_broadcast("room").await.unwrap(); + let err = match remote.track("datagrams").unwrap().fetch_group(sequence, None).await { + Ok(_) => panic!("{versions:?}: a datagram was fetched"), + Err(err) => err, + }; + match err { + moq_net::Error::Stream(moq_net::StreamError::NotFetchable) => NOT_FETCHABLE, + moq_net::Error::NotFound | moq_net::Error::Stream(moq_net::StreamError::NotFound) => NOT_FOUND, + err => panic!("{versions:?}: refused with {err:?}"), + } + } + + for (versions, expected) in [ + (&["moq-lite-05"][..], NOT_FOUND), + (&["moq-lite-06"], NOT_FOUND), + (&["moq-lite-07-wip"], NOT_FETCHABLE), + (&["moq-lite-07-wip", "moq-lite-07-wip"], NOT_FETCHABLE), + (&["moq-lite-07-wip", "moq-lite-06"], NOT_FOUND), + ] { + let refused = moq_net_sim::timeout(TEST_TIMEOUT, fetch_a_datagram(versions)) + .await + .unwrap_or_else(|_| panic!("{versions:?}: timed out")); + assert_eq!(refused, expected, "{versions:?}"); + } +} diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 2163c23ee6..de85a5dba4 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -969,8 +969,13 @@ async fn serve_fetch( }; let group = match async { crate::fetch_group(&broadcast.track(&track)?, sequence).await }.await { Ok(group) => group, - // A miss upstream arrives as the stream reset that refused the FETCH. - Err(moq_net::Error::NotFound | moq_net::Error::Stream(moq_net::StreamError::NotFound)) => { + // A miss upstream arrives as the stream reset that refused the FETCH. A datagram + // group is never cached, so it is a miss too. + Err( + moq_net::Error::NotFound + | moq_net::Error::NotFetchable + | moq_net::Error::Stream(moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable), + ) => { return Err(StatusCode::NOT_FOUND); } Err(_) => return Err(StatusCode::INTERNAL_SERVER_ERROR),