From 70adf4f3dd7afff23acb709cd5381190487d0b43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:48:37 -0700 Subject: [PATCH 1/5] quest: claim rejoin-idle-race Co-Authored-By: Claude Opus 5.5 From 895ce16e165dcd73e05b8b91b8a3fc6ab08ae8f5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:55:40 -0700 Subject: [PATCH 2/5] fix(moq-net): an IETF copy goes idle before its cancel The subscriber marked the copy idle only after cancel_subscribe returned, which waits on the publisher. The publisher stops serving once the cancel lands, so a reader rejoining in that gap took the stale cache as the live edge. Mark the copy idle first. The mock transport now delivers STOP_SENDING (and a receiver's drop) after the link latency, like stream data, so a test can open that gap. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 2 - quest/m1/test-flakes-2/rejoin-idle-race.md | 32 ----------- rs/moq-net/src/ietf/subscriber.rs | 18 ++++--- rs/moq-net/tests/rejoin.rs | 62 ++++++++++++++++++++++ rs/moq-net/tests/support/mock.rs | 32 +++++++---- 5 files changed, 94 insertions(+), 52 deletions(-) delete mode 100644 quest/m1/test-flakes-2/rejoin-idle-race.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 6e338309e9..cc60fd4673 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -38,8 +38,6 @@ Public API: none. Wire: none. - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time -- [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 - ## Related - [Archive enrollment](/quest/m1/archive/enrollment-flake.md) - the same 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 44b6e5187f..95ab44ca4d 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 afea8c5d6e..234c8cf5f1 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -325,3 +325,65 @@ 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. +#[tokio::test(start_paused = true)] +async fn rejoin_during_the_cancel_skips_the_cache() { + let pre06 = [ + "moq-lite-01", + "moq-lite-02", + "moq-lite-03", + "moq-lite-04", + "moq-lite-05", + ]; + for version in Version::names().filter(|version| !pre06.contains(version)) { + tokio::time::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 fff2efdf43..ed3b3a72b9 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -102,16 +102,17 @@ fn is_set(slot: &kio::Ref<'_, Option>) -> Poll<()> { /// wakes; the send side's `poll_closed` reads this without consuming state. #[derive(Default)] struct ClosedSignal { - /// Set once the peer signals stop or drops. Setting it wakes pending - /// `poll_closed` watches. - result: kio::Shared>>, + /// Set once the peer signals stop or drops, with when the sender learns of it. + /// Setting it wakes pending `poll_closed` watches. + result: kio::Shared)>>, } 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, delay: Duration, result: Result<(), MockError>) { let mut slot = self.result.lock(); if slot.is_none() { - *slot = Some(result); + *slot = Some((tokio::time::Instant::now() + delay, result)); } } } @@ -123,6 +124,8 @@ type Flight = (tokio::time::Instant, StreamChunk); pub struct MockSendStream { tx: Option>, closed: Arc, + /// Wakes `poll_closed` when the peer's stop lands. + landing: Option>>, park: kio::Park, /// Acknowledge the FIN as soon as it is sent, for a stream the peer's transport holds /// back from its application (see [`MockSession::hold_unis`]). @@ -168,7 +171,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?; @@ -187,9 +190,17 @@ impl poll::SendStream for MockSendStream { fn poll_closed(&mut self, cx: &mut Context<'_>) -> Poll> { let waiter = self.park.hold(cx); if !self.conn.hold_fins.load(Ordering::Relaxed) - && let Poll::Ready(result) = self.closed.result.poll(waiter, is_set) + && let Poll::Ready(slot) = self.closed.result.poll(waiter, is_set) { - return Poll::Ready(result.clone().expect("set")); + let (arrival, result) = slot.clone().expect("set"); + drop(slot); + let landing = self + .landing + .get_or_insert_with(|| Box::pin(tokio::time::sleep_until(arrival))); + landing.as_mut().reset(arrival); + if landing.as_mut().poll(cx).is_ready() { + return Poll::Ready(result); + } } drop(std::task::ready!(self.conn.close_state.poll(waiter, is_set))); Poll::Ready(Err(self.conn.error().expect("closed"))) @@ -293,7 +304,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; } @@ -326,7 +337,7 @@ 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(())); + self.closed.set(self.conn.latency(), Ok(())); self.rx.close(); } } @@ -341,6 +352,7 @@ fn new_stream_pair(conn: &Arc) -> (MockSendStream, MockRecvStre let send = MockSendStream { tx: Some(queue.clone()), closed: closed.clone(), + landing: None, park: kio::Park::default(), ack_fin: false, withhold_fin: Arc::default(), From ad94c415129d8fcb13960b8d9aea1e6f9dcb3e58 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 01:02:15 -0700 Subject: [PATCH 3/5] test(moq-net): tidy the mock's delayed STOP_SENDING Drop a redundant timer reset (a stop's arrival is set once), document that the link latency now covers STOP_SENDING too, and restore the blank line before the questline's Related heading. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 + rs/moq-net/tests/support/mock.rs | 7 +++---- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index cc60fd4673..1032e30f61 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -38,6 +38,7 @@ Public API: none. Wire: none. - [Scoped WARN capture](/quest/m1/test-flakes-2/warn-capture.md) - the drop-unfinished tests count only their own WARNs - [moq-mux debounce clock](/quest/m1/test-flakes-2/mux-debounce-clock.md) - the TS export debounce test advances on the paused clock - [js/publish audio clock](/quest/m1/test-flakes-2/publish-audio-clock.md) - the audio encoder delay test runs on mock time + ## Related - [Archive enrollment](/quest/m1/archive/enrollment-flake.md) - the same diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index ed3b3a72b9..f3f75c9b7f 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -197,7 +197,6 @@ impl poll::SendStream for MockSendStream { let landing = self .landing .get_or_insert_with(|| Box::pin(tokio::time::sleep_until(arrival))); - landing.as_mut().reset(arrival); if landing.as_mut().poll(cx).is_ready() { return Poll::Ready(result); } @@ -386,7 +385,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, } @@ -640,8 +639,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 tokio's clock, so paused-time tests advance + /// Delay stream data and STOP_SENDING sent from now on by `latency` in each direction, + /// keeping each stream in order. Measured on tokio's clock, so paused-time tests advance /// through it without sleeping. pub fn set_latency(&self, latency: Duration) { *self.side.conn.latency.lock().unwrap() = latency; From 62c30556032f43f57f65224a982165b63f8e8e00 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 02:12:18 -0700 Subject: [PATCH 4/5] test(moq-net): run the cancel-race rejoin over every version It passes on every lite and IETF version, so the hard-coded pre-lite-06 exclusion only hid future versions. Also note that the mock fails writes a link latency before it reports a dropped receiver's stop. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/rejoin.rs | 9 +-------- rs/moq-net/tests/support/mock.rs | 3 ++- 2 files changed, 3 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index 234c8cf5f1..d870b3393b 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -333,14 +333,7 @@ async fn rejoin_goes_live_without_the_join_head() { /// copy idle, or it takes the cached group as the live edge. #[tokio::test(start_paused = true)] async fn rejoin_during_the_cancel_skips_the_cache() { - let pre06 = [ - "moq-lite-01", - "moq-lite-02", - "moq-lite-03", - "moq-lite-04", - "moq-lite-05", - ]; - for version in Version::names().filter(|version| !pre06.contains(version)) { + for version in Version::names() { tokio::time::timeout(TEST_TIMEOUT, async { let publisher = produce_origin(1); let relay = produce_origin(2); diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index f3f75c9b7f..402ea69356 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -335,7 +335,8 @@ 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. + // 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(); } From 62279cf5c007a75c0800392b6fa8f5b6ffd35f4f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 02:53:49 -0700 Subject: [PATCH 5/5] chore: retrigger CI Co-Authored-By: Claude Opus 5.5