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 b9b9765d44..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,7 +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. + /// 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 { @@ -761,6 +767,7 @@ 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)) @@ -861,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)); } } @@ -945,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); @@ -964,6 +973,7 @@ impl TrackState { group: group.clone(), stamp, visible, + protection: self.protects(sequence).then(|| group.cache_protect()), }, ); if visible { @@ -7277,6 +7287,57 @@ mod test { ); } + /// 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_track_expires_a_shared_group_unless_live_edge() { + let pool = cache::Pool::new(cache::Config::default().with_expiry(Duration::from_secs(1))); + // 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(); + } + }; + 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.peek_group(0).is_some(), + "and so does the ended track's share of it" + ); + + 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!(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. #[tokio::test] async fn abort_after_the_end_settles_ends_clean() { 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")); + } +}