From ce2b0866df78d0505dcd6cf8605408ba36d0bca0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 16:35:24 -0700 Subject: [PATCH 1/5] quest: claim lite07-head-fetch-arrival Co-Authored-By: Claude Opus 5.5 From e422e109cf62cf6a06f86ab589befe11a77ec412 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 17:03:55 -0700 Subject: [PATCH 2/5] fix(net): a fetched group head keeps the live group visible A lite-07 relay pulling an open group mid-group caches it headless. A fetch of its head then went through `claim_sequence`, which dropped the live slot because it couldn't answer from frame 0, and committed the fetched copy invisible. The live group's arrival entry no longer resolved, so a fresh subscriber got nothing, and the copy the feed kept writing left the cache. A fetch no longer displaces a live group. The fetched copy rides beside it as the slot's head and is served in its place while it lasts. The live group keeps its slot and arrival entry, so an abandoned fetch leaves it servable. Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/ietf-rejoin-open-group.md | 1 - quest/m1/lite07-finalize.md | 1 - quest/m1/lite07-head-fetch-arrival.md | 66 ----------- rs/moq-net/src/model/track.rs | 139 ++++++++++++++++++---- rs/moq-net/tests/head_fetch.rs | 162 ++++++++++++++++++++++++++ 6 files changed, 278 insertions(+), 92 deletions(-) delete mode 100644 quest/m1/lite07-head-fetch-arrival.md create mode 100644 rs/moq-net/tests/head_fetch.rs diff --git a/quest/m1/README.md b/quest/m1/README.md index d9df2dea1d..0e728c3aac 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -66,7 +66,6 @@ blocks. The quests that gated m0 lines moved under them. - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges in either order and replaces FETCH; relays fill misses by range, including over moq-transport - [Live media time](/quest/m1/subscribe-live-time.md) - re-scoped against `set_live`: a lite-07 SUBSCRIBE_OK carries the publisher's current media time only if a reader still needs it - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls -- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - on lite-07, a relay fetching the head of a group it receives mid-group still delivers that group to new subscribers - [Untimed lite-07](/quest/m1/lite-untimed.md) - lite-07 carries an untimed track in both languages; lite-05/06 write send time - [Finalize moq-lite-07](/quest/m1/lite07-finalize.md) - when the maintainer cuts it, lite-07 negotiates as `moq-lite-07` and the next release ships it - [Late lower groups](/quest/m1/lite-late-lower-group.md) - a moq-lite subscriber with a floor receives a group created below the first served one, as moq-transport does diff --git a/quest/m1/ietf-rejoin-open-group.md b/quest/m1/ietf-rejoin-open-group.md index 3ef18f8a11..e2aa3d4a98 100644 --- a/quest/m1/ietf-rejoin-open-group.md +++ b/quest/m1/ietf-rejoin-open-group.md @@ -21,5 +21,4 @@ Public API: none. Wire: none. ## Related -- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - another mid-group join that loses part of a group - [Rejoin after the copy goes idle](/quest/m1/test-flakes-2/rejoin-idle-race.md) - a different IETF rejoin bug, serving a stale cache diff --git a/quest/m1/lite07-finalize.md b/quest/m1/lite07-finalize.md index 56a91b3312..14015eea2b 100644 --- a/quest/m1/lite07-finalize.md +++ b/quest/m1/lite07-finalize.md @@ -69,7 +69,6 @@ is published; older versions are unchanged. - [Live media time](/quest/m1/subscribe-live-time.md) - SUBSCRIBE_OK carries the publisher's live media time - [Untimed lite-07](/quest/m1/lite-untimed.md) - an untimed track crosses the wire untimed - [One route cost](/quest/m1/route-cost.md) - ANNOUNCE carries one cost -- [Fetched heads stay visible](/quest/m1/lite07-head-fetch-arrival.md) - a lite-07 relay still delivers a group whose head it fetched ## Related diff --git a/quest/m1/lite07-head-fetch-arrival.md b/quest/m1/lite07-head-fetch-arrival.md deleted file mode 100644 index 5c434c1eca..0000000000 --- a/quest/m1/lite07-head-fetch-arrival.md +++ /dev/null @@ -1,66 +0,0 @@ -# [S] A fetched group head keeps the live group visible on lite-07 - -## Goal - -On moq-lite-07, a relay that fetches the head of a group it is receiving -mid-group still delivers that group to new arrival-order subscribers, and -frames the publisher writes into it afterward still reach them. Today a fresh -subscriber with no start receives nothing, even after the publisher writes more -frames into the group. lite-05, lite-06, and moq-transport are unaffected. - -A long-lived open group (a catalog: snapshot at frame 0, deltas after) is the -case that hurts: a viewer joining the relay never gets the catalog. - -## Plan - -Scenario, reproduced on `main` 2704e10e2 for `moq-lite-07-wip` only: - -1. Publisher P holds open group G with a snapshot and two deltas. Relay R pulls - P; mesh peer M pulls R. -2. M subscribes to `catalog.json` with a start of `(G, 1)`, then reads its own - copy from frame 0, so it fetches G's head through R. -3. A fresh reader on R subscribes with no start: it receives no group within - 5 s of virtual time, and a frame P writes into G after 1 s doesn't help. - -Test shape (`rs/moq-net/tests`, the mock harness, paused time; run it for every -version and expect group G starting with `snapshot`): - -```rust -let start = track::Position { group: sequence, frame: 1 }; -let mut sub = peer_remote.track("catalog.json").unwrap() - .subscribe(track::Subscription::default().with_start(start)).await.unwrap(); -let mut group = sub.recv_group().await.unwrap().unwrap(); -// The peer reads its headless copy from frame 0: a fetch of G's head through R. -let _ = tokio::time::timeout(Duration::from_millis(500), group.read_frame()).await; - -// A fresh reader on R wants the latest group from its snapshot. -let mut sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap(); -let mut group = tokio::time::timeout(Duration::from_secs(5), sub.recv_group()) - .await.expect("no group").unwrap().unwrap(); -assert_eq!(group.sequence, sequence); -assert_eq!(group.read_frame().await.unwrap().unwrap().payload, b"snapshot".as_ref()); -``` - -Inferred cause, unverified: a lite-07 SUBSCRIBE_OK carries the publisher's -largest position, so `TrackServe::widen_frame_bounds` -(`rs/moq-net/src/lite/subscriber.rs`) trusts it and keeps M's mid-group start: -R's upstream subscription starts G at frame 1. M's fetch reaches -`TrackState::insert_group_request` (`rs/moq-net/src/model/track.rs`), whose -`claim_sequence` drops that headless live slot because it can't answer from -frame 0, then commits the fetched group invisible. G's arrival entry keeps the -old stamp, so `poll_recv_group` skips it, and the live copy R keeps filling -from P is no longer in its cache. Settle whether a fetch may replace a live, -visible slot at all, or must keep it arrival-visible and still writable. - -Only bites once lite-07 ships. lite-06 has frame bounds too, but no largest -(`Version::has_largest`), so `widen_frame_bounds` rounds the start to the head -of the group and R never holds a headless G. Widening on lite-07 too would hide -the bug at the cost of what the largest position is for; the fix belongs in the -cache. - -## Related - -- [#4829](https://github.com/moq-dev/moq/pull/4829) - `release`'s backport of the lite-06 head widening -- [The moq.pro mesh runs lite-07](/quest/m3/lite07-mesh.md) - when this starts to bite -- [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - lite-07 relays fill misses by range, the same cache seam -- [Late lower groups](/quest/m1/lite-late-lower-group.md) - another lite group a subscriber silently never sees diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 95d0acae27..25a2148b0a 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -186,7 +186,8 @@ pub(crate) struct TrackState { // Publisher-produced groups in arrival order as (sequence, stamp), walked by // subscriptions; an entry only resolves while its stamp matches the slot's. // Fetched backfill (`insert_group_request`) is deliberately absent: it is - // served by sequence, never replayed to arrival-order subscribers. + // served by sequence, never replayed to arrival-order subscribers. A fetched + // head of a live group is the exception, served through that group's entry. arrival: VecDeque<(u64, u32)>, // Eviction order under memory pressure as (sequence, stamp): every cached @@ -308,6 +309,35 @@ struct Slot { // Whether this incarnation came from the live publisher and can replace older // subscription content. Fetch-only backfill stays cached but never anchors drift. visible: bool, + + // A fetched copy of a live `group` that the feed started past the frames a fetch + // asked for. It runs to the end of the group too, so it is served in the live + // group's place while it lasts; `group` stays the slot, still written by the feed, + // so an abandoned fetch leaves the group servable rather than gone. Boxed: rare. + head: Option>, +} + +impl Slot { + /// Whether the live feed's group still holds this slot. + fn is_live(&self) -> bool { + self.visible && !self.group.is_aborted() + } + + /// The copy to hand readers: the fetched head while it lasts, else the live group. + fn serving(&self) -> &group::Producer { + match &self.head { + Some(head) if !head.is_aborted() => head, + _ => &self.group, + } + } + + /// Abort every copy this slot holds. + fn abort(&self, err: Error) { + if let Some(head) = &self.head { + let _ = (**head).clone().abort(err.clone()); + } + let _ = self.group.clone().abort(err); + } } /// Heap the track keeps per cached group, excluding the group itself @@ -390,9 +420,9 @@ impl TrackState { if *sequence >= min_sequence && let Some(slot) = self.lookup.get(sequence) && slot.stamp == *stamp - && !slot.group.is_aborted() + && !slot.serving().is_aborted() { - return Poll::Ready(Ok(Some((slot.group.clone(), self.offset + i)))); + return Poll::Ready(Ok(Some((slot.serving().clone(), self.offset + i)))); } } @@ -480,7 +510,7 @@ impl TrackState { .lookup .range(next_sequence.max(self.live_floor.unwrap_or(0))..) .filter(|_| self.readable()) - .map(|(_, slot)| &slot.group) + .map(|(_, slot)| slot.serving()) .take_while(|group| super::subscription::before_end(group.sequence, end_sequence)) .find(|group| !group.is_aborted()); @@ -513,8 +543,11 @@ impl TrackState { /// is a miss instead, so the fetch goes upstream for the frames that are missing. fn covering_group(&self, sequence: u64, frame_start: u64) -> Option<&group::Producer> { let slot = self.lookup.get(&sequence)?; - let first = slot.group.live_first_frame()?; - (first as u64 <= frame_start).then_some(&slot.group) + std::iter::once(&slot.group).chain(slot.head.as_deref()).find(|group| { + group + .live_first_frame() + .is_some_and(|first| first as u64 <= frame_start) + }) } /// The local retention bound, or `None` when unknown or unlimited. @@ -782,7 +815,7 @@ impl TrackState { .cache_accessed_tick(scan.gc.then_some(scan.now)) .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) { - let _ = slot.group.clone().abort(Error::Old); + slot.abort(Error::Old); } } } @@ -826,8 +859,7 @@ impl TrackState { // Take the group out of the cache and abort it, so any consumer // still reading surfaces `Error::Old` instead of blocking forever // on a frame that will never arrive. - let slot = self.lookup.remove(&sequence).unwrap(); - let _ = slot.group.abort(Error::Old); + self.lookup.remove(&sequence).unwrap().abort(Error::Old); } } @@ -976,6 +1008,7 @@ impl TrackState { group: group.clone(), stamp, visible, + head: None, }, ); if visible { @@ -1101,8 +1134,7 @@ impl TrackState { self.debt -= size; paid = paid.saturating_add(size); - let slot = self.lookup.remove(&sequence).unwrap(); - let _ = slot.group.abort(Error::Evicted); + self.lookup.remove(&sequence).unwrap().abort(Error::Evicted); } } @@ -1278,8 +1310,16 @@ impl TrackState { } let info = self.info.clone().unwrap(); - // An evicted sequence can be re-fetched; a live one is a duplicate. - self.claim_sequence(sequence, frame_start)?; + // A live group the feed started past `frame_start` keeps its slot: subscribers + // walk it in arrival order and the feed keeps writing it. The fetch fills its + // head instead. Anything else that can't answer from `frame_start` (evicted, + // aborted, or narrower backfill) is replaced; a group that can is a duplicate. + if self.covering_group(sequence, frame_start).is_some() { + return Err(Error::Duplicate); + } + if !self.lookup.get(&sequence).is_some_and(Slot::is_live) { + self.claim_sequence(sequence, frame_start)?; + } let mut group = group::Producer::new(group::Info { sequence }, info, self.cache.clone()); // Start where the request did before the group is visible: a fetch looking it up @@ -1288,10 +1328,17 @@ impl TrackState { group.start_at(frame_start)?; // A backfill exists because someone is fetching it right now: stamp that // access so the eviction walk can't kill it before the fetch resolves. - // It is also invisible to arrival-order subscribers: fetched on demand, - // not produced live by the publisher. group.cache_refresh(); - self.commit_group(&group, false); + + // Settled first, like `commit_group`, which may evict the live slot itself. + self.charge_debt(); + match self.lookup.get_mut(&sequence).filter(|slot| slot.is_live()) { + Some(slot) => slot.head = Some(Box::new(group.clone())), + // Invisible to arrival-order subscribers: fetched on demand, not produced + // live by the publisher. + None => self.insert_group(&group, false), + } + self.evict_expired(); Ok(group) } } @@ -2542,7 +2589,7 @@ impl Consumer { .lookup .range(..sequence) .rev() - .map(|(_, slot)| &slot.group) + .map(|(_, slot)| slot.serving()) .find(|group| !group.is_aborted()) .map(|group| group.consume()) } @@ -2553,11 +2600,11 @@ impl Consumer { /// miss. pub(crate) fn peek_group(&self, sequence: u64) -> Option { let state = self.state.read(); - let slot = state.lookup.get(&sequence)?; - if slot.group.is_aborted() { + let group = state.lookup.get(&sequence)?.serving(); + if group.is_aborted() { return None; } - Some(slot.group.consume()) + Some(group.consume()) } /// Poll for group `sequence` the way the live feed delivers it: `Some` once it is @@ -2566,9 +2613,9 @@ impl Consumer { pub(crate) fn poll_group(&self, sequence: u64, waiter: &kio::Waiter) -> Poll> { let res = self.state.poll(waiter, |state| { if let Some(slot) = state.lookup.get(&sequence) - && !slot.group.is_aborted() + && !slot.serving().is_aborted() { - return Poll::Ready(Some(slot.group.consume())); + return Poll::Ready(Some(slot.serving().consume())); } let passed = state.max_sequence.is_some_and(|newest| newest > sequence) || state.start_sequence.is_some_and(|start| start > sequence) @@ -3725,7 +3772,7 @@ impl Subscriber { && state .lookup .get(sequence) - .is_some_and(|slot| slot.stamp == *stamp && !slot.group.is_aborted()) + .is_some_and(|slot| slot.stamp == *stamp && !slot.serving().is_aborted()) })) } @@ -7997,6 +8044,52 @@ mod test { ); } + /// A fetch for the head of a live group the feed started partway into keeps that group + /// in arrival order: subscribers get the fetched copy while it lasts, and the live + /// group again once the fetch is gone. + #[tokio::test] + async fn fetch_fills_the_head_of_a_live_group() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap(); + live.start_at(1).unwrap(); + live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta")) + .unwrap(); + + let fetch = consumer.fetch_group(0, None); + let request = dynamic.requested_group().await.unwrap(); + let mut head = request.accept(None).unwrap(); + for payload in [&b"snapshot"[..], b"delta"] { + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload)) + .unwrap(); + } + assert_eq!(fetch.await.unwrap().index(), 0); + + // Arrival order hands out the fetched copy, from its first frame. + let mut sub = producer.subscribe(None); + let mut group = sub.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!( + group.read_frame().await.unwrap().unwrap().payload, + bytes::Bytes::from_static(b"snapshot") + ); + + // An abandoned fetch leaves the live group in its slot, still written by the feed. + head.abort(Error::Cancel).unwrap(); + live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta2")) + .unwrap(); + let mut sub = producer.subscribe(None); + let mut group = sub.recv_group().now_or_never().unwrap().unwrap().unwrap(); + // Positioning at the start clamps up to the live group's first frame. + group.start_at(0); + assert_eq!(group.index(), 1); + assert_eq!( + group.read_frame().await.unwrap().unwrap().payload, + bytes::Bytes::from_static(b"delta") + ); + } + /// Joining a queued fetch widens its range, and a caller that arrives once the range /// is already on the wire fails cleanly rather than being handed the narrower group. #[tokio::test] diff --git a/rs/moq-net/tests/head_fetch.rs b/rs/moq-net/tests/head_fetch.rs new file mode 100644 index 0000000000..54c1b19897 --- /dev/null +++ b/rs/moq-net/tests/head_fetch.rs @@ -0,0 +1,162 @@ +//! A relay that fetches the head of a group it is receiving mid-group keeps delivering +//! that group to arrival-order subscribers, along with the frames written into it after. +//! +//! A mesh peer subscribing from partway into an open group (a catalog: snapshot, then +//! deltas) has its relay pull the group headless. Reading the peer's copy from frame 0 +//! then fetches the head through the relay, whose cache must not trade the live group +//! for the fetched copy. + +mod support; + +use std::time::Duration; + +use moq_net::track::{Position, Subscription}; +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(5); + +const VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-17", + "moq-transport-22", +]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer { + let consumer = origin.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves") +} + +async fn read(group: &mut moq_net::group::Consumer) -> Vec { + tokio::time::timeout(TIMEOUT, group.read_frame()) + .await + .expect("no frame") + .expect("read_frame") + .expect("group ended") + .payload + .to_vec() +} + +/// Publisher P holds an open group, relay R pulls P, peer M pulls R from the group's +/// second frame and then reads its copy from the first. A fresh subscriber on R must +/// still get the whole group, and a frame P writes afterward. +async fn round(version: &str) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("catalog.json", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut group = track.append_group().unwrap(); + let sequence = group.sequence; + for payload in [&b"snapshot"[..], b"delta1", b"delta2"] { + group.write_frame(Timestamp::now(), payload).unwrap(); + } + + let relay = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let upstream = connect_mock(options).await; + + let peer = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(peer.clone()); + let downstream = connect_mock(options).await; + + let remote = resolve(&relay).await; + let peer_remote = resolve(&peer).await; + + let start = Position { + group: sequence, + frame: 1, + }; + let mut peer_sub = peer_remote + .track("catalog.json") + .unwrap() + .subscribe(Subscription::default().with_start(start)) + .await + .expect("peer subscribe"); + let mut peer_group = tokio::time::timeout(TIMEOUT, peer_sub.recv_group()) + .await + .expect("peer got no group") + .unwrap() + .unwrap(); + // The peer reads its headless copy from frame 0: a fetch of the head through R. + let _ = tokio::time::timeout(Duration::from_millis(500), peer_group.read_frame()).await; + + // A fresh reader on R wants the latest group from its snapshot. + let mut sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap(); + let mut got = tokio::time::timeout(TIMEOUT, sub.recv_group()) + .await + .unwrap_or_else(|_| panic!("{version}: no group")) + .unwrap() + .unwrap(); + assert_eq!(got.sequence, sequence, "{version}"); + assert_eq!(read(&mut got).await, b"snapshot", "{version}"); + assert_eq!(read(&mut got).await, b"delta1", "{version}"); + assert_eq!(read(&mut got).await, b"delta2", "{version}"); + + // The publisher keeps writing into the group: the reader on R hears it. + tokio::time::sleep(Duration::from_secs(1)).await; + group.write_frame(Timestamp::now(), &b"delta3"[..]).unwrap(); + assert_eq!( + read(&mut got).await, + b"delta3", + "{version}: frame written after the fetch" + ); + + // Every reader of the fetched copy leaves, so its fetch is abandoned, while both + // subscriptions keep the live feed running. The live group still holds R's slot: a + // later reader gets the whole group again. + drop((got, peer_group)); + tokio::time::sleep(Duration::from_secs(1)).await; + let mut late_sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap(); + let mut late = tokio::time::timeout(TIMEOUT, late_sub.recv_group()) + .await + .unwrap_or_else(|_| panic!("{version}: no group after the fetch was abandoned")) + .unwrap() + .unwrap(); + assert_eq!(late.sequence, sequence, "{version}"); + for payload in [&b"snapshot"[..], b"delta1", b"delta2", b"delta3"] { + assert_eq!( + read(&mut late).await, + payload, + "{version}: after the fetch was abandoned" + ); + } + + drop(( + late, late_sub, sub, peer_sub, group, track, broadcast, downstream, upstream, publisher, relay, peer, + )); +} + +#[tokio::test] +async fn a_fetched_head_keeps_the_live_group_visible() { + tokio::time::pause(); + let mut failures = Vec::new(); + for version in VERSIONS { + if tokio::spawn(round(version)).await.is_err() { + failures.push(*version); + } + } + assert!(failures.is_empty(), "failed: {failures:?}"); +} From ead2754bb0ecc8e58d6c3b9a4dadc7d012bf0a43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 17:09:56 -0700 Subject: [PATCH 3/5] fix(net): a slot's eviction and expiry weigh its fetched head too Reclaim, eviction, and expiry looked only at the live copy, so an aborted live group dropped a head that still held the whole group, and reads on the head did not keep the slot from expiring. A live re-delivery could also replace a slot whose head still covered it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 171 ++++++++++++++++++++++++---------- 1 file changed, 120 insertions(+), 51 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 25a2148b0a..5951ca1a37 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -331,12 +331,48 @@ impl Slot { } } + /// Every copy this slot holds, the live group first. + fn copies(&self) -> impl Iterator { + std::iter::once(&self.group).chain(self.head.as_deref()) + } + + /// Whether every copy is gone, so the slot has nothing left to serve. + fn is_aborted(&self) -> bool { + self.serving().is_aborted() + } + + /// The latest access to any copy still held: readers may be on either one. + fn cache_accessed(&self) -> u64 { + self.copies() + .filter(|group| !group.is_aborted()) + .map(group::Producer::cache_accessed) + .max() + .unwrap_or(0) + } + + /// The bytes every copy holds, all freed together. + fn cache_size(&self) -> u64 { + self.copies().map(group::Producer::cache_size).sum() + } + + /// Whether every copy still held sat idle past the scan's expiry window. Asks + /// each one, so a cleanup scan dates them all. + fn is_expired(&self, scan: &ExpiryScan) -> bool { + self.copies() + .filter(|group| !group.is_aborted()) + .fold(true, |expired, group| { + let idle = group + .cache_accessed_tick(scan.gc.then_some(scan.now)) + .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks); + expired && idle + }) + } + /// Abort every copy this slot holds. fn abort(&self, err: Error) { - if let Some(head) = &self.head { - let _ = (**head).clone().abort(err.clone()); + for group in self.copies() { + let _ = group.clone().abort(err.clone()); } - let _ = self.group.clone().abort(err); } } @@ -420,7 +456,7 @@ impl TrackState { if *sequence >= min_sequence && let Some(slot) = self.lookup.get(sequence) && slot.stamp == *stamp - && !slot.serving().is_aborted() + && !slot.is_aborted() { return Poll::Ready(Ok(Some((slot.serving().clone(), self.offset + i)))); } @@ -542,8 +578,7 @@ impl TrackState { /// it to someone who asked for the whole group would silently hand back a tail. It /// is a miss instead, so the fetch goes upstream for the frames that are missing. fn covering_group(&self, sequence: u64, frame_start: u64) -> Option<&group::Producer> { - let slot = self.lookup.get(&sequence)?; - std::iter::once(&slot.group).chain(slot.head.as_deref()).find(|group| { + self.lookup.get(&sequence)?.copies().find(|group| { group .live_first_frame() .is_some_and(|first| first as u64 <= frame_start) @@ -607,7 +642,7 @@ impl TrackState { fn holds(&self, sequence: u64, stamp: u32) -> bool { self.lookup .get(&sequence) - .is_some_and(|slot| slot.stamp == stamp && !slot.group.is_aborted()) + .is_some_and(|slot| slot.stamp == stamp && !slot.is_aborted()) } /// The furthest presentation time the group at `sequence` could still reach: where @@ -774,13 +809,7 @@ impl TrackState { if slot.stamp != stamp { continue; } - if slot.group.is_aborted() - || (!self.protects(sequence) - && slot - .group - .cache_accessed_tick(scan.gc.then_some(scan.now)) - .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks)) - { + if slot.is_aborted() || (!self.protects(sequence) && slot.is_expired(&scan)) { return true; } retained += 1; @@ -808,13 +837,7 @@ impl TrackState { let Some(slot) = self.lookup.get(sequence) else { continue; }; - if slot.stamp == *stamp - && !slot.group.is_aborted() - && slot - .group - .cache_accessed_tick(scan.gc.then_some(scan.now)) - .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) - { + if slot.stamp == *stamp && !slot.is_aborted() && slot.is_expired(&scan) { slot.abort(Error::Old); } } @@ -838,16 +861,11 @@ impl TrackState { } // Already aborted: the frames are gone, reclaim the slot so a // later fetch can serve the sequence again. - if slot.group.is_aborted() { + if slot.is_aborted() { self.lookup.remove(&sequence); continue; } - if self.protects(sequence) - || slot - .group - .cache_accessed_tick(scan.gc.then_some(scan.now)) - .is_none_or(|tick| scan.now.saturating_sub(tick) <= scan.max_ticks) - { + if self.protects(sequence) || !slot.is_expired(&scan) { // Writes keep their scan bounded. Cleanup visits the entire // queue to date pending accesses and find idle entries behind them. retained += 1; @@ -920,7 +938,8 @@ impl TrackState { /// keep the finished ones for readers still draining. A consumer that already /// pulled an open group keeps its own handle and ends with it. fn drop_open_groups(&mut self) { - self.lookup.retain(|_, slot| slot.group.is_finished()); + self.lookup + .retain(|_, slot| slot.copies().any(group::Producer::is_finished)); } /// Attach the publisher's immutable metadata without replacing it with local cache policy. @@ -954,19 +973,11 @@ impl TrackState { /// request from there, so the wider producer takes the slot. Readers already /// draining the old one keep their own handle. fn claim_sequence(&mut self, sequence: u64, frame_start: u64) -> Result<()> { - if let Some(slot) = self.lookup.get(&sequence) { - // The same question `covering_group` asks: can this slot still answer from - // `frame_start`? If it can, the sequence is taken; if it can't, it is dead - // and the caller gets to replace it. - if slot - .group - .live_first_frame() - .is_some_and(|first| first as u64 <= frame_start) - { - return Err(Error::Duplicate); - } - self.lookup.remove(&sequence); + // Either copy that can still answer from `frame_start` takes the sequence. + if self.covering_group(sequence, frame_start).is_some() { + return Err(Error::Duplicate); } + self.lookup.remove(&sequence); Ok(()) } @@ -1071,7 +1082,7 @@ impl TrackState { let Some(slot) = self.lookup.get(sequence) else { return false; }; - slot.stamp == *stamp && !slot.group.is_aborted() && slot.group.cache_accessed() <= average + slot.stamp == *stamp && !slot.is_aborted() && slot.cache_accessed() <= average } /// Abort this track's stalest groups until the outstanding debt is paid, or @@ -1105,7 +1116,7 @@ impl TrackState { // A historical hint; the live entry is elsewhere in the queue. continue; } - if slot.group.is_aborted() { + if slot.is_aborted() { // Aborted upstream: the frames are already gone, reclaim the slot. self.lookup.remove(&sequence); continue; @@ -1120,13 +1131,13 @@ impl TrackState { // Protected: accessed more recently than the average (a fresh insert, // an active reader, or a FETCH hit, which also covers a backfill still // being filled). Rotate to the back. - if slot.group.cache_accessed() > average { + if slot.cache_accessed() > average { self.evict.push_back((sequence, stamp)); continue; } // The full footprint including overhead, so even empty groups repay // their share of the budget when evicted. - let size = slot.group.cache_size(); + let size = slot.cache_size(); if size > self.debt { self.evict.push_front((sequence, stamp)); return; @@ -1314,10 +1325,11 @@ impl TrackState { // walk it in arrival order and the feed keeps writing it. The fetch fills its // head instead. Anything else that can't answer from `frame_start` (evicted, // aborted, or narrower backfill) is replaced; a group that can is a duplicate. - if self.covering_group(sequence, frame_start).is_some() { - return Err(Error::Duplicate); - } - if !self.lookup.get(&sequence).is_some_and(Slot::is_live) { + if self.lookup.get(&sequence).is_some_and(Slot::is_live) { + if self.covering_group(sequence, frame_start).is_some() { + return Err(Error::Duplicate); + } + } else { self.claim_sequence(sequence, frame_start)?; } @@ -2613,7 +2625,7 @@ impl Consumer { pub(crate) fn poll_group(&self, sequence: u64, waiter: &kio::Waiter) -> Poll> { let res = self.state.poll(waiter, |state| { if let Some(slot) = state.lookup.get(&sequence) - && !slot.serving().is_aborted() + && !slot.is_aborted() { return Poll::Ready(Some(slot.serving().consume())); } @@ -3772,7 +3784,7 @@ impl Subscriber { && state .lookup .get(sequence) - .is_some_and(|slot| slot.stamp == *stamp && !slot.serving().is_aborted()) + .is_some_and(|slot| slot.stamp == *stamp && !slot.is_aborted()) })) } @@ -8090,6 +8102,63 @@ mod test { ); } + /// A live group with a fetched head: group 0, fed from frame 1, plus a two-frame head. + async fn live_group_with_head(producer: &Producer) -> (group::Producer, group::Producer) { + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap(); + live.start_at(1).unwrap(); + live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta")) + .unwrap(); + + let fetch = consumer.fetch_group(0, None); + let request = dynamic.requested_group().await.unwrap(); + let mut head = request.accept(None).unwrap(); + for payload in [&b"snapshot"[..], b"delta"] { + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(payload)) + .unwrap(); + } + head.finish().unwrap(); + assert_eq!(fetch.await.unwrap().index(), 0); + (live, head) + } + + /// The live copy ending early (its subscription went away) leaves a fetched head + /// that still holds the whole group servable, instead of reclaiming the slot. + #[tokio::test] + async fn aborted_live_group_keeps_its_fetched_head() { + let producer = track_producer("test", None); + let (live, _head) = live_group_with_head(&producer).await; + + live.abort(Error::Cancel).unwrap(); + // The next group demotes 0 into the eviction order and runs an expiry scan. + producer.create_group(group::Info { sequence: 1 }).unwrap(); + + let mut group = producer.consume().peek_group(0).expect("the head still serves group 0"); + assert_eq!( + group.read_frame().await.unwrap().unwrap().payload, + bytes::Bytes::from_static(b"snapshot") + ); + } + + /// Reads land on the fetched head, so they keep the slot from expiring even though + /// the live copy beside it sits idle. + #[tokio::test] + async fn reading_the_head_keeps_the_slot_from_expiring() { + let producer = track_producer("test", None); + let (live, _head) = live_group_with_head(&producer).await; + live.finish().unwrap(); + producer.create_group(group::Info { sequence: 1 }).unwrap(); + + crate::model::clock::advance(cache::DEFAULT_EXPIRY + Duration::from_secs(1)); + // A fetch hit on the head, past the window the live copy was last written in. + producer.consume().fetch_group(0, None).await.unwrap(); + producer.create_group(group::Info { sequence: 2 }).unwrap(); + + assert!(producer.consume().peek_group(0).is_some(), "a read head keeps its slot"); + } + /// Joining a queued fetch widens its range, and a caller that arrives once the range /// is already on the wire fails cleanly rather than being handed the narrower group. #[tokio::test] From 6b4f066ef979617e0f1b54ff7b9c9981474a3b59 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 17:46:26 -0700 Subject: [PATCH 4/5] fix(net): a fetched head joins the pool's access average with its slot The head skipped cache_demote, so a slot left holding only its head had no access sample and could dodge pressure eviction indefinitely. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 49 ++++++++++++++++++++++++++++++++--- 1 file changed, 46 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 5951ca1a37..126bc3eab5 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -368,6 +368,14 @@ impl Slot { }) } + /// Enter every copy into the evictable population, so the pool's access average + /// samples whichever one readers use. + fn cache_demote(&self) { + for group in self.copies() { + group.cache_demote(); + } + } + /// Abort every copy this slot holds. fn abort(&self, err: Error) { for group in self.copies() { @@ -929,7 +937,7 @@ impl TrackState { if let Some(latest) = self.latest_group && let Some(slot) = self.lookup.get(&latest) { - slot.group.cache_demote(); + slot.cache_demote(); self.evict.push_back((latest, slot.stamp)); } } @@ -1002,7 +1010,7 @@ impl TrackState { && sequence > latest && let Some(prev) = self.lookup.get(&latest) { - prev.group.cache_demote(); + prev.cache_demote(); self.evict.push_back((latest, prev.stamp)); } self.latest_group = Some(sequence); @@ -1344,8 +1352,15 @@ impl TrackState { // Settled first, like `commit_group`, which may evict the live slot itself. self.charge_debt(); + let protected = self.protects(sequence); match self.lookup.get_mut(&sequence).filter(|slot| slot.is_live()) { - Some(slot) => slot.head = Some(Box::new(group.clone())), + Some(slot) => { + // Joins the slot's standing: demoted already unless it is the live edge. + if !protected { + group.cache_demote(); + } + slot.head = Some(Box::new(group.clone())); + } // Invisible to arrival-order subscribers: fetched on demand, not produced // live by the publisher. None => self.insert_group(&group, false), @@ -8159,6 +8174,34 @@ mod test { assert!(producer.consume().peek_group(0).is_some(), "a read head keeps its slot"); } + /// A head left alone in its slot (the live copy aborted) still joins the pool's + /// access average when the slot is demoted, so memory pressure can evict it even + /// when nothing else is evictable. + #[tokio::test] + async fn a_head_only_slot_yields_to_memory_pressure() { + let (producer, _pool) = pooled_producer(10_000); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap(); + live.start_at(1).unwrap(); + let fetch = consumer.fetch_group(0, None); + let mut head = dynamic.requested_group().await.unwrap().accept(None).unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000])) + .unwrap(); + head.finish().unwrap(); + drop(fetch.await.unwrap()); + live.abort(Error::Cancel).unwrap(); + + // The latest group is protected, so the head is the only evictable content. + let mut next = producer.create_group(group::Info { sequence: 1 }).unwrap(); + for _ in 0..30 { + next.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000])) + .unwrap(); + } + assert!(consumer.peek_group(0).is_none(), "pressure evicts the head"); + } + /// Joining a queued fetch widens its range, and a caller that arrives once the range /// is already on the wire fails cleanly rather than being handed the narrower group. #[tokio::test] From de7a794c6ad66ac34dda25de7079c8640c598479 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 18:15:16 -0700 Subject: [PATCH 5/5] fix(net): weigh a slot's copies by their mean access for eviction Both copies sample the pool's access average, so comparing their maximum kept a lone two-copy slot above the average forever under pressure. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 47 +++++++++++++++++++++++++---------- 1 file changed, 34 insertions(+), 13 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 126bc3eab5..eacd270e3c 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -341,13 +341,17 @@ impl Slot { self.serving().is_aborted() } - /// The latest access to any copy still held: readers may be on either one. + /// The mean access over copies still held, weighed the way each one samples the + /// pool's average: a maximum would sit above the mean forever, so a lone slot with + /// two copies could never be evicted. fn cache_accessed(&self) -> u64 { - self.copies() + let (sum, count) = self + .copies() .filter(|group| !group.is_aborted()) - .map(group::Producer::cache_accessed) - .max() - .unwrap_or(0) + .fold((0u128, 0u128), |(sum, count), group| { + (sum + group.cache_accessed() as u128, count + 1) + }); + (sum / count.max(1)) as u64 } /// The bytes every copy holds, all freed together. @@ -8174,32 +8178,49 @@ mod test { assert!(producer.consume().peek_group(0).is_some(), "a read head keeps its slot"); } - /// A head left alone in its slot (the live copy aborted) still joins the pool's - /// access average when the slot is demoted, so memory pressure can evict it even - /// when nothing else is evictable. - #[tokio::test] - async fn a_head_only_slot_yields_to_memory_pressure() { + /// Fill group 0 as a live copy plus a fetched head written after it, optionally + /// abort the live copy, then write past capacity into a protected latest group, so + /// slot 0 is the only evictable content. Returns whether pressure evicted it. + async fn pressure_evicts_a_headed_slot(abort_live: bool) -> bool { let (producer, _pool) = pooled_producer(10_000); let dynamic = producer.dynamic(); let consumer = producer.consume(); let mut live = producer.create_group(group::Info { sequence: 0 }).unwrap(); live.start_at(1).unwrap(); + live.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"delta")) + .unwrap(); + live.finish().unwrap(); let fetch = consumer.fetch_group(0, None); let mut head = dynamic.requested_group().await.unwrap().accept(None).unwrap(); head.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000])) .unwrap(); head.finish().unwrap(); drop(fetch.await.unwrap()); - live.abort(Error::Cancel).unwrap(); + if abort_live { + live.abort(Error::Cancel).unwrap(); + } - // The latest group is protected, so the head is the only evictable content. let mut next = producer.create_group(group::Info { sequence: 1 }).unwrap(); for _ in 0..30 { next.write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![0u8; 10_000])) .unwrap(); } - assert!(consumer.peek_group(0).is_none(), "pressure evicts the head"); + consumer.peek_group(0).is_none() + } + + /// A head left alone in its slot still joins the pool's access average when the + /// slot is demoted, so memory pressure can evict it. + #[tokio::test] + async fn a_head_only_slot_yields_to_memory_pressure() { + assert!(pressure_evicts_a_headed_slot(true).await); + } + + /// A slot weighs its copies the way the pool's average samples them, so two copies + /// with different access stamps can't keep it above the average on their own. + #[tokio::test] + async fn a_two_copy_slot_yields_to_memory_pressure() { + assert!(pressure_evicts_a_headed_slot(false).await); } /// Joining a queued fetch widens its range, and a caller that arrives once the range