From 616c35609e71b9f4f67a57d1aaafc3b1cea62ff2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 15:20:33 -0700 Subject: [PATCH 1/2] fix(net): fill no history below a subscription's start that its reader skips A reader with the default oldest range asked a fill for every group below SUBSCRIBE_START. On an untimed copy each looked fresh, so lite-05 and moq-transport pulled the publisher's whole cached history on every first subscribe, and lite-07 sent one FETCH per stale group for the publisher to refuse. Below a start the publisher resolved from a budget no tighter than the reader's (lite-06+), every group is stale: the reader gives the run up unasked. Below a live edge start, it fills only a floor its ranges name. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 14 +++++- rs/moq-net/src/model/subscription.rs | 5 +- rs/moq-net/src/model/subscription/window.rs | 14 ++++++ rs/moq-net/src/model/track.rs | 43 ++++++++++++++++ rs/moq-net/src/model/track/miss.rs | 44 ++++++++++++++--- rs/moq-net/tests/integration/range_anchors.rs | 8 ++- rs/moq-net/tests/integration/warm_backfill.rs | 19 +++---- .../tests/integration/widened_max_age.rs | 49 ++++++++++++++++--- 8 files changed, 165 insertions(+), 31 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 164e88ef9e8..6e61644669e 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2143,6 +2143,7 @@ mod tests { .unwrap(), id: 0, max_delay: Duration::ZERO, + requested_max_delay: Duration::ZERO, start: None, priority: 0, requested: None, @@ -2278,6 +2279,7 @@ mod tests { .unwrap(), id: 0, max_delay: Duration::ZERO, + requested_max_delay: Duration::ZERO, start: Some(floor), priority: 0, requested: Some(floor), @@ -2741,6 +2743,7 @@ mod tests { stream, id: 0, max_delay: Duration::ZERO, + requested_max_delay: Duration::ZERO, start: Some(2), priority: 0, requested: Some(2), @@ -6299,6 +6302,9 @@ struct SubStream { /// start sits elsewhere the request-tracked floor stands instead, and demand /// returning here makes the declaration valid again. requested: Option, + /// The max delay the SUBSCRIBE itself carried, fixed for the stream's life: a lite-06+ + /// SUBSCRIBE_START resolves the start from it, skipping older groups as stale. + requested_max_delay: Duration, /// The groups received for this subscription, shared with its [`TrackEntry`]. tail: kio::Producer, /// The first group the publisher serves (SUBSCRIBE_START), once declared. @@ -7059,6 +7065,7 @@ impl Establish { stream, id: self.id, max_delay: wire_max_delay(self.subscription.max_delay), + requested_max_delay: wire_max_delay(self.subscription.max_delay), start: self.subscription.start(), ranges: self.ranges.clone(), priority: self.subscription.priority, @@ -8003,7 +8010,12 @@ impl ServeLoop { // moved back restores it); elsewhere the // request-tracked floor stands rather than a guess. if active.start == active.requested { - let _ = self.serving.start_at(start.group); + let _ = match serve.subscriber.version.resolves_start() { + true => { + self.serving.start_resolved(start.group, active.requested_max_delay) + } + false => self.serving.start_at(start.group), + }; } active.served = Some(start.group); // The groups the SUBSCRIBE asked for below it are not diff --git a/rs/moq-net/src/model/subscription.rs b/rs/moq-net/src/model/subscription.rs index 96b2bf9970b..1fe7d5ebc6e 100644 --- a/rs/moq-net/src/model/subscription.rs +++ b/rs/moq-net/src/model/subscription.rs @@ -171,8 +171,9 @@ impl Range { /// Everything from the oldest group still around onward, the default. /// - /// Unlike `Absolute(0)..` this names no floor, so an older wire asks for the live edge - /// and lets [`Subscription::max_delay`] decide how far back to reach. + /// Unlike `Absolute(0)..` this names no floor: a wire whose publisher resolves the start + /// from [`Subscription::max_delay`] reaches as far back as it allows, and an older wire + /// starts at the live edge, filling nothing below it. pub fn oldest() -> Self { Self { start: Anchor::Relative(u64::MAX), diff --git a/rs/moq-net/src/model/subscription/window.rs b/rs/moq-net/src/model/subscription/window.rs index fc4b32df39e..143ce65645e 100644 --- a/rs/moq-net/src/model/subscription/window.rs +++ b/rs/moq-net/src/model/subscription/window.rs @@ -450,6 +450,20 @@ impl Window { self.spans.iter().map(|span| span.start).min() } + /// The lowest group a range names, `None` when none does or a relative anchor waits. + /// Unlike [`Self::start`], a range reaching back to the oldest group names nothing. + pub fn named(&self) -> Option { + match self.pending { + true => None, + false => self + .spans + .iter() + .filter(|span| span.floored) + .map(|span| span.start) + .min(), + } + } + /// Each range resolved, in the order given: `None` for one that is empty or still /// waits on a first group. An unsettled time end is reported open, never at its cap. pub fn resolved(&self) -> Vec> { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c583307886e..fb900b2cae9 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -435,6 +435,10 @@ pub(crate) struct TrackState { // [`Subscriber::poll_start`]) wait on it; everything else treats the floor as usual. start_pending: bool, + // The budget the publisher resolved `start_sequence` from (a lite-06+ SUBSCRIBE_START), + // skipping every older group as stale under it. Cleared by any other declaration. + start_budget: Option, + // A copy received from a remote publisher, whose declared end waits for groups still // in flight below it (see `tail`), and whose readers' relative and time anchors are // refused unless `anchored` (see `Holdings::drops_anchors`). @@ -2278,6 +2282,7 @@ impl TrackState { self.latch_settled(); self.start_sequence = start_sequence; self.start_pending = pending; + self.start_budget = None; if let Some(start) = start_sequence { self.cache.wakes().arrived(..start); let now = self.cache.pool().now_wall(); @@ -3226,6 +3231,16 @@ impl Producer { Ok(()) } + /// [`Self::start_at`] a start the publisher resolved from `budget`: every older group + /// is stale under it, so a reader whose budget is no looser gives them up unasked. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn start_resolved(&mut self, sequence: u64, budget: Duration) -> Result<()> { + let mut state = self.modify()?; + state.set_start(Some(sequence), false); + state.start_budget = Some(budget); + Ok(()) + } + /// Mark this track as a copy received from a remote publisher, before writing to it. /// /// Its declared end waits for groups still in flight below it; see [`Self::finish_at`]. @@ -6635,6 +6650,7 @@ impl Cursor { misses.judge( &state, (from, revision, cap), + window.named(), |span| { let end = cap.map_or(*span.end(), |cap| (*span.end()).min(cap.saturating_sub(1))); if cap == Some(0) || *span.start() > end { @@ -17981,6 +17997,33 @@ mod test { assert!(dynamic.poll_requested_fill(&waiter).is_pending()); } + /// Below a start its publisher resolved from a budget, a reader no looser than it gives + /// every group up unasked, while a looser one may still reach some, so it asks. + #[test] + #[cfg(feature = "async")] + fn below_a_start_resolved_from_a_budget_only_a_looser_reader_asks() { + let mut producer = track_producer("test", Info::timed()); + let dynamic = producer.requests(); + let mut narrow = producer.subscribe(replay()); + let mut wide = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(60))); + producer.start_resolved(2, Duration::from_secs(30)).unwrap(); + write_group(&producer, 2); + write_group(&producer, 3); + let waiter = kio::Waiter::noop(); + assert_eq!(drain_groups(&mut narrow), [2, 3]); + assert_eq!(drained_drops(&mut narrow), [(0, 2)]); + assert!( + dynamic.poll_requested_fill(&waiter).is_pending(), + "the narrow reader asks nothing" + ); + + assert_eq!(drain_groups(&mut wide), [2, 3]); + let Poll::Ready(Ok(fill)) = dynamic.poll_requested_fill(&waiter) else { + panic!("the wide reader asks below the start"); + }; + assert_eq!((fill.start().group, fill.end()), (0, 2)); + } + /// A reader arriving after another's fill cached the groups under the start reads them /// from the cache: the arrival walk never reaches a backfill, and no fill is asked again. #[test] diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index 8b7e4c6bbd6..9094b4d0b35 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -12,7 +12,8 @@ //! //! A start is no promise, since a group below it may still arrive late. A hole under it is //! asked at once, but a refusal is final only once the ledger gives up on it too; until -//! then the live feed may still deliver it. +//! then the live feed may still deliver it. Under a received copy's start, only what the +//! reader may still want is a hole: see [`Misses::judge`]. use std::{ collections::{BTreeMap, BTreeSet}, @@ -349,11 +350,17 @@ impl Misses { } /// Judge the holes in `target` the last probe has not, returning the runs to ask a - /// fill for. Gives up a run already stale under this reader's budget. + /// fill for. Gives up a run already stale under this reader's budget. `named` is the + /// lowest group the reader's ranges name, which may reach below the feed's start. + #[expect( + clippy::too_many_arguments, + reason = "one probe's inputs, borrowed apart from its cursor" + )] pub fn judge( &mut self, state: &TrackState, key: (u64, u64, Option), + named: Option, mut target: impl FnMut(RangeInclusive) -> RangeInclusiveSet, drift: &Drift, given_up: &mut GivenUp, @@ -491,9 +498,21 @@ impl Misses { self.judged.insert(run.clone()); } let start = state.start_sequence.unwrap_or(0); - // Below the start its publisher resolved from the budget, such a copy cannot tell a - // group a reader's budget reaches, so it leaves them to the feed's re-offer. - if state.fills_from_start && start > 0 { + let (stale_below, unasked) = match state.start_budget { + // Its publisher skipped every older group as stale under a budget no tighter + // than this reader's, so they are stale for it too. + Some(budget) if drift.budget <= budget => (start, false), + // Such a copy cannot tell a group a looser reader's budget reaches, so it leaves + // them to the feed's re-offer. + Some(_) => (0, state.fills_from_start), + // The feed began at the live edge or the floor the copy asked for, so older + // groups are history only a reader naming them asked for. + None => ( + 0, + state.fills_from_start || state.received && named.is_none_or(|named| named >= start), + ), + }; + if unasked && start > 0 { candidates.remove(0..start); } for run in candidates.iter() { @@ -556,8 +575,10 @@ impl Misses { // budget is stale throughout. let stale: Vec> = asks .iter() - .filter(|run| state.drifted(run.end - 1, &drift.edge, drift.budget)) - .cloned() + .filter_map(|run| match state.drifted(run.end - 1, &drift.edge, drift.budget) { + true => Some(run.clone()), + false => Some(run.start..run.end.min(stale_below)).filter(|run| !run.is_empty()), + }) .collect(); for run in stale { asks.remove(run.clone()); @@ -824,6 +845,7 @@ mod tests { let asks = misses.judge( &state, (0, 0, None), + None, |span| [span].into_iter().collect(), &drift, &mut given_up, @@ -862,6 +884,7 @@ mod tests { misses.judge( &state, (0, 0, None), + None, |_| panic!("the edge has not moved"), &drift, &mut given_up, @@ -885,6 +908,7 @@ mod tests { misses.judge( &state, (0, 0, None), + None, |_| panic!("recreation extends no span"), &drift, &mut given_up, @@ -925,6 +949,7 @@ mod tests { let asks = misses.judge( &state, (0, 0, None), + None, |span| [span].into_iter().collect(), &drift, &mut given_up, @@ -957,6 +982,7 @@ mod tests { let asks = misses.judge( &state, (0, 0, None), + None, |_| panic!("the edge has not moved"), &drift, &mut given_up, @@ -992,6 +1018,7 @@ mod tests { .judge( &state, (0, 0, None), + None, |span| [span].into_iter().collect(), &drift, &mut given_up, @@ -1006,6 +1033,7 @@ mod tests { .judge( &state, (0, 0, None), + None, |_| panic!("an unchanged probe rebuilds no span"), &drift, &mut given_up, @@ -1025,6 +1053,7 @@ mod tests { .judge( &state, (0, 0, None), + None, |_| panic!("a late arrival extends no span"), &drift, &mut given_up, @@ -1040,6 +1069,7 @@ mod tests { let asks = misses.judge( &state, (0, 0, None), + None, |_| panic!("a due gap extends no span"), &drift, &mut given_up, diff --git a/rs/moq-net/tests/integration/range_anchors.rs b/rs/moq-net/tests/integration/range_anchors.rs index 71850dd535b..978ede720cd 100644 --- a/rs/moq-net/tests/integration/range_anchors.rs +++ b/rs/moq-net/tests/integration/range_anchors.rs @@ -145,7 +145,7 @@ async fn each_wire_refuses_only_the_anchors_it_drops() { } /// Every wire carries the live and oldest starts, which it spells with no floor. A wire whose -/// copy starts at the live edge fills the older groups after it. +/// copy starts at the live edge reads oldest from there: no floor asks for the older groups. #[moq_clock::test] async fn every_wire_carries_the_live_and_oldest_starts() { for version in VERSIONS { @@ -153,6 +153,10 @@ async fn every_wire_carries_the_live_and_oldest_starts() { assert_eq!(live.unwrap().last(), Some(&4), "{version}: live reads the latest group"); let mut oldest = read(Some(version), Range::oldest()).await.unwrap(); oldest.sort_unstable(); - assert_eq!(oldest, [0, 1, 2, 3, 4], "{version}: oldest reads every group"); + let want: &[u64] = match *version { + "moq-lite-05" | "moq-transport-18" | "moq-transport-22" => &[4], + _ => &[0, 1, 2, 3, 4], + }; + assert_eq!(oldest, want, "{version}: oldest reads every group its start reaches"); } } diff --git a/rs/moq-net/tests/integration/warm_backfill.rs b/rs/moq-net/tests/integration/warm_backfill.rs index aeedd1ca90e..975e1cb926a 100644 --- a/rs/moq-net/tests/integration/warm_backfill.rs +++ b/rs/moq-net/tests/integration/warm_backfill.rs @@ -4,9 +4,10 @@ //! from its cache. //! //! Where the upstream starts the copy at its live edge, as moq-transport and moq-lite 05 -//! do, the backlog below that start arrives through fills, outside the live feed's arrival -//! walk. A newcomer finds those groups cached, so it neither walks nor asks for them, and -//! read only the live edge until it took the cached backfill as filled. +//! do, the backlog a reader's floor names below that start arrives through fills, outside +//! the live feed's arrival walk. A newcomer finds those groups cached, so it neither walks +//! nor asks for them, and read only the live edge until it took the cached backfill as +//! filled. use std::collections::BTreeSet; use std::time::Duration; @@ -23,13 +24,9 @@ const GROUPS: u64 = 10; /// Reaches every group, however long the readers take. const BUDGET: Duration = Duration::from_secs(100); -const VERSIONS: &[&str] = &[ - "moq-lite-05", - "moq-lite-06", - "moq-lite-07-wip", - "moq-transport-18", - "moq-transport-22", -]; +/// Moq-transport-22 is left out: its cold relay serves an absolute floor only from the +/// live edge. +const VERSIONS: &[&str] = &["moq-lite-05", "moq-lite-06", "moq-lite-07-wip", "moq-transport-18"]; fn origin(hop: u64) -> origin::Producer { let (origin, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); @@ -44,7 +41,7 @@ async fn subscribe(relay: &origin::Producer) -> track::Subscriber { remote .track("video") .unwrap() - .subscribe(Subscription::default().with_max_delay(BUDGET)) + .subscribe(Subscription::default().with_max_delay(BUDGET).with_range(0..)) .await .unwrap() } diff --git a/rs/moq-net/tests/integration/widened_max_age.rs b/rs/moq-net/tests/integration/widened_max_age.rs index c664ff65e80..d6ecfdabdd0 100644 --- a/rs/moq-net/tests/integration/widened_max_age.rs +++ b/rs/moq-net/tests/integration/widened_max_age.rs @@ -13,7 +13,7 @@ use std::time::Duration; use moq_net::track::{Info, Range, Subscription}; use moq_net::{Hop, Timescale, Timestamp, Version, origin, track}; -use crate::support::harness::{MockConnectOptions, connect_mock, peer, spawn}; +use crate::support::harness::{MockConnectOptions, MockPair, connect_mock, peer, spawn}; const TIMEOUT: Duration = Duration::from_secs(10); @@ -27,7 +27,7 @@ const NARROW: Duration = Duration::from_millis(2500); const WIDE: Duration = Duration::from_millis(6500); /// The wires whose publisher resolves the start from the budget. Lite-05 and moq-transport -/// start at the live edge and fill the backlog by range instead, which the budget does not gate. +/// start at the live edge, below which a reader fills only the floor its ranges name. const VERSIONS: &[&str] = &["moq-lite-06", "moq-lite-07-wip"]; /// The wire for the tests on one version: only lite-07 FETCHes below the start, so only it @@ -44,6 +44,7 @@ fn origin(hop: u64) -> origin::Producer { struct Chain { source: track::Producer, viewer: origin::Producer, + links: Vec, _hold: Vec>, } @@ -77,15 +78,21 @@ impl Chain { Self { source, viewer, - _hold: vec![ - Box::new(publisher), - Box::new(relay), - Box::new(links), - Box::new(broadcast), - ], + links, + _hold: vec![Box::new(publisher), Box::new(relay), Box::new(broadcast)], } } + /// The FETCH streams opened on any lite link, by their stream type 3. + fn fetches(&self) -> usize { + self.links + .iter() + .flat_map(|link| [link.client_transport.bidi_types(), link.server_transport.bidi_types()]) + .flatten() + .filter(|kind| *kind == 3) + .count() + } + /// Subscribe from the viewer with `budget`, reading from the oldest group still around. async fn subscribe(&self, budget: Duration) -> track::Subscriber { let consumer = self.viewer.consume(); @@ -108,6 +115,32 @@ async fn received(subscriber: &mut track::Subscriber) -> BTreeSet { got } +/// A fresh reader asks no fill for the history below where its feed starts, through a relay +/// or not: below a start resolved from a budget no tighter than its own, every group is +/// stale, and below a live edge start it names no floor under, none was asked for. +#[moq_clock::test] +async fn a_fresh_reader_fills_nothing_below_its_start() { + for (version, want) in [ + ("moq-lite-04", 6..GROUPS), + // Starts at the live edge. + ("moq-lite-05", GROUPS - 1..GROUPS), + ("moq-lite-06", 6..GROUPS), + ("moq-lite-07-wip", 6..GROUPS), + ("moq-transport-18", GROUPS - 1..GROUPS), + ("moq-transport-22", GROUPS - 1..GROUPS), + ] { + for relay in [false, true] { + let chain = Chain::new(version, relay).await; + let mut reader = moq_clock::timeout(TIMEOUT, chain.subscribe(NARROW)).await.unwrap(); + let got = moq_clock::timeout(TIMEOUT, received(&mut reader)).await.unwrap(); + assert_eq!(got, want.clone().collect(), "{version} relay={relay}"); + if version.starts_with("moq-lite") { + assert_eq!(chain.fetches(), 0, "{version} relay={relay}: fetched below the start"); + } + } + } +} + /// What the subscriber reads after widening its own max age, through a relay or not. #[moq_clock::test] async fn a_widened_reader_gets_what_was_skipped() { From 410a5d5224781c06b163f229145aa8786f84b783 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 15:34:08 -0700 Subject: [PATCH 2/2] docs(concept): a live edge wire reads history only from a named start Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 23d6e26371a..c5c36bf35a1 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -252,7 +252,9 @@ per-request token, so there an announcement rides the session's scope. A subscriber names a broadcast and track. Delivery starts at the oldest group it can still use, which at the default budget is the latest one, so every group must begin at a point a fresh subscriber can decode from (a keyframe, a full -JSON snapshot). Groups can also be fetched by sequence number, which is how the +JSON snapshot). A moq-lite 05 or moq-transport publisher starts at the latest +group whatever the budget, so a subscription there reads older groups only from +a starting group it names. Groups can also be fetched by sequence number, which is how the [HLS gateway](/bin/hls) and the relay's [HTTP fetch](/bin/relay/http) serve history.