From f04cce8aa8d1f0dcae2683dbb8e33b9c8771f48d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 22:16:45 -0700 Subject: [PATCH 1/4] fix(net): judge spliced staleness against the logical live edge Each segment of a spliced track measured drift against its own track's edge, so a parked warm copy only ever saw its frozen last group and a returning reader replayed a backlog a plain track would have shed. The splice now pushes the newest edge across its segments down to each segment cursor alongside the cap it already pushed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 69 ++++++++++ rs/moq-net/src/model/resume.rs | 223 ++++++++++++++++++++++----------- rs/moq-net/src/model/track.rs | 195 +++++++++++++++++++--------- 3 files changed, 354 insertions(+), 133 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 2f4c0af840..26186b4424 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6291,6 +6291,75 @@ mod tests { assert_eq!(&group.read_frame().await.unwrap().unwrap().payload[..], b"live"); } + /// A returning reader judges the warm cache against the logical track's live + /// edge, not the parked segment's own frozen one: groups the fresh source + /// has left behind by more than the budget are skipped, exactly as they would + /// be on one unspliced track. + #[tokio::test] + async fn resumed_reader_skips_warm_groups_behind_the_new_edge() { + let budget = track::Subscription::default().with_max_age(Duration::from_millis(100)); + let write = |source: &track::Producer, sequence: u64, millis: u64| { + let mut group = source.create_group(sequence.into()).unwrap(); + group + .write_frame(crate::Timestamp::from_millis(millis).unwrap(), b"x".as_ref()) + .unwrap(); + group.finish().unwrap(); + }; + let drain = |subscription: &mut track::Subscriber| { + let mut sequences = Vec::new(); + while let Poll::Ready(group) = subscription.poll_recv_group(&kio::Waiter::noop()) { + sequences.push(group.unwrap().expect("track ended").sequence); + } + sequences + }; + + let producer = origin(1).produce(); + let consumer = producer.consume(); + let broadcast = producer.publish("room/alice", Route::default()).unwrap(); + let mut dynamic = broadcast.dynamic(); + let resolved = consumer.request_broadcast("room/alice").await.expect("resolves"); + + let track = resolved.track("video").unwrap(); + let first = budget.clone(); + let subscribing = tokio::spawn(async move { track.subscribe(first).await }); + let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track()) + .await + .expect("the front asked the source") + .expect("request"); + let source = request.accept(None); + for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] { + write(&source, sequence, millis); + } + let mut subscription = subscribing.await.unwrap().expect("subscribe"); + next_group(&mut subscription).await.unwrap().expect("a cached group"); + drain(&mut subscription); + drop(subscription); + + // Parked: groups 0..=3 stay warm while the source goes idle. + tokio::time::timeout(Duration::from_secs(1), source.unused()) + .await + .expect("parked") + .expect("source open"); + drop(source); + + let track = resolved.track("video").unwrap(); + let second = budget.clone(); + let subscribing = tokio::spawn(async move { track.subscribe(second).await }); + let request = tokio::time::timeout(Duration::from_secs(1), dynamic.requested_track()) + .await + .expect("the front asked the source again") + .expect("request"); + let source = request.accept(None); + for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] { + write(&source, sequence, millis); + } + let mut subscription = subscribing.await.unwrap().expect("resubscribe"); + settle(|| subscription.latest() == Some(22)).await; + + // Groups 0..=2 reach at most 60ms against an edge at 440ms. + assert_eq!(drain(&mut subscription), [3, 20, 21, 22]); + } + /// A front serving from another front's spliced copy has no snapshot to keep: /// it still drops upstream on the unused edge, so the publisher's `unused()` /// resolves far below `TRACK_IDLE_LINGER` through the whole chain. The next diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 2fa3d0c048..a4365602ec 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -27,6 +27,7 @@ use std::ops::Bound; use std::task::{Poll, ready}; use crate::{Datagram, Error, Result, frame, group, track}; +use track::{Anchor, LiveEdge}; use super::subscription::{Cap, Position, Subscription, max_some, min_some}; @@ -219,6 +220,20 @@ impl ResumeState { } } + /// The newest live edge across the segments below the exclusive `cap`, each clamped + /// to its own range: the edge the logical track measures drift against. + fn live_edge(&self, cap: Option) -> Option { + self.segments + .iter() + .filter_map(|segment| { + let edge = segment.track.live_edge(min_some(cap, last_group(segment.end)))?; + // Out-of-range content below the segment is not its to serve. + let start = segment.start.map_or(0, |start| start.group); + (edge.sequence >= start).then_some(edge) + }) + .max_by_key(|edge| edge.sequence) + } + /// Append a segment serving the track from `start` onward, capping (or replacing) /// the previous segments so the ranges stay disjoint and ascending. fn switch(&mut self, track: track::Consumer, start: Option) -> Result<()> { @@ -591,8 +606,8 @@ impl Consumer { next_sequence: 0, min_sequence: 0, end_sequence: None, - stale_cap: None, - drift_cap: kio::Producer::new(None), + outer: Anchor::default(), + drift_anchor: kio::Producer::new(Anchor::default()), } } @@ -659,6 +674,11 @@ impl Consumer { self.state.read().resume_position() } + /// The newest live edge across the segments; see [`track::Consumer::live_edge`]. + pub(crate) fn live_edge(&self, cap: Option) -> Option { + self.state.read().live_edge(cap) + } + /// The newest cached group across every spliced segment; see /// [`track::Consumer::peek_latest`]. pub(crate) fn peek_latest(&self) -> Option { @@ -865,8 +885,8 @@ pub(crate) struct Group { state: kio::Consumer, /// The logical subscription's live max age budget. subscription: kio::Consumer, - /// The logical reader's group cap, used only to bound each route's drift anchor. - cap: kio::Consumer>, + /// The logical reader's drift anchor, used only to judge each route's copy. + anchor: kio::Consumer, /// The logical group being assembled. sequence: u64, @@ -921,7 +941,7 @@ impl Clone for Group { Self { state: self.state.clone(), subscription: self.subscription.clone(), - cap: self.cap.clone(), + anchor: self.anchor.clone(), sequence: self.sequence, index: self.index, end: self.end, @@ -936,14 +956,14 @@ impl Group { fn new( state: kio::Consumer, subscription: kio::Consumer, - cap: kio::Consumer>, + anchor: kio::Consumer, sequence: u64, index: u64, ) -> Self { Self { state, subscription, - cap, + anchor, sequence, index, end: None, @@ -1136,7 +1156,7 @@ impl Group { // `start_at` clamps up to the first frame the copy still holds, so landing // higher than asked means this route can't cover the seam after all. Treat it // like a dead copy and wait for one that can. - let mut group = track.guard_group(group, self.subscription.clone(), self.cap.clone(), bound); + let mut group = track.guard_group(group, self.subscription.clone(), self.anchor.clone(), bound); group.set_stale_meter(self.stale_stats.clone()); group.start_at(self.index); if group.index() != self.index { @@ -1295,7 +1315,7 @@ impl Group { // already includes the frames it skipped. Some(continuation) => { let mut continuation = - track.guard_group(continuation, self.subscription.clone(), self.cap.clone(), bound); + track.guard_group(continuation, self.subscription.clone(), self.anchor.clone(), bound); continuation.set_stale_meter(self.stale_stats.clone()); return continuation.poll_finished(waiter); } @@ -1437,14 +1457,15 @@ pub struct Subscriber { min_sequence: u64, /// Exclusive cap for [`Self::next_group`], set by [`Self::end_at`]. end_sequence: Option, - /// A cap imposed by a reader wrapping this subscriber (a nested splice segment), - /// folded into every drift anchor pushed onto the segments. Never bounds delivery: - /// the outer reader enforces its own window, and an inner cap would hide a - /// segment's completion from it. - stale_cap: Option, - /// Shared copy of the effective cap ([`Self::end_sequence`] and [`Self::stale_cap`] - /// combined) for groups that outlive this cursor poll. - drift_cap: kio::Producer>, + /// The anchor imposed by a reader wrapping this subscriber (a nested splice + /// segment), folded into every drift anchor pushed onto the segments. Its cap never + /// bounds delivery: the outer reader enforces its own window, and an inner cap would + /// hide a segment's completion from it. + outer: Anchor, + /// The logical drift anchor ([`Self::end_sequence`] and [`Self::outer`] combined, + /// with the newest edge across the segments), shared with groups that outlive this + /// cursor poll. Refreshed on every [`Self::poll_sync`]. + drift_anchor: kio::Producer, } impl Subscriber { @@ -1453,6 +1474,7 @@ impl Subscriber { fn poll_sync(&mut self, waiter: &kio::Waiter) { self.sync(waiter); self.reap(); + self.refresh_anchor(); } /// Reap retired cursors, then bound the live stragglers: a pruned segment's @@ -1571,7 +1593,6 @@ impl Subscriber { } self.segments.retain(|s| !s.retired()); - let anchor = self.anchor_end(); let nexts: Vec<_> = segments.iter().skip(1).map(|next| Some(next.track.clone())).collect(); let nexts = nexts.into_iter().chain(std::iter::once(None)); for (segment, next) in segments.into_iter().zip(nexts) { @@ -1581,11 +1602,10 @@ impl Subscriber { warm.next = next; } if existing.end != segment.end { + // The boundary bounds the drift anchor as well as the demand; the + // anchor follows in `refresh_anchor`. existing.end = segment.end; - let cap = Self::stale_cap(existing, anchor); if let Some(sub) = existing.stale_sub_mut() { - // The boundary bounds the drift anchor as well as the demand. - sub.set_stale_cap(cap); // Shrink the demand so the session can cap upstream. The // read bounds stay on this subscriber (see `poll_recv_group`): // an inner `end_at` would park boundary-crossing groups in the @@ -1642,7 +1662,7 @@ impl Subscriber { let spliced = Group::new( self.state.clone(), self.prefs.consume(), - self.drift_cap.consume(), + self.drift_anchor.consume(), sequence, 0, ) @@ -1650,41 +1670,66 @@ impl Subscriber { Some(group.into_spliced(spliced)) } - /// The highest sequence a segment could hand its reader: the reader's own cap and - /// the segment's boundary, whichever is lower. + /// The anchor a segment's cursor measures drift against: the logical `anchor`, with + /// its cap lowered to the segment's boundary. /// /// A segment's inner cursor is deliberately left uncapped (an inner `end_at` would /// park boundary-crossing groups where its completion can't be seen), so this is how /// both bounds reach the drift anchor. Without them a segment measures staleness /// against groups it will never surface: the route running past the boundary, or the - /// reader's own cap holding content back. `anchor_end` is [`Self::anchor_end`], so a - /// cap imposed on this subscriber from outside is included. - fn stale_cap(seg: &SegmentSub, anchor_end: Option) -> Option { - min_some(anchor_end, seg.last_group()) + /// reader's own cap holding content back. The edge is left as is: it is the newest + /// content the logical track holds, which a segment's own track never sees. + fn segment_anchor(seg: &SegmentSub, anchor: Anchor) -> Anchor { + Anchor { + cap: min_some(anchor.cap, seg.last_group()), + ..anchor + } } - /// The tightest cap any reader of this subscriber imposes: its own [`Self::end_at`] - /// and whatever a wrapping splice pushed down. What every segment's drift anchor is - /// bounded by. - fn anchor_end(&self) -> Option { - min_some(self.stale_cap, self.end_sequence) + /// The logical drift anchor, as of the last [`Self::refresh_anchor`]. + fn anchor(&self) -> Anchor { + *self.drift_anchor.read() } - /// Bound every segment's drift anchor from outside; the spliced arm of - /// [`track::Subscriber::set_stale_cap`], for this subscriber nested as a segment of - /// another splice. - pub(crate) fn set_stale_cap(&mut self, cap: Option) { - self.stale_cap = cap; - self.update_drift_cap(); - let anchor = self.anchor_end(); + /// Re-derive the logical drift anchor and push it onto every segment cursor. + /// + /// The cap is the tightest any reader of this subscriber imposes: its own + /// [`Self::end_at`] and whatever a wrapping splice pushed down. The edge is the + /// newest across the producer's segments within that cap, or a wrapping splice's if + /// newer. Resolved on every sync, since each segment is a separate track and the + /// logical edge moves whenever any of them grows. + fn refresh_anchor(&mut self) { + let cap = min_some(self.outer.cap, self.end_sequence); + let edge = self + .state + .read() + .live_edge(cap) + .into_iter() + .chain(self.outer.edge) + .max_by_key(|edge| edge.sequence); + let anchor = Anchor { cap, edge }; + // Skip a no-op write: every handed-out group's expiry watches this channel. + if self.anchor() != anchor + && let Ok(mut current) = self.drift_anchor.write() + { + *current = anchor; + } for seg in &mut self.segments { - let cap = Self::stale_cap(seg, anchor); + let anchor = Self::segment_anchor(seg, anchor); if let Some(sub) = seg.stale_sub_mut() { - sub.set_stale_cap(cap); + sub.set_anchor(anchor); } } } + /// Bound every segment's drift anchor from outside; the spliced arm of + /// [`track::Subscriber::set_anchor`], for this subscriber nested as a segment of + /// another splice. + pub(crate) fn set_anchor(&mut self, anchor: Anchor) { + self.outer = anchor; + self.refresh_anchor(); + } + /// Count each segment's seek convictions behind the deliverer's `committed` /// watermark; the spliced arm of [`track::Subscriber::commit_seek_stale`]. pub(crate) fn commit_seek_stale(&mut self, committed: u64) { @@ -1723,13 +1768,6 @@ impl Subscriber { Poll::Ready(Ok(false)) } - /// Publish the effective cap for groups that outlive this cursor poll. - fn update_drift_cap(&mut self) { - if let Ok(mut cap) = self.drift_cap.write() { - *cap = self.anchor_end(); - } - } - /// Resolve a segment's pending subscription, if any. Ready once the segment is /// `Active` or `Done`; a rejected or closed track becomes `Done` (stall, not /// error). Never consumes groups, so terminal-state pollers can share it. @@ -1737,7 +1775,7 @@ impl Subscriber { seg: &mut SegmentSub, prefs: &Subscription, min_sequence: u64, - anchor_end: Option, + anchor: Anchor, waiter: &kio::Waiter, ) -> Poll<()> { if matches!(seg.sub, SubState::Pending(_)) @@ -1770,7 +1808,7 @@ impl Subscriber { // assigned: the inner subscription resolved its own start from its // budget and floor, and this must not rewind past it. sub.raise_start_to(seg.first_group().max(min_sequence)); - sub.set_stale_cap(Self::stale_cap(seg, anchor_end)); + sub.set_anchor(Self::segment_anchor(seg, anchor)); let _ = sub.update(slice(prefs, seg.ask, seg.end)); seg.sub = SubState::Active(Box::new(sub)); } @@ -1787,13 +1825,13 @@ impl Subscriber { seg: &mut SegmentSub, prefs: &Subscription, min_sequence: u64, - anchor_end: Option, + anchor: Anchor, waiter: &kio::Waiter, ) -> Poll> { loop { match &mut seg.sub { SubState::Pending(_) => { - ready!(Self::poll_activate(seg, prefs, min_sequence, anchor_end, waiter)); + ready!(Self::poll_activate(seg, prefs, min_sequence, anchor, waiter)); } SubState::Active(sub) => match ready!(sub.poll_recv_group(waiter)) { Ok(Some(group)) => { @@ -1859,7 +1897,7 @@ impl Subscriber { self.commit_seek_stale(committed); } - let anchor = self.anchor_end(); + let anchor = self.anchor(); let mut floor = floor; 'retry: loop { let mut all_done = true; @@ -1966,7 +2004,7 @@ impl Subscriber { self.poll_sync(waiter); let end_sequence = self.end_sequence; - let anchor = self.anchor_end(); + let anchor = self.anchor(); let min_sequence = self.min_sequence; let beyond_cap = |sequence: u64| !super::subscription::before_end(sequence, end_sequence); @@ -2142,7 +2180,7 @@ impl Subscriber { // datagrams must still resolve the subscription (registering demand) and // be woken when it activates. let mut pending_activation = false; - let anchor = self.anchor_end(); + let anchor = self.anchor(); if let Some(seg) = self.segments.last_mut() { if Self::poll_activate(seg, &self.last_prefs, self.min_sequence, anchor, waiter).is_pending() { pending_activation = true; @@ -2203,7 +2241,7 @@ impl Subscriber { /// completing the segment, would steal them from a `recv_group` caller on the /// same subscriber. fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll> { - let anchor = min_some(self.stale_cap, self.end_sequence); + let anchor = self.anchor(); let Some(seg) = self.segments.last_mut() else { return Poll::Ready(None); }; @@ -2269,16 +2307,9 @@ impl Subscriber { /// re-offers it. pub fn end_at(&mut self, end: impl Into) { self.end_sequence = end.into().exclusive(); - self.update_drift_cap(); // The cap bounds each segment's drift anchor as well as this reader's own // delivery: a segment must not measure against groups this cap hides. - let anchor = self.anchor_end(); - for seg in &mut self.segments { - let cap = Self::stale_cap(seg, anchor); - if let Some(sub) = seg.stale_sub_mut() { - sub.set_stale_cap(cap); - } - } + self.refresh_anchor(); } /// The shared preferences channel, so `track::Control` can wrap it. @@ -3126,9 +3157,11 @@ mod test { "the arrival path skips straight to the live edge" ); - // Spliced: the same backlog split across a takeover boundary. - let (mut track_a, consumer_a) = track_pair("a"); - let (mut track_b, consumer_b) = track_pair("b"); + // Spliced: the same backlog split across a takeover boundary. Retained long + // enough that the replay budget below is not clamped to the default window. + let retain = track::Info::default().with_max_age(Duration::from_secs(60)); + let (mut track_a, consumer_a) = track_pair_with("a", retain.clone()); + let (mut track_b, consumer_b) = track_pair_with("b", retain); let mut producer = Producer::new(); producer.switch(&consumer_a, None).unwrap(); producer.switch(&consumer_b, Position::group(2)).unwrap(); @@ -3147,11 +3180,12 @@ mod test { }) .collect(); - // Group 1 survives where the plain cursor drops it: each segment measures drift - // against its own boundary, since a segment cannot serve content past it, and - // group 1 is the newest thing segment A has. The sequence path inherits that - // from the arrival path rather than inventing its own anchor, so the two agree. - assert_eq!(spliced, vec![1, 3], "each segment is judged within its own boundary"); + // Group 1 survives where the plain cursor drops it: every segment measures + // against the logical edge (group 3), but a group's reach is bounded only by a + // successor in its own track, and group 1's lives in segment B. The sequence + // path inherits that from the arrival path rather than inventing its own + // anchor, so the two agree. + assert_eq!(spliced, vec![1, 3], "each segment is judged against the logical edge"); let mut arrival = producer.consume().subscribe(None); let arrival: Vec = std::iter::from_fn(|| { @@ -3177,9 +3211,50 @@ mod test { assert_eq!(replayed, vec![0, 1, 2, 3], "a backlog inside the budget crosses whole"); } + /// A segment's track never sees the groups of the segments after it, so its own + /// edge freezes once a takeover caps it. Both spliced cursors judge its backlog + /// against the logical edge instead, the one a single track would have. + #[tokio::test] + async fn a_capped_segment_is_judged_against_the_newest_segment() { + let budget = Subscription::default().with_max_age(Duration::from_millis(100)); + let (mut track_a, consumer_a) = track_pair("a"); + let (mut track_b, consumer_b) = track_pair("b"); + let mut producer = Producer::new(); + producer.switch(&consumer_a, None).unwrap(); + producer.switch(&consumer_b, Position::group(4)).unwrap(); + for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] { + write_group_at(&mut track_a, sequence, "a", Duration::from_millis(millis)); + } + for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] { + write_group_at(&mut track_b, sequence, "b", Duration::from_millis(millis)); + } + + // Groups 0..=2 reach at most 60ms against an edge at 440ms. Group 3 stays: its + // successor lives in segment B, so nothing in its own track bounds its reach. + let mut arrival = producer.consume().subscribe(budget.clone()); + let arrival: Vec = std::iter::from_fn(|| { + kio::wait(|waiter| arrival.poll_recv_group(waiter)) + .now_or_never()? + .expect("should not error") + .map(|group| group.sequence) + }) + .collect(); + assert_eq!(arrival, vec![3, 20, 21, 22]); + + let mut ordered = producer.consume().subscribe(budget); + let ordered: Vec = std::iter::from_fn(|| { + kio::wait(|waiter| ordered.poll_next_group(waiter)) + .now_or_never()? + .expect("should not error") + .map(|group| group.sequence) + }) + .collect(); + assert_eq!(ordered, arrival); + } + /// A nested splice's leaves are judged within the *outer* boundary. The outer /// window reaches them two ways, through the seek's own `end` and through - /// [`track::Subscriber::set_stale_cap`] recursing into the spliced segment; without + /// [`track::Subscriber::set_anchor`] recursing into the spliced segment; without /// either, a leaf anchors drift on a group past the outer boundary and convicts /// everything the outer segment still owes its reader. #[tokio::test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 17bec9149b..2e4082577b 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -477,9 +477,8 @@ impl TrackState { /// it, so the groups it still wants aren't stale just because the route ran on. /// Fetched backfill is absent from `arrival`, so it cannot age subscription content /// as though it were a live replacement. - fn live_edge(&self, cap: Option) -> Option { - let presentation = self - .lookup + fn live_edge(&self, cap: Option) -> Option { + self.lookup .range(..) .rev() .filter(|(seq, _)| super::subscription::before_end(**seq, cap)) @@ -495,9 +494,18 @@ impl TrackState { stamp: slot.stamp, timestamp: slot.group.latest().unwrap_or(timestamp), }) - }); + }) + } - presentation.map(|presentation| Edge { presentation, cap }) + /// The edge a reader bounded by `anchor` measures drift against: this track's own + /// live edge under the anchor's cap, plus whatever newer edge the anchor carries + /// from outside (a splice's other segments). + fn drift_edge(&self, anchor: Anchor) -> Edge { + Edge { + presentation: self.live_edge(anchor.cap), + outer: anchor.edge, + cap: anchor.cap, + } } /// The furthest presentation time the group at `sequence` could still reach: where @@ -543,11 +551,9 @@ impl TrackState { /// /// The edge must sit strictly above the candidate. The live edge is never late /// against itself, and backfill or the tail of a rewound timeline can carry a high - /// timestamp on a low sequence without being an edge at all. - fn is_stale(&self, sequence: u64, edge: Option<&Edge>, budget: Duration) -> bool { - let Some(edge) = edge else { - return false; - }; + /// timestamp on a low sequence without being an edge at all. Of the edges that + /// qualify, the highest sequence is the newest content, whichever track holds it. + fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { if !self.lookup.contains_key(&sequence) { return false; } @@ -555,16 +561,27 @@ impl TrackState { // The anchor was resolved under an earlier lock, so confirm it still names // the same servable incarnation before it convicts a candidate. Failing safe // (delivering) is right, since the next poll resolves fresh anchors. - let live_edge = &edge.presentation; - let reach = self.reach(sequence, edge.cap); - live_edge.sequence > sequence - && self - .lookup - .get(&live_edge.sequence) - .is_some_and(|live| live.stamp == live_edge.stamp && !live.group.is_aborted()) - && reach.is_some_and( - |reach| matches!(live_edge.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget), - ) + let local = edge + .presentation + .filter(|live| { + self.lookup + .get(&live.sequence) + .is_some_and(|slot| slot.stamp == live.stamp && !slot.group.is_aborted()) + }) + .map(LiveEdge::from); + // An outer edge lives on another track, so there is no slot here to revalidate + // it against; it is resolved fresh on every poll of the splice that pushes it. + let Some(live_edge) = local + .into_iter() + .chain(edge.outer) + .filter(|live| live.sequence > sequence) + .max_by_key(|live| live.sequence) + else { + return false; + }; + self.reach(sequence, edge.cap).is_some_and( + |reach| matches!(live_edge.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget), + ) } /// Resolve a one-shot fetch from the track side: the cached group, or an [`Error`] @@ -1554,7 +1571,7 @@ impl Producer { let min_sequence = floor_of(&preferences); let subscription = kio::Producer::new(preferences); register_subscription(self.state.read(), &subscription); - let drift_cap = kio::Producer::new(None); + let drift_anchor = kio::Producer::new(Anchor::default()); // Hoisted: an inline `read()` guard would live to the end of the struct literal, // deadlocking against the `consume()` below. @@ -1573,7 +1590,7 @@ impl Producer { end_sequence: None, parked: BTreeMap::new(), stale_cap: None, - drift_cap, + drift_anchor, stale: stats::Content::default(), seek_pending: BTreeMap::new(), }), @@ -2369,6 +2386,15 @@ impl Consumer { } } + /// The live edge below the exclusive `cap` that drift is measured against; see + /// [`TrackState::live_edge`]. A splice reports the newest across its segments. + pub(crate) fn live_edge(&self, cap: Option) -> Option { + match &self.inner { + ConsumerKind::Plain(state) => state.read().live_edge(cap).map(LiveEdge::from), + ConsumerKind::Spliced(resume) => resume.live_edge(cap), + } + } + /// The nearest cached group below `sequence`, under the same terms as /// [`Self::peek_group`]. Walks the cache's own order, so gaps in the group numbering /// are crossed and aborted (evicted) entries are skipped. @@ -2411,7 +2437,7 @@ impl Consumer { &self, group: group::Consumer, subscription: kio::Consumer, - cap: kio::Consumer>, + anchor: kio::Consumer, bound: Option, ) -> group::Consumer { let ConsumerKind::Plain(state) = &self.inner else { @@ -2421,7 +2447,7 @@ impl Consumer { group.with_expiry(Arc::new(GroupExpiry { state: state.weak(), subscription, - cap, + anchor, bound, sequence, })) @@ -2672,7 +2698,7 @@ impl Subscribing { let info = ready!(state.poll(waiter, |state| state.poll_info())) .map_err(|e| e.abort.clone().unwrap_or(Error::Dropped))??; - let drift_cap = kio::Producer::new(None); + let drift_anchor = kio::Producer::new(Anchor::default()); let min_sequence = floor_of(&self.subscription.read()); Poll::Ready(Ok(Subscriber { name: self.name.clone(), @@ -2688,7 +2714,7 @@ impl Subscribing { end_sequence: None, parked: BTreeMap::new(), stale_cap: None, - drift_cap, + drift_anchor, stale: stats::Content::default(), seek_pending: BTreeMap::new(), }), @@ -2982,7 +3008,7 @@ enum SubscriberKind { #[derive(Clone)] struct Drift { budget: Duration, - edge: Option, + edge: Edge, } /// Keeps one handed-out group tied to the subscription whose cursor selected it. @@ -2994,7 +3020,7 @@ struct GroupExpiry { /// the last real subscriber. state: kio::ConsumerWeak, subscription: kio::Consumer, - cap: kio::Consumer>, + anchor: kio::Consumer, bound: Option, sequence: u64, } @@ -3007,19 +3033,20 @@ impl group::Expiry for GroupExpiry { Poll::<()>::Pending }); - let mut cap = None; - let _ = self.cap.poll(waiter, |current| { - cap = **current; + let mut anchor = Anchor::default(); + let _ = self.anchor.poll(waiter, |current| { + anchor = **current; Poll::<()>::Pending }); - let cap = super::subscription::min_some(cap, self.bound); + anchor.cap = super::subscription::min_some(anchor.cap, self.bound); + let cap = anchor.cap; let mut expired = false; let _ = self.state.poll(waiter, |state| { let budget = clamp_max_age(max_age, state.max_age_bound()); loop { - let edge = state.live_edge(cap); - expired = state.is_stale(self.sequence, edge.as_ref(), budget); + let edge = state.drift_edge(anchor); + expired = state.is_stale(self.sequence, &edge, budget); if expired { break; } @@ -3075,12 +3102,43 @@ impl group::Expiry for GroupExpiry { /// from whatever may occupy its sequence by the time a candidate is judged. #[derive(Clone)] struct Edge { - presentation: PresentationEdge, + /// This track's own edge, revalidated before it convicts anything. + presentation: Option, + /// A newer edge on another track of the same splice; see [`Anchor::edge`]. + outer: Option, /// The cap the edge was resolved under, so per-candidate reach lookups measure /// against the same servable window. cap: Option, } +/// How a reader wrapping a cursor bounds its drift anchor from outside: pushed by a +/// splice onto each segment's cursor, and shared with the groups a cursor hands out. +#[derive(Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct Anchor { + /// The exclusive sequence cap on what the reader could be handed; see [`servable_cap`]. + pub cap: Option, + /// The newest edge across a splice's segments. Each segment is a separate track + /// that only sees its own groups, so without this a parked segment measures against + /// its own frozen edge while the logical track has moved on. + pub edge: Option, +} + +/// The newest stamped group of a track: its sequence and the newest frame it presented. +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) struct LiveEdge { + pub sequence: u64, + pub timestamp: Timestamp, +} + +impl From for LiveEdge { + fn from(edge: PresentationEdge) -> Self { + Self { + sequence: edge.sequence, + timestamp: edge.timestamp, + } + } +} + /// The newest servable group that has presented at least one frame. #[derive(Clone, Copy)] struct PresentationEdge { @@ -3124,8 +3182,9 @@ struct PlainSubscriber { /// segment), folded into the drift anchor only. Delivery is still bounded by /// `end_sequence`, which stays unset on a segment so its completion is visible. stale_cap: Option, - /// Shared effective cap used by groups after this cursor hands them out. - drift_cap: kio::Producer>, + /// Shared effective anchor used by groups after this cursor hands them out. The + /// only copy of the outer edge a wrapping reader pushed (see [`Anchor::edge`]). + drift_anchor: kio::Producer, /// Groups the drift budget skipped since the count was last drained. Accumulated /// here rather than metered in place because the handle that owns the stats scope /// is the outer [`Subscriber`], which may be reading this cursor through a @@ -3138,9 +3197,25 @@ struct PlainSubscriber { } impl PlainSubscriber { - fn update_drift_cap(&mut self) { - if let Ok(mut cap) = self.drift_cap.write() { - *cap = servable_cap(self.end_sequence, self.stale_cap); + /// The drift anchor for a read bounded by `end`: the outer edge, with `end` folded + /// into the outer cap. + fn anchor(&self, end: Option) -> Anchor { + Anchor { + cap: servable_cap(end, self.stale_cap), + edge: self.drift_anchor.read().edge, + } + } + + fn update_drift_anchor(&mut self, edge: Option) { + let anchor = Anchor { + cap: servable_cap(self.end_sequence, self.stale_cap), + edge, + }; + // Skip a no-op write: every handed-out group's expiry watches this channel. + if *self.drift_anchor.read() != anchor + && let Ok(mut current) = self.drift_anchor.write() + { + *current = anchor; } } @@ -3177,7 +3252,7 @@ impl PlainSubscriber { /// discarding a backlog of N groups costs one scan rather than N. Only ever /// [`Poll::Ready`]; the track ending surfaces as the error the caller was going to /// get anyway. - fn poll_drift(&self, cap: Option, waiter: &kio::Waiter) -> Poll> { + fn poll_drift(&self, anchor: Anchor, waiter: &kio::Waiter) -> Poll> { let mut max_age = Duration::default(); let _ = self.subscription.poll(waiter, |subscription| { max_age = subscription.max_age; @@ -3186,7 +3261,7 @@ impl PlainSubscriber { self.poll(waiter, move |state| { Poll::Ready(Ok(Drift { budget: clamp_max_age(max_age, state.max_age_bound()), - edge: state.live_edge(cap), + edge: state.drift_edge(anchor), })) }) } @@ -3195,7 +3270,7 @@ impl PlainSubscriber { /// for this poll. fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll> { self.poll(waiter, move |state| { - Poll::Ready(Ok(state.is_stale(group.sequence, drift.edge.as_ref(), drift.budget))) + Poll::Ready(Ok(state.is_stale(group.sequence, &drift.edge, drift.budget))) }) } @@ -3204,7 +3279,7 @@ impl PlainSubscriber { group.with_expiry(Arc::new(GroupExpiry { state: self.state.weak(), subscription: self.subscription.consume(), - cap: self.drift_cap.consume(), + anchor: self.drift_anchor.consume(), bound: None, sequence, })) @@ -3232,7 +3307,7 @@ impl PlainSubscriber { .retain(|sequence, group| *sequence >= min_sequence && watch(group)); // One scan for the whole poll, so walking a backlog off stays linear in its size. - let drift = ready!(self.poll_drift(servable_cap(self.end_sequence, self.stale_cap), waiter))?; + let drift = ready!(self.poll_drift(self.anchor(self.end_sequence), waiter))?; loop { // Re-offer the lowest parked group back inside the cap once it rises, @@ -3331,7 +3406,7 @@ impl PlainSubscriber { let mut floor = floor.max(self.min_sequence); let end = super::subscription::min_some(end, self.end_sequence); // One scan for the whole poll, so walking a backlog off stays linear in its size. - let drift = ready!(self.poll_drift(servable_cap(end, self.stale_cap), waiter))?; + let drift = ready!(self.poll_drift(self.anchor(end), waiter))?; loop { let Some(producer) = ready!(self.poll(waiter, |state| state.poll_next_in_range(floor, end))?) else { @@ -3462,21 +3537,22 @@ impl Subscriber { } /// Bound the drift anchor from outside, for a reader that caps this subscriber - /// without capping its cursor. + /// without capping its cursor, and splices it with other tracks. /// /// A [`super::resume::Subscriber`] segment is deliberately left uncapped /// ([`Self::set_groups`] would park boundary-crossing groups where its completion can't /// be seen), so its own cap has to reach the anchor this way or the segment measures - /// drift against groups its reader will never be served. A spliced segment folds the - /// cap into what it pushes onto its own segments, so the bound reaches the plain - /// cursors at the leaves however deep the splices nest. - pub(crate) fn set_stale_cap(&mut self, cap: Option) { + /// drift against groups its reader will never be served. The same goes for the edge: + /// a segment's track never sees the groups of the segments after it. A spliced + /// segment folds the anchor into what it pushes onto its own segments, so it reaches + /// the plain cursors at the leaves however deep the splices nest. + pub(crate) fn set_anchor(&mut self, anchor: Anchor) { match &mut self.inner { SubscriberKind::Plain(plain) => { - plain.stale_cap = cap; - plain.update_drift_cap(); + plain.stale_cap = anchor.cap; + plain.update_drift_anchor(anchor.edge); } - SubscriberKind::Spliced(spliced) => spliced.set_stale_cap(cap), + SubscriberKind::Spliced(spliced) => spliced.set_anchor(anchor), } } @@ -3514,7 +3590,7 @@ impl Subscriber { SubscriberKind::Plain(plain) => plain, SubscriberKind::Spliced(spliced) => return spliced.poll_stale(group, waiter), }; - let drift = ready!(plain.poll_drift(servable_cap(plain.end_sequence, plain.stale_cap), waiter))?; + let drift = ready!(plain.poll_drift(plain.anchor(plain.end_sequence), waiter))?; let stale = ready!(plain.poll_stale(group, &drift, waiter))?; if stale { plain.note_stale(group); @@ -3741,7 +3817,8 @@ impl Subscriber { match &mut self.inner { SubscriberKind::Plain(plain) => { plain.end_sequence = end.exclusive(); - plain.update_drift_cap(); + let edge = plain.drift_anchor.read().edge; + plain.update_drift_anchor(edge); } SubscriberKind::Spliced(spliced) => spliced.end_at(end), } @@ -6018,10 +6095,10 @@ mod test { let state = producer.state.read(); let drift = Drift { budget: Duration::ZERO, - edge: state.live_edge(None), + edge: state.drift_edge(Anchor::default()), }; assert!( - state.is_stale(0, drift.edge.as_ref(), drift.budget), + state.is_stale(0, &drift.edge, drift.budget), "stale against a live edge" ); drop(state); @@ -6032,7 +6109,7 @@ mod test { let state = producer.state.read(); assert!( - !state.is_stale(0, drift.edge.as_ref(), drift.budget), + !state.is_stale(0, &drift.edge, drift.budget), "a vanished edge is no reason to drop what is left" ); } From 0c4c8f79351ed3c89207526cb91e615fa923d163 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 22:48:41 -0700 Subject: [PATCH 2/4] fix(net): bound a segment's last group by the next segment's start A group's reach is where its successor starts, but a segment's last group has its successor in the next segment's track, so it was never judged stale. The splice now pushes each segment the start of the first group the later segments serve past its boundary, matching a single plain track. Tests that read splice mechanics with a zero budget over groups all stamped at zero relied on that gap; they now use a replay budget, and the stale-count tests use a rewound timeline to keep one group fresh. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/origin.rs | 3 +- rs/moq-net/src/model/resume.rs | 220 +++++++++++++++++++-------------- rs/moq-net/src/model/track.rs | 84 ++++++++++--- 3 files changed, 195 insertions(+), 112 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 26186b4424..89f3fe2dbb 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6414,10 +6414,11 @@ mod tests { "every front keeps the delivered groups after releasing its source" ); + // A budget spanning the cache, so the returning reader replays it. let mut subscription = edge_resolved .track("video") .unwrap() - .subscribe(None) + .subscribe(track::Subscription::default().with_max_age(Duration::from_secs(3600))) .await .expect("resubscribe"); tokio::time::timeout(Duration::from_secs(5), track.used()) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index a4365602ec..7b2f104a28 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -26,7 +26,7 @@ use std::collections::{BTreeMap, HashSet}; use std::ops::Bound; use std::task::{Poll, ready}; -use crate::{Datagram, Error, Result, frame, group, track}; +use crate::{Datagram, Error, Result, Timestamp, frame, group, track}; use track::{Anchor, LiveEdge}; use super::subscription::{Cap, Position, Subscription, max_some, min_some}; @@ -142,6 +142,15 @@ fn slice(prefs: &Subscription, start: Option, end: Option) - } } +/// Where the first servable group in `from..cap` starts presenting, from the first of +/// `segments` holding one in its range; see [`track::Consumer::first_start`]. +fn first_start(segments: &[Segment], from: u64, cap: Option) -> Option> { + segments.iter().find_map(|segment| { + let start = segment.start.map_or(0, |start| start.group).max(from); + segment.track.first_start(start, min_some(cap, last_group(segment.end))) + }) +} + /// How many segments a logical track keeps before pruning terminal ones from the /// front: the live segment plus a couple of predecessors still draining to slow /// readers. Without a bound, every failover leaves one dead segment (pinning a @@ -234,6 +243,14 @@ impl ResumeState { .max_by_key(|edge| edge.sequence) } + /// Where the logical track continues past the exclusive group `boundary` of segment + /// `id`, below the reader's `cap`: the start of the first group the later segments + /// serve there. `None` while none is cached, or it has no frame yet. + fn successor(&self, id: u64, boundary: u64, cap: Option) -> Option { + let index = self.segments.iter().position(|segment| segment.id == id)?; + first_start(&self.segments[index + 1..], boundary, cap)? + } + /// Append a segment serving the track from `start` onward, capping (or replacing) /// the previous segments so the ranges stay disjoint and ascending. fn switch(&mut self, track: track::Consumer, start: Option) -> Result<()> { @@ -679,6 +696,12 @@ impl Consumer { self.state.read().live_edge(cap) } + /// Where the first servable group in `from..cap` starts presenting across the + /// segments; see [`track::Consumer::first_start`]. + pub(crate) fn first_start(&self, from: u64, cap: Option) -> Option> { + first_start(&self.state.read().segments, from, cap) + } + /// The newest cached group across every spliced segment; see /// [`track::Consumer::peek_latest`]. pub(crate) fn peek_latest(&self) -> Option { @@ -1337,6 +1360,9 @@ struct SegmentSub { /// A completed segment's cursor, retained while parked groups may need their /// max age budget re-evaluated after the outer cap rises. terminal: Option, + /// The drift anchor for this segment's cursor as of the last + /// [`Subscriber::refresh_anchor`], applied when a pending cursor activates. + anchor: Anchor, /// The producer dropped this segment (pruned, or replaced before producing). /// The cursor drains what it already holds, then retires; see /// [`Self::retired`]. @@ -1629,6 +1655,7 @@ impl Subscriber { ask: segment.ask, sub: SubState::Pending(sub), terminal: None, + anchor: Anchor::default(), pruned: false, parked: BTreeMap::new(), warm: segment.warm.then(|| Warm { @@ -1679,11 +1706,19 @@ impl Subscriber { /// against groups it will never surface: the route running past the boundary, or the /// reader's own cap holding content back. The edge is left as is: it is the newest /// content the logical track holds, which a segment's own track never sees. - fn segment_anchor(seg: &SegmentSub, anchor: Anchor) -> Anchor { - Anchor { - cap: min_some(anchor.cap, seg.last_group()), - ..anchor + /// + /// When the boundary lowers the cap, the reader's next group past it lives in a + /// later segment, so `state` supplies where it starts ([`Anchor::successor`]). + /// Otherwise the logical anchor's own successor (a wrapping splice's) still holds. + fn segment_anchor(seg: &SegmentSub, anchor: Anchor, state: &ResumeState) -> Anchor { + let Some(boundary) = seg.last_group() else { + return anchor; + }; + let mut capped = anchor.capped(Some(boundary)); + if capped.cap != anchor.cap { + capped.successor = state.successor(seg.id, boundary, anchor.cap); } + capped } /// The logical drift anchor, as of the last [`Self::refresh_anchor`]. @@ -1699,15 +1734,14 @@ impl Subscriber { /// newer. Resolved on every sync, since each segment is a separate track and the /// logical edge moves whenever any of them grows. fn refresh_anchor(&mut self) { - let cap = min_some(self.outer.cap, self.end_sequence); - let edge = self - .state - .read() - .live_edge(cap) + let outer = self.outer.capped(self.end_sequence); + let state = self.state.read(); + let edge = state + .live_edge(outer.cap) .into_iter() - .chain(self.outer.edge) + .chain(outer.edge) .max_by_key(|edge| edge.sequence); - let anchor = Anchor { cap, edge }; + let anchor = Anchor { edge, ..outer }; // Skip a no-op write: every handed-out group's expiry watches this channel. if self.anchor() != anchor && let Ok(mut current) = self.drift_anchor.write() @@ -1715,7 +1749,8 @@ impl Subscriber { *current = anchor; } for seg in &mut self.segments { - let anchor = Self::segment_anchor(seg, anchor); + let anchor = Self::segment_anchor(seg, anchor, &state); + seg.anchor = anchor; if let Some(sub) = seg.stale_sub_mut() { sub.set_anchor(anchor); } @@ -1771,13 +1806,7 @@ impl Subscriber { /// Resolve a segment's pending subscription, if any. Ready once the segment is /// `Active` or `Done`; a rejected or closed track becomes `Done` (stall, not /// error). Never consumes groups, so terminal-state pollers can share it. - fn poll_activate( - seg: &mut SegmentSub, - prefs: &Subscription, - min_sequence: u64, - anchor: Anchor, - waiter: &kio::Waiter, - ) -> Poll<()> { + fn poll_activate(seg: &mut SegmentSub, prefs: &Subscription, min_sequence: u64, waiter: &kio::Waiter) -> Poll<()> { if matches!(seg.sub, SubState::Pending(_)) && let Some(warm) = &seg.warm { @@ -1808,7 +1837,8 @@ impl Subscriber { // assigned: the inner subscription resolved its own start from its // budget and floor, and this must not rewind past it. sub.raise_start_to(seg.first_group().max(min_sequence)); - sub.set_anchor(Self::segment_anchor(seg, anchor)); + let anchor = seg.anchor; + sub.set_anchor(anchor); let _ = sub.update(slice(prefs, seg.ask, seg.end)); seg.sub = SubState::Active(Box::new(sub)); } @@ -1825,13 +1855,12 @@ impl Subscriber { seg: &mut SegmentSub, prefs: &Subscription, min_sequence: u64, - anchor: Anchor, waiter: &kio::Waiter, ) -> Poll> { loop { match &mut seg.sub { SubState::Pending(_) => { - ready!(Self::poll_activate(seg, prefs, min_sequence, anchor, waiter)); + ready!(Self::poll_activate(seg, prefs, min_sequence, waiter)); } SubState::Active(sub) => match ready!(sub.poll_recv_group(waiter)) { Ok(Some(group)) => { @@ -1897,7 +1926,6 @@ impl Subscriber { self.commit_seek_stale(committed); } - let anchor = self.anchor(); let mut floor = floor; 'retry: loop { let mut all_done = true; @@ -1905,14 +1933,8 @@ impl Subscriber { for index in 0..self.segments.len() { if matches!(self.segments[index].sub, SubState::Pending(_)) - && Self::poll_activate( - &mut self.segments[index], - &self.last_prefs, - self.min_sequence, - anchor, - waiter, - ) - .is_pending() + && Self::poll_activate(&mut self.segments[index], &self.last_prefs, self.min_sequence, waiter) + .is_pending() { all_done = false; continue; @@ -2004,7 +2026,6 @@ impl Subscriber { self.poll_sync(waiter); let end_sequence = self.end_sequence; - let anchor = self.anchor(); let min_sequence = self.min_sequence; let beyond_cap = |sequence: u64| !super::subscription::before_end(sequence, end_sequence); @@ -2059,13 +2080,7 @@ impl Subscriber { } loop { - let polled = Self::poll_segment( - &mut self.segments[index], - &self.last_prefs, - min_sequence, - anchor, - waiter, - ); + let polled = Self::poll_segment(&mut self.segments[index], &self.last_prefs, min_sequence, waiter); match polled { Poll::Ready(Some(group)) => { if beyond_cap(group.sequence) { @@ -2180,9 +2195,8 @@ impl Subscriber { // datagrams must still resolve the subscription (registering demand) and // be woken when it activates. let mut pending_activation = false; - let anchor = self.anchor(); if let Some(seg) = self.segments.last_mut() { - if Self::poll_activate(seg, &self.last_prefs, self.min_sequence, anchor, waiter).is_pending() { + if Self::poll_activate(seg, &self.last_prefs, self.min_sequence, waiter).is_pending() { pending_activation = true; } else if let SubState::Active(sub) = &mut seg.sub && let Ok(Some(datagram)) = ready!(sub.poll_recv_datagram(waiter)) @@ -2241,17 +2255,10 @@ impl Subscriber { /// completing the segment, would steal them from a `recv_group` caller on the /// same subscriber. fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll> { - let anchor = self.anchor(); let Some(seg) = self.segments.last_mut() else { return Poll::Ready(None); }; - ready!(Self::poll_activate( - seg, - &self.last_prefs, - self.min_sequence, - anchor, - waiter - )); + ready!(Self::poll_activate(seg, &self.last_prefs, self.min_sequence, waiter)); match &mut seg.sub { SubState::Done(count) => Poll::Ready(*count), // Observe only: the cursor may still hold groups, so the read path @@ -2354,7 +2361,7 @@ impl Subscriber { #[cfg(test)] mod test { use super::*; - use crate::{Timestamp, broadcast}; + use crate::broadcast; use futures::FutureExt; use std::sync::Arc; use std::time::Duration; @@ -2724,7 +2731,7 @@ mod test { let mut producer = Producer::new(); producer.switch(&consumer_a, None).unwrap(); - let mut sub = producer.consume().subscribe(None); + let mut sub = producer.consume().subscribe(replay()); write_group(&mut track_a, 0, "a0"); producer.switch(&consumer_b, Position::group(1)).unwrap(); @@ -3180,12 +3187,11 @@ mod test { }) .collect(); - // Group 1 survives where the plain cursor drops it: every segment measures - // against the logical edge (group 3), but a group's reach is bounded only by a - // successor in its own track, and group 1's lives in segment B. The sequence - // path inherits that from the arrival path rather than inventing its own - // anchor, so the two agree. - assert_eq!(spliced, vec![1, 3], "each segment is judged against the logical edge"); + // Segment A's groups are judged as the plain cursor judges them: against the + // logical edge (group 3), with group 1's reach bounded by its successor in + // segment B. The sequence path inherits that from the arrival path rather than + // inventing its own anchor, so the two agree. + assert_eq!(spliced, baseline, "a splice sheds the backlog like one track"); let mut arrival = producer.consume().subscribe(None); let arrival: Vec = std::iter::from_fn(|| { @@ -3211,26 +3217,45 @@ mod test { assert_eq!(replayed, vec![0, 1, 2, 3], "a backlog inside the budget crosses whole"); } - /// A segment's track never sees the groups of the segments after it, so its own - /// edge freezes once a takeover caps it. Both spliced cursors judge its backlog - /// against the logical edge instead, the one a single track would have. + /// A segment's track never sees the groups of the segments after it: its own edge + /// freezes once a takeover caps it, and its last group has no successor there. + /// Both spliced cursors judge its backlog as one plain track holding the same groups + /// would: against the logical edge, with the last group's reach bounded by where the + /// next segment picks up. #[tokio::test] - async fn a_capped_segment_is_judged_against_the_newest_segment() { + async fn a_capped_segment_is_judged_like_one_track() { let budget = Subscription::default().with_max_age(Duration::from_millis(100)); + let old = [(0, 0), (1, 20), (2, 40), (3, 60)]; + let new = [(20, 400), (21, 420), (22, 440), (23, 460), (24, 480), (25, 500)]; + + let (mut plain, _plain_consumer) = track_pair("plain"); + for (sequence, millis) in old.into_iter().chain(new) { + write_group_at(&mut plain, sequence, "p", Duration::from_millis(millis)); + } + let mut baseline = plain.subscribe(budget.clone()); + let baseline: Vec = std::iter::from_fn(|| { + baseline + .recv_group() + .now_or_never()? + .expect("should not error") + .map(|group| group.sequence) + }) + .collect(); + // Group 3 reaches group 20's start at 400ms, 100ms behind the edge at 500ms. + assert_eq!(baseline, vec![20, 21, 22, 23, 24, 25]); + let (mut track_a, consumer_a) = track_pair("a"); let (mut track_b, consumer_b) = track_pair("b"); let mut producer = Producer::new(); producer.switch(&consumer_a, None).unwrap(); producer.switch(&consumer_b, Position::group(4)).unwrap(); - for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] { + for (sequence, millis) in old { write_group_at(&mut track_a, sequence, "a", Duration::from_millis(millis)); } - for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] { + for (sequence, millis) in new { write_group_at(&mut track_b, sequence, "b", Duration::from_millis(millis)); } - // Groups 0..=2 reach at most 60ms against an edge at 440ms. Group 3 stays: its - // successor lives in segment B, so nothing in its own track bounds its reach. let mut arrival = producer.consume().subscribe(budget.clone()); let arrival: Vec = std::iter::from_fn(|| { kio::wait(|waiter| arrival.poll_recv_group(waiter)) @@ -3239,7 +3264,7 @@ mod test { .map(|group| group.sequence) }) .collect(); - assert_eq!(arrival, vec![3, 20, 21, 22]); + assert_eq!(arrival, baseline); let mut ordered = producer.consume().subscribe(budget); let ordered: Vec = std::iter::from_fn(|| { @@ -3249,7 +3274,7 @@ mod test { .map(|group| group.sequence) }) .collect(); - assert_eq!(ordered, arrival); + assert_eq!(ordered, baseline); } /// A nested splice's leaves are judged within the *outer* boundary. The outer @@ -3260,30 +3285,33 @@ mod test { #[tokio::test] async fn nested_splice_judges_within_the_outer_boundary() { let stamp = |sequence: u64| Duration::from_secs(10 * sequence); + let retain = track::Info::default().with_max_age(Duration::from_secs(60)); + let budget = Subscription::default().with_max_age(Duration::from_secs(15)); // Inner splice: one segment carrying groups 0..=2. Group 2 sits past the - // outer boundary below, so it is exactly the anchor the outer window must - // hide from the inner leaves. - let (mut track_a, consumer_a) = track_pair("a"); + // outer boundary below and rewinds to 0s, so it is exactly the successor the + // outer window must hide from the inner leaves: taken as group 1's, it would + // collapse group 1's reach to 0s, 30s behind the edge. + let (mut track_a, consumer_a) = track_pair_with("a", retain.clone()); let mut inner = Producer::new(); inner.switch(&consumer_a, None).unwrap(); write_group_at(&mut track_a, 0, "a0", stamp(0)); write_group_at(&mut track_a, 1, "a1", stamp(1)); - write_group_at(&mut track_a, 2, "a2", stamp(2)); + write_group_at(&mut track_a, 2, "a2", stamp(0)); // Outer splice: the inner spliced track up to group 2, then a plain track. let inner_track = track::Consumer::spliced("inner".into(), Arc::new(broadcast::Info::default()), inner.consume()); - let (mut track_b, consumer_b) = track_pair("b"); + let (mut track_b, consumer_b) = track_pair_with("b", retain); let mut outer = Producer::new(); outer.switch(&inner_track, None).unwrap(); outer.switch(&consumer_b, Position::group(2)).unwrap(); write_group_at(&mut track_b, 2, "b2", stamp(2)); write_group_at(&mut track_b, 3, "b3", stamp(3)); - // Group 1 is the newest group the outer window can serve from the nested - // segment, so it is its own live edge there and survives a zero budget. - let mut sub = outer.consume().subscribe(None); + // Group 0 is 20s behind the edge, past the 15s budget. Group 1 reaches B's + // group 2 at 20s, only 10s behind, so it survives. + let mut sub = outer.consume().subscribe(budget.clone()); let sequences: Vec = std::iter::from_fn(|| { kio::wait(|waiter| sub.poll_next_group(waiter)) .now_or_never()? @@ -3293,11 +3321,11 @@ mod test { .collect(); assert_eq!( sequences, - vec![1, 3], + vec![1, 2, 3], "the nested segment is judged within the outer boundary" ); - let mut arrival = outer.consume().subscribe(None); + let mut arrival = outer.consume().subscribe(budget); let arrival: Vec = std::iter::from_fn(|| { kio::wait(|waiter| arrival.poll_recv_group(waiter)) .now_or_never()? @@ -3314,7 +3342,9 @@ mod test { /// the group after all, and delivered content never counts. #[tokio::test] async fn seek_conviction_counts_only_once_committed() { - let stamp = |sequence: u64| Duration::from_secs(10 * sequence); + // Group 2 starts after group 3 (a rewind), so group 1's reach runs past the live + // edge and it survives a zero budget, while groups 0 and 2 are convicted. + let stamp = |sequence: u64| Duration::from_secs([0, 10, 40, 30][sequence as usize]); let (mut track_a, consumer_a) = track_pair("a"); let (mut track_b, consumer_b) = track_pair("b"); @@ -3359,7 +3389,9 @@ mod test { /// jumped past it, and still be deliverable (uncounted) once the budget widens. #[tokio::test] async fn a_reversible_floor_does_not_commit_a_conviction() { - let stamp = |sequence: u64| Duration::from_secs(10 * sequence); + // Group 2 starts after group 3 (a rewind), so group 1's reach runs past the live + // edge and it survives a zero budget, while groups 0 and 2 are convicted. + let stamp = |sequence: u64| Duration::from_secs([0, 10, 40, 30][sequence as usize]); let (mut track_a, consumer_a) = track_pair("a"); let (mut track_b, consumer_b) = track_pair("b"); @@ -3408,7 +3440,9 @@ mod test { /// that content, so the conviction must be discarded rather than counted. #[tokio::test] async fn a_delivered_continuation_is_not_counted_stale() { - let stamp = |sequence: u64| Duration::from_secs(10 * sequence); + // Group 2 starts after group 3 (a rewind), so group 1's reach runs past the live + // edge and it survives a zero budget, while groups 0 and 2 are convicted. + let stamp = |sequence: u64| Duration::from_secs([0, 10, 40, 30][sequence as usize]); let (mut track_a, consumer_a) = track_pair("a"); let (mut track_b, consumer_b) = track_pair("b"); @@ -3432,9 +3466,9 @@ mod test { .sequence }; - // Group 1 wins from segment A (it holds the head copy) while segment B has - // convicted both its continuation copy of 1 and group 2. Delivering 1 must - // discard B's copy, not count it. + // Group 1 wins from segment A (it holds the head copy) while segment B convicts + // group 2. B's continuation copy of 1 is delivered through the head, so it + // must not be counted. assert_eq!(next(&mut sub), 1); assert_eq!(next(&mut sub), 3); assert_eq!( @@ -3451,12 +3485,16 @@ mod test { /// commit may count; a conviction never committed is dropped with the cursor. #[tokio::test] async fn a_finalized_segment_flushes_nothing_without_a_delivery() { - let stamp = |sequence: u64| Duration::from_secs(10 * sequence); + // Group 1 starts after group 3 (a rewind), so group 0's reach runs past the live + // edge and it survives a zero budget, while groups 1 and 2 are convicted. + let stamp = |sequence: u64| Duration::from_secs([0, 40, 20, 30][sequence as usize]); // C owns group 0, A owns group 1's head, finalized B owns its tail onward. - let (mut track_c, consumer_c) = track_pair("c"); - let (mut track_a, consumer_a) = track_pair("a"); - let (mut track_b, consumer_b) = track_pair("b"); + // Retained long enough that the widened budget below is not clamped. + let retain = track::Info::default().with_max_age(Duration::from_secs(60)); + let (mut track_c, consumer_c) = track_pair_with("c", retain.clone()); + let (mut track_a, consumer_a) = track_pair_with("a", retain.clone()); + let (mut track_b, consumer_b) = track_pair_with("b", retain); let mut producer = Producer::new(); producer.switch(&consumer_c, None).unwrap(); producer.switch(&consumer_a, Position::group(1)).unwrap(); @@ -4348,7 +4386,7 @@ mod test { #[tokio::test] async fn capped_subscriber_bounds_parked_segments() { let mut producer = Producer::new(); - let mut sub = producer.consume().subscribe(None); + let mut sub = producer.consume().subscribe(replay()); sub.end_at(..1); // Every round parks one group beyond the cap, then fails over to a live @@ -5157,8 +5195,8 @@ mod test { let mut producer = Producer::new(); producer.switch(&consumer_a, None).unwrap(); let consumer = producer.consume(); - let mut sub1 = consumer.subscribe(None); - let mut sub2 = consumer.subscribe(None); + let mut sub1 = consumer.subscribe(replay()); + let mut sub2 = consumer.subscribe(replay()); recv_pending(&mut sub1); recv_pending(&mut sub2); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 2e4082577b..9e8d6f0efa 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -505,6 +505,7 @@ impl TrackState { presentation: self.live_edge(anchor.cap), outer: anchor.edge, cap: anchor.cap, + successor: anchor.successor, } } @@ -518,14 +519,27 @@ impl TrackState { /// so a later stamped group proves nothing about where an unstamped successor will /// begin, and shrinking the bound is the unsafe direction. An unstamped successor /// therefore leaves the reach unbounded until it presents its first frame. - fn reach(&self, sequence: u64, cap: Option) -> Option { - let successor = self + /// + /// With no servable successor below `cap`, the reader's next group is past the cap, + /// and `beyond` is where it starts when another track serves it (a splice's next + /// segment; see [`Anchor::successor`]). + fn reach(&self, sequence: u64, cap: Option, beyond: Option) -> Option { + match self.first_start(sequence.saturating_add(1), cap) { + Some(start) => start, + None => beyond, + } + } + + /// Where the first servable group in `from..cap` starts presenting: `None` when no + /// such group is cached, `Some(None)` while it has no frame yet. + fn first_start(&self, from: u64, cap: Option) -> Option> { + let slot = self .lookup - .range(sequence.saturating_add(1)..) + .range(from..) .map(|(_, slot)| slot) .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) .find(|slot| slot.visible && !slot.group.is_aborted())?; - successor.group.timestamp() + Some(slot.group.timestamp()) } /// Whether the group at `sequence` has drifted further behind `edge` than `budget` @@ -579,7 +593,7 @@ impl TrackState { else { return false; }; - self.reach(sequence, edge.cap).is_some_and( + self.reach(sequence, edge.cap, edge.successor).is_some_and( |reach| matches!(live_edge.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget), ) } @@ -2386,6 +2400,16 @@ impl Consumer { } } + /// Where the first servable group in `from..cap` starts presenting: `None` when no + /// such group is cached, `Some(None)` while it has no frame yet. A splice answers from + /// the first segment holding one. + pub(crate) fn first_start(&self, from: u64, cap: Option) -> Option> { + match &self.inner { + ConsumerKind::Plain(state) => state.read().first_start(from, cap), + ConsumerKind::Spliced(resume) => resume.first_start(from, cap), + } + } + /// The live edge below the exclusive `cap` that drift is measured against; see /// [`TrackState::live_edge`]. A splice reports the newest across its segments. pub(crate) fn live_edge(&self, cap: Option) -> Option { @@ -3038,7 +3062,7 @@ impl group::Expiry for GroupExpiry { anchor = **current; Poll::<()>::Pending }); - anchor.cap = super::subscription::min_some(anchor.cap, self.bound); + let anchor = anchor.capped(self.bound); let cap = anchor.cap; let mut expired = false; @@ -3109,6 +3133,8 @@ struct Edge { /// The cap the edge was resolved under, so per-candidate reach lookups measure /// against the same servable window. cap: Option, + /// Bounds the reach of the last group below `cap`; see [`Anchor::successor`]. + successor: Option, } /// How a reader wrapping a cursor bounds its drift anchor from outside: pushed by a @@ -3121,6 +3147,24 @@ pub(crate) struct Anchor { /// that only sees its own groups, so without this a parked segment measures against /// its own frozen edge while the logical track has moved on. pub edge: Option, + /// Where the reader's next group past `cap` starts presenting, when another track + /// serves it (a splice's next segment). The last group below the cap has no + /// successor in its own track, so without this nothing bounds its reach and it is + /// never judged stale. `None` while unknown or unstamped. + pub successor: Option, +} + +impl Anchor { + /// This anchor under a further `cap`. A lower cap drops the successor: it named + /// where the reader continues past the old cap, which is no longer served. + pub fn capped(mut self, cap: Option) -> Self { + let capped = servable_cap(cap, self.cap); + if capped != self.cap { + self.cap = capped; + self.successor = None; + } + self + } } /// The newest stamped group of a track: its sequence and the newest frame it presented. @@ -3197,20 +3241,15 @@ struct PlainSubscriber { } impl PlainSubscriber { - /// The drift anchor for a read bounded by `end`: the outer edge, with `end` folded - /// into the outer cap. + /// The drift anchor for a read bounded by `end`. Every read folds `end_sequence` + /// into `end`, so capping the shared anchor (which already holds it) is exact. fn anchor(&self, end: Option) -> Anchor { - Anchor { - cap: servable_cap(end, self.stale_cap), - edge: self.drift_anchor.read().edge, - } + self.drift_anchor.read().capped(end) } - fn update_drift_anchor(&mut self, edge: Option) { - let anchor = Anchor { - cap: servable_cap(self.end_sequence, self.stale_cap), - edge, - }; + /// Publish the `outer` anchor under this cursor's own cap. + fn update_drift_anchor(&mut self, outer: Anchor) { + let anchor = outer.capped(self.end_sequence); // Skip a no-op write: every handed-out group's expiry watches this channel. if *self.drift_anchor.read() != anchor && let Ok(mut current) = self.drift_anchor.write() @@ -3550,7 +3589,7 @@ impl Subscriber { match &mut self.inner { SubscriberKind::Plain(plain) => { plain.stale_cap = anchor.cap; - plain.update_drift_anchor(anchor.edge); + plain.update_drift_anchor(anchor); } SubscriberKind::Spliced(spliced) => spliced.set_anchor(anchor), } @@ -3817,8 +3856,13 @@ impl Subscriber { match &mut self.inner { SubscriberKind::Plain(plain) => { plain.end_sequence = end.exclusive(); - let edge = plain.drift_anchor.read().edge; - plain.update_drift_anchor(edge); + // A successor dropped by a lower cap stays dropped until the wrapping + // reader pushes its anchor again, which it does on every poll. + let outer = Anchor { + cap: plain.stale_cap, + ..*plain.drift_anchor.read() + }; + plain.update_drift_anchor(outer); } SubscriberKind::Spliced(spliced) => spliced.end_at(end), } From adcd6735d2685005988ed76bbfae034660a34af2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 06:46:09 -0700 Subject: [PATCH 3/4] fix(net): revalidate a pushed edge and successor before they convict A splice caches the logical live edge and the next segment's start on each cursor. Judging against that cache after an eviction still expired older groups, including a one-segment splice's own edge, which skipped the local stamp check. Carry the slot stamp and track, and revalidate both outside the candidate track's lock. Co-Authored-By: Grok 4.7 --- rs/moq-net/src/model/resume.rs | 49 ++--- rs/moq-net/src/model/track.rs | 329 +++++++++++++++++++++++++++------ 2 files changed, 298 insertions(+), 80 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 7b2f104a28..e2227a17f1 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -26,8 +26,10 @@ use std::collections::{BTreeMap, HashSet}; use std::ops::Bound; use std::task::{Poll, ready}; -use crate::{Datagram, Error, Result, Timestamp, frame, group, track}; -use track::{Anchor, LiveEdge}; +#[cfg(test)] +use crate::Timestamp; +use crate::{Datagram, Error, Result, frame, group, track}; +use track::{Anchor, LiveEdge, Successor}; use super::subscription::{Cap, Position, Subscription, max_some, min_some}; @@ -142,12 +144,15 @@ fn slice(prefs: &Subscription, start: Option, end: Option) - } } -/// Where the first servable group in `from..cap` starts presenting, from the first of -/// `segments` holding one in its range; see [`track::Consumer::first_start`]. -fn first_start(segments: &[Segment], from: u64, cap: Option) -> Option> { +/// The first servable group in `from..cap` across `segments`, with the slot identity a +/// later judgment needs. An unstamped group stops the search: skipping it for a later +/// start would shrink a reach that is not yet proven. +fn served_start(segments: &[Segment], from: u64, cap: Option) -> Option { segments.iter().find_map(|segment| { let start = segment.start.map_or(0, |start| start.group).max(from); - segment.track.first_start(start, min_some(cap, last_group(segment.end))) + segment + .track + .served_start(start, min_some(cap, last_group(segment.end))) }) } @@ -246,9 +251,9 @@ impl ResumeState { /// Where the logical track continues past the exclusive group `boundary` of segment /// `id`, below the reader's `cap`: the start of the first group the later segments /// serve there. `None` while none is cached, or it has no frame yet. - fn successor(&self, id: u64, boundary: u64, cap: Option) -> Option { + fn successor(&self, id: u64, boundary: u64, cap: Option) -> Option { let index = self.segments.iter().position(|segment| segment.id == id)?; - first_start(&self.segments[index + 1..], boundary, cap)? + served_start(&self.segments[index + 1..], boundary, cap) } /// Append a segment serving the track from `start` onward, capping (or replacing) @@ -696,10 +701,10 @@ impl Consumer { self.state.read().live_edge(cap) } - /// Where the first servable group in `from..cap` starts presenting across the - /// segments; see [`track::Consumer::first_start`]. - pub(crate) fn first_start(&self, from: u64, cap: Option) -> Option> { - first_start(&self.state.read().segments, from, cap) + /// Where the first servable group in `from..cap` starts, with the identity to + /// revalidate it; see [`track::Consumer::served_start`]. + pub(crate) fn served_start(&self, from: u64, cap: Option) -> Option { + served_start(&self.state.read().segments, from, cap) } /// The newest cached group across every spliced segment; see @@ -1714,16 +1719,17 @@ impl Subscriber { let Some(boundary) = seg.last_group() else { return anchor; }; + let cap = anchor.cap; let mut capped = anchor.capped(Some(boundary)); - if capped.cap != anchor.cap { - capped.successor = state.successor(seg.id, boundary, anchor.cap); + if capped.cap != cap { + capped.successor = state.successor(seg.id, boundary, cap); } capped } /// The logical drift anchor, as of the last [`Self::refresh_anchor`]. fn anchor(&self) -> Anchor { - *self.drift_anchor.read() + self.drift_anchor.read().clone() } /// Re-derive the logical drift anchor and push it onto every segment cursor. @@ -1734,23 +1740,23 @@ impl Subscriber { /// newer. Resolved on every sync, since each segment is a separate track and the /// logical edge moves whenever any of them grows. fn refresh_anchor(&mut self) { - let outer = self.outer.capped(self.end_sequence); + let outer = self.outer.clone().capped(self.end_sequence); let state = self.state.read(); let edge = state .live_edge(outer.cap) .into_iter() - .chain(outer.edge) + .chain(outer.edge.clone()) .max_by_key(|edge| edge.sequence); let anchor = Anchor { edge, ..outer }; // Skip a no-op write: every handed-out group's expiry watches this channel. if self.anchor() != anchor && let Ok(mut current) = self.drift_anchor.write() { - *current = anchor; + *current = anchor.clone(); } for seg in &mut self.segments { - let anchor = Self::segment_anchor(seg, anchor, &state); - seg.anchor = anchor; + let anchor = Self::segment_anchor(seg, anchor.clone(), &state); + seg.anchor = anchor.clone(); if let Some(sub) = seg.stale_sub_mut() { sub.set_anchor(anchor); } @@ -1837,8 +1843,7 @@ impl Subscriber { // assigned: the inner subscription resolved its own start from its // budget and floor, and this must not rewind past it. sub.raise_start_to(seg.first_group().max(min_sequence)); - let anchor = seg.anchor; - sub.set_anchor(anchor); + sub.set_anchor(seg.anchor.clone()); let _ = sub.update(slice(prefs, seg.ask, seg.end)); seg.sub = SubState::Active(Box::new(sub)); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 9e8d6f0efa..ba83c78728 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -497,18 +497,25 @@ impl TrackState { }) } - /// The edge a reader bounded by `anchor` measures drift against: this track's own - /// live edge under the anchor's cap, plus whatever newer edge the anchor carries - /// from outside (a splice's other segments). - fn drift_edge(&self, anchor: Anchor) -> Edge { + /// This track's own edge under the exclusive `cap`, for measuring drift. An outer + /// edge and a successor live on other tracks; the caller revalidates those before + /// taking this lock and passes them in, so the locks never nest. + fn drift_edge(&self, cap: Option, outer: Option<(u64, Timestamp)>, successor: Option) -> Edge { Edge { - presentation: self.live_edge(anchor.cap), - outer: anchor.edge, - cap: anchor.cap, - successor: anchor.successor, + presentation: self.live_edge(cap), + outer, + cap, + successor, } } + /// Whether `sequence` still holds the servable incarnation `stamp`. + fn holds(&self, sequence: u64, stamp: u32) -> bool { + self.lookup + .get(&sequence) + .is_some_and(|slot| slot.stamp == stamp && !slot.group.is_aborted()) + } + /// The furthest presentation time the group at `sequence` could still reach: where /// its immediate successor begins, or `None` when nothing proves where it stops. /// @@ -542,6 +549,24 @@ impl TrackState { Some(slot.group.timestamp()) } + /// The first servable group's start in `from..cap`, with the slot identity a later + /// judgment needs to tell that group from whatever replaces it. `None` when no such + /// group is cached or it has no frame yet: an unstamped successor leaves reach + /// unbounded, and this does not skip past it to a later group. + fn served_start(&self, from: u64, cap: Option) -> Option { + let slot = self + .lookup + .range(from..) + .map(|(_, slot)| slot) + .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) + .find(|slot| slot.visible && !slot.group.is_aborted())?; + Some(ServedStart { + sequence: slot.group.sequence, + stamp: slot.stamp, + timestamp: slot.group.timestamp()?, + }) + } + /// Whether the group at `sequence` has drifted further behind `edge` than `budget` /// tolerates, so a subscriber should skip it rather than hand it over. /// @@ -566,7 +591,8 @@ impl TrackState { /// The edge must sit strictly above the candidate. The live edge is never late /// against itself, and backfill or the tail of a rewound timeline can carry a high /// timestamp on a low sequence without being an edge at all. Of the edges that - /// qualify, the highest sequence is the newest content, whichever track holds it. + /// qualify, the highest sequence is the newest content, whichever track holds it: + /// this one's, or the `outer` edge a splice pushed from another segment. fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { if !self.lookup.contains_key(&sequence) { return false; @@ -577,25 +603,21 @@ impl TrackState { // (delivering) is right, since the next poll resolves fresh anchors. let local = edge .presentation - .filter(|live| { - self.lookup - .get(&live.sequence) - .is_some_and(|slot| slot.stamp == live.stamp && !slot.group.is_aborted()) - }) - .map(LiveEdge::from); - // An outer edge lives on another track, so there is no slot here to revalidate - // it against; it is resolved fresh on every poll of the splice that pushes it. - let Some(live_edge) = local + .filter(|live| self.holds(live.sequence, live.stamp)) + .map(|live| (live.sequence, live.timestamp)); + // `outer` and `successor` were revalidated on their own tracks before this lock + // was taken ([`LiveEdge::is_live`], [`Successor::start`]). There is no slot for + // them here, and taking their locks here would nest. + let Some((_, timestamp)) = local .into_iter() .chain(edge.outer) - .filter(|live| live.sequence > sequence) - .max_by_key(|live| live.sequence) + .filter(|(live, _)| *live > sequence) + .max_by_key(|(live, _)| *live) else { return false; }; - self.reach(sequence, edge.cap, edge.successor).is_some_and( - |reach| matches!(live_edge.timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget), - ) + self.reach(sequence, edge.cap, edge.successor) + .is_some_and(|reach| matches!(timestamp.checked_sub(reach), Ok(age) if Duration::from(age) >= budget)) } /// Resolve a one-shot fetch from the track side: the cached group, or an [`Error`] @@ -2400,25 +2422,40 @@ impl Consumer { } } - /// Where the first servable group in `from..cap` starts presenting: `None` when no - /// such group is cached, `Some(None)` while it has no frame yet. A splice answers from - /// the first segment holding one. - pub(crate) fn first_start(&self, from: u64, cap: Option) -> Option> { - match &self.inner { - ConsumerKind::Plain(state) => state.read().first_start(from, cap), - ConsumerKind::Spliced(resume) => resume.first_start(from, cap), - } - } - /// The live edge below the exclusive `cap` that drift is measured against; see /// [`TrackState::live_edge`]. A splice reports the newest across its segments. pub(crate) fn live_edge(&self, cap: Option) -> Option { match &self.inner { - ConsumerKind::Plain(state) => state.read().live_edge(cap).map(LiveEdge::from), + ConsumerKind::Plain(state) => { + let edge = state.read().live_edge(cap)?; + Some(LiveEdge { + sequence: edge.sequence, + timestamp: edge.timestamp, + stamp: edge.stamp, + track: state.weak(), + }) + } ConsumerKind::Spliced(resume) => resume.live_edge(cap), } } + /// Where the first servable group in `from..cap` starts, with enough identity to + /// revalidate it later. A splice answers from the first segment holding one. + pub(crate) fn served_start(&self, from: u64, cap: Option) -> Option { + match &self.inner { + ConsumerKind::Plain(state) => { + let served = state.read().served_start(from, cap)?; + Some(Successor { + sequence: served.sequence, + timestamp: served.timestamp, + stamp: served.stamp, + track: state.weak(), + }) + } + ConsumerKind::Spliced(resume) => resume.served_start(from, cap), + } + } + /// The nearest cached group below `sequence`, under the same terms as /// [`Self::peek_group`]. Walks the cache's own order, so gaps in the group numbering /// are crossed and aborted (evicted) entries are skipped. @@ -3028,11 +3065,14 @@ enum SubscriberKind { /// One poll's view of how far this subscription may drift: the clamped budget and the /// live edge to measure a candidate group against. Resolved once, then applied to every -/// group that poll considers. +/// group that poll considers. `outer` and `successor` are revalidated per candidate, +/// outside this track's lock. #[derive(Clone)] struct Drift { budget: Duration, edge: Edge, + outer: Option, + successor: Option, } /// Keeps one handed-out group tied to the subscription whose cursor selected it. @@ -3059,17 +3099,23 @@ impl group::Expiry for GroupExpiry { let mut anchor = Anchor::default(); let _ = self.anchor.poll(waiter, |current| { - anchor = **current; + anchor = (**current).clone(); Poll::<()>::Pending }); let anchor = anchor.capped(self.bound); let cap = anchor.cap; + // Before this track's lock: both may name another track, and nesting deadlocks. + let outer = anchor + .edge + .filter(LiveEdge::is_live) + .map(|live| (live.sequence, live.timestamp)); + let successor = anchor.successor.as_ref().and_then(Successor::start); let mut expired = false; let _ = self.state.poll(waiter, |state| { let budget = clamp_max_age(max_age, state.max_age_bound()); loop { - let edge = state.drift_edge(anchor); + let edge = state.drift_edge(cap, outer, successor); expired = state.is_stale(self.sequence, &edge, budget); if expired { break; @@ -3124,34 +3170,37 @@ impl group::Expiry for GroupExpiry { /// The group a poll's drift is measured against, identified well enough to tell it apart /// from whatever may occupy its sequence by the time a candidate is judged. -#[derive(Clone)] +#[derive(Clone, Copy)] struct Edge { /// This track's own edge, revalidated before it convicts anything. presentation: Option, - /// A newer edge on another track of the same splice; see [`Anchor::edge`]. - outer: Option, + /// A newer edge on another track, already revalidated: its sequence and the newest + /// frame it had presented. See [`Anchor::edge`]. + outer: Option<(u64, Timestamp)>, /// The cap the edge was resolved under, so per-candidate reach lookups measure /// against the same servable window. cap: Option, - /// Bounds the reach of the last group below `cap`; see [`Anchor::successor`]. + /// Where the next group past `cap` starts, already revalidated. See [`Anchor::successor`]. successor: Option, } /// How a reader wrapping a cursor bounds its drift anchor from outside: pushed by a /// splice onto each segment's cursor, and shared with the groups a cursor hands out. -#[derive(Clone, Copy, Default, PartialEq, Eq)] +#[derive(Clone, Default, PartialEq)] pub(crate) struct Anchor { /// The exclusive sequence cap on what the reader could be handed; see [`servable_cap`]. pub cap: Option, /// The newest edge across a splice's segments. Each segment is a separate track /// that only sees its own groups, so without this a parked segment measures against - /// its own frozen edge while the logical track has moved on. + /// its own frozen edge while the logical track has moved on. Revalidated on its own + /// track before it convicts anything, since it may be judged long after it was pushed. pub edge: Option, /// Where the reader's next group past `cap` starts presenting, when another track /// serves it (a splice's next segment). The last group below the cap has no /// successor in its own track, so without this nothing bounds its reach and it is - /// never judged stale. `None` while unknown or unstamped. - pub successor: Option, + /// never judged stale. `None` while unknown or unstamped. Revalidated like `edge`: + /// a cached start must not convict once that group is gone. + pub successor: Option, } impl Anchor { @@ -3167,19 +3216,74 @@ impl Anchor { } } -/// The newest stamped group of a track: its sequence and the newest frame it presented. -#[derive(Clone, Copy, PartialEq, Eq)] +/// The newest stamped group of a track: its sequence and the newest frame it presented, +/// plus enough identity for a reader on another track to revalidate it. +#[derive(Clone)] pub(crate) struct LiveEdge { pub sequence: u64, pub timestamp: Timestamp, + stamp: u32, + track: kio::ConsumerWeak, } -impl From for LiveEdge { - fn from(edge: PresentationEdge) -> Self { - Self { - sequence: edge.sequence, - timestamp: edge.timestamp, +impl LiveEdge { + /// Whether the edge still names the same servable group on its own track, the + /// check [`TrackState::is_stale`] runs on a local edge. An eviction or abort since + /// the splice resolved it must not convict anything. Takes that track's lock, so + /// never call it under another's. + fn is_live(&self) -> bool { + self.track.read().holds(self.sequence, self.stamp) + } +} + +impl PartialEq for LiveEdge { + fn eq(&self, other: &Self) -> bool { + self.sequence == other.sequence + && self.timestamp == other.timestamp + && self.stamp == other.stamp + && self.track.same_channel(&other.track) + } +} + +/// The first servable group past a segment boundary: where it starts, and which slot +/// that start was read from. +struct ServedStart { + sequence: u64, + stamp: u32, + timestamp: Timestamp, +} + +/// A successor pushed onto another track's cursor. The timestamp alone is not enough: +/// once the group is evicted, a later group can keep the outer edge valid while this +/// start is no longer where the track continues. +#[derive(Clone)] +pub(crate) struct Successor { + sequence: u64, + timestamp: Timestamp, + stamp: u32, + track: kio::ConsumerWeak, +} + +impl Successor { + /// The start this still names, re-read from its own track, or `None` once that + /// group is gone or no longer stamped. Takes that track's lock, so never call it + /// under another's. + fn start(&self) -> Option { + let state = self.track.read(); + let slot = state.lookup.get(&self.sequence)?; + if slot.stamp != self.stamp || slot.group.is_aborted() { + return None; } + slot.group.timestamp() + } +} + +impl PartialEq for Successor { + fn eq(&self, other: &Self) -> bool { + self.sequence == other.sequence + && self.timestamp == other.timestamp + && self.stamp == other.stamp + && self.track.same_channel(&other.track) } } @@ -3244,7 +3348,7 @@ impl PlainSubscriber { /// The drift anchor for a read bounded by `end`. Every read folds `end_sequence` /// into `end`, so capping the shared anchor (which already holds it) is exact. fn anchor(&self, end: Option) -> Anchor { - self.drift_anchor.read().capped(end) + self.drift_anchor.read().clone().capped(end) } /// Publish the `outer` anchor under this cursor's own cap. @@ -3297,10 +3401,17 @@ impl PlainSubscriber { max_age = subscription.max_age; Poll::<()>::Pending }); - self.poll(waiter, move |state| { + let cap = anchor.cap; + let outer = anchor.edge; + let successor = anchor.successor; + self.poll(waiter, |state| { + // Local edge only. The pushed edge and successor are revalidated in + // [`Self::poll_stale`], outside this lock. Poll::Ready(Ok(Drift { budget: clamp_max_age(max_age, state.max_age_bound()), - edge: state.drift_edge(anchor), + edge: state.drift_edge(cap, None, None), + outer: outer.clone(), + successor: successor.clone(), })) }) } @@ -3308,8 +3419,25 @@ impl PlainSubscriber { /// Whether the drift budget says to skip `group`, against a [`Drift`] already resolved /// for this poll. fn poll_stale(&self, group: &group::Consumer, drift: &Drift, waiter: &kio::Waiter) -> Poll> { + // Revalidate before this track's lock. Both can name another track, including + // one whose own judgment is waiting on this one. + let outer = drift + .outer + .as_ref() + .filter(|live| live.is_live()) + .map(|live| (live.sequence, live.timestamp)); + let successor = drift.successor.as_ref().and_then(Successor::start); + let presentation = drift.edge.presentation; + let cap = drift.edge.cap; + let budget = drift.budget; self.poll(waiter, move |state| { - Poll::Ready(Ok(state.is_stale(group.sequence, &drift.edge, drift.budget))) + let edge = Edge { + presentation, + outer, + cap, + successor, + }; + Poll::Ready(Ok(state.is_stale(group.sequence, &edge, budget))) }) } @@ -3860,7 +3988,7 @@ impl Subscriber { // reader pushes its anchor again, which it does on every poll. let outer = Anchor { cap: plain.stale_cap, - ..*plain.drift_anchor.read() + ..plain.drift_anchor.read().clone() }; plain.update_drift_anchor(outer); } @@ -6139,7 +6267,9 @@ mod test { let state = producer.state.read(); let drift = Drift { budget: Duration::ZERO, - edge: state.drift_edge(Anchor::default()), + edge: state.drift_edge(None, None, None), + outer: None, + successor: None, }; assert!( state.is_stale(0, &drift.edge, drift.budget), @@ -6158,6 +6288,89 @@ mod test { ); } + /// The same holds for an edge a splice pushed from another segment: a handed-out + /// group judges it long after the splice resolved it, so it is revalidated on its own + /// track before it convicts anything. + #[tokio::test] + async fn an_evicted_outer_edge_convicts_nothing() { + let mut producer = track_producer("a", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1))); + let mut open = producer.append_group().unwrap(); + open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a")) + .unwrap(); + let mut group = subscriber.recv_group().await.unwrap().expect("group"); + assert!(group.read_frame().await.unwrap().is_some()); + // A successor bounds the open group's reach, without being late against it. + append_at(&mut producer, 10); + + let mut next = track_producer("b", None); + append_at(&mut next, 30_000); + let edge = append_at(&mut next, 30_010); + subscriber.set_anchor(Anchor { + cap: None, + edge: next.consume().live_edge(None), + successor: None, + }); + + let mut control = group.clone(); + assert!( + matches!(control.read_frame().now_or_never(), Some(Ok(None))), + "the open group ends against the outer edge" + ); + + // The outer edge dies before the group is judged again. + let slot = next.modify().unwrap().lookup.remove(&edge).unwrap(); + let _ = slot.group.abort(Error::Evicted); + + assert!( + group.read_frame().now_or_never().is_none(), + "a vanished outer edge is no reason to drop what is left" + ); + assert!(!group.latency_expired()); + open.finish().unwrap(); + } + + /// A pushed successor is the same kind of cached fact. Evicting it must not keep + /// bounding the previous segment's last group while a later group still anchors the + /// outer edge. + #[tokio::test] + async fn an_evicted_successor_convicts_nothing() { + let producer = track_producer("a", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_secs(1))); + let mut open = producer.append_group().unwrap(); + open.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"a")) + .unwrap(); + let mut group = subscriber.recv_group().await.unwrap().expect("group"); + assert!(group.read_frame().await.unwrap().is_some()); + + let mut next = track_producer("b", None); + // Early enough that group 0 is already past the budget, and not itself the edge. + let successor = append_at(&mut next, 10); + append_at(&mut next, 30_000); + let consumer = next.consume(); + subscriber.set_anchor(Anchor { + cap: None, + edge: consumer.live_edge(None), + successor: consumer.served_start(successor, None), + }); + + let mut control = group.clone(); + assert!( + matches!(control.read_frame().now_or_never(), Some(Ok(None))), + "the open group ends against the successor" + ); + + let slot = next.modify().unwrap().lookup.remove(&successor).unwrap(); + let _ = slot.group.abort(Error::Evicted); + + assert!( + group.read_frame().now_or_never().is_none(), + "a vanished successor is no reason to drop what is left" + ); + assert!(!group.latency_expired()); + open.finish().unwrap(); + } + #[tokio::test] async fn a_lower_sequence_is_never_the_live_edge() { let producer = track_producer("test", None); From 6598de83916408f8e701542ea76d69adf70cb304 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 08:09:58 -0700 Subject: [PATCH 4/4] test(net): judge the warm-cache repro on a served front A local source is released when its reader leaves, so the old repro dropped the cache before staleness ran. A served front still parks it. The copy resolves at the cached edge, and the budget keeps group 3 while shedding the groups behind it. Co-Authored-By: Grok 4.7 --- rs/moq-net/src/model/origin.rs | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 89f3fe2dbb..52c06cc5f6 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -6295,14 +6295,19 @@ mod tests { /// edge, not the parked segment's own frozen one: groups the fresh source /// has left behind by more than the budget are skipped, exactly as they would /// be on one unspliced track. + /// + /// A local source is released rather than parked (it keeps its own cache), so + /// this goes through a served front, which is what actually holds the warm copy. + /// The copy resolves at the cached edge, not past it: resolving past it drops + /// the cache outright, which is a different case. #[tokio::test] async fn resumed_reader_skips_warm_groups_behind_the_new_edge() { + let ms = |v: u64| crate::Timestamp::from_millis(v).unwrap(); + let (_server, _upstream, mut dynamic, resolved) = served_front().await; let budget = track::Subscription::default().with_max_age(Duration::from_millis(100)); let write = |source: &track::Producer, sequence: u64, millis: u64| { let mut group = source.create_group(sequence.into()).unwrap(); - group - .write_frame(crate::Timestamp::from_millis(millis).unwrap(), b"x".as_ref()) - .unwrap(); + group.write_frame(ms(millis), b"x".as_ref()).unwrap(); group.finish().unwrap(); }; let drain = |subscription: &mut track::Subscriber| { @@ -6313,12 +6318,6 @@ mod tests { sequences }; - let producer = origin(1).produce(); - let consumer = producer.consume(); - let broadcast = producer.publish("room/alice", Route::default()).unwrap(); - let mut dynamic = broadcast.dynamic(); - let resolved = consumer.request_broadcast("room/alice").await.expect("resolves"); - let track = resolved.track("video").unwrap(); let first = budget.clone(); let subscribing = tokio::spawn(async move { track.subscribe(first).await }); @@ -6326,7 +6325,7 @@ mod tests { .await .expect("the front asked the source") .expect("request"); - let source = request.accept(None); + let source = request.resolving_start().accept(None); for (sequence, millis) in [(0, 0), (1, 20), (2, 40), (3, 60)] { write(&source, sequence, millis); } @@ -6335,7 +6334,6 @@ mod tests { drain(&mut subscription); drop(subscription); - // Parked: groups 0..=3 stay warm while the source goes idle. tokio::time::timeout(Duration::from_secs(1), source.unused()) .await .expect("parked") @@ -6349,14 +6347,18 @@ mod tests { .await .expect("the front asked the source again") .expect("request"); - let source = request.accept(None); + let mut source = request.resolving_start().accept(None); for (sequence, millis) in [(20, 400), (21, 420), (22, 440)] { write(&source, sequence, millis); } + // At the cached edge, not past it, so the warm copy stays and the budget + // decides. Past it, the copy has already judged the cache stale. + source.start_at(3).unwrap(); let mut subscription = subscribing.await.unwrap().expect("resubscribe"); settle(|| subscription.latest() == Some(22)).await; - // Groups 0..=2 reach at most 60ms against an edge at 440ms. + // Groups 0..=2 reach at most 60ms against an edge at 440ms. Group 3 reaches + // where group 20 starts, 40ms behind that edge, inside the 100ms budget. assert_eq!(drain(&mut subscription), [3, 20, 21, 22]); }