diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 16de2758ed..8f2e0e8a4f 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/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 diff --git a/quest/m1/test-flakes-2/rejoin-idle-race.md b/quest/m1/test-flakes-2/rejoin-idle-race.md deleted file mode 100644 index 4437939535..0000000000 --- a/quest/m1/test-flakes-2/rejoin-idle-race.md +++ /dev/null @@ -1,32 +0,0 @@ -# [S] Rejoin after the copy goes idle - -## Goal - -A reader rejoining a parked IETF track never gets the stale warm cache -first, and moq-tokio `broadcast_rejoin_skips_a_stale_warm_cache` -(`rs/moq-tokio/tests/broadcast.rs`) passes in every full-suite run. - -## Plan - -It failed once under a loaded `just check` ("moq-transport-17: a rejoining -reader was served the stale cache first: group 3") and passed 8/8 alone. - -Likely cause: on idle, the IETF subscriber awaits `cancel_subscribe` before -it calls `track.set_idle()` (`rs/moq-net/src/ietf/subscriber.rs`, around -line 2178 at time of writing). Draft-17 has no UNSUBSCRIBE, so the cancel is -a STOP_SENDING followed by a `close().await` that waits for the peer. -The publisher sees the cancel and its demand drops, which is what the test -waits on, about a round trip before the copy is marked idle. A resubscribe -in that gap finds a copy that still looks live. Its newest group, 3, has no -successor, so it isn't stale, and it is served. - -Decided 2026-10-05 11:39 +0200: fix the product, since a real client -rejoining in that gap gets the stale group too. Reproduce the race first, -then mark the copy idle as soon as `End::Idle` is decided, before the -awaited cancel. The test then waits on an event that comes after idle and -needs no change. Rejected: only making the test wait for the subscriber's -idle, which leaves the window open for clients. Check the other drafts and -lite for the same ordering. Add a deterministic regression test if one is -easy. - -Public API: none. Wire: none. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e972304807..7ae0971013 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/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, diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index ad95ffbeda..38ed299ace 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/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::().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")); + } +} diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 5c05682583..ddcf384d2b 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/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, 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>, - /// 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, } @@ -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;