fix(net): a subscriber hands its cursors off to a park's cache

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

@@ -6080,7 +6080,8 @@ mod tests {
 	/// next route only for the frames past the break. The front keeps the head it
 	/// spliced, so a reader arriving afterwards still gets the whole group: whether the
 	/// old route dies before the new one takes over (a reconnect) or after (a cheaper
-	/// route preempting it).
+	/// route preempting it). A reader joining mid-outage gets it too, and the reader
+	/// already mid-group is not handed it a second time.
 	#[tokio::test]
 	async fn a_takeover_keeps_the_open_group_head_for_later_readers() {
 		tokio::time::pause();
@@ -6124,8 +6125,18 @@ mod tests {
 		// Either the first route's session ends and reconnects, announcing the same route
 		// afresh while the old one is still held, or a cheaper route preempts the live one.
 		let mut first = Some((group, source, upstream, dynamic));
+		let mut joining = None;
 		if dies_first {
 			drop(first.take());
+			// The reader sees its route die and stalls. Polled in place: yielding would let
+			// the front give up on a path with no route left before the replacement lands.
+			assert!(futures::FutureExt::now_or_never(subscription.recv_group()).is_none());
+			let track = resolved.track("log").unwrap();
+			joining = Some(tokio::spawn(async move {
+				let mut joined = track.subscribe(None).await.unwrap();
+				let group = joined.recv_group().await.unwrap().expect("track ended");
+				(joined, group)
+			}));
 		}
 		let cost = if dies_first { 5 } else { 0 };
 		let replacement_server = producer
@@ -6171,13 +6182,37 @@ mod tests {
 		for expect in [b"a", b"b", b"c"] {
 			assert_eq!(read(&mut late).await, expect, "dies_first={dies_first}");
 		}
+		let mut joined = None;
+		if let Some(joining) = joining {
+			let (subscription, mut group) = tokio::time::timeout(Duration::from_secs(1), joining)
+				.await
+				.expect("the reader joining mid-outage never got the group")
+				.unwrap();
+			assert_eq!(group.sequence, 0);
+			for expect in [b"a", b"b", b"c"] {
+				assert_eq!(read(&mut group).await, expect);
+			}
+			joined = Some((subscription, group));
+		}
+
+		// The reader that was mid-group already has it.
+		assert!(
+			tokio::time::timeout(Duration::from_secs(1), subscription.recv_group())
+				.await
+				.is_err(),
+			"dies_first={dies_first}: the group was handed out twice"
+		);
 
 		// The continuation still flows to both.
 		next_group.write_frame(crate::Timestamp::ZERO, b"d".as_ref()).unwrap();
 		assert_eq!(read(&mut reading).await, b"d");
 		assert_eq!(read(&mut late).await, b"d");
+		if let Some((_, group)) = &mut joined {
+			assert_eq!(read(group).await, b"d");
+		}
 		next_group.finish().unwrap();
 		drop((
+			joined,
 			reading,
 			late,
 			subscription,

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

@@ -1532,6 +1532,9 @@ struct SegmentSub {
 	id: u64,
 	start: Option<Position>,
 	end: Option<Position>,
+	/// Groups below this were handed out before the cursor subscribed: a warm cache
+	/// taking over a reader's cursors must not surface them again.
+	floor: u64,
 	/// Where the source is asked to start; see [`Segment::ask`].
 	ask: Option<Position>,
 	sub: SubState,
@@ -1583,7 +1586,7 @@ impl SegmentSub {
 
 	/// The first group this segment can serve, for the underlying read cursor.
 	fn first_group(&self) -> u64 {
-		self.start.map_or(0, |start| start.group)
+		self.start.map_or(0, |start| start.group).max(self.floor)
 	}
 
 	/// The exclusive group cap this segment can serve, for the underlying read
@@ -1794,6 +1797,17 @@ impl Subscriber {
 		for s in &mut self.segments {
 			s.pruned = !segments.iter().any(|n| n.id == s.id);
 		}
+		// A park replaces every segment with a cache of what they delivered (see
+		// [`Producer::park`]). Cursors still draining the replaced segments would hand
+		// that content out a second time, so the cache takes over from them, past what
+		// this reader was already handed. Groups they parked at the cap are in the
+		// cache too, and park again from there.
+		let parked = segments
+			.iter()
+			.any(|n| n.warm && !self.segments.iter().any(|s| s.id == n.id));
+		if parked {
+			self.segments.retain(|s| !s.pruned);
+		}
 		self.segments.retain(|s| !s.retired());
 
 		let nexts: Vec<_> = segments.iter().skip(1).map(|next| Some(next.track.clone())).collect();
@@ -1829,6 +1843,8 @@ impl Subscriber {
 						id: segment.id,
 						start: segment.start,
 						end: segment.end,
+						// A cache taking over (see above) starts past what this reader has.
+						floor: if segment.warm { self.next_sequence } else { 0 },
 						ask: segment.ask,
 						sub: SubState::Pending(sub),
 						terminal: None,
@@ -3250,6 +3266,85 @@ mod test {
 		assert_eq!(recv(&mut sub), 1);
 	}
 
+	/// A park that replaces a reader's segments with a cache of what they delivered
+	/// hands that reader nothing twice, however many times one open group (a catalog
+	/// taking deltas) changes route. A newcomer still gets the whole group, head first,
+	/// after the takeovers have pruned every segment that started at its head.
+	#[tokio::test]
+	async fn repeated_parks_mid_group_hand_the_head_out_once() {
+		let mut producer = Producer::new();
+		let (track, consumer) = track_pair("a");
+		producer.takeover(&consumer).unwrap();
+		let mut sub = producer.consume().subscribe(None);
+		let mut group = track.create_group(group::Info { sequence: 0 }).unwrap();
+		group.write_frame(Timestamp::ZERO, b"0".to_vec()).unwrap();
+		let mut reading = sub.recv_group().now_or_never().unwrap().unwrap().unwrap();
+		assert_eq!(read(&mut reading), b"0");
+
+		let mut routes = vec![(track, consumer, group)];
+		let mut heads = Vec::new();
+		let rounds = MAX_SEGMENTS as u64 + 1;
+		for frame in 1..=rounds {
+			// The cache holds the whole group so far, the way the origin rebuilds it.
+			let (cache, cache_consumer) = track_pair("cache");
+			let mut whole = cache.create_group(group::Info { sequence: 0 }).unwrap();
+			for f in 0..frame {
+				whole.write_frame(Timestamp::ZERO, f.to_string().into_bytes()).unwrap();
+			}
+			producer.park(&cache_consumer).unwrap();
+
+			let (next, next_consumer) = track_pair("next");
+			producer.takeover(&next_consumer).unwrap();
+			let mut continuation = next.create_group(group::Info { sequence: 0 }).unwrap();
+			continuation.start_at(frame).unwrap();
+			continuation
+				.write_frame(Timestamp::ZERO, frame.to_string().into_bytes())
+				.unwrap();
+
+			assert_eq!(read(&mut reading), frame.to_string().as_bytes());
+			recv_pending(&mut sub);
+			routes.push((next, next_consumer, continuation));
+			// The origin finishes a cache once the takeover lands, but holds its open
+			// group: dropping that would clear its frames.
+			cache.finish().unwrap();
+			heads.push(whole);
+		}
+
+		let mut late = producer.consume().subscribe(None);
+		let mut group = late.recv_group().now_or_never().unwrap().unwrap().unwrap();
+		for frame in 0..=rounds {
+			assert_eq!(read(&mut group), frame.to_string().as_bytes());
+		}
+	}
+
+	/// A group held back at the reader's cap when a park replaces its segments comes
+	/// from the cache once the cap rises, exactly once.
+	#[tokio::test]
+	async fn a_park_reoffers_a_group_held_at_the_cap_once() {
+		let (mut track_a, consumer_a) = track_pair("a");
+		let mut producer = Producer::new();
+		producer.takeover(&consumer_a).unwrap();
+		let mut sub = producer.consume().subscribe(replay());
+		sub.end_at(..1);
+		write_group(&mut track_a, 0, "0");
+		write_group(&mut track_a, 1, "1");
+		assert_eq!(recv(&mut sub), 0);
+		recv_pending(&mut sub);
+
+		let (mut cache, cache_consumer) = track_pair("cache");
+		write_group(&mut cache, 0, "0");
+		write_group(&mut cache, 1, "1");
+		producer.park(&cache_consumer).unwrap();
+		let (mut track_b, consumer_b) = track_pair("b");
+		producer.takeover(&consumer_b).unwrap();
+		write_group(&mut track_b, 2, "2");
+
+		sub.end_at(..3);
+		assert_eq!(recv(&mut sub), 1);
+		assert_eq!(recv(&mut sub), 2);
+		recv_pending(&mut sub);
+	}
+
 	/// A parked beyond-cap group must not block in-range groups that arrive
 	/// behind it: a relay can ingest a burst micro-reordered (newest first).
 	#[tokio::test]

rs/moq-relay/tests/goaway_cluster.rs

@@ -118,6 +118,105 @@ async fn drain_session_with_zero_timeout_closes_at_once_inner() {
 	let _ = within("client observes the close", client_connection.closed()).await;
 }
 
+#[test]
+fn cluster_continues_a_group_split_by_goaway() {
+	run_cluster_test(cluster_continues_a_group_split_by_goaway_inner());
+}
+
+/// A GOAWAY that lands mid-group hands the rest of the group to the reader already
+/// holding it, and the next group after it, never the split group a second time.
+async fn cluster_continues_a_group_split_by_goaway_inner() {
+	let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+
+	let upstream_origin = moq_tokio::origin::spawn();
+	let broadcast = upstream_origin.create_broadcast("cam").expect("create broadcast");
+	broadcast.announce(Default::default()).expect("announce");
+	let track = broadcast.create_track("video", None).expect("create track");
+
+	let (port_a, mut accepted_a, _handle_a) = spawn_upstream(upstream_origin.clone()).await;
+	let (port_b, mut accepted_b, _handle_b) = spawn_upstream(upstream_origin.clone()).await;
+
+	let mut client_config = moq_tokio::connect::Config::default();
+	client_config.tls.insecure = Some(true);
+	client_config.goaway.handover = Duration::from_secs(2);
+	let client = client_config.init(Default::default()).expect("client init");
+
+	let mut cluster_config = cluster::Config::default();
+	cluster_config.connect = vec![Peer::new(format!("tcp://127.0.0.1:{port_a}/"))];
+	let cluster = cluster::Cluster::new(cluster::Options::new(cluster_config))
+		.expect("cluster init")
+		.with_client(client);
+	let started = cluster.clone().start().await.expect("cluster start");
+	let cluster_run = tokio::spawn(started.run());
+
+	let session_a = within("A accepts", accepted_a.recv()).await.expect("A accepts");
+	let consumer = cluster.origin.consume();
+	within("routed via A", consumer.routed("cam")).await.expect("routed");
+	let bc = consumer.request_broadcast("cam").await.expect("broadcast resolves");
+	let mut sub = within("subscribe", bc.track("video").expect("track").subscribe(None))
+		.await
+		.expect("subscribe");
+
+	let write = |group: &mut moq_net::group::Producer, payload: &'static str| {
+		group
+			.write_frame(moq_net::Timestamp::ZERO, payload.as_bytes())
+			.expect("write frame");
+	};
+	// A group's sequence and its next frame, or `None` at its end.
+	async fn read(group: &mut moq_net::group::Consumer) -> (u64, Option<bytes::Bytes>) {
+		let frame = within("read frame", group.read_frame()).await.expect("read");
+		(group.sequence, frame.map(|frame| frame.payload))
+	}
+
+	// The first group is open, its first frame read, when A redirects to B.
+	let mut split = track.append_group().expect("append group");
+	write(&mut split, "a");
+	let mut reading = within("recv the split group", sub.recv_group())
+		.await
+		.expect("recv")
+		.expect("track ended");
+	assert_eq!(read(&mut reading).await, (0, Some("a".into())));
+
+	// The cursor replays A's route first. After the GOAWAY, A's re-prices to DRAIN, so
+	// the next route at any other cost is B's.
+	let mut announced = consumer.announced();
+	within("A's route", announced.next()).await.expect("announce update");
+	session_a
+		.drain()
+		.send(moq_net::goaway::Goaway::redirect(format!("tcp://127.0.0.1:{port_b}/")))
+		.expect("send goaway");
+	let _session_b = within("B accepts", accepted_b.recv()).await.expect("B accepts");
+	within("routed via B", async {
+		loop {
+			let update = announced.next().await.expect("announce update");
+			if update.kind.is_active() && update.route.cost != moq_net::origin::Cost::DRAIN {
+				break;
+			}
+		}
+	})
+	.await;
+
+	write(&mut split, "b");
+	split.finish().expect("finish");
+	assert_eq!(read(&mut reading).await, (0, Some("b".into())));
+	assert_eq!(read(&mut reading).await, (0, None));
+
+	let mut after = track.append_group().expect("append group");
+	write(&mut after, "c");
+	after.finish().expect("finish");
+	let mut next = within("recv the next group", sub.recv_group())
+		.await
+		.expect("recv")
+		.expect("track ended");
+	assert_eq!(
+		read(&mut next).await,
+		(1, Some("c".into())),
+		"the split group came back"
+	);
+
+	cluster_run.abort();
+}
+
 #[test]
 fn cluster_reconnects_on_empty_uri_goaway() {
 	run_cluster_test(cluster_reconnects_on_empty_uri_goaway_inner());