From b04a182a5cbef7a647a7ad75bfa4a9976e119cf3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 21:14:22 -0700 Subject: [PATCH 1/4] chore: claim release FETCH backport quest Co-Authored-By: GPT-6.1 Sol From 190cc02289f301b0cc15ef4896e019891d15f308 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 21:54:57 -0700 Subject: [PATCH 2/4] fix(net): backport single-group FETCH on draft-20 and later Backport #4971 to the 0.17 release line, adapting the existing FETCH regression tests to the release codec, clock, and spliced track model. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: GPT-6.1 Sol --- doc/concept/standard.md | 10 +- rs/moq-net/src/ietf/publisher.rs | 424 +++++++++++++++++++++++++----- rs/moq-net/src/ietf/subscriber.rs | 308 ++++++++++++++++------ rs/moq-net/src/model/resume.rs | 6 + rs/moq-net/src/model/track.rs | 9 + rs/moq-tokio/tests/broadcast.rs | 35 +-- 6 files changed, 620 insertions(+), 172 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 37a959a3ac..7b691ec635 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -50,12 +50,14 @@ to model priority 127, where higher values are served first. A track that never sets a priority is 127 as well, so it goes out as 128 on IETF and 127 on moq-lite. -On drafts 14–19, the Rust publisher answers a standalone `FETCH` within one +On drafts 14–22, the Rust publisher answers a standalone `FETCH` within one group from the cache. A relay fetches a missing group upstream with a `FETCH` of that one whole group, and an upstream refusal is the refusal the fetcher -sees. A range touching several groups is refused with `NOT_SUPPORTED`, as is -any `FETCH` on draft-20 and later, which moved the range into -`LOCATION_FILTER`. A standalone `FETCH` +sees. A range touching several groups is refused with `NOT_SUPPORTED`, as is a +draft-20+ `LOCATION_FILTER` that ends at Largest Object instead of an absolute +End Location. On draft-20 and later, `FETCH_OK` reports the inclusive requested +end, capped at Largest Object when known. An empty range gets an empty fetch +stream; a start past Largest Object is refused with `INVALID_RANGE`. A standalone `FETCH` carries no timestamps, since no `SUBSCRIBE_OK` declared a timescale for it. On drafts 14–19, the Rust publisher also serves relative and absolute joining diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 1a32051088..599781d077 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1238,8 +1238,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 @@ -1247,19 +1247,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 @@ -1285,7 +1272,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, @@ -1328,7 +1354,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, _ => { @@ -1353,7 +1379,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") @@ -1398,6 +1424,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; + } let (end_location, end_of_track) = if joined { // The subscription starts at the saved Largest Object, so the prefix of that @@ -1409,7 +1466,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 { @@ -1423,6 +1479,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 @@ -4091,30 +4168,29 @@ mod serve_tests { } } - #[tokio::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!(u64::decode(&mut buf, Version::Draft20).unwrap(), ietf::RequestError::ID); - assert_eq!( - ietf::RequestError::decode(&mut buf, Version::Draft20) - .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) { @@ -4129,7 +4205,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(); @@ -4141,12 +4246,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, }, @@ -4229,53 +4329,257 @@ 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. - #[tokio::test] - async fn a_standalone_fetch_of_the_last_group_reports_the_end_of_track() { + /// A range ending inside its group stops at its End Object, which draft 20 counts + /// inclusively and older drafts count as one past. + #[tokio::test(start_paused = true)] + 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}"); } } - /// Draft-20 carries the range in LOCATION_FILTER, which is not read yet. - #[tokio::test] - async fn a_draft20_fetch_is_refused() { + /// 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. + #[tokio::test(start_paused = true)] + 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 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::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. + #[tokio::test(start_paused = true)] + 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::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:?}"); + } + } + } + } + } + + /// 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. + #[tokio::test(start_paused = true)] + 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. + #[tokio::test(start_paused = true)] + 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::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. + #[tokio::test(start_paused = true)] + 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. + #[tokio::test(start_paused = true)] + 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 5139713e4d..ccef616f82 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2990,8 +2990,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 @@ -3010,11 +3010,7 @@ where request.reject(Error::GoingAway); return; } - // 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, @@ -3030,6 +3026,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.encode(&ietf::Fetch::ID).await?; stream @@ -3038,19 +3064,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, }) @@ -3069,9 +3083,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; @@ -3080,15 +3116,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); // Only a track nothing has subscribed to yet takes this info, as SUBSCRIBE_OK would @@ -3224,7 +3259,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, @@ -3282,8 +3318,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, @@ -7416,37 +7453,42 @@ mod stitch_tests { } } - /// 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. - #[tokio::test] - async fn a_group_fetch_must_reach_its_end_location() { + /// 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). + #[tokio::test(start_paused = true)] + 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"), } } } @@ -7890,35 +7932,114 @@ 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. + #[tokio::test(start_paused = true)] + 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, + }, + }, + VERSION, + ); + + let session = ScriptedSession::per_stream_eof(vec![ok]); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let subscriber = Subscriber::new( + crate::time::Clock::tokio(), + 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(), + ); + 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. #[tokio::test(start_paused = true)] 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, }, - VERSION, + version, ); // The fetch stream answering our first request id, from object START on. let mut objects = bytes::BytesMut::new(); - ietf::FetchHeader::TYPE.encode(&mut objects, VERSION).unwrap(); + ietf::FetchHeader::TYPE.encode(&mut objects, version).unwrap(); ietf::FetchHeader { request_id: RequestId(1), } - .encode(&mut objects, VERSION) + .encode(&mut objects, version) .unwrap(); for (index, payload) in [b"c", b"d"].iter().enumerate() { let first = index == 0; @@ -7929,9 +8050,9 @@ mod joining_fetch_tests { priority: first.then_some(0), properties: None, } - .encode(&mut objects, VERSION) + .encode(&mut objects, version) .unwrap(); - 1u64.encode(&mut objects, VERSION).unwrap(); + 1u64.encode(&mut objects, version).unwrap(); objects.extend_from_slice(&payload[..]); } @@ -7946,7 +8067,7 @@ mod joining_fetch_tests { peer::PeerSetup::default(), crate::Hop::new(1).unwrap(), None, - VERSION, + version, tasks, Default::default(), ); @@ -7972,7 +8093,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"); @@ -7984,27 +8105,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 = fetch.1.clone(); - 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 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/resume.rs b/rs/moq-net/src/model/resume.rs index c8b06fae5b..9c2c9285f8 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -731,6 +731,12 @@ impl Consumer { self.state.read().latest() } + /// Whether the newest segment still has its live producer. + pub(crate) fn is_live(&self) -> bool { + let track = self.state.read().segments.last().map(|segment| segment.track.clone()); + track.is_some_and(|track| track.is_live()) + } + /// The newest segment's declared exclusive end, where fetches are routed. pub(crate) fn final_sequence(&self) -> Option { // Copied out: the segment's track takes its own lock. diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 5b9284a6e9..be18e351c2 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2844,6 +2844,15 @@ impl Consumer { } } + /// Whether the cache is fed live, so the newest object it holds is the track's largest. + /// A relay's warm cache whose upstream producer closed is not. + pub(crate) fn is_live(&self) -> bool { + match &self.inner { + ConsumerKind::Plain(state) => !state.read().closed, + ConsumerKind::Spliced(resume) => resume.is_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-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index 2ab11f6844..c1c0b410e3 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,31 @@ 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; +} + +#[tracing_test::traced_test] +#[tokio::test] +async fn fetch_moq_transport_22() { + transport_fetch_roundtrip("moq-transport-22").await; } /// A fetch must be served while a live subscription is active on the same track. From d2e23191c8365b96559ebb97269166272c7f5fcc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 13 Oct 2026 11:51:45 -0700 Subject: [PATCH 3/4] fix(net): a fetch-only cache is not a live feed A track served only through its fetch handler, never accepted, was classified live because it was not closed, so its newest fetched object was taken as Largest Object and a draft-20 FETCH past it was refused or capped. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 38 +++++++++++++++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index be18e351c2..e1c20656ef 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2845,10 +2845,13 @@ impl Consumer { } /// Whether the cache is fed live, so the newest object it holds is the track's largest. - /// A relay's warm cache whose upstream producer closed is not. + /// A cache only fetches fill, or a relay's warm cache whose upstream producer closed, is not. pub(crate) fn is_live(&self) -> bool { match &self.inner { - ConsumerKind::Plain(state) => !state.read().closed, + ConsumerKind::Plain(state) => { + let state = state.read(); + state.published && !state.closed + } ConsumerKind::Spliced(resume) => resume.is_live(), } } @@ -8951,6 +8954,37 @@ mod test { assert!(pool.used() <= 13_000, "usage converges: {}", pool.used()); } + /// A cache only a fetch handler fills has no live feed, so its newest fetched object + /// is not the track's largest. Accepting the request feeds it live until the producer goes. + #[tokio::test] + async fn a_fetch_only_cache_is_not_live() { + let request = Request::new(Arc::new(broadcast::Info::default()), "test"); + let dynamic = request.dynamic(); + let consumer = request.consume(); + + let pending = consumer.fetch_group(2, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let mut fetched = req.accept(None).unwrap(); + fetched + .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"2-0")) + .unwrap(); + fetched.finish().unwrap(); + pending.await.unwrap(); + assert!(!consumer.is_live(), "a fetch-only cache is not live"); + + let producer = request.accept(None); + assert!(consumer.is_live(), "an accepted track is live"); + + // The fetch handler keeps the track alive like a producer clone, so both go. + drop(dynamic); + drop(producer); + assert!(!consumer.is_live(), "a track whose producer went is not live"); + } + /// Re-serving a sequence many times must not accumulate eviction hints: stale /// hints die on stamp mismatch and compaction reclaims them. #[tokio::test] From 648a1c6e1290b664afa7d2c4d549f765bdd3a17e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 13 Oct 2026 13:09:59 -0700 Subject: [PATCH 4/4] fix(net): a parked warm cache is not a live feed A spliced track whose newest segment is a parked warm cache still read as live, since the cache is an open producer, so its stale edge was taken as Largest Object for a draft-20 FETCH. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 9c2c9285f8..d4895f9f6d 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -731,9 +731,16 @@ impl Consumer { self.state.read().latest() } - /// Whether the newest segment still has its live producer. + /// Whether the newest segment still has its live producer. A parked warm cache is an + /// open producer with no upstream feed behind it, so it never is. pub(crate) fn is_live(&self) -> bool { - let track = self.state.read().segments.last().map(|segment| segment.track.clone()); + let track = self + .state + .read() + .segments + .last() + .filter(|segment| !segment.warm) + .map(|segment| segment.track.clone()); track.is_some_and(|track| track.is_live()) } @@ -3324,6 +3331,25 @@ mod test { } } + /// A parked track's warm cache is an open producer with no upstream feed, so the track + /// is not live until a copy is spliced after it. + #[tokio::test] + async fn a_parked_warm_cache_is_not_live() { + let (_track, consumer) = track_pair("a"); + let mut producer = Producer::new(); + producer.takeover(&consumer).unwrap(); + let resumed = producer.consume(); + assert!(resumed.is_live(), "a live copy is live"); + + let (_cache, cache_consumer) = track_pair("cache"); + producer.park(&cache_consumer).unwrap(); + assert!(!resumed.is_live(), "a parked warm cache is not live"); + + let (_next, next_consumer) = track_pair("next"); + producer.takeover(&next_consumer).unwrap(); + assert!(resumed.is_live(), "the copy spliced after it is live"); + } + /// A group held back at the reader's cap when a park replaces its segments comes /// from the cache once the cap rises, exactly once. #[tokio::test]