From 946c67abcb508911e3e13cb754c4fb64c5d8c7df Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 07:09:01 -0700 Subject: [PATCH 1/3] fix(net): a relay resuming mid-group asks upstream for the group's head A relay whose upstream SUBSCRIBE starts mid-group (every takeover after a route change resumes there) held a copy without the group's first frames. A later reader that needed frame 0 was spliced from that copy, lagged, and parked until the group ended. A hang catalog or a compressed stats track is one long-lived group, so new viewers got no catalog and dashboards no stats, sometimes for minutes, on any edge downstream of a relay restart. No SUBSCRIBE_OK on this line reports the largest position, so ask from the head of the group instead; the requester's own start still skips the frames below it. This is the lite-06 half of #4741's widening on main. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 14 +- rs/moq-net/tests/catalog_resume_snapshot.rs | 145 ++++++++++++++++++++ 2 files changed, 155 insertions(+), 4 deletions(-) create mode 100644 rs/moq-net/tests/catalog_resume_snapshot.rs diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 41bf3ce820..8c048cb61c 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2009,10 +2009,10 @@ mod tests { assert_eq!((msg.start_frame, msg.end_frame), (0, None)); } - /// The same demand on a lite-06 peer keeps its frame offsets, so the widening is - /// version-gated rather than unconditional. + /// A lite-06 peer keeps a mid-group end, but the start widens to the head of its + /// group, so this copy can still serve frame 0 to a later reader. #[tokio::test] - async fn frame_bounds_survive_on_a_lite06_peer() { + async fn frame_bounds_start_at_the_group_head_on_a_lite06_peer() { let mut h = Harness::new(Version::Lite06); let mut sub = Sub::None; @@ -2034,7 +2034,7 @@ mod tests { lite::ControlType::Subscribe ); let msg = lite::Subscribe::decode(&mut wire, Version::Lite06).unwrap(); - assert_eq!((msg.start_frame, msg.end_frame), (3, Some(7))); + assert_eq!((msg.start_frame, msg.end_frame), (0, Some(7))); } /// The model's exclusive end maps back to the wire's inclusive pair. @@ -3231,6 +3231,12 @@ struct TrackServe { impl TrackServe { fn widen_frame_bounds(&self, subscription: &mut Subscription) { + // No SUBSCRIBE_OK here reports the largest position, so a mid-group start would + // leave this copy without the group's head: a later reader that needs frame 0 + // parks until the group ends, which a catalog or a compressed stats track never + // does. Ask from the head of the group instead; the original requester's own + // start still skips the frames below it. + subscription.start = subscription.start.map(|start| Position::group(start.group)); if self.subscriber.version.has_frame_bounds() { return; } diff --git a/rs/moq-net/tests/catalog_resume_snapshot.rs b/rs/moq-net/tests/catalog_resume_snapshot.rs new file mode 100644 index 0000000000..bc9bf3ac81 --- /dev/null +++ b/rs/moq-net/tests/catalog_resume_snapshot.rs @@ -0,0 +1,145 @@ +//! A long-lived open group (a catalog: snapshot at frame 0, deltas after) relayed to a +//! fresh reader after a mid-group resume created the relay's upstream subscription. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version, track}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TEST_TIMEOUT: Duration = Duration::from_secs(30); +/// How long the fresh reader may wait for the snapshot, in virtual time. +const SNAPSHOT_WAIT: Duration = Duration::from_secs(5); + +const VERSIONS: &[&str] = &[ + "moq-lite-05", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-17", + "moq-transport-22", +]; + +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())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Returns `Ok(first payload of the group the fresh reader got)` or `Err(reason)`. +async fn scenario(version: &str, resume_at: Option, peer_reads: bool) -> Result<(u64, Vec), String> { + let publisher = produce_origin(1); + let relay = produce_origin(2); + + let broadcast = publisher.create_broadcast("demo").unwrap(); + let track = broadcast.create_track("catalog.json", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let ts = |ms| Timestamp::from_millis(ms).unwrap(); + // The catalog: one group that never ends, a snapshot then deltas. + let mut open = track.append_group().unwrap(); + open.write_frame(ts(0), b"snapshot".as_ref()).unwrap(); + open.write_frame(ts(10), b"delta1".as_ref()).unwrap(); + open.write_frame(ts(20), b"delta2".as_ref()).unwrap(); + let sequence = open.sequence; + + let version: Version = version.parse().unwrap(); + // The relay pulls from the publisher; a mesh peer pulls from the relay. + 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 consumer = relay.consume(); + consumer.routed("demo").await.unwrap(); + let remote = consumer.request_broadcast("demo").await.unwrap(); + + // A mesh peer resumes the catalog mid-group over a session: it already holds frame 0. + let _resumed = if let Some(resume_frame) = resume_at { + let peer = produce_origin(3); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(peer.clone()); + let link = connect_mock(options).await; + + let peer_consumer = peer.consume(); + peer_consumer.routed("demo").await.unwrap(); + let peer_remote = peer_consumer.request_broadcast("demo").await.unwrap(); + let start = track::Position { + group: sequence, + frame: resume_frame, + }; + let mut sub = peer_remote + .track("catalog.json") + .unwrap() + .subscribe(track::Subscription::default().with_start(start)) + .await + .unwrap(); + // The peer only forwards demand, as a relay serving its own resumed subscriber + // does: reading its copy here would make it fetch the head through the relay. + let mut group = sub.recv_group().await.unwrap().unwrap(); + if peer_reads { + let _ = tokio::time::timeout(Duration::from_millis(500), group.read_frame()).await; + } + Some((link, peer, sub, group)) + } else { + None + }; + + // A brand-new reader wants the latest group from its snapshot. + let fresh = async { + let mut sub = remote.track("catalog.json").unwrap().subscribe(None).await.unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + let frame = group.read_frame().await; + (group.sequence, frame) + }; + match tokio::time::timeout(SNAPSHOT_WAIT, fresh).await { + Err(_) => Err("fresh reader received nothing (timed out)".into()), + Ok((seq, Ok(Some(frame)))) => Ok((seq, frame.payload.to_vec())), + Ok((seq, Ok(None))) => Err(format!("group {seq} ended empty")), + Ok((seq, Err(err))) => Err(format!("group {seq} errored: {err}")), + } +} + +async fn run(resume_at: Option, peer_reads: bool) { + let mut failures = Vec::new(); + for version in VERSIONS { + let result = tokio::time::timeout(TEST_TIMEOUT, scenario(version, resume_at, peer_reads)) + .await + .unwrap_or_else(|_| Err("scenario timed out".into())); + match result { + Ok((0, payload)) if payload == b"snapshot" => {} + Ok((seq, payload)) => failures.push(format!( + "{version}: got group {seq} starting with {:?}", + String::from_utf8_lossy(&payload) + )), + Err(err) => failures.push(format!("{version}: {err}")), + } + } + assert!(failures.is_empty(), "\n{}", failures.join("\n")); +} + +/// The regression: a resume at (G, 1) must not keep a fresh reader from G's snapshot. +#[tokio::test(start_paused = true)] +async fn fresh_reader_gets_snapshot_after_mid_group_resume() { + run(Some(1), false).await; +} + +/// Control: with no prior resume the fresh reader gets the snapshot. +#[tokio::test(start_paused = true)] +async fn fresh_reader_gets_snapshot_without_resume() { + run(None, false).await; +} + +/// Control: the same resume from the head of the group, (G, 0). +#[tokio::test(start_paused = true)] +async fn fresh_reader_gets_snapshot_after_head_resume() { + run(Some(0), false).await; +} + +/// Variant: the peer also reads its own copy before the fresh reader arrives, so the +/// relay serves the peer's read and the fresh reader from the same upstream copy. +#[tokio::test(start_paused = true)] +async fn fresh_reader_gets_snapshot_after_peer_fetches_head() { + run(Some(1), true).await; +} From a152fe5c1449bc38447216cb360b42c16e1390a9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 08:30:53 -0700 Subject: [PATCH 2/3] fix(net): widen only frame-bounded starts to the head; pin the peer's own read Moves the head widening into the frame-bounds branch so older peers keep their existing rounding and debug log, and asserts the resumed peer reads its group from the snapshot (it lagged on lite-06/07 before the fix). Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 12 ++++++------ rs/moq-net/tests/catalog_resume_snapshot.rs | 12 +++++++++--- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 8c048cb61c..08f3d2293c 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3231,13 +3231,13 @@ struct TrackServe { impl TrackServe { fn widen_frame_bounds(&self, subscription: &mut Subscription) { - // No SUBSCRIBE_OK here reports the largest position, so a mid-group start would - // leave this copy without the group's head: a later reader that needs frame 0 - // parks until the group ends, which a catalog or a compressed stats track never - // does. Ask from the head of the group instead; the original requester's own - // start still skips the frames below it. - subscription.start = subscription.start.map(|start| Position::group(start.group)); if self.subscriber.version.has_frame_bounds() { + // No SUBSCRIBE_OK here reports the largest position, so a mid-group start + // would leave this copy without the group's head: a later reader that needs + // frame 0 parks until the group ends, which a catalog or a compressed stats + // track never does. Ask from the head of the group instead; the original + // requester's own start still skips the frames below it. + subscription.start = subscription.start.map(|start| Position::group(start.group)); return; } diff --git a/rs/moq-net/tests/catalog_resume_snapshot.rs b/rs/moq-net/tests/catalog_resume_snapshot.rs index bc9bf3ac81..0a9c016179 100644 --- a/rs/moq-net/tests/catalog_resume_snapshot.rs +++ b/rs/moq-net/tests/catalog_resume_snapshot.rs @@ -79,7 +79,13 @@ async fn scenario(version: &str, resume_at: Option, peer_reads: bool) -> Re // does: reading its copy here would make it fetch the head through the relay. let mut group = sub.recv_group().await.unwrap().unwrap(); if peer_reads { - let _ = tokio::time::timeout(Duration::from_millis(500), group.read_frame()).await; + // A local group reader starts at the head, so the peer's own copy must be able + // to serve it too, not just the relay's. + let read = tokio::time::timeout(Duration::from_millis(500), group.read_frame()).await; + match read { + Ok(Ok(Some(frame))) if frame.payload.as_ref() == b"snapshot" => {} + other => return Err(format!("peer's first read was {other:?}, not the snapshot")), + } } Some((link, peer, sub, group)) } else { @@ -137,8 +143,8 @@ async fn fresh_reader_gets_snapshot_after_head_resume() { run(Some(0), false).await; } -/// Variant: the peer also reads its own copy before the fresh reader arrives, so the -/// relay serves the peer's read and the fresh reader from the same upstream copy. +/// Variant: the peer also reads its own copy from the head before the fresh reader +/// arrives, so both the peer's copy and the relay's must hold frame 0. #[tokio::test(start_paused = true)] async fn fresh_reader_gets_snapshot_after_peer_fetches_head() { run(Some(1), true).await; From 606bd04012b3529271bf767cd9a53117c37df06a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 5 Oct 2026 08:58:40 -0700 Subject: [PATCH 3/3] test(net): cover a frame-precise resume arriving at the relay itself A peer running the fix widens its own start, so the relay's widening was never exercised. A local subscriber resuming at (G, 1) on the relay stands in for a not-yet-upgraded downstream peer; it lags on lite-06/07 without the fix. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 4 +- rs/moq-net/tests/catalog_resume_snapshot.rs | 51 +++++++++++++++++---- 2 files changed, 45 insertions(+), 10 deletions(-) diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 08f3d2293c..92b01d1423 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3235,8 +3235,8 @@ impl TrackServe { // No SUBSCRIBE_OK here reports the largest position, so a mid-group start // would leave this copy without the group's head: a later reader that needs // frame 0 parks until the group ends, which a catalog or a compressed stats - // track never does. Ask from the head of the group instead; the original - // requester's own start still skips the frames below it. + // track never does. Ask from the head of the group instead; a downstream + // session's own start still skips the frames below it. subscription.start = subscription.start.map(|start| Position::group(start.group)); return; } diff --git a/rs/moq-net/tests/catalog_resume_snapshot.rs b/rs/moq-net/tests/catalog_resume_snapshot.rs index 0a9c016179..9b01e49d11 100644 --- a/rs/moq-net/tests/catalog_resume_snapshot.rs +++ b/rs/moq-net/tests/catalog_resume_snapshot.rs @@ -27,7 +27,7 @@ fn produce_origin(hop: u64) -> moq_net::origin::Producer { } /// Returns `Ok(first payload of the group the fresh reader got)` or `Err(reason)`. -async fn scenario(version: &str, resume_at: Option, peer_reads: bool) -> Result<(u64, Vec), String> { +async fn scenario(version: &str, resume: Resume, peer_reads: bool) -> Result<(u64, Vec), String> { let publisher = produce_origin(1); let relay = produce_origin(2); @@ -54,8 +54,27 @@ async fn scenario(version: &str, resume_at: Option, peer_reads: bool) -> Re consumer.routed("demo").await.unwrap(); let remote = consumer.request_broadcast("demo").await.unwrap(); + // A frame-precise resume on the relay itself, as a downstream peer without the head + // widening sends on the wire: the relay's own upstream SUBSCRIBE must widen it. + let _local = if let Resume::Local(resume_frame) = resume { + let start = track::Position { + group: sequence, + frame: resume_frame, + }; + let mut sub = remote + .track("catalog.json") + .unwrap() + .subscribe(track::Subscription::default().with_start(start)) + .await + .unwrap(); + let group = sub.recv_group().await.unwrap().unwrap(); + Some((sub, group)) + } else { + None + }; + // A mesh peer resumes the catalog mid-group over a session: it already holds frame 0. - let _resumed = if let Some(resume_frame) = resume_at { + let _resumed = if let Resume::Peer(resume_frame) = resume { let peer = produce_origin(3); let mut options = MockConnectOptions::new(version); options.server_publish = Some(relay.consume()); @@ -107,10 +126,20 @@ async fn scenario(version: &str, resume_at: Option, peer_reads: bool) -> Re } } -async fn run(resume_at: Option, peer_reads: bool) { +/// Where the mid-group resume comes from, if any. +#[derive(Clone, Copy)] +enum Resume { + None, + /// A mesh peer over a session, which widens its own upstream start. + Peer(u64), + /// A subscriber on the relay itself, so only the relay's widening can help. + Local(u64), +} + +async fn run(resume: Resume, peer_reads: bool) { let mut failures = Vec::new(); for version in VERSIONS { - let result = tokio::time::timeout(TEST_TIMEOUT, scenario(version, resume_at, peer_reads)) + let result = tokio::time::timeout(TEST_TIMEOUT, scenario(version, resume, peer_reads)) .await .unwrap_or_else(|_| Err("scenario timed out".into())); match result { @@ -128,24 +157,30 @@ async fn run(resume_at: Option, peer_reads: bool) { /// The regression: a resume at (G, 1) must not keep a fresh reader from G's snapshot. #[tokio::test(start_paused = true)] async fn fresh_reader_gets_snapshot_after_mid_group_resume() { - run(Some(1), false).await; + run(Resume::Peer(1), false).await; +} + +/// The same resume arriving frame-precise at the relay, as from a not-yet-upgraded peer. +#[tokio::test(start_paused = true)] +async fn fresh_reader_gets_snapshot_after_local_mid_group_resume() { + run(Resume::Local(1), false).await; } /// Control: with no prior resume the fresh reader gets the snapshot. #[tokio::test(start_paused = true)] async fn fresh_reader_gets_snapshot_without_resume() { - run(None, false).await; + run(Resume::None, false).await; } /// Control: the same resume from the head of the group, (G, 0). #[tokio::test(start_paused = true)] async fn fresh_reader_gets_snapshot_after_head_resume() { - run(Some(0), false).await; + run(Resume::Peer(0), false).await; } /// Variant: the peer also reads its own copy from the head before the fresh reader /// arrives, so both the peer's copy and the relay's must hold frame 0. #[tokio::test(start_paused = true)] async fn fresh_reader_gets_snapshot_after_peer_fetches_head() { - run(Some(1), true).await; + run(Resume::Peer(1), true).await; }