fix(ietf): skip FETCH holes to the next cached group when nothing is upstream

quest/m1/moxygen/README.md

@@ -1,4 +1,4 @@
-# Moxygen compatibility
+# [S] Moxygen compatibility
 
 ## Goal
 
@@ -42,10 +42,6 @@ Decided in the review of
 
 Docs stay inline in the change that makes them stale. No new guide.
 
-## Required
-
-- [Sparse FETCH ranges](/quest/m1/moxygen/fetch-span.md) - a FETCH costs the groups it returns, not the span of its range
-
 ## Related
 
 - [JavaScript FETCH](/quest/m1/js-fetch.md) - the browser publisher answers a FETCH this relay forwards

rs/moq-net/Cargo.toml

@@ -16,8 +16,8 @@ categories = ["multimedia", "network-programming", "web-programming"]
 ignored = ["getrandom"]
 
 [features]
-# Exposes the wire codecs to the `fuzz/` harness and the announce bench through the
-# hidden `fuzz` module. Off by default and hidden from the docs.
+# Exposes the wire codecs to the `fuzz/` harness, and private paths to the announce and
+# fetch benches, through the hidden `fuzz` module. Off by default and hidden from the docs.
 fuzz = []
 
 [dependencies]
@@ -75,6 +75,12 @@ name = "announce"
 harness = false
 required-features = ["fuzz"]
 
+# Reaches the private FETCH walk through the hidden `fuzz` module.
+[[bench]]
+name = "fetch"
+harness = false
+required-features = ["fuzz"]
+
 [[bench]]
 name = "group"
 harness = false

rs/moq-net/benches/fetch.rs

@@ -0,0 +1,94 @@
+//! Standalone FETCH walk over a sparse track: what one range request costs a
+//! publisher answering it from its own cache, with nothing upstream to ask.
+//!
+//! `fetch/span` holds the cached groups fixed and widens the range they are spread
+//! across; `fetch/present` holds the range fixed and adds groups. The walk should
+//! cost the groups it returns, so `span` stays flat and `present` grows linearly.
+//! Throughput counts the groups returned.
+//!
+//! Run with `cargo bench -p moq-net --features fuzz --bench fetch`.
+
+use std::time::Duration;
+
+use bytes::Bytes;
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use moq_net::{Timestamp, broadcast, cache, group, track};
+
+/// Range widths, from dense to sparse enough that stepping through it never ends.
+const SPANS: [u64; 4] = [1 << 6, 1 << 12, 1 << 24, 1 << 40];
+
+/// Cached group counts spread across the range.
+const PRESENT: [u64; 4] = [1, 16, 256, 4096];
+
+/// Held fixed while the other axis sweeps.
+const FIXED_PRESENT: u64 = 64;
+const FIXED_SPAN: u64 = 1 << 24;
+
+/// Keeps the ownership chain alive around the track being fetched.
+struct Sparse {
+	_broadcast: broadcast::Producer,
+	_track: track::Producer,
+	consumer: track::Consumer,
+	span: u64,
+}
+
+impl Sparse {
+	/// `present` one-frame groups spread evenly across `0..span`.
+	fn new(span: u64, present: u64) -> Self {
+		let mut info = broadcast::Info::default();
+		// Nothing may expire or evict mid-measurement, or later samples walk fewer groups.
+		let config = cache::Config::default()
+			.with_capacity(1 << 30)
+			.with_expiry(Duration::from_secs(24 * 60 * 60));
+		info.pool = cache::Pool::new(config);
+		let broadcast = broadcast::Producer::new(info);
+		let track = broadcast.create_track("bench", None).unwrap();
+
+		let step = span / present;
+		for sequence in (0..present).map(|i| i * step) {
+			let mut group = track.create_group(group::Info { sequence }).unwrap();
+			group
+				.write_frame(Timestamp::ZERO, Bytes::from_static(&[0; 64]))
+				.unwrap();
+			group.finish().unwrap();
+		}
+
+		Self {
+			consumer: track.consume(),
+			_broadcast: broadcast,
+			_track: track,
+			span,
+		}
+	}
+
+	fn walk(&self) -> usize {
+		futures::executor::block_on(moq_net::fuzz::walk_fetch(&self.consumer, 0, self.span)).unwrap()
+	}
+}
+
+fn bench_span(c: &mut Criterion) {
+	let mut group = c.benchmark_group("fetch/span");
+	group.throughput(Throughput::Elements(FIXED_PRESENT));
+	for span in SPANS {
+		let sparse = Sparse::new(span, FIXED_PRESENT);
+		assert_eq!(sparse.walk() as u64, FIXED_PRESENT);
+		group.bench_function(BenchmarkId::from_parameter(format!("2^{}", span.ilog2())), |b| {
+			b.iter(|| sparse.walk())
+		});
+	}
+	group.finish();
+}
+
+fn bench_present(c: &mut Criterion) {
+	let mut group = c.benchmark_group("fetch/present");
+	for present in PRESENT {
+		group.throughput(Throughput::Elements(present));
+		let sparse = Sparse::new(FIXED_SPAN, present);
+		assert_eq!(sparse.walk() as u64, present);
+		group.bench_function(BenchmarkId::from_parameter(present), |b| b.iter(|| sparse.walk()));
+	}
+	group.finish();
+}
+
+criterion_group!(benches, bench_span, bench_present);
+criterion_main!(benches);

rs/moq-net/src/fuzz.rs

@@ -7,6 +7,8 @@
 //! here also lets the tests below replay them on stable, which is what turns a crash
 //! the fuzzer found into a regression `just check` runs.
 //!
+//! The announce and fetch benches borrow a few `pub` helpers here for the same reason.
+//!
 //! Compiled only under `cfg(test)` or the `fuzz` feature, so none of this is part of
 //! the published API. See `fuzz/README.md` for the workflow.
 
@@ -252,6 +254,17 @@ pub fn decode_announces(mut data: &[u8], compress: bool) -> Vec<Announced> {
 	resolved
 }
 
+/// Walk the whole groups `start..end` out of `track` as a standalone FETCH does,
+/// returning how many it read. The fetch bench's way into the private walk.
+pub async fn walk_fetch(track: &crate::track::Consumer, start: u64, end: u64) -> Result<usize, crate::Error> {
+	let start = ietf::Location {
+		group: start,
+		object: 0,
+	};
+	let end = ietf::Location { group: end, object: 0 };
+	Ok(ietf::walk_fetch(track, start, end, 0).await?.groups.len())
+}
+
 /// Feed a lite-07 announce stream through the stateful decoder, then check that our
 /// encoder's compression of what it resolved reads back as the same announcements.
 ///

rs/moq-net/src/ietf/mod.rs

@@ -47,6 +47,8 @@ pub use parameters::*;
 pub use properties::Properties;
 pub use publish::*;
 pub use publish_namespace::*;
+#[cfg(any(test, feature = "fuzz"))]
+pub(crate) use publisher::walk_fetch;
 use publisher::*;
 pub use request::*;
 pub use session::*;

rs/moq-net/src/ietf/publisher.rs

@@ -57,7 +57,7 @@ enum FetchPrior {
 
 /// One group of a FETCH answer, read out of the cache so a later eviction cannot
 /// truncate what FETCH_OK already promised.
-struct FetchedGroup {
+pub(crate) struct FetchedGroup {
 	sequence: u64,
 	/// The Object ID of the first frame.
 	first: u64,
@@ -85,8 +85,8 @@ impl FetchedGroup {
 }
 
 /// A FETCH range read out of a track.
-struct Walked {
-	groups: Vec<FetchedGroup>,
+pub(crate) struct Walked {
+	pub(crate) groups: Vec<FetchedGroup>,
 	/// Why the walk stopped before the end of the range, when it did.
 	stopped: Option<Error>,
 }
@@ -97,12 +97,22 @@ struct Walked {
 /// A group that does not exist below the newest one is a hole and is skipped; one at or
 /// past it is the end of the track, so the walk stops there. Any other failure answers
 /// the whole FETCH.
-async fn walk_fetch(track: &track::Consumer, start: Location, end: Location, priority: u8) -> Result<Walked, Error> {
+///
+/// With nothing upstream to ask, the cache is the whole track, so a hole skips straight
+/// to the next cached group: the walk costs the groups it returns, not the range's span.
+pub(crate) async fn walk_fetch(
+	track: &track::Consumer,
+	start: Location,
+	end: Location,
+	priority: u8,
+) -> Result<Walked, Error> {
 	let mut groups = Vec::new();
-	for sequence in start.group..=end.group {
-		if sequence == end.group && end.object == 0 {
+	let mut next = Some(start.group);
+	while let Some(sequence) = next {
+		if sequence > end.group || (sequence == end.group && end.object == 0) {
 			break;
 		}
+		next = sequence.checked_add(1);
 
 		let skip = match sequence == start.group {
 			true => start.object,
@@ -116,13 +126,22 @@ async fn walk_fetch(track: &track::Consumer, start: Location, end: Location, pri
 
 		let mut group = match track.fetch_group(sequence, fetch).await {
 			Ok(group) => group,
-			Err(Error::NotFound) if track.latest().is_some_and(|latest| sequence < latest) => continue,
-			Err(err @ Error::NotFound) => {
-				return Ok(Walked {
-					groups,
-					stopped: Some(err),
-				});
-			}
+			Err(err @ Error::NotFound) => match track.latest() {
+				Some(latest) if sequence < latest => {
+					// Capped at the newest group, whose own probe judges where the track ends.
+					if !track.fetches_misses() {
+						let cached = track.next_cached(sequence + 1);
+						next = Some(cached.map_or(latest, |cached| cached.min(latest)));
+					}
+					continue;
+				}
+				_ => {
+					return Ok(Walked {
+						groups,
+						stopped: Some(err),
+					});
+				}
+			},
 			Err(err) => return Err(err),
 		};
 
@@ -3904,6 +3923,44 @@ mod serve_tests {
 		}
 	}
 
+	/// With nothing upstream, a hole skips to the next cached group instead of stepping
+	/// through every missing sequence, which would never finish here.
+	#[tokio::test]
+	async fn a_standalone_fetch_skips_a_sparse_span() {
+		const FAR: u64 = 1 << 40;
+		for version in [Version::Draft16, Version::Draft18] {
+			let h = serve(version);
+			for sequence in [0, 5, FAR] {
+				let mut group = h.track.create_group(group::Info { sequence }).unwrap();
+				for object in 0..2 {
+					group
+						.write_frame(timestamp(), format!("{sequence}-{object}").into_bytes())
+						.unwrap();
+				}
+				group.finish().unwrap();
+			}
+			settle().await;
+
+			let buf = standalone_fetch(
+				&h,
+				Location { group: 0, object: 0 },
+				Location { group: FAR, object: 0 },
+				GroupOrder::Ascending,
+			)
+			.await;
+			let (ok, objects) = fetch_answer(buf, version);
+			assert_eq!(
+				ok.end_location,
+				Location {
+					group: FAR + 1,
+					object: 0
+				},
+				"{version}"
+			);
+			assert_eq!(objects, pairs([0, 5, FAR]), "{version}");
+		}
+	}
+
 	/// Read a REQUEST_ERROR (or draft-14 FETCH_ERROR) off a refused FETCH, as its code.
 	fn fetch_refusal(mut buf: bytes::Bytes, version: Version) -> u64 {
 		let id = u64::decode(&mut buf, version).unwrap();

rs/moq-net/src/model/resume.rs

@@ -738,6 +738,28 @@ impl Consumer {
 		self.peek_below(Some(sequence))
 	}
 
+	/// The lowest sequence at or above `from` any segment caches; see
+	/// [`track::Consumer::next_cached`].
+	///
+	/// Segment ranges are left to the fetch, so this is a superset of what
+	/// [`Self::cached_group`] serves and skipping to it never passes a servable group.
+	pub(crate) fn next_cached(&self, from: u64) -> Option<u64> {
+		let segments = self.state.read().segments.clone();
+		segments
+			.iter()
+			.filter_map(|segment| segment.track.next_cached(from))
+			.min()
+	}
+
+	/// Whether the newest segment, where fetches go, serves cache misses; see
+	/// [`track::Consumer::fetches_misses`]. With no segment yet, the first route to
+	/// serve the track may bring a handler.
+	pub(crate) fn fetches_misses(&self) -> bool {
+		// Copied out: the segment's track takes its own lock.
+		let track = self.state.read().segments.last().map(|segment| segment.track.clone());
+		track.is_none_or(|track| track.fetches_misses())
+	}
+
 	/// Walk the segments newest-first for the first one holding a group in range.
 	///
 	/// Every segment, not just the newest: a takeover splices in a fresh track with no

rs/moq-net/src/model/track.rs

@@ -2579,6 +2579,31 @@ impl Consumer {
 		}
 	}
 
+	/// The lowest sequence at or above `from` holding a live cached group, crossing gaps
+	/// in the group numbering. A splice answers from whichever segment caches one first.
+	pub(crate) fn next_cached(&self, from: u64) -> Option<u64> {
+		match &self.inner {
+			ConsumerKind::Plain(state) => state
+				.read()
+				.lookup
+				.range(from..)
+				.find(|(_, slot)| !slot.group.is_aborted())
+				.map(|(sequence, _)| *sequence),
+			ConsumerKind::Spliced(resume) => resume.next_cached(from),
+		}
+	}
+
+	/// Whether a group missing from the cache can still be fetched: a [`Dynamic`] (a
+	/// relay's upstream) serves misses. Without one, the cache is all the track has.
+	///
+	/// A spliced track answers for its newest segment, which is where fetches go.
+	pub(crate) fn fetches_misses(&self) -> bool {
+		match &self.inner {
+			ConsumerKind::Plain(state) => state.read().fetch.lock().has_handlers(),
+			ConsumerKind::Spliced(resume) => resume.fetches_misses(),
+		}
+	}
+
 	/// A cached group by sequence, under the same terms as [`Self::peek_latest`]. Unlike a
 	/// fetch, a peek does not refresh the group's cache standing, so it never keeps a
 	/// group alive over one a subscriber actually read; an aborted (evicted) group is a