From 70adf4f3dd7afff23acb709cd5381190487d0b43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:48:37 -0700 Subject: [PATCH 01/15] quest: claim rejoin-idle-race Co-Authored-By: Claude Opus 5.5 From e07bff74ee796c107a0a11ff561fe8c4064eb2d2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 23:49:28 -0700 Subject: [PATCH 02/15] chore: claim media late-join quest 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 03/15] 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 a7ce0a541af17c1492f93ab35ce180380769c5cd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:39:06 -0700 Subject: [PATCH 04/15] quest(test-flakes-2): trace the media late join to a relay's idle copy The late joiner is handed a GOP the relay cached before a demand gap, because the copy goes live on the route's first group header while that group has no frame yet. Record the reproduction and the open decision. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/media-late-join.md | 76 ++++++++++++++++------- 1 file changed, 55 insertions(+), 21 deletions(-) diff --git a/quest/m1/test-flakes-2/media-late-join.md b/quest/m1/test-flakes-2/media-late-join.md index 1e5c92dd93..b63a937edb 100644 --- a/quest/m1/test-flakes-2/media-late-join.md +++ b/quest/m1/test-flakes-2/media-late-join.md @@ -2,29 +2,63 @@ ## Goal -Late join holds its one-GOP budget under load in both `just test media` and -the interop browser lane, or the regression behind it is fixed. +A viewer joining a broadcast through a relay starts at the newest GOP, +never an older one the relay cached, and the `just test media` late-join +check passes under load without a wider budget. ## Plan It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181): "joined at frame 111, 16 frames behind 127", against a budget of one GOP -(15). First decide whether 16 frames is a real regression (the player -joining at the previous GOP's keyframe) or an off-by-one in how the fixture -samples the live edge. Fix whichever it is; don't widen the budget without a -reason. - -The interop browser check "late join starts live" -(`test/interop/clients/js/media.ts`, `MAX_LATE_JOIN_LAG_FRAMES`, one GOP -compared with `<=`) fails the same way on `main`: 16 frames behind with a -15-frame GOP on 09-24, 09-25, 09-28, and in -[#4577](https://github.com/moq-dev/moq/pull/4577). It samples `live` just -before the page opens. Treat it as the same flake (decided 2026-10-01). - -Recorded in the 2026-10-05 audit: on the test-flakes-2 line, -[#4719](https://github.com/moq-dev/moq/pull/4719) replaced -`MAX_LATE_JOIN_LAG_FRAMES` in the interop browser check with a -published-keyframe check: the latecomer's presented timestamp must be at or -after the published track's newest keyframe. That fixes the browser lane's -sampling. The original 16-frame failure was never reproduced, so what remains -is the Rust `just test media` late join under load. +(15). The interop browser check "late join starts live" failed the same way +on `main` (09-24, 09-25, 09-28, and +[#4577](https://github.com/moq-dev/moq/pull/4577)). +[#4719](https://github.com/moq-dev/moq/pull/4719), landing with the +test-flakes-2 line, compares the latecomer against the published keyframe +instead of the painted counter. `just test media` is that same browser lane; +there is no separate Rust media test. + +Reproduced 2026-10-06 on a loaded machine (load 40 to 65 on 24 cores) by +looping `bun media.ts --cases late-join` against a local `moq-relay`. In 3 of +16 runs the latecomer presented pre-gap media, and 2 of them failed both the +old check (16 and 18 frames behind) and #4719's keyframe check. It is a real +regression, not fixture sampling, and its cause is in the Rust relay +(`moq-net`): + +- The old viewer leaves, so the relay cancels its upstream subscription and + keeps its copy idle. The cancel resets the group in flight. +- The latecomer subscribes with max age 0. The relay re-subscribes from its + newest cached group, and the browser publisher resumes. +- The route's first group header makes the copy live + (`lite/subscriber.rs` `GroupRecv`, `set_live` before `create_group`), but + its first frame is still in flight. Under max age 0 an unstamped successor + never expires the group before it (`TrackState::reach`, and the + Expiration section of the moq-lite draft), and an aborted one is skipped. +- So the latecomer's cursor hands over the cached GOP before the newest one + (relay debug log: groups 7, 8, 9 served within 0.2ms), and + the player presents it. + +Open decision (maintainer): how a relay's copy stops exposing an older group +as the newest while the route's next group has no frame yet. + +1. Recommended: a received group stays out of arrival-order delivery until + its first frame lands or its stream ends, and an idle copy goes live at + that point. This matches origin publishers, which append and write in one + step. It also matches the `idle_newest` doc ("its first frame says whether + the cache is current"). Model semantics and the draft stay as they are. + Cost: a reveal step in `track.rs`, used by the relay receive paths. +2. At max age 0, any newer servable group expires older ones, stamped or + not. It's about five lines in Rust and JS, and matches the SUBSCRIBE text + ("immediately stale once a newer group arrives"). It contradicts the + Expiration section, changes zero-budget delivery on unstamped tracks, and + breaks 11 `moq-net` model tests that encode the current rule. Tried + alone, it still let 3 of 30 runs join pre-gap, through the + `set_live`/`create_group` window. + +Either way, add a deterministic `rs/moq-net/tests/rejoin.rs` case: a copy +going idle mid-group, then a zero-budget rejoin whose route answers with a +group whose header lands before its frame. That needs the mock transport to +split a stream's header from its first frame. + +Public API: none. Wire: none for option 1; option 2 edits the moq-lite +Expiration rule. From ab9dcfea55895dceb0c388358ef98971115da18c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:47:49 -0700 Subject: [PATCH 05/15] wip: withhold a received group until its first frame --- rs/moq-net/src/lite/subscriber.rs | 53 ++++++++---- rs/moq-net/src/model/group.rs | 5 ++ rs/moq-net/src/model/track.rs | 132 +++++++++++++++++++++++++----- 3 files changed, 154 insertions(+), 36 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ea801d25c8..09c6ea8db1 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -468,11 +468,13 @@ impl Subscriber { if let Ok(mut tail) = entry.tail.write() { tail.account(dg.sequence..dg.sequence.saturating_add(1), self.runtime.now()); } - // A datagram says where the live feed is as well as a group does. - if !entry.producer.is_live() { + // A datagram says where the live feed is as well as a group does. Inserted first, so + // the copy goes live with it already showing. + let live = entry.producer.is_live(); + entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?; + if !live { entry.producer.set_live(Some(Position::group(dg.sequence))); } - entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?; Ok(()) } @@ -786,6 +788,8 @@ enum GroupRecvState { group: crate::recv::Group, track: track::Producer, ingest: FrameIngest, + /// Whether readers see the group yet; see [`track::Producer::receive_group`]. + shown: bool, _reading: Reading, }, Done, @@ -813,19 +817,21 @@ impl GroupRecv { // The subscription's end waits until this stream is read. let reading = Reading::open(&entry.tail, Some(hdr.sequence), self.subscriber.runtime.now()); + let group_info = group::Info { sequence: hdr.sequence }; + // Stats (groups/frames/bytes) are counted in the model as the group + // is written, through the tagged `track::Producer`. Withheld from + // readers until its first frame lands. + let received = entry.producer.receive_group(group_info); // A route's first group says where its live feed is, when its answer did // not: even one the cache already holds, as an idle copy asked from its - // head gets back. + // head gets back. The copy goes live once the cache shows that group. if !entry.producer.is_live() { entry.producer.set_live(Some(Position { group: hdr.sequence, frame: hdr.frame_start, })); } - let group_info = group::Info { sequence: hdr.sequence }; - // Stats (groups/frames/bytes) are counted in the model as the group - // is written, through the tagged `track::Producer`. - let mut group = match entry.producer.create_group(group_info) { + let mut group = match received { Ok(group) => group, // The group is at or past the end the publisher declared, which no // later stream can repair. Before lite-05 only a local finish sets @@ -859,11 +865,16 @@ impl GroupRecv { group: crate::recv::Group::new(group), track, ingest: FrameIngest::new(&self.subscriber, timescale), + shown: false, _reading: reading, }; } GroupRecvState::Serve { - group, track, ingest, .. + group, + track, + ingest, + shown, + .. } => { // The track or group dying cancels the stream; the peer's own close // arrives through the ingest's reads. @@ -874,7 +885,13 @@ impl GroupRecv { if let Poll::Ready(err) = group.poll_closed(waiter) { break 'serve Err(err); } - match ingest.poll(&mut self.reader, group, waiter) { + let res = ingest.poll(&mut self.reader, group, waiter); + // Shown once its first frame is stamped, even mid-payload. + if !*shown && group.timestamp().is_some() { + track.reveal_group(group); + *shown = true; + } + match res { Poll::Ready(res) => break 'serve res, Poll::Pending => return Poll::Pending, } @@ -884,27 +901,35 @@ impl GroupRecv { // short drops into an already-aborted group instead of reporting a loss. let GroupRecvState::Serve { group, + track, ingest: _ingest, + shown, _reading, - .. } = std::mem::replace(&mut self.state, GroupRecvState::Done) else { unreachable!() }; - match res { + // A group that ended without a frame shows once it settles. + let handle = (!shown).then(|| (*group).clone()); + let res = match res { Ok(()) => { let _ = group.finish(); + Ok(()) } Err(err @ (Error::Cancel | Error::Stream(crate::StreamError::Cancel))) => { let _ = group.abort(err); + Ok(()) } Err(err) => { tracing::debug!(%err, group = %group.sequence, "group error"); let _ = group.abort(err.clone()); - return Poll::Ready(Err(err)); + Err(err) } + }; + if let Some(handle) = handle { + track.reveal_group(&handle); } - return Poll::Ready(Ok(())); + return Poll::Ready(res); } GroupRecvState::Done => return Poll::Ready(Ok(())), } diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 0c9f9edd75..c6136f4d35 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -859,6 +859,11 @@ impl Producer { (state.committed > state.offset).then_some(state.committed) } + /// Whether `other` is a handle to this same group. + pub(crate) fn is_clone(&self, other: &Self) -> bool { + self.state.same_channel(&other.state) + } + /// Where the group starts in presentation time: its first frame's timestamp, /// or `None` while no frame has been opened. /// diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 86dcb76c37..a3610cd28d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -155,6 +155,20 @@ impl Info { } } +/// Whether a track's cache reflects its live feed; see [`TrackState::set_idle`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Feed { + /// Readers take from the cache. + #[default] + Live, + /// The upstream subscription ended with the copy still held, so how stale the cache + /// is cannot be told until the route answers again. + Idle, + /// The route answered that its feed reaches group `from`, which the cache cannot show + /// yet: until it can, the newest group cached would read as the live edge. + Answered { from: u64 }, +} + #[derive(Default)] pub(crate) struct TrackState { // The publisher's properties, once known; always Some for Subscriber/Producer. @@ -257,8 +271,8 @@ pub(crate) struct TrackState { // Whether the cache reflects the live feed: readers get nothing from it while it does // not, since how stale it is cannot be told; fetches still do. A track is live from // creation, and only a session's copy goes idle: when its upstream subscription ends - // with the copy still held, until the route answers again. - live: bool, + // with the copy still held, until the route's answer shows in the cache. + feed: Feed, // Readers start at this group: the route's live feed went on past a gap after // everything cached, so nothing bounds how old the cache below it is. live_floor: Option, @@ -310,6 +324,10 @@ struct Slot { // subscription content. Fetch-only backfill stays cached but never anchors drift. visible: bool, + // A live group received from a route, withheld from readers until its first frame + // lands or its stream ends; see [`Producer::receive_group`]. + pending: bool, + // A fetched copy of a live `group` that the feed started past the frames a fetch // asked for. It runs to the end of the group too, so it is served in the live // group's place while it lasts; `group` stays the slot, still written by the feed, @@ -323,6 +341,11 @@ impl Slot { self.visible && !self.group.is_aborted() } + /// Whether readers see this slot as live content: [`Self::is_live`], and not withheld. + fn is_shown(&self) -> bool { + self.is_live() && !self.pending + } + /// The copy to hand readers: the fetched head while it lasts, else the live group. fn serving(&self) -> &group::Producer { match &self.head { @@ -457,7 +480,7 @@ impl TrackState { /// Whether readers may take from the cache: it reflects the live feed, or the track /// ended, which makes the cache the whole of it. fn readable(&self) -> bool { - self.live || self.sealed || self.final_sequence.is_some() || self.abort.is_some() + self.feed == Feed::Live || self.sealed || self.final_sequence.is_some() || self.abort.is_some() } fn poll_recv_group(&self, index: usize, min_sequence: u64) -> Poll>> { @@ -513,7 +536,9 @@ impl TrackState { self.datagrams.pop_front(); self.datagram_offset += 1; } + let sequence = datagram.sequence; self.datagrams.push_back(datagram); + self.shown(sequence); } /// Find the smallest-sequence cached group satisfying @@ -628,7 +653,7 @@ impl TrackState { .rev() .filter(|(seq, _)| super::subscription::before_end(**seq, cap)) .find_map(|(_, slot)| { - if !slot.visible || slot.group.is_aborted() { + if !slot.is_shown() { return None; } // The map is ordered by sequence, so the first stamped group from the @@ -678,7 +703,7 @@ impl TrackState { .range(from..) .map(|(_, slot)| slot) .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) - .find(|slot| slot.visible && !slot.group.is_aborted()) + .find(|slot| slot.is_shown()) .map(|slot| &slot.group) } @@ -968,7 +993,6 @@ impl TrackState { fn spawn(broadcast: Arc) -> kio::Producer { let state = kio::Producer::new(Self { broadcast: broadcast.clone(), - live: true, ..Default::default() }); let cache = cache::Track::new(broadcast.pool.clone(), state.downgrade()); @@ -998,8 +1022,9 @@ impl TrackState { /// Updates the live edge, demoting the previous latest into the eviction order; /// the current latest is never enqueued, which is what protects it from /// eviction. `visible` controls arrival-order delivery: publisher-produced - /// groups reach subscribers, fetched backfill is served by sequence only. - fn insert_group(&mut self, group: &group::Producer, visible: bool) { + /// groups reach subscribers, fetched backfill is served by sequence only. A + /// `pending` group joins arrival order only once [`Self::reveal`] shows it. + fn insert_group(&mut self, group: &group::Producer, visible: bool, pending: bool) { let sequence = group.sequence; self.next_stamp = self.next_stamp.wrapping_add(1); let stamp = self.next_stamp; @@ -1031,20 +1056,48 @@ impl TrackState { group: group.clone(), stamp, visible, + pending, head: None, }, ); - if visible { + if visible && !pending { self.arrival.push_back((sequence, stamp)); + self.shown(sequence); + } + } + + /// Show a [`Producer::receive_group`] group to readers, if `group` still holds its slot. + fn reveal(&mut self, group: &group::Producer) { + let sequence = group.sequence; + let Some(slot) = self.lookup.get_mut(&sequence) else { + return; + }; + if !slot.pending || !slot.group.is_clone(group) { + return; + } + slot.pending = false; + self.arrival.push_back((sequence, slot.stamp)); + // A group reset before its first frame shows nothing of the route's feed. + if !slot.group.is_aborted() { + self.shown(sequence); + } + } + + /// Content at `sequence` became readable: an answer waiting on it now shows. + fn shown(&mut self, sequence: u64) { + if let Feed::Answered { from } = self.feed + && sequence >= from + { + self.feed = Feed::Live; } } /// Admit a freshly-created group: settle eviction debt first (so the newcomer /// can never be a victim of the very write that created it), insert it, then /// expire idle groups. - fn commit_group(&mut self, group: &group::Producer, visible: bool) { + fn commit_group(&mut self, group: &group::Producer, pending: bool) { self.charge_debt(); - self.insert_group(group, visible); + self.insert_group(group, true, pending); self.evict_expired(); } @@ -1245,12 +1298,12 @@ impl TrackState { /// stale, so readers get nothing from it until the route answers again. Buffered /// datagrams go too, since a reader returning later must not be handed them. fn set_idle(&mut self) { - self.live = false; + self.feed = Feed::Idle; self.idle_newest = self .lookup .iter() .rev() - .find(|(_, slot)| slot.visible && !slot.group.is_aborted()) + .find(|(_, slot)| slot.is_shown()) .map(|(sequence, _)| *sequence); self.datagram_offset += self.datagrams.len(); self.datagrams.clear(); @@ -1260,11 +1313,13 @@ impl TrackState { /// is current up to there. A feed that went on past a gap after everything cached /// leaves the cache unjudgeable, since nothing bounds how far an old group reached, so /// readers skip it; whatever the feed delivers stays. + /// + /// Readers wait until the cache shows that position: before then its newest group would + /// read as the live edge, though the route already holds a newer one. fn set_live(&mut self, largest: Option) { - if self.live { + if self.feed != Feed::Idle { return; } - self.live = true; let cached = self.idle_newest.take(); self.live_floor = match (largest, cached) { (Some(largest), Some(cached)) if largest.group > cached.saturating_add(1) => Some(cached + 1), @@ -1272,11 +1327,21 @@ impl TrackState { (None, Some(cached)) => Some(cached.saturating_add(1)), _ => self.live_floor, }; + self.feed = match largest { + Some(largest) if !self.shows(largest.group) => Feed::Answered { from: largest.group }, + _ => Feed::Live, + }; + } + + /// Whether readers can already see content at or past `sequence`. + fn shows(&self, sequence: u64) -> bool { + self.lookup.range(sequence..).any(|(_, slot)| slot.is_shown()) + || self.datagrams.iter().any(|datagram| datagram.sequence >= sequence) } /// The newest live group the cache holds, and the number of frames it has so far. fn newest(&self) -> Option<(u64, u64)> { - let (sequence, slot) = self.lookup.iter().rev().find(|(_, slot)| slot.visible)?; + let (sequence, slot) = self.lookup.iter().rev().find(|(_, slot)| slot.visible && !slot.pending)?; Some((*sequence, slot.group.frame_count() as u64)) } @@ -1367,7 +1432,7 @@ impl TrackState { } // Invisible to arrival-order subscribers: fetched on demand, not produced // live by the publisher. - None => self.insert_group(&group, false), + None => self.insert_group(&group, false, false), } self.evict_expired(); Ok(group) @@ -1465,6 +1530,29 @@ impl Producer { /// Create a new group with the given sequence number. pub fn create_group(&self, group: group::Info) -> Result { + self.insert(group, false) + } + + /// Create a group a session is receiving from a route, withheld from readers until + /// [`Self::reveal_group`]. + /// + /// A publisher writes a group's first frame as it creates it, but a route's group + /// header lands before its first frame. Shown in between, it would leave the group + /// before it with no stamped successor, so a joiner at the live edge would be handed + /// that older group first. + pub(crate) fn receive_group(&self, group: group::Info) -> Result { + self.insert(group, true) + } + + /// Show a [`Self::receive_group`] group to readers: its first frame landed, or its + /// stream ended. A no-op once another incarnation took its sequence. + pub(crate) fn reveal_group(&self, group: &group::Producer) { + if let Ok(mut state) = self.modify() { + state.reveal(group); + } + } + + fn insert(&self, group: group::Info, pending: bool) -> Result { let mut state = self.modify()?; if let Some(fin) = state.final_sequence && group.sequence >= fin @@ -1477,7 +1565,7 @@ impl Producer { state.claim_sequence(group.sequence, 0)?; let group = group::Producer::new(group, track, state.cache.clone()).with_meter(self.stats.meter()); - state.commit_group(&group, true); + state.commit_group(&group, pending); Ok(group) } @@ -1499,7 +1587,7 @@ impl Producer { let group = group::Producer::new(group::Info { sequence }, track, state.cache.clone()).with_meter(self.stats.meter()); - state.commit_group(&group, true); + state.commit_group(&group, false); Ok(group) } @@ -1693,14 +1781,14 @@ impl Producer { /// Whether readers may take from the cache; see [`Self::set_idle`]. pub(crate) fn is_live(&self) -> bool { - self.state.read().live + self.state.read().feed == Feed::Live } /// While idle, the newest group cached when the track went idle: a route asked from /// its head sends it again, and its first frame says whether the cache is current. pub(crate) fn idle_newest(&self) -> Option { let state = self.state.read(); - state.idle_newest.filter(|_| !state.live) + state.idle_newest.filter(|_| state.feed == Feed::Idle) } /// Declare the floor a subscription asked for while the serving session has yet to @@ -3354,7 +3442,7 @@ impl group::Expiry for GroupExpiry { .range((std::ops::Bound::Excluded(past), std::ops::Bound::Unbounded)) .map(|(_, slot)| slot) .take_while(|slot| super::subscription::before_end(slot.group.sequence, cap)) - .filter(|slot| slot.visible && !slot.group.is_aborted()) + .filter(|slot| slot.is_shown()) .map(|slot| &slot.group); for group in successor.into_iter().chain(beyond) { if group.timestamp().is_none() From be14c95442e83643d8ef75e6c0ee4814b9009b8d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 00:57:28 -0700 Subject: [PATCH 06/15] fix(net): show a relay's received group once its first frame lands A relay's idle copy went live on the route's first group header, before that group had a frame. Under max age 0 the cached group before it then had no stamped successor, so a rejoining reader was handed that older group first: the browser late join presented the GOP before the newest. A received group now stays out of arrival-order delivery until its first frame lands or its stream ends, and an answered copy goes live only once the cache shows the answered position (groups, datagrams, and lite-07 SUBSCRIBE_START alike). The mock transport can now deliver a uni stream's first write ahead of the rest, which the new rejoin test uses. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 7 +++- rs/moq-net/src/model/track.rs | 3 +- rs/moq-net/tests/rejoin.rs | 66 ++++++++++++++++++++++++++++++++ rs/moq-net/tests/support/mock.rs | 42 +++++++++++++++++++- 4 files changed, 114 insertions(+), 4 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index bf3817736c..b3dd452c6f 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -315,8 +315,11 @@ reader or the next fetch finds its cache. That copy is not live meanwhile: readers get nothing from its cache until the source answers again, since how stale it is cannot be told. lite-07 and moq-transport answer with their largest position, and older lite versions are asked from the head of the newest cached -group, so their first frame says where their feed is. A feed that has moved on -past everything cached leaves the cache below it to fetches. +group, so their first frame says where their feed is. Readers resume once the +cache shows the answered position: a received group stays hidden until its +first frame lands or its stream ends, since a bare group header would leave the +older cached group looking like the live edge. A feed that has moved on past +everything cached leaves the cache below it to fetches. The publisher may declare a retention window per track. Omission sets no limit; zero keeps only the live edge. The origin cache ceiling and cache pool may still diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index a3610cd28d..3b36a01d5e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1772,7 +1772,8 @@ impl Producer { } /// The route answered with its largest position, `None` for nothing yet; see - /// [`Self::set_idle`]. A no-op while live. + /// [`Self::set_idle`]. Readers resume once the cache shows that position. A no-op + /// unless idle. pub(crate) fn set_live(&mut self, largest: Option) { if let Ok(mut state) = self.modify() { state.set_live(largest); diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index 234c8cf5f1..be25cc3cb1 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -387,3 +387,69 @@ async fn rejoin_during_the_cancel_skips_the_cache() { .unwrap_or_else(|_| panic!("{version}: timed out")); } } + +/// A rejoining reader is not handed the cached group while the route's next group has no +/// frame yet. +/// +/// The publisher moved on by one group while nobody subscribed, so the route answers the +/// rejoin with that group, right after the one the relay cached. Its header lands before its +/// first frame. Shown in between, it would leave the cached group without a stamped +/// successor, and a reader at the live edge would be handed that older group first. +#[tokio::test(start_paused = true)] +async fn rejoin_waits_for_the_answers_first_frame() { + // Lite01/02 carry no max age, so the publisher serves the relay's cached head again and + // that answer is what makes the cache current. + let unbudgeted = ["moq-lite-01", "moq-lite-02"]; + for version in Version::names().filter(|version| !unbudgeted.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()); + 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(); + + // Held for fetches only: it keeps the relay's copy without subscribing. + let _held = remote.track("video").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(); + + let mut group = track.append_group().unwrap(); + group.write_frame(ts(500), b"new".as_ref()).unwrap(); + group.finish().unwrap(); + + pair.server_transport.split_unis(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + let early = tokio::time::timeout(Duration::from_secs(1), sub.recv_group()).await; + assert!( + early.is_err(), + "{version}: the rejoining reader got the cached group {:?} first", + early.map(|group| group.unwrap().unwrap().sequence) + ); + + pair.server_transport.release_split(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.sequence, 1, "{version}"); + assert_eq!(read_all(&mut group).await.unwrap(), [b"new".to_vec()], "{version}"); + }) + .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 ed3b3a72b9..c99674eaa2 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -230,6 +230,11 @@ pub struct MockRecvStream { buf: Bytes, /// Whether we hit FIN or reset. done: bool, + /// Once released, delivers what follows this stream's first write; see + /// [`MockSession::split_unis`]. + split: Option>>, + /// Data chunks handed to the reader so far. + chunks: usize, /// Shared signal to notify the peer's send-side `poll_closed`. closed: Arc, park: kio::Park, @@ -240,6 +245,17 @@ impl MockRecvStream { /// Pop the next chunk, mapping queue closure to an implicit FIN. Once everything /// sent before a CONNECTION_CLOSE is read, an unfinished stream fails with it. fn poll_chunk(&mut self, cx: &mut Context<'_>) -> Poll> { + if let Some(split) = &self.split + && self.chunks > 0 + && split + .poll(self.park.hold(cx), |released| match **released { + true => Poll::Ready(()), + false => Poll::Pending, + }) + .is_pending() + { + return Poll::Pending; + } if self.flight.is_none() { let waiter = self.park.hold(cx); // Register on the close first so one racing the pop still wakes this poll. @@ -281,6 +297,7 @@ impl poll::RecvStream for MockRecvStream { match std::task::ready!(self.poll_chunk(cx)) { Some(StreamChunk::Data(data)) => { + self.chunks += 1; let n = dst.len().min(data.len()); dst[..n].copy_from_slice(&data[..n]); if n < data.len() { @@ -364,6 +381,8 @@ fn new_stream_pair(conn: &Arc) -> (MockSendStream, MockRecvStre landing: None, buf: Bytes::new(), done: false, + split: None, + chunks: 0, closed, park: kio::Park::default(), conn: conn.clone(), @@ -426,6 +445,9 @@ struct SessionSide { held: Mutex>>, /// Whether the peer has withheld uni stream credit, parking every open. withheld: Mutex, + /// While set, the uni streams this side opens hold back everything after their first + /// write until it is released; see [`MockSession::split_unis`]. + split: Mutex>>>, /// Whether the datagrams this side sends are lost. lossy: Mutex, /// Whether this side drops the FIN of the bidi streams it opens. @@ -503,7 +525,8 @@ impl poll::Session for MockSession { return Poll::Ready(Err(self.close_error())); } - let (mut our_send, peer_recv) = new_stream_pair(&self.side.conn); + let (mut our_send, mut peer_recv) = new_stream_pair(&self.side.conn); + peer_recv.split = self.side.split.lock().unwrap().clone(); if let Some(held) = self.side.held.lock().unwrap().as_mut() { our_send.ack_fin = true; @@ -635,6 +658,21 @@ impl MockSession { *self.side.withheld.lock().unwrap() = true; } + /// Deliver only the first write of each uni stream this side opens from now on, holding + /// the rest until [`Self::release_split`]. A group stream's first write is its header, + /// so the header lands before the group's first frame, as QUIC may split them across + /// packets. + pub fn split_unis(&self) { + *self.side.split.lock().unwrap() = Some(Arc::default()); + } + + /// Deliver what [`Self::split_unis`] held, and stop splitting. + pub fn release_split(&self) { + if let Some(split) = self.side.split.lock().unwrap().take() { + *split.lock() = true; + } + } + /// Lose every datagram this side sends from now on. pub fn lose_datagrams(&self) { *self.side.lossy.lock().unwrap() = true; @@ -684,6 +722,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, conn: conn.clone(), held: Mutex::default(), withheld: Mutex::default(), + split: Mutex::default(), lossy: Mutex::default(), withhold_bidi_fins: Arc::default(), }); @@ -699,6 +738,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, conn, held: Mutex::default(), withheld: Mutex::default(), + split: Mutex::default(), lossy: Mutex::default(), withhold_bidi_fins: 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 07/15] 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 08/15] 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 09/15] chore: retrigger CI Co-Authored-By: Claude Opus 5.5 From ca6e4dac8e84081fcb8d57d26cf4624d6e9beae3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 09:42:50 -0700 Subject: [PATCH 10/15] test(net): run the first-frame rejoin test on the simulated clock Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/rejoin.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index b6e54cce48..a3a8d4d0ca 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -388,13 +388,13 @@ async fn rejoin_during_the_cancel_skips_the_cache() { /// rejoin with that group, right after the one the relay cached. Its header lands before its /// first frame. Shown in between, it would leave the cached group without a stamped /// successor, and a reader at the live edge would be handed that older group first. -#[tokio::test(start_paused = true)] +#[moq_net_sim::test] async fn rejoin_waits_for_the_answers_first_frame() { // Lite01/02 carry no max age, so the publisher serves the relay's cached head again and // that answer is what makes the cache current. let unbudgeted = ["moq-lite-01", "moq-lite-02"]; for version in Version::names().filter(|version| !unbudgeted.contains(version)) { - tokio::time::timeout(TEST_TIMEOUT, async { + moq_net_sim::timeout(TEST_TIMEOUT, async { let publisher = produce_origin(1); let relay = produce_origin(2); @@ -430,7 +430,7 @@ async fn rejoin_waits_for_the_answers_first_frame() { pair.server_transport.split_unis(); let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); - let early = tokio::time::timeout(Duration::from_secs(1), sub.recv_group()).await; + let early = moq_net_sim::timeout(Duration::from_secs(1), sub.recv_group()).await; assert!( early.is_err(), "{version}: the rejoining reader got the cached group {:?} first", From 8ca3a737a97c54c61ca42fc9d3e76337895e2699 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 09:58:40 -0700 Subject: [PATCH 11/15] fix(net): judge a rejoin's answer by the groups that survived the leave The cancel that idles a relay's copy resets the group in flight, so an answer one past the snapshot can still leave a gap after the newest group the copy holds. Measured from the snapshot it read as contiguous, and the cached group's reach ran to the answer, fresh to any positive budget. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 39 ++++++++++++++++++++++++++++++++++- 1 file changed, 38 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 2b078a3bc6..bcf41ea521 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1322,7 +1322,15 @@ impl TrackState { if self.feed != Feed::Idle { return; } - let cached = self.idle_newest.take(); + // Judged by what survived: the cancel that idled the copy resets the group in flight, + // and an answer past it leaves a gap after the newest group still cached. + let cached = self.idle_newest.take().and_then(|newest| { + self.lookup + .range(..=newest) + .rev() + .find(|(_, slot)| slot.is_shown()) + .map(|(sequence, _)| *sequence) + }); self.live_floor = match (largest, cached) { (Some(largest), Some(cached)) if largest.group > cached.saturating_add(1) => Some(cached + 1), // The route has nothing, so whatever is cached is not its feed. @@ -5622,6 +5630,35 @@ mod test { sequences } + /// A rejoined copy judges the route's answer by the groups that survived the leave. + /// + /// The cancel that idles a copy resets the group in flight. The route then answers past + /// it, so the newest group still cached is two behind: a gap, which the cached group + /// cannot be measured across. Judged by the group the copy held when it went idle, the + /// answer would read as contiguous, and the cached group would reach all the way to the + /// answer, fresh to any positive budget. + #[test] + fn an_answer_past_a_reset_group_skips_the_cache() { + let mut producer = track_producer("test", None); + append_at(&mut producer, 0); + let mut open = producer.append_group().unwrap(); + open.write_frame(Timestamp::from_millis(500).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + + producer.set_idle(); + open.abort(Error::Cancel).unwrap(); + + let mut answer = producer.receive_group(group::Info { sequence: 2 }).unwrap(); + producer.set_live(Some(Position::group(2))); + answer + .write_frame(Timestamp::from_millis(533).unwrap(), bytes::Bytes::from_static(b"")) + .unwrap(); + producer.reveal_group(&answer); + + let mut subscriber = producer.subscribe(Subscription::default().with_max_age(Duration::from_millis(400))); + assert_eq!(drain(&mut subscriber), vec![2]); + } + #[test] fn real_time_skips_a_backlog_to_the_live_edge() { let mut producer = track_producer("test", None); From 763cd531c1f5df241e91ad9512569ff66b303a60 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 10:00:24 -0700 Subject: [PATCH 12/15] quest(test-flakes-2): record the relay fixes and the joiner's remaining reach-back Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/media-late-join.md | 108 ++++++++++------------ 1 file changed, 51 insertions(+), 57 deletions(-) diff --git a/quest/m1/test-flakes-2/media-late-join.md b/quest/m1/test-flakes-2/media-late-join.md index b63a937edb..9c8f0b376f 100644 --- a/quest/m1/test-flakes-2/media-late-join.md +++ b/quest/m1/test-flakes-2/media-late-join.md @@ -2,63 +2,57 @@ ## Goal -A viewer joining a broadcast through a relay starts at the newest GOP, -never an older one the relay cached, and the `just test media` late-join -check passes under load without a wider budget. +A viewer joining a broadcast through a relay after a demand gap presents the +newest GOP, never an older one, and the `just test media` late-join check +("late join starts live") passes under load without a wider budget. ## Plan -It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181): -"joined at frame 111, 16 frames behind 127", against a budget of one GOP -(15). The interop browser check "late join starts live" failed the same way -on `main` (09-24, 09-25, 09-28, and -[#4577](https://github.com/moq-dev/moq/pull/4577)). -[#4719](https://github.com/moq-dev/moq/pull/4719), landing with the -test-flakes-2 line, compares the latecomer against the published keyframe -instead of the painted counter. `just test media` is that same browser lane; -there is no separate Rust media test. - -Reproduced 2026-10-06 on a loaded machine (load 40 to 65 on 24 cores) by -looping `bun media.ts --cases late-join` against a local `moq-relay`. In 3 of -16 runs the latecomer presented pre-gap media, and 2 of them failed both the -old check (16 and 18 frames behind) and #4719's keyframe check. It is a real -regression, not fixture sampling, and its cause is in the Rust relay -(`moq-net`): - -- The old viewer leaves, so the relay cancels its upstream subscription and - keeps its copy idle. The cancel resets the group in flight. -- The latecomer subscribes with max age 0. The relay re-subscribes from its - newest cached group, and the browser publisher resumes. -- The route's first group header makes the copy live - (`lite/subscriber.rs` `GroupRecv`, `set_live` before `create_group`), but - its first frame is still in flight. Under max age 0 an unstamped successor - never expires the group before it (`TrackState::reach`, and the - Expiration section of the moq-lite draft), and an aborted one is skipped. -- So the latecomer's cursor hands over the cached GOP before the newest one - (relay debug log: groups 7, 8, 9 served within 0.2ms), and - the player presents it. - -Open decision (maintainer): how a relay's copy stops exposing an older group -as the newest while the route's next group has no frame yet. - -1. Recommended: a received group stays out of arrival-order delivery until - its first frame lands or its stream ends, and an idle copy goes live at - that point. This matches origin publishers, which append and write in one - step. It also matches the `idle_newest` doc ("its first frame says whether - the cache is current"). Model semantics and the draft stay as they are. - Cost: a reveal step in `track.rs`, used by the relay receive paths. -2. At max age 0, any newer servable group expires older ones, stamped or - not. It's about five lines in Rust and JS, and matches the SUBSCRIBE text - ("immediately stale once a newer group arrives"). It contradicts the - Expiration section, changes zero-budget delivery on unstamped tracks, and - breaks 11 `moq-net` model tests that encode the current rule. Tried - alone, it still let 3 of 30 runs join pre-gap, through the - `set_live`/`create_group` window. - -Either way, add a deterministic `rs/moq-net/tests/rejoin.rs` case: a copy -going idle mid-group, then a zero-budget rejoin whose route answers with a -group whose header lands before its frame. That needs the mock transport to -split a stream's header from its first frame. - -Public API: none. Wire: none for option 1; option 2 edits the moq-lite -Expiration rule. +It failed once after [#4181](https://github.com/moq-dev/moq/pull/4181) ("16 +frames behind" a 15-frame GOP), and the interop browser check failed the +same way on `main`. `just test media` is that browser lane through a Rust +`moq-relay`; there is no separate Rust media test. +[#4719](https://github.com/moq-dev/moq/pull/4719) now compares the +latecomer against the published keyframe instead of the painted counter. + +It is real, not fixture sampling: the latecomer is handed a GOP from before +the demand gap. Reproduced by looping `bun media.ts --cases late-join` +against a local relay with debug logs. In the fixture, the old viewer +leaves, the publisher pauses (closing its open GOP and writing an empty +marker group), and the latecomer joins about a second later. + +[#4914](https://github.com/moq-dev/moq/pull/4914) fixed two relay windows, +each with a deterministic test: + +- A received group stays hidden from readers until its first frame lands or + its stream ends, and an idle copy goes live only once the cache shows the + route's answer (`rejoin_waits_for_the_answers_first_frame`). +- A rejoin's answer is judged against the groups that survived the leave, + not the idle snapshot, since the cancel resets the group in flight + (`an_answer_past_a_reset_group_skips_the_cache`). + +What remains (1 of 30 runs after both fixes): the latecomer's max age rises +to its playout delay (about 145ms) right after it subscribes. When the +pre-gap GOP was short, the GOP before it still reaches within that budget of +the marker, so the relay hands it over by design. The player then presents +from that GOP's keyframe, not from its playout position. Relay log: groups +7, 8, 9 served in 1ms with a 145ms budget, and the player shows frames 360ms +behind the published keyframe. + +Open decision (maintainer): what keeps a joiner off pre-gap media that its +budget legitimately reaches. + +1. Recommended: in live (unbuffered) mode, the watch player presents nothing + older than its playout position (newest timestamp minus delay) and decodes + the rest silently. This holds however the groups interleave: the relay + served the cached GOP and the marker within 1ms of each other, so whatever + decides has to run at presentation, after both have landed. Buffered + playback still starts at the head of what its budget can play. +2. The hang consumer (`js/hang` `Container.Consumer`, and its Rust mirror) + drops every group before a discontinuity marker that lands before its + first frame is handed out. The marker already exists for this ("a later + subscriber resumes ... without the pre-gap group reading as live"). It's + narrower, but it loses the race whenever the cached GOP's keyframe is + handed out before the marker lands. + +Public API: none. Wire: none. From af547166602be25b3f90a32a67311681111c10fa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 10:03:00 -0700 Subject: [PATCH 13/15] style(net): format the withheld-group scan Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index bcf41ea521..1d8834894d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1351,7 +1351,11 @@ impl TrackState { /// The newest live group the cache holds, and the number of frames it has so far. fn newest(&self) -> Option<(u64, u64)> { - let (sequence, slot) = self.lookup.iter().rev().find(|(_, slot)| slot.visible && !slot.pending)?; + let (sequence, slot) = self + .lookup + .iter() + .rev() + .find(|(_, slot)| slot.visible && !slot.pending)?; Some((*sequence, slot.group.frame_count() as u64)) } From dec94e702402d77300c93a9dde1229f63c36aeb7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 10:16:16 -0700 Subject: [PATCH 14/15] quest(test-flakes-2): late join checks catch-up, not the first keyframe Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 2 +- quest/m1/test-flakes-2/media-late-join.md | 34 ++++++++++------------- 2 files changed, 16 insertions(+), 20 deletions(-) diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 8f2e0e8a4f..305a7dadd2 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -31,4 +31,4 @@ Public API: none. Wire: none. ## Required - [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 +- [Media late join](/quest/m1/test-flakes-2/media-late-join.md) - a late joiner shows video promptly and catches up to live, and the check asserts that under load diff --git a/quest/m1/test-flakes-2/media-late-join.md b/quest/m1/test-flakes-2/media-late-join.md index 9c8f0b376f..c9c1a8032d 100644 --- a/quest/m1/test-flakes-2/media-late-join.md +++ b/quest/m1/test-flakes-2/media-late-join.md @@ -1,10 +1,10 @@ -# [S] Media late join within one GOP +# [S] Media late join catches up to live ## Goal -A viewer joining a broadcast through a relay after a demand gap presents the -newest GOP, never an older one, and the `just test media` late-join check -("late join starts live") passes under load without a wider budget. +A viewer joining a broadcast through a relay after a demand gap shows video +promptly and reaches the live edge within a bound, and the late-join check +asserts exactly that and passes under load. ## Plan @@ -39,20 +39,16 @@ from that GOP's keyframe, not from its playout position. Relay log: groups 7, 8, 9 served in 1ms with a 145ms budget, and the player shows frames 360ms behind the published keyframe. -Open decision (maintainer): what keeps a joiner off pre-gap media that its -budget legitimately reaches. - -1. Recommended: in live (unbuffered) mode, the watch player presents nothing - older than its playout position (newest timestamp minus delay) and decodes - the rest silently. This holds however the groups interleave: the relay - served the cached GOP and the marker within 1ms of each other, so whatever - decides has to run at presentation, after both have landed. Buffered - playback still starts at the head of what its budget can play. -2. The hang consumer (`js/hang` `Container.Consumer`, and its Rust mirror) - drops every group before a discontinuity marker that lands before its - first frame is handed out. The marker already exists for this ("a later - subscriber resumes ... without the pre-gap group reading as live"). It's - narrower, but it loses the race whenever the cached GOP's keyframe is - handed out before the marker lands. +Decided 2026-10-06 by the maintainer: a joiner showing the reachable pre-gap +GOP at once and then fast-forwarding to live is the wanted behavior, so no +player or consumer change. The check is what's too strict: it asserts the +first presented frame is at or after the newest published keyframe. + +Remaining work: the late-join check (`just test media` and the interop +browser lane) asserts what a viewer cares about instead: video shows within a +start bound, and the player reaches the live edge within a catch-up bound. +Set both bounds from measurements, looped under synthetic CPU load (the +maintainer approved a bounded load generator such as `stress-ng` for these +runs), and keep the relay fixes' deterministic tests as the regression guard. Public API: none. Wire: none. From 75a536c07138d2299f662d15c903dbdf18be10a3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 22:23:32 -0700 Subject: [PATCH 15/15] fix(net): revealing a group presents its frames again Its frames present while it is hidden, so a parked read they wake can judge the older edge and park again on the same deadline. The reveal's landing wake only reaches reads from the group's nearest shown predecessor up, so an older expired read stayed parked. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/track.rs | 54 +++++++++++++++++++++++++++++++++-- 1 file changed, 52 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 227d6f64e9..30bcb2a697 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1112,8 +1112,15 @@ impl TrackState { slot.pending = false; self.arrival.push_back((sequence, slot.stamp)); // A group reset before its first frame shows nothing of the route's feed. - if !slot.group.is_aborted() { - self.landed(sequence); + if slot.group.is_aborted() { + return; + } + let latest = slot.group.latest(); + self.landed(sequence); + // Its frames presented while hidden, so a read woken by them may have judged the + // older edge and parked again: present them once more now that they count. + if let Some(latest) = latest { + self.cache.wakes().presented(latest); } } @@ -6326,6 +6333,49 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } + /// A received group's frames present while it is hidden, so a parked read they wake + /// still sees the older edge and parks again. Revealing the group has to wake it once + /// more, though the read sits below the group's nearest shown predecessor. + #[test] + fn revealing_a_presented_group_wakes_a_read_it_expires() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(500))); + let mut head = producer.append_group().unwrap(); + head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + append_at(&mut producer, 1000); // the successor bounds the head's reach + + let mut held = subscriber + .recv_group() + .now_or_never() + .unwrap() + .unwrap() + .expect("head group"); + assert_eq!(held.sequence, 0); + assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); + + let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); + let mut cx = std::task::Context::from_waker(&waker); + let mut next = std::pin::pin!(held.read_frame()); + assert!(next.as_mut().poll(&mut cx).is_pending()); + + let mut received = producer.receive_group(group::Info { sequence: 2 }).unwrap(); + received + .write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + woken.store(false, Ordering::SeqCst); + assert!( + next.as_mut().poll(&mut cx).is_pending(), + "the hidden group moves no edge" + ); + + producer.reveal_group(&received); + assert!(woken.load(Ordering::SeqCst), "the reveal wakes the expired read"); + let result = next.as_mut().poll(&mut cx); + assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); + } + /// The ordinary live case, at the default real-time budget: 2s GOPs produced one at /// a time and read as they arrive. The budget must take the live edge without /// shortening the group the reader is already on, so every frame of every group is