fix(moq-net): an IETF copy goes idle before its cancel

quest/m1/test-flakes-2/README.md

@@ -32,4 +32,3 @@ Public API: none. Wire: none.
 
 - [Subscription cut by disconnect](/quest/m1/test-flakes-2/subscription-cut.md) - a publisher disconnect never ends a subscription clean
 - [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - `just test media` late join stays within one GOP, or the regression is fixed
-- [Rejoin after idle](/quest/m1/test-flakes-2/rejoin-idle-race.md) - an IETF rejoin never gets the stale warm cache first, because the copy goes idle before the cancel

rs/moq-net/src/ietf/subscriber.rs

@@ -2221,20 +2221,22 @@ where
 		// Clean up
 		let aborted = self.remove_subscribe(request_id).is_none();
 
-		match idle {
-			true => self.cancel_subscribe(stream, request_id).await,
+		if !idle {
 			// The publisher already ended the request, so a FIN is all we owe it.
-			false => {
-				stream.writer.finish().ok();
-				return None;
-			}
+			stream.writer.finish().ok();
+			return None;
+		}
+		// What the copy cached goes stale from here. Marked before the cancel, which waits
+		// on the publisher: it stops serving as soon as the cancel lands, and a reader
+		// returning in between must not take the cache as the live edge.
+		if !aborted {
+			track.set_idle();
 		}
+		self.cancel_subscribe(stream, request_id).await;
 		// A session abort took the copy too.
 		if aborted {
 			return None;
 		}
-		// What the copy cached goes stale from here.
-		track.set_idle();
 		Some(Idle {
 			demand: track.demand(),
 			track,

rs/moq-net/tests/rejoin.rs

@@ -325,3 +325,58 @@ async fn rejoin_goes_live_without_the_join_head() {
 		.unwrap_or_else(|_| panic!("{version}: timed out"));
 	}
 }
+
+/// A reader rejoining while the relay is still cancelling upstream is not handed the stale cache.
+///
+/// Cancelling waits on the publisher, a round trip on a slow link, but the publisher stops
+/// serving as soon as the cancel lands. A reader returning in between must already find the
+/// copy idle, or it takes the cached group as the live edge.
+#[moq_net_sim::test]
+async fn rejoin_during_the_cancel_skips_the_cache() {
+	for version in Version::names() {
+		moq_net_sim::timeout(TEST_TIMEOUT, async {
+			let publisher = produce_origin(1);
+			let relay = produce_origin(2);
+
+			let broadcast = publisher.create_broadcast("bench").unwrap();
+			let track = broadcast.create_track("video", None).unwrap();
+			broadcast.announce(Default::default()).unwrap();
+
+			let mut options = MockConnectOptions::new(version.parse::<Version>().unwrap());
+			options.server_publish = Some(publisher.consume());
+			options.client_subscribe = Some(relay.clone());
+			options.latency = Duration::from_millis(50);
+			let _pair = connect_mock(options).await;
+
+			let consumer = relay.consume();
+			consumer.routed("bench").await.unwrap();
+			let remote = consumer.request_broadcast("bench").await.unwrap();
+			let ts = |ms| Timestamp::from_millis(ms).unwrap();
+
+			let mut group = track.append_group().unwrap();
+			group.write_frame(ts(0), b"old".as_ref()).unwrap();
+			group.finish().unwrap();
+			let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap();
+			let mut group = sub.recv_group().await.unwrap().unwrap();
+			assert_eq!(read_all(&mut group).await.unwrap(), [b"old".to_vec()], "{version}");
+			drop((group, sub));
+			track.demand().unused().await.unwrap();
+
+			// The publisher moves on while the relay's cancel is still in flight.
+			for sequence in 1..=3u64 {
+				let mut group = track.append_group().unwrap();
+				group.write_frame(ts(sequence * 1000), b"new".as_ref()).unwrap();
+				group.finish().unwrap();
+			}
+
+			let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap();
+			let group = sub.recv_group().await.unwrap().unwrap();
+			assert_eq!(
+				group.sequence, 3,
+				"{version}: the rejoining reader got the stale cache first"
+			);
+		})
+		.await
+		.unwrap_or_else(|_| panic!("{version}: timed out"));
+	}
+}

rs/moq-net/tests/support/mock.rs

@@ -106,7 +106,17 @@ struct ClosedSignal {
 }
 
 impl ClosedSignal {
-	fn set(&self, result: Result<(), MockError>) {
+	/// Signal the sender once `delay` has passed: a STOP_SENDING crosses the link like data.
+	fn set(self: &Arc<Self>, delay: Duration, result: Result<(), MockError>) {
+		if !delay.is_zero() {
+			// Only a test sets a latency, so the stop lands on the simulated clock.
+			let signal = self.clone();
+			drop(moq_net_sim::spawn(async move {
+				moq_net_sim::sleep(delay).await;
+				signal.set(Duration::ZERO, result);
+			}));
+			return;
+		}
 		let mut slot = self.result.lock();
 		if slot.is_none() {
 			*slot = Some(result);
@@ -166,7 +176,7 @@ impl poll::SendStream for MockSendStream {
 			// trusts this signal ahead of the connection error.
 			let pushed = self.push(StreamChunk::Fin);
 			if pushed.is_ok() && (self.ack_fin || self.conn.ack_fins.load(Ordering::Relaxed)) {
-				self.closed.set(Ok(()));
+				self.closed.set(Duration::ZERO, Ok(()));
 			}
 			self.tx = None;
 			pushed?;
@@ -297,7 +307,7 @@ impl poll::RecvStream for MockRecvStream {
 	}
 
 	fn stop(&mut self, _code: u32) {
-		self.closed.set(Ok(()));
+		self.closed.set(self.conn.latency(), Ok(()));
 		self.done = true;
 	}
 
@@ -329,8 +339,9 @@ impl poll::RecvStream for MockRecvStream {
 impl Drop for MockRecvStream {
 	fn drop(&mut self) {
 		// Signal the paired SendStream that the receiver is gone (implicit STOP),
-		// and fail its future writes.
-		self.closed.set(Ok(()));
+		// and fail its future writes. Unlike QUIC, writes fail at once, a link
+		// latency before `poll_closed` reports the stop.
+		self.closed.set(self.conn.latency(), Ok(()));
 		self.rx.close();
 	}
 }
@@ -378,7 +389,7 @@ struct ConnectionState {
 	/// Set once by whichever side closes first.
 	/// Setting it wakes both sides.
 	close_state: kio::Shared<Option<(u32, String)>>,
-	/// One-way delay for stream data in each direction. Zero by default.
+	/// One-way delay for stream data and STOP_SENDING in each direction. Zero by default.
 	latency: Mutex<Duration>,
 }
 
@@ -632,8 +643,8 @@ impl MockSession {
 		*self.side.lossy.lock().unwrap() = true;
 	}
 
-	/// Delay stream data sent from now on by `latency` in each direction, keeping
-	/// each stream in order. Measured on the simulated clock, so tests advance
+	/// Delay stream data and STOP_SENDING sent from now on by `latency` in each direction,
+	/// keeping each stream in order. Measured on the simulated clock, so tests advance
 	/// through it without sleeping.
 	pub fn set_latency(&self, latency: Duration) {
 		*self.side.conn.latency.lock().unwrap() = latency;