fix(net): fill no history below a subscription's start that its reader skips

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.
 

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<u64>,
+	/// 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<Tail>,
 	/// 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

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),

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<u64> {
+		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<Option<Resolved>> {

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<Duration>,
+
 	// 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]

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<u64>),
+		named: Option<u64>,
 		mut target: impl FnMut(RangeInclusive<u64>) -> RangeInclusiveSet<u64>,
 		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<ops::Range<u64>> = 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,

rs/moq-net/tests/integration/range_anchors.rs

@@ -145,14 +145,18 @@ 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 {
 		let live = read(Some(version), Range::live()).await;
 		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");
 	}
 }

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()
 }