From 7b2c2941829a19c870cfe4843c80b9730c28e66c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 20:13:39 -0700 Subject: [PATCH 1/4] quest: claim ts-passthrough-jump Co-Authored-By: Claude Opus 5.5 From 18d5e1750e54e989270ff5f41550f98e72a31bb2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 21:04:38 -0700 Subject: [PATCH 2/4] test(moq-cli): the TS passthrough recording starts before the second copy On moq-transport-14 the recording joins the publisher's current group when its SUBSCRIBE arrives through the relay. Under load that landed after the second copy's first group (4), so the joining FETCH skipped group 2, the first copy's tail, and "both copies crossed" failed. Take the first group before decoding the second copy, so the read always starts behind it. Co-Authored-By: Claude Opus 5.5 --- quest/m1/test-flakes-2/README.md | 1 - quest/m1/test-flakes-2/ts-passthrough-jump.md | 21 ------------ rs/moq-cli/src/publish.rs | 32 +++++++++++++------ 3 files changed, 22 insertions(+), 32 deletions(-) delete mode 100644 quest/m1/test-flakes-2/ts-passthrough-jump.md diff --git a/quest/m1/test-flakes-2/README.md b/quest/m1/test-flakes-2/README.md index 1003d8d701..f1fd9f1620 100644 --- a/quest/m1/test-flakes-2/README.md +++ b/quest/m1/test-flakes-2/README.md @@ -37,5 +37,4 @@ Public API: none. Wire: none. - [Interop close code](/quest/m1/test-flakes-2/interop-close-code.md) - the browser close-code test reliably sees `unauthorized` for a refused session - [TS duration fidelity](/quest/m1/test-flakes-2/ts-duration-fidelity.md) - TS compliance captures the whole round-tripped stream - [Cluster burst overrun](/quest/m1/test-flakes-2/cluster-burst-overrun.md) - the impaired cluster burst drill always overruns its bottleneck, so it always exercises it -- [TS passthrough jump](/quest/m1/test-flakes-2/ts-passthrough-jump.md) - moq-cli's flagged-jump TS passthrough test passes under load without wall-clock pacing - [Media audio tone](/quest/m1/media-audio-tone.md) - the `just test media` audio-tone check passes under load, fixed at its cause diff --git a/quest/m1/test-flakes-2/ts-passthrough-jump.md b/quest/m1/test-flakes-2/ts-passthrough-jump.md deleted file mode 100644 index 50ba913edd..0000000000 --- a/quest/m1/test-flakes-2/ts-passthrough-jump.md +++ /dev/null @@ -1,21 +0,0 @@ -# [S] TS passthrough jump test holds under load - -## Goal - -`publish::tests::ts_passthrough_crosses_a_relay_through_a_flagged_jump` -(`rs/moq-cli`) passes under a loaded `just check`, fixed at its cause. - -## Plan - -Seen 2026-10-09 by three PRs on unmodified `main` (#5146, #5155, #5153): the -assertion "moq-transport-14: both copies crossed" fails, so the recording -held at most one copy of the fixture. It failed 2 of 3 loaded runs and -passes alone. The test came in with #5003. It paces its input with a 15 ms -wall-clock sleep and waits up to 10 s. - -Decided 2026-10-09: find why the second copy goes missing under load before -changing anything, then replace the wall-clock pacing with mocked time or -observable events where the fixture allows. Never widen the wait or add a -retry, per this questline's rules. - -Public API: none. Wire: none. diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 446684f680..1510a26beb 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -1008,14 +1008,21 @@ pub(crate) mod tests { .expect("a catalog"); let name = catalog.m2ts.expect("the m2ts section").track; + // Subscribe and take the first group, so a read starts behind anything decoded + // afterwards: an IETF subscription joins the publisher's current group when its + // SUBSCRIBE arrives, not when it is sent. + let start = async |subscription: moq_net::track::Subscription| { + let mut track = consumer.track(&name).unwrap().subscribe(subscription).await.unwrap(); + let first = tokio::time::timeout(WAIT, track.recv_group()).await.unwrap().unwrap(); + (track, first) + }; // Each group's sequence and frames, read until the track ends. A read cut off as // stale leaves `None` for that group's frames. - let read = |subscription: moq_net::track::Subscription| { - let track = consumer.track(&name).unwrap(); + let read = |(mut track, first): (moq_net::track::Subscriber, Option)| { tokio::spawn(tokio::time::timeout(WAIT, async move { - let mut track = track.subscribe(subscription).await.unwrap(); let mut groups = Vec::new(); - while let Some(mut group) = track.recv_group().await.unwrap() { + let mut next = first; + while let Some(mut group) = next { let mut frames = Some(Vec::new()); loop { match group.read_frame().await { @@ -1033,13 +1040,17 @@ pub(crate) mod tests { } } groups.push((group.sequence, frames)); + next = track.recv_group().await.unwrap(); } groups.sort_by_key(|(sequence, _)| *sequence); groups })) }; - let recording = read(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_AGE)); - // About twice real time, so the recording keeps up and only staleness can cut it. + let recording = + read(start(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_AGE)).await); + // About twice real time, so the recording keeps up and only staleness can cut it. A + // burst ends the track while the relay still pulls the older groups, and the relay + // drops its upstream subscription with those groups unfinished. for chunk in second.chunks(8 * 1024) { decoder.decode_chunk(chunk).unwrap(); tokio::time::sleep(Duration::from_millis(15)).await; @@ -1080,10 +1091,11 @@ pub(crate) mod tests { // the budget behind. An IETF subscription starts at the newest group, so only lite // has the groups behind it to read. if version.starts_with("moq-lite") { - let late = read(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))) - .await - .unwrap() - .expect("the late read ends with the track"); + let late = + read(start(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))).await) + .await + .unwrap() + .expect("the late read ends with the track"); let whole: Vec = late .iter() .filter(|(_, frames)| frames.is_some()) From 867762c828b76fd0a70f88cac8ef0c2b9a759df5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 22:05:32 -0700 Subject: [PATCH 3/4] fix(net): a lite publisher keeps its subscription until in-flight groups finish Once a lite subscription reached the track's end, the publisher dropped its track subscription and only then drained the group streams still in flight. On a relay that was the copy's last demand, so the relay cancelled its upstream subscription as idle while those groups' tails were still coming, and a reader behind the relay hung on the stranded group. The subscription now stays until the drain finishes. The TS passthrough relay test decodes its second copy all at once again, which is what hit this. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/publish.rs | 9 +- rs/moq-net/src/lite/publisher.rs | 51 ++++---- rs/moq-net/tests/relay_end_in_flight.rs | 167 ++++++++++++++++++++++++ 3 files changed, 192 insertions(+), 35 deletions(-) create mode 100644 rs/moq-net/tests/relay_end_in_flight.rs diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 1510a26beb..6a584a735b 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -1048,13 +1048,8 @@ pub(crate) mod tests { }; let recording = read(start(moq_net::track::Subscription::default().with_max_delay(RECORDING_MAX_AGE)).await); - // About twice real time, so the recording keeps up and only staleness can cut it. A - // burst ends the track while the relay still pulls the older groups, and the relay - // drops its upstream subscription with those groups unfinished. - for chunk in second.chunks(8 * 1024) { - decoder.decode_chunk(chunk).unwrap(); - tokio::time::sleep(Duration::from_millis(15)).await; - } + // All at once: the track ends while the relay is still pulling the older groups. + decoder.decode_chunk(&second).unwrap(); decoder.finish().unwrap(); let recording: Vec<_> = recording diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index a7679fa926..767eba5148 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1451,11 +1451,10 @@ enum SubscribeServe { latest: Option, update: Option, }, - /// Streaming groups and datagrams. Boxed: by far the largest state, and the enum - /// is moved on every transition. + /// Streaming groups and datagrams, then draining the in-flight group streams before + /// the FIN. Boxed: by far the largest state, and the enum is moved on every + /// transition. Run(Box>), - /// The track finished: draining the in-flight group streams before the FIN. - Drain { children: kio::Tasks> }, } impl Request for SubscribeServe { @@ -1500,8 +1499,6 @@ impl Request for SubscribeServe { match self { Self::Confirm { update: pending, .. } => *pending = Some(update), Self::Run(run) => run.update(update), - // The track is over: nothing left to steer. - Self::Drain { .. } => {} } } @@ -1513,19 +1510,14 @@ impl Request for SubscribeServe { ) -> Poll, Error>> { loop { match self { - Self::Confirm { subscribing, .. } => { + Self::Confirm { + subscribing, + msg, + latest, + update, + } => { let mut track = ready!(subscribing.poll_ok(waiter))?; - let Self::Confirm { - msg, latest, update, .. - } = std::mem::replace( - self, - Self::Drain { - children: Default::default(), - }, - ) - else { - unreachable!() - }; + let (latest, update) = (*latest, update.take()); // Per-frame timestamps require a wire format that carries them. Lite05+ // prefixes every frame with a zigzag-delta timestamp at the track's @@ -1580,17 +1572,17 @@ impl Request for SubscribeServe { *self = Self::Run(Box::new(run)); } Self::Run(run) => { - // The live edge reached the boundary; SUBSCRIBE_END was already sent (or - // the version predates the track stream). Drain the in-flight group - // machines, then FIN. - if ready!(run.poll_step(writer, waiter))?.is_continue() { - return Poll::Ready(Ok(ControlFlow::Continue(()))); + // Once the live edge reaches the boundary (SUBSCRIBE_END was already sent, + // or the version predates the track stream), drain the in-flight group + // machines, then FIN. The subscription stays until they finish, or a relay + // would let its upstream go and strand their tails. + if !run.ended { + if ready!(run.poll_step(writer, waiter))?.is_continue() { + return Poll::Ready(Ok(ControlFlow::Continue(()))); + } + run.ended = true; } - let children = std::mem::take(&mut run.children); - *self = Self::Drain { children }; - } - Self::Drain { children } => { - ready!(children.poll(waiter)); + ready!(run.children.poll(waiter)); return Poll::Ready(Ok(ControlFlow::Break(()))); } } @@ -2700,6 +2692,8 @@ struct TrackRun { // fallback: otherwise off. datagrams: bool, children: kio::Tasks>, + // The live edge reached the track's boundary: only the in-flight groups are left. + ended: bool, } impl TrackRun { @@ -2731,6 +2725,7 @@ impl TrackRun { count_streams, datagrams, children: kio::Tasks::new(), + ended: false, } } diff --git a/rs/moq-net/tests/relay_end_in_flight.rs b/rs/moq-net/tests/relay_end_in_flight.rs new file mode 100644 index 0000000000..8d958e0544 --- /dev/null +++ b/rs/moq-net/tests/relay_end_in_flight.rs @@ -0,0 +1,167 @@ +//! A track's end reaching a relay while an earlier group is still in flight upstream +//! never strands that group: the subscriber behind the relay reads it whole. +//! +//! The subscriber is reading the older group when a newer group and the track's end +//! arrive, and only then does the older group's tail follow. The relay hands out every +//! group up to the end, but must keep pulling the older group's tail for the reader +//! still on it rather than let its upstream subscription go. +//! +//! Deterministic: simulated time, and every "let the sessions send" step is an +//! idle-advance rather than a race. + +mod support; + +use std::time::Duration; + +use moq_net::track::{Info, Position, Subscription}; +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// A cache no group outlives on the publisher. +const FOREVER: Duration = Duration::from_millis((1 << 53) - 1); + +/// The reader's budget, well inside the timeout: moq-lite-03 cannot say where a track +/// ends, so its reader waits this long for a group it cannot account for. +const BUDGET: Duration = Duration::from_secs(1); + +const VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-17", + "moq-transport-22", +]; + +const FRAMES: [&[u8]; 2] = [b"frame-a", b"frame-b"]; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +/// Let the drivers run until nothing is runnable: simulated time advances only once +/// every task is idle, so this returns once every queued write has been sent. +async fn settle() { + moq_net_sim::sleep(Duration::from_millis(10)).await; +} + +/// Publish two groups through a relay, the older one's tail written after the newer +/// group and the track's end. Returns each group's sequence and frame count as read behind +/// the relay and the error the read ended with, or `None` when the read hung. +async fn round(version: &str) -> Option<(Vec<(u64, usize)>, Option)> { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast + .create_track("video", Info::default().with_max_age(FOREVER)) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let relay = produce_origin(2); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let upstream = connect_mock(options).await; + + let subscriber = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(subscriber.clone()); + let downstream = connect_mock(options).await; + + let consumer = subscriber.consume(); + moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let reader = moq_net_sim::spawn(async move { + let subscription = Subscription::default() + .with_max_delay(BUDGET) + .with_start(Position::group(0)); + let mut sub = remote + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let mut got = Vec::new(); + loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => break (got, None), + Err(err) => break (got, Some(err)), + }; + let mut frames = 0; + let err = loop { + match group.read_frame().await { + Ok(Some(_)) => frames += 1, + Ok(None) => break None, + Err(err) => break Some(err), + } + }; + got.push((group.sequence, frames)); + if err.is_some() { + break (got, err); + } + } + }); + + moq_net_sim::timeout(TIMEOUT, track.demand().used()) + .await + .expect("no subscriber appeared") + .unwrap(); + settle().await; + + // The older group's head reaches the reader. + let mut old = track.append_group().unwrap(); + old.write_frame(Timestamp::now(), FRAMES[0]).unwrap(); + settle().await; + + // A newer group and the end arrive while the older group is still open. + let mut new = track.append_group().unwrap(); + for frame in FRAMES { + new.write_frame(Timestamp::now(), frame).unwrap(); + } + new.finish().unwrap(); + track.finish().unwrap(); + settle().await; + + // Then the older group's tail. + old.write_frame(Timestamp::now(), FRAMES[1]).unwrap(); + old.finish().unwrap(); + + let result = moq_net_sim::timeout(TIMEOUT, reader) + .await + .ok() + .map(|res| res.expect("reader panicked")); + drop(( + old, new, track, upstream, downstream, relay, broadcast, publisher, subscriber, + )); + result.map(|(mut got, err)| { + got.sort(); + (got, err) + }) +} + +#[moq_net_sim::test] +async fn an_end_never_strands_a_group_in_flight_through_a_relay() { + let mut failures = Vec::new(); + for version in VERSIONS { + match round(version).await { + Some((got, None)) if got == [(0, FRAMES.len()), (1, FRAMES.len())] => {} + Some((got, err)) => failures.push(format!("{version}: got {got:?}, err {err:?}")), + None => failures.push(format!("{version}: the read behind the relay hung")), + } + } + assert!(failures.is_empty(), "{failures:#?}"); +} From 90edb78fd37c32a82e26b1c5ad420dced4fa4d9f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 12 Oct 2026 08:01:54 -0700 Subject: [PATCH 4/4] test(net): drop relay_end_in_flight, covered by finite_relay since #4225 main's relay_keeps_groups_in_flight_past_the_end already fails without the publisher drain fix, on simulated time, across the same versions. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/relay_end_in_flight.rs | 167 ------------------------ 1 file changed, 167 deletions(-) delete mode 100644 rs/moq-net/tests/relay_end_in_flight.rs diff --git a/rs/moq-net/tests/relay_end_in_flight.rs b/rs/moq-net/tests/relay_end_in_flight.rs deleted file mode 100644 index 8d958e0544..0000000000 --- a/rs/moq-net/tests/relay_end_in_flight.rs +++ /dev/null @@ -1,167 +0,0 @@ -//! A track's end reaching a relay while an earlier group is still in flight upstream -//! never strands that group: the subscriber behind the relay reads it whole. -//! -//! The subscriber is reading the older group when a newer group and the track's end -//! arrive, and only then does the older group's tail follow. The relay hands out every -//! group up to the end, but must keep pulling the older group's tail for the reader -//! still on it rather than let its upstream subscription go. -//! -//! Deterministic: simulated time, and every "let the sessions send" step is an -//! idle-advance rather than a race. - -mod support; - -use std::time::Duration; - -use moq_net::track::{Info, Position, Subscription}; -use moq_net::{Hop, Timestamp, Version}; -use support::harness::{MockConnectOptions, connect_mock}; - -const TIMEOUT: Duration = Duration::from_secs(10); - -/// A cache no group outlives on the publisher. -const FOREVER: Duration = Duration::from_millis((1 << 53) - 1); - -/// The reader's budget, well inside the timeout: moq-lite-03 cannot say where a track -/// ends, so its reader waits this long for a group it cannot account for. -const BUDGET: Duration = Duration::from_secs(1); - -const VERSIONS: &[&str] = &[ - "moq-lite-03", - "moq-lite-05", - "moq-lite-06", - "moq-lite-07-wip", - "moq-transport-14", - "moq-transport-17", - "moq-transport-22", -]; - -const FRAMES: [&[u8]; 2] = [b"frame-a", b"frame-b"]; - -fn produce_origin(hop: u64) -> moq_net::origin::Producer { - let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); - support::harness::spawn(driver); - producer -} - -/// Let the drivers run until nothing is runnable: simulated time advances only once -/// every task is idle, so this returns once every queued write has been sent. -async fn settle() { - moq_net_sim::sleep(Duration::from_millis(10)).await; -} - -/// Publish two groups through a relay, the older one's tail written after the newer -/// group and the track's end. Returns each group's sequence and frame count as read behind -/// the relay and the error the read ended with, or `None` when the read hung. -async fn round(version: &str) -> Option<(Vec<(u64, usize)>, Option)> { - let version: Version = version.parse().unwrap(); - let publisher = produce_origin(1); - let broadcast = publisher.create_broadcast("bcast").unwrap(); - let track = broadcast - .create_track("video", Info::default().with_max_age(FOREVER)) - .unwrap(); - broadcast.announce(Default::default()).unwrap(); - - let relay = produce_origin(2); - let mut options = MockConnectOptions::new(version); - options.server_publish = Some(publisher.consume()); - options.client_subscribe = Some(relay.clone()); - let upstream = connect_mock(options).await; - - let subscriber = produce_origin(3); - let mut options = MockConnectOptions::new(version); - options.server_publish = Some(relay.consume()); - options.client_subscribe = Some(subscriber.clone()); - let downstream = connect_mock(options).await; - - let consumer = subscriber.consume(); - moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) - .await - .expect("announce timeout") - .expect("routed"); - let remote = moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) - .await - .expect("resolve timeout") - .expect("broadcast resolves"); - - let reader = moq_net_sim::spawn(async move { - let subscription = Subscription::default() - .with_max_delay(BUDGET) - .with_start(Position::group(0)); - let mut sub = remote - .track("video") - .unwrap() - .subscribe(subscription) - .await - .expect("subscribe"); - let mut got = Vec::new(); - loop { - let mut group = match sub.recv_group().await { - Ok(Some(group)) => group, - Ok(None) => break (got, None), - Err(err) => break (got, Some(err)), - }; - let mut frames = 0; - let err = loop { - match group.read_frame().await { - Ok(Some(_)) => frames += 1, - Ok(None) => break None, - Err(err) => break Some(err), - } - }; - got.push((group.sequence, frames)); - if err.is_some() { - break (got, err); - } - } - }); - - moq_net_sim::timeout(TIMEOUT, track.demand().used()) - .await - .expect("no subscriber appeared") - .unwrap(); - settle().await; - - // The older group's head reaches the reader. - let mut old = track.append_group().unwrap(); - old.write_frame(Timestamp::now(), FRAMES[0]).unwrap(); - settle().await; - - // A newer group and the end arrive while the older group is still open. - let mut new = track.append_group().unwrap(); - for frame in FRAMES { - new.write_frame(Timestamp::now(), frame).unwrap(); - } - new.finish().unwrap(); - track.finish().unwrap(); - settle().await; - - // Then the older group's tail. - old.write_frame(Timestamp::now(), FRAMES[1]).unwrap(); - old.finish().unwrap(); - - let result = moq_net_sim::timeout(TIMEOUT, reader) - .await - .ok() - .map(|res| res.expect("reader panicked")); - drop(( - old, new, track, upstream, downstream, relay, broadcast, publisher, subscriber, - )); - result.map(|(mut got, err)| { - got.sort(); - (got, err) - }) -} - -#[moq_net_sim::test] -async fn an_end_never_strands_a_group_in_flight_through_a_relay() { - let mut failures = Vec::new(); - for version in VERSIONS { - match round(version).await { - Some((got, None)) if got == [(0, FRAMES.len()), (1, FRAMES.len())] => {} - Some((got, err)) => failures.push(format!("{version}: got {got:?}, err {err:?}")), - None => failures.push(format!("{version}: the read behind the relay hung")), - } - } - assert!(failures.is_empty(), "{failures:#?}"); -}