From 6c2b733bbb6fd37d529c5539c626bf88b312e49a Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 10:12:22 +0200 Subject: [PATCH 1/3] fix(net): a warm copy's expiry no longer aborts the relay's latest group A front that parks a track adopts the relay copy's groups into a warm copy, sharing their producers. Once the warm copy ends, expire_closed aborted every idle group, including the latest group the relay's live copy still serves, so new subscribers skipped it and waited for a group that a steady catalog never writes. Abort only groups the ended track alone owns. Ports #4923's regression test to release's rejoin.rs on tokio's paused clock. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 6 ++- rs/moq-net/tests/rejoin.rs | 82 +++++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b9b9765d44..641edba19c 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -753,7 +753,9 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. + /// and leave the slots, which every read path already skips. A group still shared with a + /// live track (a warm copy adopts the relay copy's groups) is left to that track, whose + /// own eviction keeps its latest group. pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -766,7 +768,7 @@ impl TrackState { .cache_accessed_tick(scan.gc.then_some(scan.now)) .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) { - let _ = slot.group.clone().abort(Error::Old); + let _ = slot.group.abort_if_last(Error::Old); } } } diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index a7391e4620..a0bd6702e5 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -108,3 +108,85 @@ async fn rejoin_recovers_the_group_reset_on_leave() { } } } + +/// A reader leaving long after its last read leaves the relay holding the latest group. +/// +/// A catalog's latest group can sit unread past the cache's idle expiry. A downstream +/// session that read it and then leaves must not take the relay's copy of that group with +/// it: a reader that joins afterwards needs it, and no newer group may ever come. +#[tokio::test(start_paused = true)] +async fn leaving_after_the_cache_window_keeps_the_latest_group() { + for version in [ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-19", + "moq-transport-22", + ] { + tokio::time::timeout(TEST_TIMEOUT + moq_net::cache::DEFAULT_EXPIRY * 3, async { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + + let broadcast = publisher.create_broadcast("bench").unwrap(); + let track = broadcast.create_track("catalog", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"c0".as_ref()).unwrap(); + group.finish().unwrap(); + + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let _upstream = connect_mock(options).await; + + // Each session gets its own hop, so the relay serves it through its own front. + let session = |hop| { + let relay = relay.clone(); + async move { + let client = produce_origin(hop); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(client.clone()); + let pair = connect_mock(options).await; + let consumer = client.consume(); + consumer.routed("bench").await.unwrap(); + let remote = consumer.request_broadcast("bench").await.unwrap(); + (remote, pair) + } + }; + + // Holds the relay's upstream subscription, and its copy, open throughout. + let (holder, _holder) = session(3).await; + let mut held = holder.track("catalog").unwrap().subscribe(None).await.unwrap(); + assert_eq!(held.recv_group().await.unwrap().unwrap().sequence, 0, "{version}"); + + let (leaver, leaver_session) = session(4).await; + let mut sub = leaver.track("catalog").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(read_all(&mut group).await.unwrap(), [b"c0"], "{version}"); + tokio::time::sleep(moq_net::cache::DEFAULT_EXPIRY * 2).await; + drop((group, sub, leaver, leaver_session)); + + // Right after the leave, and again once the leaver's front has lingered and let go + // of the track (`IDLE_LINGER` is as long as the cache window). + for (hop, after) in [ + (5, Duration::from_millis(100)), + (6, moq_net::cache::DEFAULT_EXPIRY + Duration::from_secs(1)), + ] { + tokio::time::sleep(after).await; + let (later, _later) = session(hop).await; + let mut sub = later.track("catalog").unwrap().subscribe(None).await.unwrap(); + let group = tokio::time::timeout(Duration::from_secs(1), sub.recv_group()).await; + let mut group = group + .unwrap_or_else(|_| panic!("{version}: reader {hop} never got the latest group")) + .unwrap() + .unwrap(); + assert_eq!(group.sequence, 0, "{version}"); + assert_eq!(read_all(&mut group).await.unwrap(), [b"c0"], "{version}"); + } + }) + .await + .unwrap_or_else(|_| panic!("{version}: timed out")); + } +} From 2564923af42194bf57fe845c1b2a5f3f3fd18420 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 11:43:14 +0200 Subject: [PATCH 2/3] test(net): pin when ended tracks sharing a group release it Two ended tracks that share a group each see the other's handle, so neither aborts it; it expires once one of them lets go. Say so where expire_closed skips a shared group. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 44 ++++++++++++++++++++++++++++++++--- 1 file changed, 41 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 641edba19c..e129507b9e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -753,9 +753,10 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. A group still shared with a - /// live track (a warm copy adopts the relay copy's groups) is left to that track, whose - /// own eviction keeps its latest group. + /// and leave the slots, which every read path already skips. A group another handle + /// still owns (a warm copy shares the relay copy's) is left to that owner, which keeps + /// it while it is a live track's latest; a later sweep retries it once this track holds + /// the last handle. pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -7279,6 +7280,43 @@ mod test { ); } + /// A group two ended tracks share expires once one of them lets go. Until then each + /// sees the other's handle, so neither aborts it for both. + #[tokio::test] + async fn closed_tracks_sharing_a_group_expire_it_once_one_lets_go() { + let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); + + let first = track_producer_pooled("first", pool.clone()); + let group = first.append_group().unwrap(); + group.finish().unwrap(); + let mut second = track_producer_pooled("second", pool.clone()); + second.adopt_group(group.clone(), true).unwrap(); + drop(group); + + first.finish().unwrap(); + second.finish().unwrap(); + let stale_first = first.consume(); + let stale_second = second.consume(); + drop((first, second)); + + // The first pass dates the activity it has not seen yet; the next one expires it. + let sweep = || { + for _ in 0..2 { + crate::model::clock::advance(Duration::from_secs(2)); + pool.sweep(); + } + }; + sweep(); + assert!( + stale_second.peek_group(0).is_some(), + "a shared group outlives the sweep" + ); + + drop(stale_first); + sweep(); + assert!(stale_second.peek_group(0).is_none(), "the last owner expires it"); + } + /// An abort after every group below the declared end finished leaves the end standing. #[tokio::test] async fn abort_after_the_end_settles_ends_clean() { From f9b7982e018f40a225149579dd4ca61ecf9cef13 Mon Sep 17 00:00:00 2001 From: OneTooMany Date: Fri, 9 Oct 2026 12:02:59 +0200 Subject: [PATCH 3/3] fix(net): an ended track expires a shared group unless it is a live edge abort_if_last kept an ended warm copy from aborting the relay's live edge, but it also kept every group two ended tracks share: each saw the other's handle, so stale consumers on both pinned it. Count on the group the live tracks holding it as their latest, through a guard in the latest slot that goes on demotion, removal or close_cache. expire_closed aborts an idle group unless that count is above zero, which is the pool's rule that only a live track's latest group is exempt, applied across the tracks that share it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/cache.rs | 27 ++++++++++++ rs/moq-net/src/model/group.rs | 10 +++++ rs/moq-net/src/model/track.rs | 77 ++++++++++++++++++++++------------- 3 files changed, 86 insertions(+), 28 deletions(-) diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index b4f85cea3c..dd86719ca3 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -672,6 +672,8 @@ pub(crate) struct Charge { pub(crate) struct Access { stamp: AtomicU64, expires: AtomicU64, + // Live tracks holding the group as their latest; see [`Protection`]. + protected: AtomicUsize, } impl Access { @@ -679,9 +681,15 @@ impl Access { Self { stamp: AtomicU64::new(stamp), expires: AtomicU64::new(u64::MAX), + protected: AtomicUsize::new(0), } } + /// Whether a live track holds the group as its latest. + pub(crate) fn is_protected(&self) -> bool { + self.protected.load(Ordering::Relaxed) > 0 + } + /// The stamp, tie-breaking bits included. pub(crate) fn get(&self) -> u64 { self.stamp.load(Ordering::Relaxed) @@ -715,6 +723,25 @@ impl Access { } } +/// A live track holding a group as its latest, for as long as this lives. +/// +/// Tracks can share a group (a warm copy adopts the relay copy's), so the group itself +/// counts these: an ended track sharing it must not expire another track's live edge. +pub(crate) struct Protection(Arc); + +impl Protection { + pub(crate) fn new(access: &Arc) -> Self { + access.protected.fetch_add(1, Ordering::Relaxed); + Self(access.clone()) + } +} + +impl Drop for Protection { + fn drop(&mut self) { + self.0.protected.fetch_sub(1, Ordering::Relaxed); + } +} + impl Charge { /// Charge `n` more payload bytes, counting them as written. /// diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index db1e74f5a7..2f16d11182 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -911,6 +911,16 @@ impl Producer { self.alive.access.tick(now) } + /// Hold the group as a live track's latest until the guard drops. + pub(crate) fn cache_protect(&self) -> cache::Protection { + cache::Protection::new(&self.alive.access) + } + + /// Whether a live track holds the group as its latest. + pub(crate) fn cache_protected(&self) -> bool { + self.alive.access.is_protected() + } + /// Enter the group into the evictable population: demoted from the live edge, /// or inserted behind it. Idempotent; a no-op once the group is closed. pub(crate) fn cache_demote(&self) { diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e129507b9e..4b258985b9 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -284,6 +284,10 @@ struct Slot { // Whether this incarnation came from the live publisher and can replace older // subscription content. Fetch-only backfill stays cached but never anchors drift. visible: bool, + + // Held while this is the live edge (see `TrackState::protects`), so an ended track + // sharing the group leaves it alone. + protection: Option, } /// Heap the track keeps per cached group, excluding the group itself @@ -753,10 +757,9 @@ impl TrackState { /// Expire an ended track's idle groups, whose closed channel refuses the write /// [`Self::evict_expired_scan`] takes: abort them in place, releasing their frames, - /// and leave the slots, which every read path already skips. A group another handle - /// still owns (a warm copy shares the relay copy's) is left to that owner, which keeps - /// it while it is a live track's latest; a later sweep retries it once this track holds - /// the last handle. + /// and leave the slots, which every read path already skips. A group another track + /// still holds as its live edge stays, since tracks can share a group (a warm copy + /// adopts the relay copy's). pub(super) fn expire_closed(&self, scan: ExpiryScan) { for (sequence, stamp) in &self.evict { let Some(slot) = self.lookup.get(sequence) else { @@ -764,12 +767,13 @@ impl TrackState { }; if slot.stamp == *stamp && !slot.group.is_aborted() + && !slot.group.cache_protected() && slot .group .cache_accessed_tick(scan.gc.then_some(scan.now)) .is_some_and(|tick| scan.now.saturating_sub(tick) > scan.max_ticks) { - let _ = slot.group.abort_if_last(Error::Old); + let _ = slot.group.clone().abort(Error::Old); } } } @@ -864,9 +868,10 @@ impl TrackState { return; } if let Some(latest) = self.latest_group - && let Some(slot) = self.lookup.get(&latest) + && let Some(slot) = self.lookup.get_mut(&latest) { slot.group.cache_demote(); + slot.protection = None; self.evict.push_back((latest, slot.stamp)); } } @@ -948,9 +953,10 @@ impl TrackState { // pool's access average) like any other cached group. if let Some(latest) = self.latest_group && sequence > latest - && let Some(prev) = self.lookup.get(&latest) + && let Some(prev) = self.lookup.get_mut(&latest) { prev.group.cache_demote(); + prev.protection = None; self.evict.push_back((latest, prev.stamp)); } self.latest_group = Some(sequence); @@ -967,6 +973,7 @@ impl TrackState { group: group.clone(), stamp, visible, + protection: self.protects(sequence).then(|| group.cache_protect()), }, ); if visible { @@ -7280,25 +7287,12 @@ mod test { ); } - /// A group two ended tracks share expires once one of them lets go. Until then each - /// sees the other's handle, so neither aborts it for both. + /// An ended track that shares a group with a live one leaves it while it is the live + /// track's latest, and expires it like any other idle group once it is not. A group + /// only ended tracks share expires even while stale consumers keep them all. #[tokio::test] - async fn closed_tracks_sharing_a_group_expire_it_once_one_lets_go() { + async fn closed_track_expires_a_shared_group_unless_live_edge() { let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); - - let first = track_producer_pooled("first", pool.clone()); - let group = first.append_group().unwrap(); - group.finish().unwrap(); - let mut second = track_producer_pooled("second", pool.clone()); - second.adopt_group(group.clone(), true).unwrap(); - drop(group); - - first.finish().unwrap(); - second.finish().unwrap(); - let stale_first = first.consume(); - let stale_second = second.consume(); - drop((first, second)); - // The first pass dates the activity it has not seen yet; the next one expires it. let sweep = || { for _ in 0..2 { @@ -7306,15 +7300,42 @@ mod test { pool.sweep(); } }; + let adopt = |name: &str, group: &group::Producer| { + let mut track = track_producer_pooled(name, pool.clone()); + track.adopt_group(group.clone(), true).unwrap(); + track.finish().unwrap(); + let stale = track.consume(); + drop(track); + stale + }; + + let live = track_producer_pooled("live", pool.clone()); + let group = live.append_group().unwrap(); + group.finish().unwrap(); + let stale = adopt("ended", &group); + drop(group); + sweep(); + assert!(live.consume().peek_group(0).is_some(), "the live edge stays"); assert!( - stale_second.peek_group(0).is_some(), - "a shared group outlives the sweep" + stale.peek_group(0).is_some(), + "and so does the ended track's share of it" ); - drop(stale_first); + live.append_group().unwrap().finish().unwrap(); + sweep(); + assert!(stale.peek_group(0).is_none(), "once demoted, it expires"); + + let group = live.append_group().unwrap(); + group.finish().unwrap(); + let (first, second) = (adopt("first", &group), adopt("second", &group)); + drop(group); + live.finish().unwrap(); + drop(live); + sweep(); - assert!(stale_second.peek_group(0).is_none(), "the last owner expires it"); + assert!(first.peek_group(2).is_none(), "ended tracks alone don't keep it"); + assert!(second.peek_group(2).is_none(), "for each other"); } /// An abort after every group below the declared end finished leaves the end standing.