diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 97d777e363..1ecede7dda 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -26,7 +26,7 @@ maps everything else to "not supported" or a harmless equivalent. The subset drops. What a peer actually observes against this implementation: - **Pull.** Subscribers ask. Single-track `PUBLISH` offers are declined; announce a namespace and serve the resulting subscriptions. Announcements go out unsolicited, and we also ask for every prefix we may discover. The solicit `SETUP` option makes us wait to be asked. Without it, a draft 16+ peer also gets each match as a `NAMESPACE` on its `SUBSCRIBE_NAMESPACE` stream, so it hears the namespace twice. We never send `PUBLISH`: on drafts 16 and 17 a `SUBSCRIBE_NAMESPACE` asking only for `PUBLISH` is refused, one asking for both gets only `NAMESPACE`, and any other Subscribe Options value closes the session. -- **History is one group.** On drafts 14 through 19 a `FETCH` returns one whole group from the cache, or the saved prefix of the group a new subscription just joined. A range of groups is refused, as is any `FETCH` on draft 20 and later. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed. +- **History is one group.** A `FETCH` returns one group from the cache, or on drafts 14 through 19 the saved prefix of the group a new subscription just joined. A range of groups is refused, as is a draft 20+ `LOCATION_FILTER` that ends at Largest Object rather than an absolute End Location. On draft 20 and later, `FETCH_OK`'s End Location is the requested end, capped at Largest Object: a range holding no objects gets an empty fetch stream, and only a start past Largest Object is refused, with `INVALID_RANGE`. A relay's copy with no upstream subscription can't know Largest Object short of the track's end, so it echoes the requested end instead. JavaScript publishing refuses every `FETCH`. Datagrams are never fetchable. A Rust reader that only fetches learns the track from `TRACK_STATUS` instead of subscribing, so a finished track stays fetchable; draft 17 still subscribes, since its `TRACK_STATUS` answer cannot say whether the track is timed. - **Timing.** A track whose `SUBSCRIBE_OK` declares no `TIMESCALE` is untimed, as is every track on drafts 14 through 16. A Rust reader that never subscribed takes the units from `TRACK_STATUS` on draft 18 and later. - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session. Message parameters follow the negotiated draft's lists: drafts 14 and 15 ignore an unknown or misplaced parameter, draft 16 ignores one defined only for another message but closes the session on an unknown one, and draft 17 and later close it on both. - **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A token attached to an individual request is ignored. An alias reference or a second token closes the session. diff --git a/quest/m1/README.md b/quest/m1/README.md index 620bdfa56b..cea2a787c2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -53,7 +53,6 @@ were done or not worth their cost. - [Bump web-transport-iroh for the capsule close](/quest/m1/iroh-capsule-bump.md) - condition: moq-dev/web-transport#419 ships in a release, then moq's iroh HTTP/3 client reports a peer's close capsule - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it - [lite-07 Live flag](/quest/m1/lite-live.md) - a separate `Live` field on lite-07 SUBSCRIBE, so merged floors never starve a subscriber; late lower groups build on it -- [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - a draft-20+ FETCH within one group is served from its LOCATION_FILTER, as older drafts are - [FETCH_OK properties](/quest/m1/fetch-ok-properties.md) - our FETCH_OK carries the track properties SUBSCRIBE_OK does, as draft 16+ requires - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport - [Cross-relay bursts re-run](/quest/m1/cross-relay-bursts.md) - condition: the #4349 reporter re-runs their A/B/C comparison against current cdn.moq.pro diff --git a/quest/m1/fetch-ok-properties.md b/quest/m1/fetch-ok-properties.md index 35364c36af..95bb2acde0 100644 --- a/quest/m1/fetch-ok-properties.md +++ b/quest/m1/fetch-ok-properties.md @@ -32,9 +32,7 @@ opts out with INCLUDE_PROPERTIES = 0. Other fetch-only readers still need it. subscriber takes the units only from SUBSCRIBE_OK, and a fetch-only reader is untimed. Rejected: omitting timescale from FETCH_OK and leaving fetch-only readers untimed. -- Honour INCLUDE_PROPERTIES (0x35) on FETCH from draft 20, once - [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) serves draft-20 FETCH at - all (today our publisher refuses every one). It defaults to sending the +- Honour INCLUDE_PROPERTIES (0x35) on FETCH from draft 20. It defaults to sending the properties; at 0 the block is present but empty. `fetch.rs` decodes it into `Fetch::properties_wanted`, which the publisher does not honour yet. - Test per draft range: FETCH_OK round-trips the properties SUBSCRIBE_OK would @@ -46,8 +44,3 @@ change how a fetch-only reader learns the track's properties. Public API: none. Wire: FETCH_OK gains its properties on drafts that define the block. Interop: run `just test interop --all`. - -## Required - -- [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - draft-20 FETCH is served at all, and edits the same `run_fetch_stream` - diff --git a/quest/m1/ietf-fetch-location.md b/quest/m1/ietf-fetch-location.md deleted file mode 100644 index 944d3f479a..0000000000 --- a/quest/m1/ietf-fetch-location.md +++ /dev/null @@ -1,26 +0,0 @@ -# [S] Draft-20 FETCH - -## Goal - -On drafts 20 and later, a FETCH whose LOCATION_FILTER stays within one group -is answered the way drafts 14 to 19 answer a standalone FETCH: from cache, with -a miss fetched upstream, and an upstream refusal passed through. A range -touching several groups follows the same group-span rule as drafts 14 to 19, -through the shared path: refused `NOT_SUPPORTED` until -[moq-transport ranges](/quest/m1/subscribe-ranges/ietf.md) lifts that -refusal for every draft. - -## Plan - -Today both sides refuse FETCH on draft-20+ `NOT_SUPPORTED`: the publisher in -`run_fetch_stream`, and the relay's upstream group fill in `run_group_fetch`. -The codec already decodes and encodes the draft-20 layout as -`FetchType::Filtered`, so drop both refusals, and flip -`fetch_moq_transport_20` in `rs/moq-tokio/tests` back to a served fetch. A -FETCH carrying Range Filters stays refused. - -Update the draft-20 note in `doc/concept/standard.md`. - -## Related - -- [moq-transport ranges](/quest/m1/subscribe-ranges/ietf.md) - lifts the one-group refusal in `run_fetch_stream` for every draft, draft-20 included diff --git a/quest/m1/subscribe-ranges/ietf.md b/quest/m1/subscribe-ranges/ietf.md index 556d3130d5..df78bd37c5 100644 --- a/quest/m1/subscribe-ranges/ietf.md +++ b/quest/m1/subscribe-ranges/ietf.md @@ -21,10 +21,10 @@ SUBSCRIBE, as open #4974 does for fetch-only demand. Serving downstream lifts the one-group refusal ("FETCH spanning several groups not supported") in `run_fetch_stream` (`rs/moq-net/src/ietf/publisher.rs`) for every draft, including draft-20's -`FetchType::Filtered` once [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) -serves it. +`FetchType::Filtered`. Capping at the Largest Object also lifts the draft-20 +refusal of a filter bounded by it ("FETCH relative to Largest Object not +supported"): no filter, a relative start, or an absolute start with no end. ## Required - [Model ranges](/quest/m1/subscribe-ranges/model.md) - the range requests this answers -- [Draft-20 FETCH](/quest/m1/ietf-fetch-location.md) - serves draft-20 FETCH through the same group-span path this widens (#4971) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index b2ea00c2f8..fc841391b1 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1336,8 +1336,8 @@ where } } - /// Answer a FETCH within one group: a standalone range of the named track, or a - /// joining FETCH's prefix of its subscription's group. + /// Answer a FETCH within one group: a standalone or filtered range of the named track, + /// or a joining FETCH's prefix of its subscription's group. /// /// The answer is buffered before replying: FETCH_OK names where the response ends, /// which a range running past the track only learns by reading it, and a refusal can @@ -1345,19 +1345,6 @@ where async fn run_fetch_stream(mut self, mut stream: Stream, msg: ietf::Fetch<'_>) -> Result<(), Error> { let priority = super::priority::from_wire(msg.subscriber_priority); - // Draft-20 moved the range into LOCATION_FILTER, which decodes to - // `FetchType::Filtered` but is not served yet. - if Filter::is_draft20(self.version) { - return self - .reject_fetch( - stream, - msg.request_id, - &Error::Unsupported, - "FETCH not supported on draft-20", - ) - .await; - } - // Serving a Range Filter unfiltered would deliver objects the subscriber excluded. if msg.range_filters { return self @@ -1383,7 +1370,46 @@ where .await; } - let (track, start, end, timescale, joined) = match msg.fetch_type { + // Draft-20's LOCATION_FILTER is answered as the standalone range it spells. + let fetch_type = match msg.fetch_type { + FetchType::Filtered { + namespace, + track, + filter: Filter::Absolute { start, end: Some(end) }, + } => { + // The filter's End Object is inclusive, where a standalone one is the last + // object plus one. Without one, both include the whole End Group. + let Some(object) = end.object.map_or(Some(0), |object| object.checked_add(1)) else { + return self + .reject_fetch(stream, msg.request_id, &Error::InvalidRange, "End Object overflows") + .await; + }; + FetchType::Standalone { + namespace, + track, + start, + end: Location { + group: end.group, + object, + }, + } + } + // Every other filter ends at Largest Object, so the one-group rule below cannot + // check it without resolving that first. + FetchType::Filtered { .. } => { + return self + .reject_fetch( + stream, + msg.request_id, + &Error::Unsupported, + "FETCH relative to Largest Object not supported", + ) + .await; + } + other => other, + }; + + let (track, start, end, timescale, joined) = match fetch_type { FetchType::Standalone { namespace, track, @@ -1425,7 +1451,7 @@ where Ok(joined) => joined, Err((err, reason)) => return self.reject_fetch(stream, msg.request_id, &err, reason).await, }; - let start = match msg.fetch_type { + let start = match fetch_type { FetchType::RelativeJoining { group_offset, .. } => end.group.saturating_sub(group_offset), FetchType::AbsoluteJoining { group_id, .. } if group_id <= end.group => group_id, _ => { @@ -1450,7 +1476,7 @@ where true, ) } - // Refused above: only draft-20 decodes this form. + // Rewritten as standalone or refused above. FetchType::Filtered { .. } => { return self .reject_fetch(stream, msg.request_id, &Error::Unsupported, "not supported") @@ -1495,6 +1521,37 @@ where Some(Err(err)) => return self.reject_fetch(stream, msg.request_id, &err, &err.to_string()).await, None => return Ok(()), }; + let draft20 = Filter::is_draft20(self.version); + let end_of_track = !joined && group.complete && track.final_sequence() == group.sequence.checked_add(1); + // Draft-20 caps the response at Largest Object, and refuses a start past it (section + // 10.13). Both only bite when Largest Object is in the fetched group or behind it, so + // this is its Object ID there, or `Some(None)` when the group holds nothing at or + // past it. The cache knows it at the track's end, or when a live feed's newest object + // it can name is in or behind this group. Otherwise (a relay's copy with no upstream + // subscription, or a newest group it cannot read) it neither caps nor refuses: the + // read waits out an unfinished group, so echoing the requested end over a finished + // one only says objects it never held do not exist. + let largest = match end_of_track { + true => Some(group.end().checked_sub(1)), + false if !joined && track.is_live() => match live_edge(&track).largest { + Some(largest) if largest.group == group.sequence => Some(Some(largest.object)), + // Behind a group that holds nothing: every start in it is past Largest Object. + Some(largest) if largest.group < group.sequence && group.frames.is_empty() => Some(None), + _ => None, + }, + false => None, + }; + let past_largest = largest.is_some_and(|largest| largest.is_none_or(|object| start.object > object)); + if draft20 && past_largest { + return self + .reject_fetch( + stream, + msg.request_id, + &Error::InvalidRange, + "start past Largest Object", + ) + .await; + } // A standalone FETCH keeps each object's Timestamp, in the track's own units, as a // subscriber learned them from SUBSCRIBE_OK. FETCH_OK doesn't declare them yet. @@ -1515,7 +1572,6 @@ where } (end, false) } else { - let end_of_track = group.complete && track.final_sequence() == group.sequence.checked_add(1); // A group that ends the track ends the response at its last object; otherwise // the response ends where it was asked to. match end_of_track { @@ -1529,6 +1585,27 @@ where false => (end, false), } }; + // Draft-20's End Location is inclusive: the requested end, capped at Largest Object + // (section 10.14). Objects missing before it do not exist. A whole-group request has + // no End Object to report, so it covers what the finished group holds, and at least + // the start an empty answer covered. + let end_location = match draft20 { + true => { + let requested = match until { + Some(until) => until - 1, + None => group.end().saturating_sub(1).max(start.object), + }; + let object = match largest.flatten() { + Some(largest) => requested.min(largest), + None => requested, + }; + Location { + group: group.sequence, + object, + } + } + false => end_location, + }; // FETCH_OK on every draft, never REQUEST_OK: section 5.2 allows exactly one FETCH_OK or // REQUEST_ERROR in answer to a FETCH, and REQUEST_OK's own definition lists the other @@ -4919,33 +4996,29 @@ mod serve_tests { } } - #[moq_net_sim::test] - async fn a_joining_fetch_is_not_supported_on_draft20() { - let h = serve(Version::Draft20); - let mut buf = bytes::Bytes::from(joining_fetch(&h, 0).await.unwrap()); - assert_eq!( - crate::coding::decode_varint(&mut buf, Version::Draft20).unwrap(), - ietf::RequestError::ID - ); - assert_eq!( - crate::coding::decode_buf(&mut buf, Version::Draft20, ietf::RequestError::decode) - .unwrap() - .error_code, - 0x3 - ); - assert!(buf.is_empty()); - } - - /// Every draft that carries a standalone FETCH. - const FETCH_DRAFTS: [Version; 6] = [ + /// Every draft, each carrying a standalone FETCH or draft-20's filtered one. + const FETCH_DRAFTS: [Version; 9] = [ Version::Draft14, Version::Draft15, Version::Draft16, Version::Draft17, Version::Draft18, Version::Draft19, + Version::Draft20, + Version::Draft21, + Version::Draft22, ]; + /// FETCH_OK's End Location for a response whose last object is `last`: one past it + /// before draft 20, and the object itself from then on. + fn end_location(version: Version, group: u64, last: u64) -> Location { + let object = match Filter::is_draft20(version) { + true => last, + false => last + 1, + }; + Location { group, object } + } + /// Groups `0..count`, each holding `g-0` and `g-1`, skipping `hole`. fn publish_pairs(h: &mut Serve, count: u64, hole: Option) { for sequence in (0..count).filter(|sequence| Some(*sequence) != hole) { @@ -4960,7 +5033,36 @@ mod serve_tests { } /// Run a standalone FETCH of `room/video`, returning what the peer reads back. + /// + /// `end` is spelled as drafts 14 to 19 do: the last object plus one, or 0 for the whole + /// End Group. From draft 20 the same range goes out as an inclusive LOCATION_FILTER. async fn standalone_fetch(h: &Serve, start: Location, end: Location, group_order: GroupOrder) -> bytes::Bytes { + let namespace = crate::Path::new("room"); + let track = "video".into(); + let fetch_type = match Filter::is_draft20(h.publisher.version) { + true => FetchType::Filtered { + namespace, + track, + filter: Filter::Absolute { + start, + end: Some(EndLocation { + group: end.group, + object: end.object.checked_sub(1), + }), + }, + }, + false => FetchType::Standalone { + namespace, + track, + start, + end, + }, + }; + run_fetch(h, fetch_type, group_order).await + } + + /// Run a FETCH, returning what the peer reads back. + async fn run_fetch(h: &Serve, fetch_type: FetchType<'static>, group_order: GroupOrder) -> bytes::Bytes { let version = h.publisher.version; let mark = h.log.writes.lock().unwrap().len(); let stream = Stream::open(&mut h.session.clone(), version).await.unwrap(); @@ -4972,12 +5074,7 @@ mod serve_tests { request_id: FETCH_ID, subscriber_priority: 128, group_order, - fetch_type: FetchType::Standalone { - namespace: crate::Path::new("room"), - track: "video".into(), - start, - end, - }, + fetch_type, range_filters: false, fill_timeout: false, properties_wanted: true, @@ -5058,53 +5155,279 @@ mod serve_tests { ) .await; let (ok, objects) = fetch_answer(buf, version); - assert_eq!(ok.end_location, Location { group: 3, object: 0 }, "{version}"); + let end = match Filter::is_draft20(version) { + true => Location { group: 2, object: 1 }, + false => Location { group: 3, object: 0 }, + }; + assert_eq!(ok.end_location, end, "{version}"); assert!(!ok.end_of_track, "{version}"); assert_eq!(objects, pairs([2]), "{version}"); assert!(h.log.resets().is_empty(), "{version}"); } } - /// The last group of a finished track ends the response one past its last object, - /// with End of Track set. + /// A range ending inside its group stops at its End Object, which draft 20 counts + /// inclusively and older drafts count as one past. #[moq_net_sim::test] - async fn a_standalone_fetch_of_the_last_group_reports_the_end_of_track() { + async fn a_standalone_fetch_stops_at_its_end_object() { for version in FETCH_DRAFTS { let mut h = serve(version); publish_pairs(&mut h, 5, None); - h.track.finish().unwrap(); settle().await; let buf = standalone_fetch( &h, - Location { group: 4, object: 1 }, - Location { group: 4, object: 0 }, - GroupOrder::Any, + Location { group: 2, object: 0 }, + Location { group: 2, object: 1 }, + GroupOrder::Ascending, ) .await; let (ok, objects) = fetch_answer(buf, version); - assert_eq!(ok.end_location, Location { group: 4, object: 2 }, "{version}"); - assert!(ok.end_of_track, "{version}"); - assert_eq!(objects, pairs([4])[1..].to_vec(), "{version}"); + assert_eq!(ok.end_location, end_location(version, 2, 0), "{version}"); + assert!(!ok.end_of_track, "{version}"); + assert_eq!(objects, pairs([2])[..1].to_vec(), "{version}"); + } + } + + /// A group holding no objects at or past the start is answered empty, on draft 20 with + /// an End Location covering the range asked for (section 10.13). The whole group covers + /// at least its start. + #[moq_net_sim::test] + async fn a_standalone_fetch_past_the_last_object_is_empty() { + for version in FETCH_DRAFTS { + for (end, covered) in [(0, 2), (6, 5)] { + let mut h = serve(version); + publish_pairs(&mut h, 5, None); + settle().await; + + let buf = standalone_fetch( + &h, + Location { group: 2, object: 2 }, + Location { group: 2, object: end }, + GroupOrder::Ascending, + ) + .await; + let (ok, objects) = fetch_answer(buf, version); + assert_eq!(objects, Vec::new(), "{version} end={end}"); + if Filter::is_draft20(version) { + assert_eq!( + ok.end_location, + Location { + group: 2, + object: covered + }, + "{version}" + ); + } + } + } + } + + /// How much a draft-20 publisher knows of Largest Object. + #[derive(Clone, Copy, Debug)] + enum Largest { + /// The track is finished: its last object. + Finished, + /// A live feed: the newest object, here behind a newer group with no objects yet. + Live, + /// A relay's copy with no upstream subscription: unknown. + Idle, + /// An idle copy that learned a later end of track: still unknown. + IdleEnded, + /// A live feed whose newest group is a datagram the cache cannot read: unknown. + Datagram, + /// A live feed whose newest group was aborted mid-write: unknown. + Aborted, + } + + /// Groups 0 to 4 of two objects each, with Largest Object known as `largest` says. + fn publish_largest(h: &mut Serve, largest: Largest) -> Option { + publish_pairs(h, 5, None); + match largest { + Largest::Finished => { + h.track.finish().unwrap(); + None + } + Largest::Live => Some(h.track.create_group(group::Info { sequence: 5 }).unwrap()), + Largest::Idle => { + h.track.set_idle(); + None + } + Largest::IdleEnded => { + h.track.set_idle(); + h.track.finish_at(10).unwrap(); + None + } + Largest::Datagram => { + h.track.append_datagram(timestamp(), b"d".as_slice()).unwrap(); + None + } + Largest::Aborted => { + let mut newer = h.track.create_group(group::Info { sequence: 5 }).unwrap(); + newer.write_frame(timestamp(), b"5-0".to_vec()).unwrap(); + newer.abort(Error::Cancel).unwrap(); + None + } + } + } + + /// On draft 20, a start past Largest Object is refused INVALID_RANGE (section 10.13), + /// wherever the publisher knows it. Where it does not, it answers empty. + #[moq_net_sim::test] + async fn a_draft20_fetch_past_the_largest_object_is_refused() { + for version in [Version::Draft20, Version::Draft21, Version::Draft22] { + for largest in [ + Largest::Finished, + Largest::Live, + Largest::Idle, + Largest::IdleEnded, + Largest::Datagram, + Largest::Aborted, + ] { + let mut h = serve(version); + let _newer = publish_largest(&mut h, largest); + settle().await; + + let buf = standalone_fetch( + &h, + Location { group: 4, object: 2 }, + Location { group: 4, object: 6 }, + GroupOrder::Ascending, + ) + .await; + match largest { + Largest::Finished | Largest::Live => assert_eq!( + fetch_refusal(buf, version), + invalid_range(version), + "{version} {largest:?}" + ), + _ => { + let (ok, objects) = fetch_answer(buf, version); + assert_eq!( + ok.end_location, + Location { group: 4, object: 5 }, + "{version} {largest:?}" + ); + assert_eq!(objects, Vec::new(), "{version} {largest:?}"); + } + } + } } } - /// Draft-20 carries the range in LOCATION_FILTER, which is not read yet. + /// On a live feed whose newest group finished with no objects, Largest Object sits in the + /// group before it, so a draft-20 FETCH of the empty group starts past it. #[moq_net_sim::test] - async fn a_draft20_fetch_is_refused() { + async fn a_draft20_fetch_of_an_empty_newest_group_is_refused() { for version in [Version::Draft20, Version::Draft21, Version::Draft22] { let mut h = serve(version); - publish_pairs(&mut h, 3, None); + publish_pairs(&mut h, 5, None); + let empty = h.track.create_group(group::Info { sequence: 5 }).unwrap(); + empty.finish().unwrap(); settle().await; let buf = standalone_fetch( &h, - Location { group: 1, object: 0 }, - Location { group: 1, object: 0 }, + Location { group: 5, object: 0 }, + Location { group: 5, object: 4 }, GroupOrder::Ascending, ) .await; - assert_eq!(fetch_refusal(buf, version), 0x3, "{version}"); + assert_eq!(fetch_refusal(buf, version), invalid_range(version), "{version}"); + } + } + + /// On draft 20, an End Object past a finished group's last object is still the End + /// Location reported (section 10.14), unless the group holds Largest Object, which caps + /// it. A relay's idle copy does not know Largest Object, so it echoes the request. + #[moq_net_sim::test] + async fn a_draft20_fetch_past_a_group_end_reports_the_requested_end() { + for version in [Version::Draft20, Version::Draft21, Version::Draft22] { + for (group, largest, reported) in [ + (2, Largest::Live, 7), + (4, Largest::Live, 1), + (4, Largest::Finished, 1), + (4, Largest::Idle, 7), + (4, Largest::IdleEnded, 7), + (4, Largest::Datagram, 7), + (4, Largest::Aborted, 7), + ] { + let mut h = serve(version); + let _newer = publish_largest(&mut h, largest); + settle().await; + + let buf = standalone_fetch( + &h, + Location { group, object: 0 }, + Location { group, object: 8 }, + GroupOrder::Ascending, + ) + .await; + let (ok, objects) = fetch_answer(buf, version); + assert_eq!( + ok.end_location, + Location { + group, + object: reported + }, + "{version} group {group} {largest:?}" + ); + assert_eq!(ok.end_of_track, matches!(largest, Largest::Finished), "{version}"); + assert_eq!(objects, pairs([group]), "{version}"); + } + } + } + + /// The last group of a finished track ends the response at its last object, with End + /// of Track set. + #[moq_net_sim::test] + async fn a_standalone_fetch_of_the_last_group_reports_the_end_of_track() { + for version in FETCH_DRAFTS { + let mut h = serve(version); + publish_pairs(&mut h, 5, None); + h.track.finish().unwrap(); + settle().await; + + let buf = standalone_fetch( + &h, + Location { group: 4, object: 1 }, + Location { group: 4, object: 0 }, + GroupOrder::Any, + ) + .await; + let (ok, objects) = fetch_answer(buf, version); + assert_eq!(ok.end_location, end_location(version, 4, 1), "{version}"); + assert!(ok.end_of_track, "{version}"); + assert_eq!(objects, pairs([4])[1..].to_vec(), "{version}"); + } + } + + /// A draft-20 filter bounded by Largest Object is refused, even when that bound would + /// land in the start group: the one-group rule needs it resolved first. + #[moq_net_sim::test] + async fn a_draft20_fetch_relative_to_largest_object_is_refused() { + for version in [Version::Draft20, Version::Draft21, Version::Draft22] { + for filter in [ + Filter::Unfiltered, + Filter::NextObject, + Filter::Relative(1), + Filter::Absolute { + start: Location { group: 2, object: 0 }, + end: None, + }, + ] { + let mut h = serve(version); + publish_pairs(&mut h, 3, None); + settle().await; + + let fetch_type = FetchType::Filtered { + namespace: crate::Path::new("room"), + track: "video".into(), + filter, + }; + let buf = run_fetch(&h, fetch_type, GroupOrder::Ascending).await; + assert_eq!(fetch_refusal(buf, version), 0x3, "{version}: {filter:?}"); + } } } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 6479900078..d57fc40419 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3579,8 +3579,8 @@ where Ok(true) } - /// Fetch a group from the publisher to fill a cache miss, with a standalone FETCH for - /// this subscription's track. It asks from the frame the reader wants through the end + /// Fetch a group from the publisher to fill a cache miss, with a standalone (or + /// draft-20 filtered) FETCH for this subscription's track. It asks from the frame the reader wants through the end /// of the group, so a publisher that evicted the prefix can still answer. /// /// The group is accepted only once FETCH_OK arrives, so a refusal reaches every @@ -3595,11 +3595,7 @@ where ) { let sequence = request.sequence(); let start = request.frame_start(); - // Our FETCH still encodes the Fetch Type field that draft-20 removed. - if Filter::is_draft20(self.version) { - request.reject(Error::Unsupported); - return; - } + let draft20 = Filter::is_draft20(self.version); let fetch_id = match self.control.next_request_id(&self.runtime).await { Ok(id) => id, @@ -3615,6 +3611,36 @@ where Err(err) => return request.reject(err), }; + let namespace = broadcast.clone(); + let track = name.as_str().into(); + let from = ietf::Location { + group: sequence, + object: start, + }; + let fetch_type = match draft20 { + // No End Object includes the whole End Group. + true => FetchType::Filtered { + namespace, + track, + filter: Filter::Absolute { + start: from, + end: Some(ietf::EndLocation { + group: sequence, + object: None, + }), + }, + }, + // An End Object of 0 is the whole End Group. + false => FetchType::Standalone { + namespace, + track, + start: from, + end: ietf::Location { + group: sequence, + object: 0, + }, + }, + }; let res = async { stream.writer.varint(ietf::Fetch::ID).await?; stream @@ -3623,19 +3649,7 @@ where request_id: fetch_id, subscriber_priority: super::priority::to_wire(request.priority()), group_order: GroupOrder::Ascending, - // An End Object of 0 is the whole End Group. - fetch_type: FetchType::Standalone { - namespace: broadcast.clone(), - track: name.as_str().into(), - start: ietf::Location { - group: sequence, - object: start, - }, - end: ietf::Location { - group: sequence, - object: 0, - }, - }, + fetch_type, range_filters: false, fill_timeout: false, // The copy learned the track before any fetch, from SUBSCRIBE_OK or @@ -3677,9 +3691,31 @@ where } }; + // Draft-20's End Location is inclusive, so step one past it as older drafts spell it. + let mut end = ok.end_location; + if draft20 { + let Some(object) = end.object.checked_add(1) else { + request.reject(Error::ProtocolViolation); + let _ = stream.writer.close().await; + return; + }; + end.object = object; + } + + // From draft-20 an empty answer still covers its start, so an End Location before it + // is malformed and closes the session (section 10.14), before it can mark the track's end. + if draft20 && (end.group, end.object) <= (sequence, start) { + let err = Error::ProtocolViolation; + tracing::warn!(group = sequence, ?end, "FETCH_OK's End Location is before its start"); + self.session + .clone() + .close(SessionError::from(&err).to_code(), err.to_string().as_ref()); + request.reject(err); + return; + } + // The publisher knows where the track ends, which a range FETCH downstream needs. if ok.end_of_track { - let end = ok.end_location; let Some(final_sequence) = end.group.checked_add(u64::from(end.object > 0)) else { request.reject(Error::ProtocolViolation); let _ = stream.writer.close().await; @@ -3688,15 +3724,14 @@ where request.finish_track_at(final_sequence); } - // An empty answer opens no fetch stream at all. - let end = ok.end_location; + // Before draft-20, an empty answer opens no fetch stream at all. if (end.group, end.object) <= (sequence, start) { request.reject(Error::NotFound); let _ = stream.writer.close().await; return; } - // An exclusive End Location inside the group is its Largest Object plus one, so - // the stream owes every object up to it. One past the group covers it whole. + // An exclusive End Location inside the group bounds the stream; before draft-20 the + // stream also owes every object up to it. One past the group covers it whole. let end = (end.group == sequence).then_some(end.object); // Joined fetches still count until they pick the accepted group up from the cache. @@ -3884,7 +3919,8 @@ where } /// Decode one group's objects: all in the producer's group, numbered from `start` with - /// no gaps, and through `end` (exclusive) when FETCH_OK named one inside the group. + /// no gaps, and never past `end` (exclusive) when FETCH_OK named one inside the group. + /// Before draft-20 they must also reach it. async fn recv_group_fetch_objects( &self, stream: &mut Reader, @@ -3946,8 +3982,9 @@ where } // A clean FIN short of the promised end would otherwise cache a truncated group as - // a complete one. - if end.is_some_and(|end| next < end) { + // a complete one. Draft-20's End Location is the range covered, not the last object + // sent: objects missing before it do not exist (section 10.13). + if !Filter::is_draft20(self.version) && end.is_some_and(|end| next < end) { tracing::warn!( sequence, next, @@ -8733,37 +8770,42 @@ mod stitch_tests { assert!(matches!(res, Err(Error::MalformedTrack)), "{res:?}"); } - /// FETCH_OK's End Location inside the group promises every object before it. A stream - /// that FINs short of it, or runs past it, fails the group instead of caching it. + /// FETCH_OK's End Location inside the group bounds the stream: one that runs past it + /// fails the group instead of caching it. Before draft-20 the End Location also promises + /// every object before it, so a stream that FINs short fails too. From draft-20 it is the + /// range covered, and the objects missing before it do not exist (section 10.13). #[moq_net_sim::test] - async fn a_group_fetch_must_reach_its_end_location() { + async fn a_group_fetch_stays_within_its_end_location() { const END: u64 = 3; - for (count, complete) in [(2, false), (3, true), (4, false)] { - let payloads: Vec<&[u8]> = [b"a", b"b", b"c", b"d"][..count].iter().map(|p| &p[..]).collect(); - let mut run = GroupFetchRun::new(VERSION, group_fetch_objects(SEQUENCE, 0, &payloads)).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: Some(END), - }); - let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; - assert_eq!(res.is_ok(), complete, "{count} objects: {res:?}"); - - let mut read = 0; - let end = loop { - match consumer.read_frame().await { - Ok(Some(_)) => read += 1, - Ok(None) => break Ok(read), - Err(err) => break Err(err), + for version in [Version::Draft19, VERSION] { + let short = Filter::is_draft20(version); + for (count, complete) in [(2, short), (3, true), (4, false)] { + let payloads: Vec<&[u8]> = [b"a", b"b", b"c", b"d"][..count].iter().map(|p| &p[..]).collect(); + let mut run = GroupFetchRun::new(version, group_fetch_objects(SEQUENCE, 0, &payloads)).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: Some(END), + }); + let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; + assert_eq!(res.is_ok(), complete, "{version} {count} objects: {res:?}"); + + let mut read = 0; + let end = loop { + match consumer.read_frame().await { + Ok(Some(_)) => read += 1, + Ok(None) => break Ok(read), + Err(err) => break Err(err), + } + }; + match complete { + true => assert_eq!(end.expect("a complete group"), count as u64, "{version}"), + false => assert!(end.is_err(), "{version} {count} objects: the group must fail, not end"), } - }; - match complete { - true => assert_eq!(end.expect("a complete group"), END), - false => assert!(end.is_err(), "{count} objects: the group must fail, not end"), } } } @@ -9646,38 +9688,118 @@ mod joining_fetch_tests { assert!(fetch.await.is_err(), "the accepted group was aborted"); } + /// From draft-20 an End Location before the FETCH's start is malformed (section 10.14): + /// the session closes with PROTOCOL_VIOLATION rather than reading it as an empty answer. + #[moq_net_sim::test] + async fn a_draft20_end_location_before_the_start_closes_the_session() { + const VERSION: Version = Version::Draft20; + const GROUP: u64 = 4; + const START: u64 = 3; + + let ok = message_bytes( + ietf::FetchOk::ID, + &ietf::FetchOk { + request_id: None, + group_order: GroupOrder::Ascending, + end_of_track: false, + end_location: ietf::Location { + group: GROUP, + object: START - 1, + }, + properties: Default::default(), + }, + VERSION, + ); + + let session = ScriptedSession::per_stream_eof(vec![ok]); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let subscriber = Subscriber::new( + crate::time::Clock::sim(), + session.clone(), + crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + Control::new(None, false), + None, + peer::PeerSetup::default(), + crate::Hop::new(1).unwrap(), + None, + VERSION, + tasks, + Default::default(), + ); + + let track = track::Producer::new( + std::sync::Arc::new(crate::broadcast::Info::default()), + "video", + track::Info::default().with_timescale(None), + ); + let dynamic = track.dynamic(); + let consumer = track.consume(); + let mut fetch = std::pin::pin!(consumer.fetch_group(GROUP, group::Fetch::default().with_frame_start(START))); + assert!(futures::poll!(fetch.as_mut()).is_pending()); + let request = dynamic.requested_group().await.expect("no group requested"); + + subscriber + .clone() + .run_group_fetch(Path::new("broadcast").to_owned(), "video".into(), request, None) + .await; + + assert!(matches!(fetch.await, Err(Error::ProtocolViolation))); + let violation = SessionError::from(&Error::ProtocolViolation).to_code(); + assert!( + session.log.closes().iter().any(|(code, _)| *code == violation), + "{:?}", + session.log.closes() + ); + } + /// A cache miss for a group's tail asks upstream from the frame the reader wants and /// numbers what arrives from there, so a publisher that evicted the prefix can answer. #[moq_net_sim::test] async fn a_group_fetch_asks_from_the_wanted_frame() { - const VERSION: Version = Version::Draft19; + for version in [Version::Draft19, Version::Draft20, Version::Draft22] { + group_fetch_from_frame(version).await; + } + } + + async fn group_fetch_from_frame(version: Version) { const GROUP: u64 = 4; const START: u64 = 2; + const LAST: u64 = START + 1; + let draft20 = Filter::is_draft20(version); + // The whole group, as each draft answers it: draft 20 names the last object, and + // older drafts one past the group. + let end_location = match draft20 { + true => ietf::Location { + group: GROUP, + object: LAST, + }, + false => ietf::Location { + group: GROUP + 1, + object: 0, + }, + }; let ok = message_bytes( ietf::FetchOk::ID, &ietf::FetchOk { request_id: None, group_order: GroupOrder::Ascending, end_of_track: false, - end_location: ietf::Location { - group: GROUP + 1, - object: 0, - }, + end_location, properties: Default::default(), }, - VERSION, + version, ); // The fetch stream answering our first request id, from object START on. let mut objects = Vec::new(); - crate::coding::Encoder::new(&mut objects, VERSION.into()) + crate::coding::Encoder::new(&mut objects, version.into()) .varint(ietf::FetchHeader::TYPE) .unwrap(); ietf::FetchHeader { request_id: RequestId(1), } - .encode(&mut crate::coding::Encoder::new(&mut objects, VERSION.into()), VERSION) + .encode(&mut crate::coding::Encoder::new(&mut objects, version.into()), version) .unwrap(); for (index, payload) in [b"c", b"d"].iter().enumerate() { let first = index == 0; @@ -9688,9 +9810,9 @@ mod joining_fetch_tests { priority: first.then_some(0), properties: None, } - .encode(&mut crate::coding::Encoder::new(&mut objects, VERSION.into()), VERSION) + .encode(&mut crate::coding::Encoder::new(&mut objects, version.into()), version) .unwrap(); - crate::coding::Encoder::new(&mut objects, VERSION.into()) + crate::coding::Encoder::new(&mut objects, version.into()) .varint(1u64) .unwrap(); objects.extend_from_slice(&payload[..]); @@ -9707,7 +9829,7 @@ mod joining_fetch_tests { peer::PeerSetup::default(), crate::Hop::new(1).unwrap(), None, - VERSION, + version, tasks, Default::default(), ); @@ -9733,7 +9855,7 @@ mod joining_fetch_tests { // The fetch stream, as the peer would open it. let (_, recv) = session.clone().open_bi().await.unwrap(); - let mut stream = Reader::new(recv, VERSION); + let mut stream = Reader::new(recv, version); subscriber.clone().recv_fill(&mut stream).await.expect("group fetch"); serving.await.expect("run_group_fetch"); @@ -9745,27 +9867,40 @@ mod joining_fetch_tests { } assert_eq!(payloads, [b"c".to_vec(), b"d".to_vec()]); - let messages = decode_messages(&session.log, VERSION); + let messages = decode_messages(&session.log, version); let fetch = messages.iter().find(|(id, _)| *id == ietf::Fetch::ID).expect("FETCH"); - let mut body = Decoder::new(&fetch.1, VERSION.into()); - let msg = ietf::Fetch::decode_msg(&mut body, VERSION).unwrap(); - let FetchType::Standalone { start, end, .. } = msg.fetch_type else { - panic!("a group fetch is standalone: {:?}", msg.fetch_type); + let mut body = Decoder::new(&fetch.1, version.into()); + let msg = ietf::Fetch::decode_msg(&mut body, version).unwrap(); + let from = ietf::Location { + group: GROUP, + object: START, }; - assert_eq!( - start, - ietf::Location { - group: GROUP, - object: START + match msg.fetch_type { + // No End Object is the whole End Group. + FetchType::Filtered { filter, .. } if draft20 => assert_eq!( + filter, + Filter::Absolute { + start: from, + end: Some(ietf::EndLocation { + group: GROUP, + object: None + }), + }, + "{version}: through the end of the group" + ), + // An End Object of 0 is the whole End Group. + FetchType::Standalone { start, end, .. } if !draft20 => { + assert_eq!(start, from, "{version}"); + assert_eq!( + end, + ietf::Location { + group: GROUP, + object: 0 + }, + "{version}: through the end of the group" + ); } - ); - assert_eq!( - end, - ietf::Location { - group: GROUP, - object: 0 - }, - "through the end of the group" - ); + other => panic!("{version}: unexpected group fetch: {other:?}"), + } } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 649818bcec..c6c0716d64 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2923,6 +2923,15 @@ impl Consumer { routes.serving() } + /// Whether the cache is fed live, so the newest object it holds is the track's largest. + /// A relay's copy with no upstream subscription is not, even once it learns the end. + pub(crate) fn is_live(&self) -> bool { + if let Some(serving) = self.serving() { + return serving.is_live(); + } + self.state.read().feed == Feed::Live + } + /// Fetching a single past group, without holding a live subscription. /// /// Returns a [`kio::Pending`] that resolves to the [`group::Consumer`]: diff --git a/rs/moq-net/tests/fetch_only.rs b/rs/moq-net/tests/fetch_only.rs index 60c991f483..21880e230f 100644 --- a/rs/moq-net/tests/fetch_only.rs +++ b/rs/moq-net/tests/fetch_only.rs @@ -81,19 +81,18 @@ fn frames(sequence: u64) -> Vec> { (0..2).map(|frame| format!("{sequence}-{frame}").into_bytes()).collect() } -/// Drafts whose FETCH we serve, and the newest, whose FETCH we refuse until draft-20 -/// FETCH lands: it must still not subscribe. Draft-17 is absent: its TRACK_STATUS answer -/// cannot say whether the track is timed, so it still SUBSCRIBEs to learn the track. -const VERSIONS: &[(&str, bool)] = &[ - ("moq-transport-14", true), - ("moq-transport-15", true), - ("moq-transport-16", true), - ("moq-transport-18", true), - ("moq-transport-19", true), - ("moq-transport-22", false), +/// Draft-17 is absent: its TRACK_STATUS answer cannot say whether the track is timed, +/// so it still SUBSCRIBEs to learn the track. +const VERSIONS: &[&str] = &[ + "moq-transport-14", + "moq-transport-15", + "moq-transport-16", + "moq-transport-18", + "moq-transport-19", + "moq-transport-22", ]; -async fn direct(version: &str, served: bool) { +async fn direct(version: &str) { let version: Version = version.parse().unwrap(); let publisher = produce_origin(1); let (broadcast, subscribed) = publish_finished(&publisher); @@ -105,11 +104,7 @@ async fn direct(version: &str, served: bool) { let pair = connect_mock(options).await; let remote = resolve(&client).await; - let fetched = fetch(&remote, 1).await; - match served { - true => assert_eq!(fetched.expect("fetch"), frames(1), "{version}"), - false => assert!(fetched.is_err(), "{version}: draft-20 FETCH is refused"), - } + assert_eq!(fetch(&remote, 1).await.expect("fetch"), frames(1), "{version}"); assert!(!subscribed.load(Ordering::SeqCst), "{version}: the fetch subscribed"); drop((remote, pair, broadcast, client, publisher)); @@ -119,8 +114,8 @@ async fn direct(version: &str, served: bool) { #[moq_net_sim::test] async fn a_fetch_learns_the_track_without_subscribing() { let mut failures = Vec::new(); - for (version, served) in VERSIONS { - if moq_net_sim::spawn(direct(version, *served)).await.is_err() { + for version in VERSIONS { + if moq_net_sim::spawn(direct(version)).await.is_err() { failures.push(*version); } } @@ -208,8 +203,8 @@ async fn a_subscribe_after_track_status_keeps_timestamps() { #[moq_net_sim::test] async fn a_relay_fetches_without_subscribing_upstream() { let mut failures = Vec::new(); - for (version, served) in VERSIONS { - if *served && moq_net_sim::spawn(relayed(version)).await.is_err() { + for version in VERSIONS { + if moq_net_sim::spawn(relayed(version)).await.is_err() { failures.push(*version); } } diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index 6db94c3d7e..d78b806f1b 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -376,10 +376,10 @@ async fn broadcast_moq_lite_05_fetch_webtransport() { lite05_fetch_roundtrip("https").await; } -/// A cache miss over moq-transport is a standalone FETCH of the one group, served from -/// the publisher's cache, and a group the publisher lacks comes back as the publisher's -/// own refusal. Without `served`, the version has no FETCH and the miss is refused. -async fn transport_fetch_roundtrip(version: &str, served: bool) { +/// A cache miss over moq-transport is a FETCH of the one group, served from the +/// publisher's cache, and a group the publisher lacks comes back as the publisher's own +/// refusal. +async fn transport_fetch_roundtrip(version: &str) { let pub_origin = moq_tokio::origin::spawn(); let broadcast = pub_origin.create_broadcast("test").expect("failed to create broadcast"); broadcast @@ -444,19 +444,6 @@ async fn transport_fetch_roundtrip(version: &str, served: bool) { let fetched = tokio::time::timeout(TIMEOUT, async { bc.track("video").unwrap().fetch_group(1, None).await }) .await .expect("fetch timed out"); - if !served { - assert!( - matches!(fetched, Err(moq_net::Error::Unsupported)), - "expected an unsupported FETCH, got {:?}", - fetched.err() - ); - drop(connection); - server_handle - .await - .expect("server task panicked") - .expect("server task failed"); - return; - } let mut group = fetched.expect("fetch failed"); assert_eq!(group.sequence, 1); for frame in 0..2 { @@ -492,25 +479,25 @@ async fn transport_fetch_roundtrip(version: &str, served: bool) { #[tracing_test::traced_test] #[tokio::test] async fn fetch_moq_transport_14() { - transport_fetch_roundtrip("moq-transport-14", true).await; + transport_fetch_roundtrip("moq-transport-14").await; } #[tracing_test::traced_test] #[tokio::test] async fn fetch_moq_transport_16() { - transport_fetch_roundtrip("moq-transport-16", true).await; + transport_fetch_roundtrip("moq-transport-16").await; } #[tracing_test::traced_test] #[tokio::test] async fn fetch_moq_transport_18() { - transport_fetch_roundtrip("moq-transport-18", true).await; + transport_fetch_roundtrip("moq-transport-18").await; } #[tracing_test::traced_test] #[tokio::test] async fn fetch_moq_transport_20() { - transport_fetch_roundtrip("moq-transport-20", false).await; + transport_fetch_roundtrip("moq-transport-20").await; } /// A fetch must be served while a live subscription is active on the same track.