fix(net): a warm copy's expiry no longer aborts the relay's latest group

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

@@ -672,16 +672,24 @@ 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 {
 	fn new(stamp: u64) -> Self {
 		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<Access>);
+
+impl Protection {
+	pub(crate) fn new(access: &Arc<Access>) -> 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.
 	///

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) {

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<cache::Protection>,
 }
 
 /// Heap the track keeps per cached group, excluding the group itself
@@ -753,14 +757,17 @@ 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 {
 				continue;
 			};
 			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() {

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"));
+	}
+}