From 85864e6ecb7f4e6f77427d0687438f8b4ef5eeaf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:30:23 -0800 Subject: [PATCH 1/9] chore(quest): claim quest/m1/ietf-resume-behind.md Quest-Claim: d25e6667b04c7fe7 From c08121a002c30b46d3f18b7399577a9ec96c369f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:43:13 -0800 Subject: [PATCH 2/9] fix(net): keep cached history owed across resumed feeds --- quest/m1/README.md | 1 - quest/m1/ietf-resume-behind.md | 33 --------------------------------- 2 files changed, 34 deletions(-) delete mode 100644 quest/m1/ietf-resume-behind.md diff --git a/quest/m1/README.md b/quest/m1/README.md index d8a307f3e3e..2d4a72b1cf7 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -190,7 +190,6 @@ per-frame CPU growth since #5926. - [Authored tracks default to a 30 s max-age](/quest/m1/max-age-defaults.md) - raw and media tracks advertise 30 s unless their author chooses otherwise, with explicit unlimited in every binding - [HLS export lists a same-codec reconfigure's first group under the new init](/quest/m1/hls-reinit-same-codec.md) - the hand-over pick, whichever of the catalog and the records arrives first, after the catalog `hls` removal is planned - [The resume model passes its long run again](/quest/m1/resume-model-7542.md) - fix #7542's regression that drops groups a long-budget reader can still get -- [A resume more than one group behind never hangs](/quest/m1/ietf-resume-behind.md) - a resumed read gets every group or a drop on every wire version - [moq-quic's idle connections keep no burst-sized state](/quest/m1/quic-readable-dedupe.md) - readable events are deduplicated and a quiet connection keeps no per-peak-stream bytes - [Stats name the route instance's epoch with a type](/quest/m1/stats-epoch-type.md) - `GroupFrame::epoch` gets a dedicated type and tests stop mislabelling the instance diff --git a/quest/m1/ietf-resume-behind.md b/quest/m1/ietf-resume-behind.md deleted file mode 100644 index c2f417d81d3..00000000000 --- a/quest/m1/ietf-resume-behind.md +++ /dev/null @@ -1,33 +0,0 @@ -# [S] A resume more than one group behind never hangs - -## Goal - -A resumed read that starts more than one group behind the latest one gets -every group or a drop for it on every wire version, never a hang. The fix is -the simplest one that works; moq-transport FETCH/SUBSCRIBE edge cases are not -worth draft-specific fallbacks. - -## Plan - -Planned 2026-11-02 (maintainer, follow-up planning after the 2026-11-02 /quest-complete run), from [moq-dev/moq#7631](https://github.com/moq-dev/moq/pull/7631), which -found it and kept its two direct bounded-resume tests on draft 22 only. - -- It hangs on `main` too, even with an open-ended range, over a draft-18 - upstream. Two causes: our moq-transport publisher answers only one-group - FETCHes, so the fill for the missing run never arrives (a one-group range - works); and the live join can skip a group that is already cached - (`TrackState::answered` in `rs/moq-net/src/model/track.rs`). -- Also check whether the moq-lite publisher has the same wait when it serves a - bounded SUBSCRIBE from a moq-transport copy that was only fetched. -- Multi-group past reads between our own moq-transport nodes are refused until - hole markers land (decided on - [moq-dev/moq#7625](https://github.com/moq-dev/moq/pull/7625)); a refused run - must come back as dropped, not hang. -- Enable the draft-18 cases of #7631's bounded-resume tests as the regression. -- A relay loses a group on a history SUBSCRIBE (folded in 2026-11-02, - maintainer, from [moq-dev/moq#7625](https://github.com/moq-dev/moq/pull/7625)): - with the `fetched_copy` fixture in `fetch_only` on draft-22, where the relay - already fetched and cached group 2, a fresh session's `1..3` read gets only - group 2. The relay asks upstream for groups 1 to 2 and upstream sends only - group 2, so group 1 is lost silently (`main` hangs instead). Group 1 must - arrive or come back as dropped. The `1..2` case #7631 tests is fine. From 3cb8031afd50805c7f17f344d128b007c82338c1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:43:21 -0800 Subject: [PATCH 3/9] fix(net): judge resumed history against each reader delay --- rs/moq-net/benches/track.rs | 28 +-- rs/moq-net/src/model/track.rs | 84 ++++++++- rs/moq-net/src/model/track/miss.rs | 2 +- rs/moq-net/tests/integration/fetch_only.rs | 201 ++++++++++++--------- 4 files changed, 210 insertions(+), 105 deletions(-) diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index b31e74d69d3..df2c7dd5604 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -1915,7 +1915,7 @@ struct Fills { } impl Fills { - fn new(readers: usize, runs: u64) -> Self { + fn new(readers: usize, runs: u64, budget: Option) -> Self { let broadcast = broadcast::Info::default().produce(); let mut track = broadcast.create_track("bench", timed()).unwrap(); let dynamic = track.requests(); @@ -1933,7 +1933,7 @@ impl Fills { .chain([track::Range::groups(start..)]); let subscription = track::Subscription::default() .with_ranges(ranges) - .with_delay(track::Delay::default().with_max(AGED_BUDGET)); + .with_delay(track::Delay::default().with_max(budget)); let readers = (0..readers).map(|_| track.subscribe(subscription.clone())).collect(); Self { _broadcast: broadcast, @@ -1975,17 +1975,23 @@ impl Fills { /// swept over the readers and the runs. Each reader must pay the same per run, so a cost /// that grows with the audience or the runs shows up as a slope. fn bench_fill(c: &mut Criterion) { - let mut group = c.benchmark_group("track_fill"); - group.sample_size(10); - for readers in [1usize, 8, 64] { - for runs in [1u64, 64, 1024] { - group.throughput(Throughput::Elements(runs * readers as u64)); - group.bench_function(BenchmarkId::new(format!("{readers}r"), runs), |b| { - b.iter_batched_ref(|| Fills::new(readers, runs), Fills::run, BatchSize::PerIteration); - }); + for (name, budget) in [("track_fill", Some(AGED_BUDGET)), ("track_fill_unlimited", None)] { + let mut group = c.benchmark_group(name); + group.sample_size(10); + for readers in [1usize, 8, 64] { + for runs in [1u64, 64, 1024] { + group.throughput(Throughput::Elements(runs * readers as u64)); + group.bench_function(BenchmarkId::new(format!("{readers}r"), runs), |b| { + b.iter_batched_ref( + || Fills::new(readers, runs, budget), + Fills::run, + BatchSize::PerIteration, + ); + }); + } } + group.finish(); } - group.finish(); } /// Readers whose disjoint ranges all fall below where the feed starts, and the handler that diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c1f90a1c3aa..16b938a4933 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -526,8 +526,8 @@ pub(crate) struct TrackState { // state, never judged from the cache. A bounded read's holes are then misses up to its // end at once, with nothing to wait for (see [`Self::nothing_arrives`]). unfed: bool, - // Readers start at this group: the route's live feed starts past a gap after - // everything cached, so nothing bounds how old the cache below it is. + // The route's live feed starts past a gap after everything cached, so the cache + // below this group is unjudgeable for readers with a finite delay budget. live_floor: Option, // Where production stopped, snapshotted when the open groups are released (an @@ -1267,7 +1267,6 @@ impl TrackState { ) -> Poll>> { let start = index.saturating_sub(self.offset); let readable = self.serves_bounded(window); - let min_sequence = min_sequence.max(self.live_floor.unwrap_or(0)); for (i, (sequence, stamp)) in self.arrival.iter().enumerate().skip(start).filter(|_| readable) { // A cut group is handed out too, so its reader sees why it ended. if *sequence >= min_sequence @@ -1395,7 +1394,7 @@ impl TrackState { } let best = window - .runs(next_sequence.max(self.live_floor.unwrap_or(0))) + .runs(next_sequence) .filter(|_| self.serves_bounded(window)) .find_map(|run| { // A cut group is handed out too, so its reader sees why it ended. @@ -1747,6 +1746,11 @@ impl TrackState { /// [`Self::is_stale`] for group `sequence`, whether this track holds it or not: its /// successor here starts a full `budget` behind the edge, or arrived a full `budget` ago. fn drifted(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { + // A gap after an idle cache gives its old groups no known reach. A finite + // budget cannot admit them, but an unlimited history read still owes them. + if budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) { + return true; + } if self.superseded(sequence, &edge.servable, budget) { return true; } @@ -1778,6 +1782,9 @@ impl TrackState { superseded: Option<&Superseded>, waiter: &kio::Waiter, ) -> bool { + if budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) { + return true; + } // Unbounded until a group lands above. Registered before judging, so a racing abort // is observed or wakes us, and armed for whichever successor this pass selected, so // re-selecting re-arms it. @@ -2622,9 +2629,8 @@ impl TrackState { if !self.received || self.closed || !self.readable() { return None; } - let floor = groups.start.max(self.live_floor.unwrap_or(0)); let asked = self.start_sequence.is_some(); - self.find_landable(floor..groups.end, asked, self.arrival_grace(), |open| { + self.find_landable(groups, asked, self.arrival_grace(), |open| { window.clip(open).map(|run| run.start) }) } @@ -2907,7 +2913,8 @@ impl TrackState { /// is current up to there. `start` is the first group the feed serves, when the answer /// names it; otherwise the largest stands in. A feed that starts past a gap after /// everything cached leaves the cache unjudgeable, since nothing bounds how far an old - /// group reached, so readers skip it; whatever the feed delivers stays. + /// group reached, so readers with a finite delay budget skip it; an unlimited read + /// still serves it. Whatever the feed delivers stays. /// /// Readers wait until the cache shows that position: before then its newest group would /// read as the live edge, though the route already holds a newer one. @@ -7246,14 +7253,13 @@ impl Cursor { if !state.serves_bounded(&self.filter.window) { return None; } - let low = state.live_floor.unwrap_or(0); loop { let run = match ascending { true => self.filter.owed.first(), false => self.filter.owed.last(), } .cloned()?; - let from = (*run.start()).max(low); + let from = *run.start(); let found = (from <= *run.end()) .then(|| { let mut slots = state.lookup.range(from..=*run.end()).map(|(_, slot)| slot); @@ -8590,7 +8596,7 @@ impl Subscriber { Inner::Resume(resume, _) => return resume.has_unread_group(unread), }; let state = cursor.state.read(); - let floor = cursor.floor().max(state.live_floor.unwrap_or(0)).max(unread.start); + let floor = cursor.floor().max(unread.start); let eligible = |sequence: u64| { sequence >= floor && super::subscription::before_end(sequence, unread.end) @@ -12272,6 +12278,64 @@ mod test { assert_eq!(drain(&mut subscriber), vec![2]); } + /// An idle cache's old groups and the holes after them stay owed by unlimited reads. + /// A refused run reports every missing sequence and releases a bounded reader, while + /// an open reader keeps receiving the live feed. Both cursor orders use the same rule. + #[test] + #[cfg(feature = "async")] + fn a_resumed_history_read_accounts_for_cached_and_refused_groups() { + for ordered in [false, true] { + for bounded in [false, true] { + let mut producer = track_producer("test", Info::default()); + let requests = producer.requests(); + producer.received(); + framed(&producer, 0, 1).finish().unwrap(); + producer.set_idle(); + let subscription = Subscription::default() + .with_delay(Delay::default().with_max(None)) + .with_range(( + Bound::Included(0), + if bounded { Bound::Excluded(3) } else { Bound::Unbounded }, + )); + let mut reader = producer.subscribe(subscription); + let waiter = kio::Waiter::noop(); + assert!(reader.poll_drop(&waiter).is_pending()); + assert_eq!(read_now(&mut reader, ordered), None, "the copy is idle"); + + producer.start_at(3).unwrap(); + producer.set_live(Some(Position::group(3)), None); + framed(&producer, 3, 1).finish().unwrap(); + assert_eq!(read_now(&mut reader, ordered), Some(Some(0)), "cached history"); + // An arrival-order reader may take the live group before its history. + if !bounded && !ordered { + assert_eq!(read_now(&mut reader, ordered), Some(Some(3))); + } + assert_eq!(read_now(&mut reader, ordered), None, "the missing run is owed"); + let Poll::Ready(Ok(fill)) = requests.poll_requested_fill(&waiter) else { + panic!("the missing history was never requested"); + }; + assert_eq!((fill.start(), fill.end()), (Position::group(1), 3)); + fill.refuse_groups(1..3, Error::Unsupported); + assert!( + reader.poll_drop(&waiter).is_pending(), + "a late live arrival may still answer" + ); + elapse(&producer, ledger::GRACE); + assert_eq!(reader.poll_drop(&waiter), Poll::Ready((1..3, StreamError::NotFound))); + assert!(reader.poll_drop(&waiter).is_pending(), "each drop is reported once"); + if bounded { + assert_eq!(read_now(&mut reader, ordered), Some(None), "nothing remains owed"); + } else { + if ordered { + assert_eq!(read_now(&mut reader, ordered), Some(Some(3))); + } + framed(&producer, 4, 1).finish().unwrap(); + assert_eq!(read_now(&mut reader, ordered), Some(Some(4)), "the feed survives"); + } + } + } + } + /// The cancel that idles a copy resets the group in flight, which the publisher never /// gave up: no reader hears it dropped, and the next feed sends it again. #[test] diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 5fd4639dd4b..f3e09c6df26 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -657,7 +657,7 @@ impl Misses { }) || self.budget != Some(drift.budget) || self.grace != Some(grace) || self.ended != ended; - let low = state.live_floor.unwrap_or(0).max(key.0); + let low = key.0; let top = newest.min(u64::MAX - 1); let mut candidates = RangeSet::new(); if terms { diff --git a/rs/moq-net/tests/integration/fetch_only.rs b/rs/moq-net/tests/integration/fetch_only.rs index 3e38135fae9..0e5c23e68fb 100644 --- a/rs/moq-net/tests/integration/fetch_only.rs +++ b/rs/moq-net/tests/integration/fetch_only.rs @@ -469,100 +469,135 @@ async fn a_relay_answers_a_live_subscribe_on_a_fetched_copy() { /// A copy kept after a fetch resumes a bounded subscription that ends at or below the /// upstream's new Largest Object, and reads it, though no group past its end arrives. -/// Draft 22 only: a draft-18 join is live, so these groups come from one fill of the run, -/// and our publisher answers only a one-group FETCH. +/// A draft-18 join fills the missing run with one FETCH, which our publisher refuses +/// until hole markers are supported. That refusal ends the read instead of hanging. #[kio::test] async fn a_resumed_bounded_subscription_reads_below_the_answered_edge() { - kio::clock::timeout(TIMEOUT, async { - let version: Version = "moq-transport-22".parse().unwrap(); - let publisher = produce_origin(1); - let (broadcast, track) = publish_live(&publisher, 2); - - let client = produce_origin(2); - let mut options = MockConnectOptions::new(version); - options.server_publish = Some(publisher.consume()); - options.client_subscribe = Some(client.clone()); - let pair = connect_mock(options).await; - - // The fetch learns Largest Object (1, 1), and leaves the copy unfed. - let remote = resolve(&client).await; - assert_eq!(fetch(&remote, 0).await.expect("fetch"), frames(0)); - for sequence in 2..4 { - write_group(&track, sequence); - } + for version in ["moq-transport-18", "moq-transport-20", "moq-transport-22"] { + kio::clock::timeout(TIMEOUT, async { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let (broadcast, track) = publish_live(&publisher, 2); - // Past the reported edge, so it subscribes; the answer names group 3, which the - // bounded feed never delivers. - let copy = remote.track("video").unwrap(); - let subscription = moq_net::track::Subscription::default() - .with_order(moq_net::track::Order::Ascending) - .with_delay(moq_net::track::Delay::default().with_max(None)) - .with_range(1..3); - let mut sub = kio::clock::timeout(TIMEOUT, copy.subscribe(subscription)) - .await - .expect("subscribe timed out") - .expect("subscribe"); - // Read in arrival order, whichever group lands first. - let mut read = vec![next_frames(&mut sub).await, next_frames(&mut sub).await]; - read.sort(); - assert_eq!(read, [Some(frames(1)), Some(frames(2))]); - assert!(track.subscription().is_some(), "the read never subscribed"); - - drop((sub, copy, remote, pair, track, broadcast, client, publisher)); - }) - .await - .expect("timed out"); + let client = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(client.clone()); + let pair = connect_mock(options).await; + + // The fetch learns Largest Object (1, 1), and leaves the copy unfed. + let remote = resolve(&client).await; + assert_eq!(fetch(&remote, 0).await.expect("fetch"), frames(0)); + for sequence in 2..4 { + write_group(&track, sequence); + } + + // Past the reported edge, so it subscribes; the answer names group 3, which the + // bounded feed never delivers. + let copy = remote.track("video").unwrap(); + let subscription = moq_net::track::Subscription::default() + .with_order(moq_net::track::Order::Ascending) + .with_delay(moq_net::track::Delay::default().with_max(None)) + .with_range(1..3); + let mut sub = kio::clock::timeout(TIMEOUT, copy.subscribe(subscription)) + .await + .expect("subscribe timed out") + .expect("subscribe"); + // Read in arrival order, whichever group lands first. A refused multi-group + // FETCH gives up the whole run; it never falls back to fetching each group. + let mut read = Vec::new(); + while let Some(group) = next_frames(&mut sub).await { + read.push(group); + } + read.sort(); + if version == "moq-transport-18".parse().unwrap() { + assert!(read.is_empty(), "the unsupported multi-group FETCH was refused"); + } else { + assert_eq!(read, [frames(1), frames(2)]); + } + assert!(track.subscription().is_some(), "the read never subscribed"); + + drop((sub, copy, remote, pair, track, broadcast, client, publisher)); + }) + .await + .expect("timed out"); + } } /// A copy that went idle holding a group resumes a bounded subscription of that group after /// the upstream moved on, and still reads it: the feed starts where it asked, not at the -/// answer's Largest, so the cached group stays below no floor. Draft 22 only: a draft-18 -/// join is live, so it starts at the answer's Largest whatever it asked. +/// answer's Largest. A live join also keeps the cached history for an unlimited read. #[kio::test] async fn a_resumed_bounded_subscription_keeps_its_cached_start() { - kio::clock::timeout(TIMEOUT, async { - let version: Version = "moq-transport-22".parse().unwrap(); - let publisher = produce_origin(1); - let (broadcast, track) = publish_live(&publisher, 1); - - let client = produce_origin(2); - let mut options = MockConnectOptions::new(version); - options.server_publish = Some(publisher.consume()); - options.client_subscribe = Some(client.clone()); - let pair = connect_mock(options).await; - - // A live read caches group 0, then leaves, so the copy goes idle holding it. - let remote = resolve(&client).await; - let copy = remote.track("video").unwrap(); - let mut live = kio::clock::timeout(TIMEOUT, copy.subscribe(None)) - .await - .expect("subscribe timed out") - .expect("subscribe"); - assert_eq!(next_frames(&mut live).await, Some(frames(0))); - drop(live); - let mut watch = track.clone(); - while watch.subscription().is_some() { - kio::clock::timeout(TIMEOUT, watch.subscription_changed()) + let versions = moq_net::Versions::all(); + for version in versions.iter().copied().chain(["moq-lite-07-wip".parse().unwrap()]) { + kio::clock::timeout(TIMEOUT, async { + let publisher = produce_origin(1); + let (broadcast, track) = publish_live(&publisher, 1); + + let client = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(client.clone()); + let pair = connect_mock(options).await; + + // A live read caches group 0, then leaves, so the copy goes idle holding it. + let remote = resolve(&client).await; + let copy = remote.track("video").unwrap(); + let mut live = kio::clock::timeout(TIMEOUT, copy.subscribe(None)) .await - .expect("unsubscribe timed out") - .unwrap(); - } - for sequence in 1..4 { - write_group(&track, sequence); - } + .expect("subscribe timed out") + .expect("subscribe"); + assert_eq!(next_frames(&mut live).await, Some(frames(0))); + drop(live); + let mut watch = track.clone(); + while watch.subscription().is_some() { + kio::clock::timeout(TIMEOUT, watch.subscription_changed()) + .await + .expect("unsubscribe timed out") + .unwrap(); + } + for sequence in 1..4 { + write_group(&track, sequence); + } + + // Past the edge the live read learned, so it subscribes; the answer names group 3. + let subscription = moq_net::track::Subscription::default() + .with_delay(moq_net::track::Delay::default().with_max(None)) + .with_range(0..1); + let mut sub = kio::clock::timeout(TIMEOUT, copy.subscribe(subscription)) + .await + .expect("subscribe timed out") + .expect("subscribe"); + assert_eq!(next_frames(&mut sub).await, Some(frames(0))); - // Past the edge the live read learned, so it subscribes; the answer names group 3. + drop((sub, copy, remote, pair, track, broadcast, client, publisher)); + }) + .await + .expect("timed out"); + } +} + +/// A relay's retained group cannot hide a missing earlier group in the same read. +#[kio::test] +async fn a_relay_accounts_for_every_group_in_multi_group_history() { + for (upstream, downstream) in [ + ("moq-transport-18", "moq-transport-22"), + ("moq-transport-22", "moq-transport-22"), + ("moq-transport-18", "moq-lite-07-wip"), + ("moq-transport-22", "moq-lite-07-wip"), + ] { + let copy = fetched_copy(upstream.parse().unwrap(), downstream.parse().unwrap()).await; let subscription = moq_net::track::Subscription::default() + .with_order(moq_net::track::Order::Ascending) .with_delay(moq_net::track::Delay::default().with_max(None)) - .with_range(0..1); - let mut sub = kio::clock::timeout(TIMEOUT, copy.subscribe(subscription)) - .await - .expect("subscribe timed out") - .expect("subscribe"); - assert_eq!(next_frames(&mut sub).await, Some(frames(0))); - - drop((sub, copy, remote, pair, track, broadcast, client, publisher)); - }) - .await - .expect("timed out"); + .with_range(1..3); + let mut sub = copy.fresh.subscribe(subscription).await.unwrap(); + let mut groups = Vec::new(); + while let Some(group) = next_frames(&mut sub).await { + groups.push(group); + } + groups.sort(); + assert_eq!(groups, [frames(1), frames(2)], "{upstream} to {downstream}"); + } } From 31ac24446e443b948de7e34bfc6aa31cc459788b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:51:23 -0800 Subject: [PATCH 4/9] fix(net): release bounded lite history at its answered start --- rs/moq-net/src/lite/subscriber.rs | 10 +++++++++- rs/moq-net/tests/integration/fetch_only.rs | 2 +- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 8a10609c858..d35fb36d4d4 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -9536,7 +9536,15 @@ impl ServeLoop { let served = start.starts.iter().flatten().min().copied(); // Where the live feed is, on versions whose answer says. if serve.subscriber.version.has_largest() { - self.serving.set_live(start.largest, served); + // A bounded feed may never deliver Largest. Its START resolves + // what it owes; the range and arrival ledger wait for those + // groups, rather than an edge outside the requested history. + match start.largest { + Some(largest) if bounded(&active.ranges) => { + self.serving.set_live_ahead(largest, served) + } + largest => self.serving.set_live(largest, served), + } if let Some(largest) = start.largest { self.serving.report_largest(largest); } diff --git a/rs/moq-net/tests/integration/fetch_only.rs b/rs/moq-net/tests/integration/fetch_only.rs index 0e5c23e68fb..59399a0d070 100644 --- a/rs/moq-net/tests/integration/fetch_only.rs +++ b/rs/moq-net/tests/integration/fetch_only.rs @@ -569,7 +569,7 @@ async fn a_resumed_bounded_subscription_keeps_its_cached_start() { .await .expect("subscribe timed out") .expect("subscribe"); - assert_eq!(next_frames(&mut sub).await, Some(frames(0))); + assert_eq!(next_frames(&mut sub).await, Some(frames(0)), "{version}"); drop((sub, copy, remote, pair, track, broadcast, client, publisher)); }) From 08a4b56b3ea3b464088956c701127a0c13c5c705 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 00:56:43 -0800 Subject: [PATCH 5/9] fix(net): keep resumed cache admission scoped to subscriptions --- rs/moq-net/src/model/track.rs | 75 ++++++++++++++++-------------- rs/moq-net/src/model/track/miss.rs | 7 ++- 2 files changed, 45 insertions(+), 37 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 16b938a4933..cacf57ce885 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1740,17 +1740,19 @@ impl TrackState { /// against itself, and a group received from past its first frame carries a later /// frame's timestamp on a low sequence without being an edge at all. fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { - self.lookup.contains_key(&sequence) && self.drifted(sequence, edge, budget) + self.lookup.contains_key(&sequence) + && (self.unjudgeable(sequence, budget) || self.drifted(sequence, edge, budget)) } - /// [`Self::is_stale`] for group `sequence`, whether this track holds it or not: its - /// successor here starts a full `budget` behind the edge, or arrived a full `budget` ago. + /// A finite subscription cannot admit a cache whose reach its resumed feed cannot + /// bound. Unlimited history still owes it, and one-shot fetches judge their own budget. + fn unjudgeable(&self, sequence: u64, budget: Duration) -> bool { + budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) + } + + /// Whether group `sequence`, cached or not, has a successor a full `budget` behind + /// the edge, or one that arrived a full `budget` ago. fn drifted(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { - // A gap after an idle cache gives its old groups no known reach. A finite - // budget cannot admit them, but an unlimited history read still owes them. - if budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) { - return true; - } if self.superseded(sequence, &edge.servable, budget) { return true; } @@ -1782,9 +1784,6 @@ impl TrackState { superseded: Option<&Superseded>, waiter: &kio::Waiter, ) -> bool { - if budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) { - return true; - } // Unbounded until a group lands above. Registered before judging, so a racing abort // is observed or wakes us, and armed for whichever successor this pass selected, so // re-selecting re-arms it. @@ -7259,17 +7258,12 @@ impl Cursor { false => self.filter.owed.last(), } .cloned()?; - let from = *run.start(); - let found = (from <= *run.end()) - .then(|| { - let mut slots = state.lookup.range(from..=*run.end()).map(|(_, slot)| slot); - match ascending { - true => slots.find(|slot| slot.is_read()), - false => slots.rev().find(|slot| slot.is_read()), - } - .map(Slot::serving) - }) - .flatten(); + let mut slots = state.lookup.range(run.clone()).map(|(_, slot)| slot); + let found = match ascending { + true => slots.find(|slot| slot.is_read()), + false => slots.rev().find(|slot| slot.is_read()), + } + .map(Slot::serving); let Some(group) = found else { self.filter.owed.remove(run); continue; @@ -7414,7 +7408,9 @@ impl Cursor { let Some(owed) = state.owed(groups.clone(), window) else { return Poll::Ready(Ok(())); }; - if state.poll_drifted(owed, &drift.edge.servable, drift.budget, None, waiter) { + if state.unjudgeable(owed, drift.budget) + || state.poll_drifted(owed, &drift.edge.servable, drift.budget, None, waiter) + { return Poll::Ready(Ok(())); } // The ledger may stop owing it before the budget gives up on it. @@ -7905,18 +7901,18 @@ impl Cursor { let _ = state.cache.wakes().watch_landing(sequence, waiter); // Judged whether or not the cache still holds the headless copy: an evicted // one goes stale all the same, and counts as a group of unknown size. - state - .poll_drifted(sequence, &servable, drift.budget, None, waiter) - .then(|| { - let content = match state.lookup.get(&sequence) { - Some(slot) => slot.group.consume().content(), - None => stats::Content { - groups: 1, - ..Default::default() - }, - }; - (sequence, content) - }) + (state.unjudgeable(sequence, drift.budget) + || state.poll_drifted(sequence, &servable, drift.budget, None, waiter)) + .then(|| { + let content = match state.lookup.get(&sequence) { + Some(slot) => slot.group.consume().content(), + None => stats::Content { + groups: 1, + ..Default::default() + }, + }; + (sequence, content) + }) }) .collect(); Poll::Ready(Ok(stale)) @@ -12276,6 +12272,15 @@ mod test { let mut subscriber = producer .subscribe(Subscription::default().with_delay(Delay::default().with_max(Duration::from_millis(400)))); assert_eq!(drain(&mut subscriber), vec![2]); + // A one-shot fetch still serves that cache: an unknown reach is no proof + // that the requested group exceeded the fetch's budget. + let mut fetched = producer + .consume() + .fetch_group(0, group::Fetch::default().with_max_delay(Duration::from_millis(400))) + .now_or_never() + .expect("the cache answers") + .expect("cached fetch"); + assert!(fetched.read_frame().now_or_never().unwrap().unwrap().is_some()); } /// An idle cache's old groups and the holes after them stay owed by unlimited reads. diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index f3e09c6df26..aed78ff61c9 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -918,7 +918,9 @@ impl Misses { // budget is stale throughout. let stale: Vec> = asks .iter() - .filter(|run| state.drifted(run.end - 1, &drift.edge, drift.budget)) + .filter(|run| { + state.unjudgeable(run.end - 1, drift.budget) || state.drifted(run.end - 1, &drift.edge, drift.budget) + }) .cloned() .collect(); for run in stale { @@ -962,7 +964,8 @@ impl Misses { let mut next = self.wanted.iter().next().map(|run| run.start); while let Some(from) = next { let bound = state.arrival(from, servable).map(|slot| slot.group.sequence); - let drifted = state.poll_drifted(from, servable, drift.budget, None, waiter); + let drifted = + state.unjudgeable(from, drift.budget) || state.poll_drifted(from, servable, drift.budget, None, waiter); next = self.stretches(state, from, bound, drifted, &mut stale, &mut arrived); } (stale, arrived) From 07f443e086aa07ad7a40b82309226f39d3da9b45 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 01:09:16 -0800 Subject: [PATCH 6/9] fix(net): wait above discarded resume history and verify refusal drops --- rs/moq-net/src/ietf/subscriber.rs | 54 +++++++++++++++++++++++++++++++ rs/moq-net/src/model/track.rs | 40 ++++++++++++++++++++--- 2 files changed, 90 insertions(+), 4 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 36c16c997a8..acb4b18bbff 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -15194,6 +15194,60 @@ mod stitch_tests { assert!(run.track.consume().peek_group(SEQUENCE).is_none(), "nothing is cached"); } + /// A refused multi-group FETCH names every owed group dropped, without retrying + /// each group or ending the live track. This exercises the wire refusal decoder. + #[kio::test] + async fn a_refused_range_fetch_reports_every_owed_group_dropped() { + for version in [Version::Draft18, Version::Draft20, Version::Draft22] { + let mut refusal = Vec::new(); + let mut writer = Encoder::new(&mut refusal, version.into()); + writer.varint(ietf::RequestError::ID).unwrap(); + ietf::RequestError { + error_code: request::NOT_SUPPORTED, + reason_phrase: "FETCH spans more than one group".into(), + retry_interval: 0, + } + .encode(&mut writer, version) + .unwrap(); + let session = ScriptedSession::per_stream_eof(vec![vec![], refusal]); + let mut run = GroupFetchRun::with_session(version, session.clone()).await; + let now = kio::clock::now(); + run.track.broadcast().pool.poll(now, &kio::Waiter::noop()); + run.track.received(); + run.track.start_at(3).unwrap(); + run.track.create_group(3u64.into()).unwrap().finish().unwrap(); + let requests = run.track.requests(); + let mut reader = run.track.subscribe( + track::Subscription::default() + .with_range(1..3) + .with_delay(track::Delay::default().with_max(None)), + ); + let waiter = kio::Waiter::noop(); + assert!(reader.poll_drop(&waiter).is_pending()); + assert!(reader.poll_recv_group(&waiter).is_pending()); + let Poll::Ready(Ok(fill)) = requests.poll_requested_fill(&waiter) else { + panic!("{version}: missing history was not requested"); + }; + assert_eq!((fill.start().group, fill.end()), (1, 3)); + run.subscriber + .clone() + .run_group_fetch(Path::from_static("broadcast"), "video".into(), fill, None) + .await; + // The peer may still deliver a late subgroup stream during the arrival grace. + run.track.broadcast().pool.poll(now + crate::ledger::GRACE, &waiter); + assert_eq!( + reader.poll_drop(&waiter), + Poll::Ready((1..3, crate::StreamError::NotFound)), + "{version}" + ); + assert!(reader.poll_drop(&waiter).is_pending(), "each drop is reported once"); + assert!(matches!(reader.poll_recv_group(&waiter), Poll::Ready(Ok(None)))); + assert!(!run.track.is_closed(), "the live track survives"); + // The harness opened its unused data stream, then the one range FETCH. + assert_eq!(session.log.bi_opens(), 2, "no per-group fallback"); + } + } + /// A third-party publisher may answer a range even though ours refuses it. The subscriber /// caches the supplied groups and reports the gaps between them as misses. #[kio::test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index cacf57ce885..6f97c325395 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -7405,12 +7405,18 @@ impl Cursor { } let window = &self.filter.window; self.poll(waiter, |state| { - let Some(owed) = state.owed(groups.clone(), window) else { + // An unjudgeable old gap cannot stand in for a fresh gap above the floor. + let floor = match drift.budget == Duration::MAX { + true => groups.start, + false => groups.start.max(state.live_floor.unwrap_or(0)), + }; + if floor >= groups.end { + return Poll::Ready(Ok(())); + } + let Some(owed) = state.owed(floor..groups.end, window) else { return Poll::Ready(Ok(())); }; - if state.unjudgeable(owed, drift.budget) - || state.poll_drifted(owed, &drift.edge.servable, drift.budget, None, waiter) - { + if state.poll_drifted(owed, &drift.edge.servable, drift.budget, None, waiter) { return Poll::Ready(Ok(())); } // The ledger may stop owing it before the budget gives up on it. @@ -12283,6 +12289,32 @@ mod test { assert!(fetched.read_frame().now_or_never().unwrap().unwrap().is_some()); } + /// An old gap below a resumed cache's floor never hides a fresh group still owed + /// above it: an ordered finite-budget reader must wait for that fresh group. + #[test] + #[cfg(feature = "async")] + fn an_unjudgeable_gap_does_not_hide_a_fresh_ordered_gap() { + let mut producer = track_producer("test", Info::default()); + producer.received(); + for sequence in [0, 4] { + framed(&producer, sequence, 1).finish().unwrap(); + } + producer.set_idle(); + let mut reader = producer.subscribe( + Subscription::default() + .with_delay(Delay::default().with_max(Duration::from_millis(400))) + .with_range(0..), + ); + producer.set_live(Some(Position::group(8)), None); + for sequence in [6, 8] { + framed(&producer, sequence, 1).finish().unwrap(); + } + assert_eq!(read_now(&mut reader, true), None, "group 5 is still owed"); + framed(&producer, 5, 1).finish().unwrap(); + assert_eq!(read_now(&mut reader, true), Some(Some(5))); + assert_eq!(read_now(&mut reader, true), Some(Some(6))); + } + /// An idle cache's old groups and the holes after them stay owed by unlimited reads. /// A refused run reports every missing sequence and releases a bounded reader, while /// an open reader keeps receiving the live feed. Both cursor orders use the same rule. From d78b3e4cf7910f61c22e2f8f1d62185df07a5f43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 01:33:22 -0800 Subject: [PATCH 7/9] fix(net): split missing runs at the reader cache floor --- rs/moq-net/src/model/track.rs | 70 ++++++++++++++++++++++++++---- rs/moq-net/src/model/track/miss.rs | 33 +++++++++++--- 2 files changed, 88 insertions(+), 15 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 6f97c325395..b1bed618eb6 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1741,13 +1741,13 @@ impl TrackState { /// frame's timestamp on a low sequence without being an edge at all. fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { self.lookup.contains_key(&sequence) - && (self.unjudgeable(sequence, budget) || self.drifted(sequence, edge, budget)) + && (sequence < self.cache_floor(budget) || self.drifted(sequence, edge, budget)) } /// A finite subscription cannot admit a cache whose reach its resumed feed cannot /// bound. Unlimited history still owes it, and one-shot fetches judge their own budget. - fn unjudgeable(&self, sequence: u64, budget: Duration) -> bool { - budget != Duration::MAX && self.live_floor.is_some_and(|floor| sequence < floor) + fn cache_floor(&self, budget: Duration) -> u64 { + self.live_floor.filter(|_| budget != Duration::MAX).unwrap_or(0) } /// Whether group `sequence`, cached or not, has a successor a full `budget` behind @@ -7406,10 +7406,7 @@ impl Cursor { let window = &self.filter.window; self.poll(waiter, |state| { // An unjudgeable old gap cannot stand in for a fresh gap above the floor. - let floor = match drift.budget == Duration::MAX { - true => groups.start, - false => groups.start.max(state.live_floor.unwrap_or(0)), - }; + let floor = groups.start.max(state.cache_floor(drift.budget)); if floor >= groups.end { return Poll::Ready(Ok(())); } @@ -7907,7 +7904,7 @@ impl Cursor { let _ = state.cache.wakes().watch_landing(sequence, waiter); // Judged whether or not the cache still holds the headless copy: an evicted // one goes stale all the same, and counts as a group of unknown size. - (state.unjudgeable(sequence, drift.budget) + (sequence < state.cache_floor(drift.budget) || state.poll_drifted(sequence, &servable, drift.budget, None, waiter)) .then(|| { let content = match state.lookup.get(&sequence) { @@ -12315,6 +12312,63 @@ mod test { assert_eq!(read_now(&mut reader, true), Some(Some(6))); } + /// Cache pressure can make one missing run span the old cache floor. A finite + /// reader gives up only the unjudgeable prefix, keeping the fresh suffix's fill. + #[test] + #[cfg(feature = "async")] + fn a_fill_crossing_the_cache_floor_keeps_its_fresh_suffix() { + let budget = Duration::from_millis(400); + for cached in [false, true] { + let mut producer = track_producer("test", Info::default()); + let requests = producer.requests(); + producer.received(); + for sequence in 0..3 { + framed(&producer, sequence, 1).finish().unwrap(); + } + producer.set_idle(); + producer.start_at(6).unwrap(); + producer.set_live(Some(Position::group(6)), None); + framed(&producer, 6, 1).finish().unwrap(); + if !cached { + for sequence in 0..3 { + evict(&producer, sequence); + } + } + let mut reader = producer.subscribe( + Subscription::default() + .with_delay(Delay::default().with_max(budget)) + .with_range(0..6), + ); + let waiter = kio::Waiter::noop(); + assert_eq!(read_now(&mut reader, false), None); + let Poll::Ready(Ok(fill)) = requests.poll_requested_fill(&waiter) else { + panic!("cached: {cached}: missing history must be requested"); + }; + assert_eq!( + (fill.start().group, fill.end()), + (3, 6), + "only fresh history is fetched" + ); + assert_eq!(read_now(&mut reader, false), None, "the fill is still owed"); + assert_eq!(fill.wanted(), Some(5), "the fresh suffix stays wanted"); + for sequence in 3..6 { + let mut group = fill.accept(sequence, None).unwrap(); + group.write_frame(None, &b"x"[..]).unwrap(); + group.finish().unwrap(); + } + let mut got = Vec::new(); + loop { + match read_now(&mut reader, false) { + Some(Some(sequence)) => got.push(sequence), + Some(None) => break, + None => panic!("cached: {cached}: the bounded read still owes history"), + } + } + got.sort_unstable(); + assert_eq!(got, [3, 4, 5], "cached: {cached}"); + } + } + /// An idle cache's old groups and the holes after them stay owed by unlimited reads. /// A refused run reports every missing sequence and releases a bounded reader, while /// an open reader keeps receiving the live feed. Both cursor orders use the same rule. diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index aed78ff61c9..93f7313b931 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -914,13 +914,22 @@ impl Misses { state.cache.watch_wall(at, waiter); } + // An old cache's floor is a boundary inside a missing run, not a drift verdict + // for that whole run. Discard only its prefix before asking for the fresh suffix. + let floor = state.cache_floor(drift.budget); + let old: Vec<_> = asks + .overlapping(&(0..floor)) + .map(|run| run.start..run.end.min(floor)) + .collect(); + for run in old { + asks.remove(run.clone()); + self.give_up(run, given_up); + } // Starts never rewind, so a run whose newest group is stale under this reader's // budget is stale throughout. let stale: Vec> = asks .iter() - .filter(|run| { - state.unjudgeable(run.end - 1, drift.budget) || state.drifted(run.end - 1, &drift.edge, drift.budget) - }) + .filter(|run| state.drifted(run.end - 1, &drift.edge, drift.budget)) .cloned() .collect(); for run in stale { @@ -959,13 +968,23 @@ impl Misses { /// What [`Self::expire`] gives up as stale, and what arrived live, parking `waiter`. fn expired(&self, state: &TrackState, drift: &Drift, waiter: &kio::Waiter) -> (RangeSet, RangeSet) { - let (mut stale, mut arrived) = (RangeSet::new(), RangeSet::new()); + let floor = state.cache_floor(drift.budget); + // A floor raised after this fill was asked can cut through its wanted run. + let mut stale: RangeSet = self + .wanted + .overlapping(&(0..floor)) + .map(|run| run.start..run.end.min(floor)) + .collect(); + let mut arrived = RangeSet::new(); let servable = &drift.edge.servable; - let mut next = self.wanted.iter().next().map(|run| run.start); + let mut next = self + .wanted + .overlapping(&(floor..u64::MAX)) + .next() + .map(|run| run.start.max(floor)); while let Some(from) = next { let bound = state.arrival(from, servable).map(|slot| slot.group.sequence); - let drifted = - state.unjudgeable(from, drift.budget) || state.poll_drifted(from, servable, drift.budget, None, waiter); + let drifted = state.poll_drifted(from, servable, drift.budget, None, waiter); next = self.stretches(state, from, bound, drifted, &mut stale, &mut arrived); } (stale, arrived) From faaf98210fd5dea0daea23e8d9bac19acd6d3a1c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 07:17:14 -0800 Subject: [PATCH 8/9] fix(net): judge the cache floor without the cache, and in unread checks A finite reader picks a group below the cache floor, then judges it with the track lock released in between. An eviction in that window used to turn the skip into a delivery, since the floor verdict was gated on the cache still holding the group. The floor needs no cached group, so it now convicts first. A replaced copy's unread check also counted that skipped cache as unread for a finite budget, keeping the copy waiting until the budget expired. It now applies the same floor as delivery. Adds regressions for both, and for a fill whose wanted run a later floor cuts through. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 118 +++++++++++++++++++++++++++++++++- 1 file changed, 115 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b1bed618eb6..8c9741c31cb 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1739,9 +1739,12 @@ impl TrackState { /// The edge must sit strictly above the candidate. The live edge is never late /// against itself, and a group received from past its first frame carries a later /// frame's timestamp on a low sequence without being an edge at all. + /// + /// A group below [`Self::cache_floor`] is stale whether or not the cache still holds + /// it: an eviction between selecting it and this verdict must not hand it out. fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { - self.lookup.contains_key(&sequence) - && (sequence < self.cache_floor(budget) || self.drifted(sequence, edge, budget)) + sequence < self.cache_floor(budget) + || (self.lookup.contains_key(&sequence) && self.drifted(sequence, edge, budget)) } /// A finite subscription cannot admit a cache whose reach its resumed feed cannot @@ -8594,8 +8597,11 @@ impl Subscriber { Inner::Plain(cursor) => cursor, Inner::Resume(resume, _) => return resume.has_unread_group(unread), }; + let budget = cursor.subscription.read().budget(); let state = cursor.state.read(); - let floor = cursor.floor().max(unread.start); + // A finite budget skips the cache below its floor at once, so it is not unread. + let budget = clamp_max_delay(budget, state.max_age_bound()); + let floor = cursor.floor().max(state.cache_floor(budget)).max(unread.start); let eligible = |sequence: u64| { sequence >= floor && super::subscription::before_end(sequence, unread.end) @@ -12369,6 +12375,112 @@ mod test { } } + /// A budget narrowed while a fill is outstanding raises the cache floor through the + /// fill's wanted run. Only the unjudgeable prefix is given up; the fresh suffix stays + /// wanted and is delivered once the fill answers. + #[test] + #[cfg(feature = "async")] + fn a_floor_raised_under_an_asked_fill_keeps_its_fresh_suffix() { + let mut producer = track_producer("test", Info::default()); + let requests = producer.requests(); + producer.received(); + for sequence in 0..3 { + framed(&producer, sequence, 1).finish().unwrap(); + } + producer.set_idle(); + producer.start_at(6).unwrap(); + producer.set_live(Some(Position::group(6)), None); + framed(&producer, 6, 1).finish().unwrap(); + for sequence in 0..3 { + evict(&producer, sequence); + } + let unlimited = Subscription::default() + .with_delay(Delay::default().with_max(None)) + .with_range(0..6); + let mut reader = producer.subscribe(unlimited.clone()); + let waiter = kio::Waiter::noop(); + assert_eq!(read_now(&mut reader, true), None); + let Poll::Ready(Ok(fill)) = requests.poll_requested_fill(&waiter) else { + panic!("missing history must be requested"); + }; + assert_eq!( + (fill.start().group, fill.end()), + (0, 6), + "unlimited history is fetched whole" + ); + assert!(reader.poll_drop(&waiter).is_pending()); + + reader + .update(unlimited.with_delay(Delay::default().with_max(Duration::from_millis(400)))) + .unwrap(); + assert_eq!(read_now(&mut reader, true), None, "the fresh suffix is still owed"); + assert_eq!( + reader.poll_drop(&waiter), + Poll::Ready((0..3, StreamError::NotFound)), + "the unjudgeable prefix is given up" + ); + assert!(fill.wanted().is_some(), "the fresh suffix stays wanted"); + for sequence in 3..6 { + let mut group = fill.accept(sequence, None).unwrap(); + group.write_frame(None, &b"x"[..]).unwrap(); + group.finish().unwrap(); + } + for sequence in 3..6 { + assert_eq!(read_now(&mut reader, true), Some(Some(sequence))); + } + assert_eq!(read_now(&mut reader, true), Some(None)); + } + + /// A finite budget skips the cache below the floor at once, so a copy handing over + /// waits on none of it; an unlimited read still owes it. + #[test] + #[cfg(feature = "async")] + fn the_cache_below_the_floor_is_unread_only_to_unlimited_reads() { + let mut producer = track_producer("test", Info::default()); + producer.received(); + for sequence in 0..3 { + framed(&producer, sequence, 1).finish().unwrap(); + } + producer.set_idle(); + producer.set_live(Some(Position::group(6)), None); + framed(&producer, 6, 1).finish().unwrap(); + let delivered = Delivered::default(); + let unread = Unread { + start: 0, + end: Some(6), + delivered: &delivered, + }; + for (max, owed) in [(Some(Duration::from_millis(400)), false), (None, true)] { + let reader = producer.subscribe( + Subscription::default() + .with_delay(Delay::default().with_max(max)) + .with_range(0..), + ); + assert_eq!(reader.has_unread_group(&unread), owed, "max delay {max:?}"); + } + } + + /// A finite reader selects a group below the floor, then judges it with the lock + /// released in between. An eviction in that window must not turn the skip into a delivery. + #[test] + #[cfg(feature = "async")] + fn an_evicted_group_below_the_floor_stays_stale() { + let mut producer = track_producer("test", Info::default()); + producer.received(); + framed(&producer, 0, 1).finish().unwrap(); + producer.set_idle(); + producer.set_live(Some(Position::group(6)), None); + framed(&producer, 6, 1).finish().unwrap(); + evict(&producer, 0); + let state = producer.state.read(); + let edge = state.drift_edge(&Servable::ALL); + assert!(state.is_stale(0, &edge, Duration::from_millis(400))); + assert!( + !state.is_stale(0, &edge, Duration::MAX), + "unlimited history still owes it" + ); + } + /// An idle cache's old groups and the holes after them stay owed by unlimited reads. /// A refused run reports every missing sequence and releases a bounded reader, while /// an open reader keeps receiving the live feed. Both cursor orders use the same rule. From b1f8ac0ad5cdb2b012e57e76794ab0f35d7f19fe Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 12:00:11 -0800 Subject: [PATCH 9/9] fix(net): go live ahead only when a bounded lite feed never serves Largest A bounded lite-07 resume went live at its SUBSCRIBE_START answer for every bounded subscription. When a resolved range still covers the reported Largest, the feed does deliver it, and a finite read must judge an old cached group against that edge rather than against the cache's own newest group. Wait for Largest then, as `set_live` did, and keep the ahead transition for ranges that stop short of it, matching the IETF subscriber's rule. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 59 +++++++++++++++++++++++++------ 1 file changed, 49 insertions(+), 10 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 4f9c99b27db..787ac6fc9c1 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2555,6 +2555,35 @@ mod tests { ); } + /// A bounded feed goes live at its answer alone only when its resolved ranges stop short + /// of Largest. One that serves Largest waits for it, so a finite read judges an old + /// cached group against that edge rather than against itself. + #[kio::test] + async fn a_bounded_feed_goes_live_ahead_only_without_largest() { + let tail = kio::Producer::new(Tail::default()); + let Sub::Active(mut active) = feed_sub(&tail).await else { + unreachable!("feed_sub is active") + }; + let largest = Some(Position::group(10)); + active.ranges = vec![lite::Range::span(Some(0), 0, Some(10), None)]; + active.served = Some(0); + active.starts = vec![Some(0)]; + assert_eq!(active.never_serves(largest), None, "group 10 is in 0..11"); + + active.ranges = vec![lite::Range::span(Some(0), 0, Some(3), None)]; + assert_eq!(active.never_serves(largest), largest, "0..4 never reaches group 10"); + + active.ranges.push(lite::Range::span(Some(10), 0, Some(12), None)); + active.starts = vec![Some(0), Some(10)]; + assert_eq!(active.never_serves(largest), None, "a later range serves it"); + active.starts[1] = None; + assert_eq!(active.never_serves(largest), largest, "that range served nothing"); + + active.ranges[1].end = None; + assert_eq!(active.never_serves(largest), None, "an open feed reaches it"); + assert_eq!(active.never_serves(None), None, "nothing to go live at"); + } + /// QUIC does not order streams, so group 2's header can reach the relay before group 1's. /// A fetch held for group 1 keeps waiting through the grace and is served when it lands. #[kio::test] @@ -7578,6 +7607,19 @@ impl SubStream { Some(start..end) } + /// `largest`, when this bounded subscription's ranges, as SUBSCRIBE_START resolved them, + /// stop short of it, so the feed never delivers it. The range and arrival ledger then wait + /// for the groups it owes instead of an edge outside the requested history. A feed that + /// serves Largest waits for it, so an old cache is judged against that edge. + fn never_serves(&self, largest: Option) -> Option { + largest.filter(|largest| { + bounded(&self.ranges) + && self + .covered() + .is_none_or(|spans| !spans.iter().any(|span| span.contains(&largest.group))) + }) + } + /// The groups the publisher owed once it FINs a covered bounded lite-07 subscription /// without SUBSCRIBE_END: each range from its absolute start, or else from where /// SUBSCRIBE_START resolved delivery, up to its end. Groups below SUBSCRIBE_START were @@ -9585,16 +9627,15 @@ impl ServeLoop { } // No range starting anywhere leaves nothing owed. let served = start.starts.iter().flatten().min().copied(); + active.served = served; + active.starts = start.starts.clone(); // Where the live feed is, on versions whose answer says. if serve.subscriber.version.has_largest() { - // A bounded feed may never deliver Largest. Its START resolves - // what it owes; the range and arrival ledger wait for those - // groups, rather than an edge outside the requested history. - match start.largest { - Some(largest) if bounded(&active.ranges) => { - self.serving.set_live_ahead(largest, served) - } - largest => self.serving.set_live(largest, served), + // A bounded feed that never delivers Largest goes live at the + // answer alone; one that does waits for it to show. + match active.never_serves(start.largest) { + Some(largest) => self.serving.set_live_ahead(largest, served), + None => self.serving.set_live(start.largest, served), } if let Some(largest) = start.largest { self.serving.report_largest(largest); @@ -9616,8 +9657,6 @@ impl ServeLoop { _ => self.serving.start_at(served), }; } - active.served = served; - active.starts = start.starts.clone(); // The groups the SUBSCRIBE asked for below it are not // waited for, whatever the demand asks later. One that // still arrives is delivered, so they are not dropped.