From 51bb5325b3069584c6fdb601af61e7e27e59c0be Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:05:16 -0700 Subject: [PATCH 1/4] chore: claim subscribe-ok-start quest From 93cd6f215ae343f7ec98ae935b62aaf945e4b955 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:09:09 -0700 Subject: [PATCH 2/4] fix(net): resolve subscribe starts past skipped ranges Co-Authored-By: GPT-6.1-Sol --- quest/m1/README.md | 1 - quest/m1/archive/tracks.md | 1 - quest/m1/subscribe-ok-start.md | 34 ----------- rs/moq-net/benches/session.rs | 34 ++++++++++- rs/moq-net/src/lite/publisher.rs | 99 +++++++++++++++++++++++++++----- 5 files changed, 117 insertions(+), 52 deletions(-) delete mode 100644 quest/m1/subscribe-ok-start.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 3c6a0aa7325..af0f84394b8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -104,7 +104,6 @@ lands before the cut. - [Timed-out requests clean up](/quest/m1/legacy-request-timeout-cleanup.md) - on drafts 14-16, Rust and JS send PUBLISH_NAMESPACE_DONE or UNSUBSCRIBE_NAMESPACE for a timed-out request before its local reset - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails -- [SUBSCRIBE_OK names where delivery starts](/quest/m1/subscribe-ok-start.md) - a warm relay names the same start group as a cold one instead of groups it skipped as stale - [Rust resumed reads](/quest/m1/resume-ordered.md) - a Rust reader takes the lowest group across route copies in both orders, as JS does - [Resumed close](/quest/m1/resume-closed.md) - a resumed JS track keeps its declared end across copies, and its `closed` and `largest()` settle on a clean end - [Resumed in-flight groups](/quest/m1/resume-in-flight.md) - a replaced route copy stays while a group below its bound is in flight or it backs a handed-out group late, and no longer than the reader's budget diff --git a/quest/m1/archive/tracks.md b/quest/m1/archive/tracks.md index df0264ee119..ea9339efad0 100644 --- a/quest/m1/archive/tracks.md +++ b/quest/m1/archive/tracks.md @@ -207,5 +207,4 @@ answer it. Wire: a new lite-07-wip stream. ## Related - [SUBSCRIBE_TRACKS](/quest/m1/archive/subscribe-tracks.md) - moq-transport answers and learns listings through this hook -- [SUBSCRIBE_OK names where delivery starts](/quest/m1/subscribe-ok-start.md) - a warm relay answers like a cold one, found while building SUBSCRIBE_ALL - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - the subscription shape each per-track SUBSCRIBE uses diff --git a/quest/m1/subscribe-ok-start.md b/quest/m1/subscribe-ok-start.md deleted file mode 100644 index fcb998fb77e..00000000000 --- a/quest/m1/subscribe-ok-start.md +++ /dev/null @@ -1,34 +0,0 @@ -# [S] SUBSCRIBE_OK names where delivery starts - -## Goal - -SUBSCRIBE_OK's `Group` is the first group the publisher actually delivers, -as the lite draft defines it, so a warm relay and a cold one answer the same -subscription the same way. Today a publisher whose read cursor gives up -cached groups as stale before the start resolves still names the cache's -oldest group, claiming groups it never sends. Only a group skipped for a -missing head is excluded now. - -## Plan - -Found while building SUBSCRIBE_ALL -([moq-dev/moq#5931](https://github.com/moq-dev/moq/pull/5931), commit -`f074d09e7`), which fixed it in `TrackRun` (`rs/moq-net/src/lite/publisher.rs`) -and covered it with `the_start_names_past_the_stale_groups_it_skips`. That -PR is being replaced, so port the fix and its test on their own: record -every range the cursor gives up before the start resolves (a missing head, -stale, or dropped upstream), and resolve the start past them. Check whether -the moq-transport publisher's SUBSCRIBE_OK and the JS publisher have the same -gap, and fix them here if so. - -Add a relay test on mock time: the same SUBSCRIBE through a cold relay and a -warm one whose cache holds groups older than the subscriber's max age gets -the same `Group`. - -Public API: none. Wire: no encoding change; SUBSCRIBE_OK's `Group` changes -behavior on every lite version with a resolved start (lite-05 on) to match -its definition. - -## Related - -- [TRACKS](/quest/m1/archive/tracks.md) - the line of work that surfaced it diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 9edebb67b78..f0a98b9afbb 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -105,6 +105,8 @@ struct Shape { /// Whole groups each watched broadcast writes per round, so a serve loop finds a /// backlog ready: the group rate a round stands in for. burst: usize, + /// Cached groups per track before a join, whose older groups the live budget skips. + history: usize, /// Bytes queued before a mock stream applies backpressure. window: usize, /// Read each group via a fresh FETCH rather than a live subscription. @@ -123,6 +125,7 @@ impl Shape { frame: 64, paced: false, burst: 1, + history: 1, window: 1024 * 1024, fetch: false, }; @@ -149,7 +152,7 @@ impl Shape { fn id(&self, version: &str) -> String { format!( - "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}{}{}{}", + "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}{}{}{}{}{}", self.relays, self.publishers, self.broadcasts, @@ -172,6 +175,11 @@ impl Shape { String::new() }, if self.fetch { "/fetch" } else { "" }, + if self.history > 1 { + format!("/history={}", self.history) + } else { + String::new() + }, ) } } @@ -725,7 +733,15 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) let cluster = Cluster::new(version, *shape).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { - write_group(track, &payload, FRAMES); + for sequence in 0..shape.history { + let mut group = track.append_group().unwrap(); + for _ in 0..FRAMES { + group + .write_frame(Timestamp::from_millis(sequence as u64).unwrap(), payload.clone()) + .unwrap(); + } + group.finish().unwrap(); + } } cluster }) @@ -1227,6 +1243,20 @@ fn session(c: &mut Criterion) { }) }), ); + // A live join skips a cached prefix. Sweep publishers against that prefix so + // resolving one start cannot hide a scan of unrelated publishers or group IDs. + join( + c, + "stale", + [1, 8, 32].into_iter().flat_map(|publishers| { + [1, 64, 1024].map(move |history| Shape { + publishers, + history, + ..base + }) + }), + ); + join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); churn(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 })); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 7a524912caa..c482e33ccb7 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1,7 +1,7 @@ use crate::{SessionError, announce, frame, group, origin, track}; use rangemap::RangeSet; use std::{ - collections::{BTreeSet, HashMap}, + collections::HashMap, ops::{Bound, ControlFlow, Range}, sync::{ Arc, Mutex, @@ -3328,8 +3328,8 @@ struct TrackRun { start: Option, // The first servable group, held until the source resolves where its feed starts. first: Option, - // Groups skipped for a missing head before the start resolved, which it must not name. - skipped: BTreeSet, + // Groups given up before the start resolved, which it must not name. + skipped: RangeSet, end_sent: bool, // Lite07+ sends SUBSCRIBE_END with the stream count instead of as soon as the // boundary is known, once every group below it has opened its stream. @@ -3369,7 +3369,7 @@ impl TrackRun { emit_range, start: None, first: None, - skipped: BTreeSet::new(), + skipped: RangeSet::new(), end_sent: false, count_streams, served, @@ -3543,7 +3543,7 @@ impl TrackRun { // only be reset; the next servable group resolves the start. tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence = group.sequence, "skipping group with a missing head"); if self.emit_range && self.start.is_none() { - self.skipped.insert(group.sequence); + self.skipped.insert(group.sequence..group.sequence.saturating_add(1)); } else if self.start.is_some() && let Some(end) = group.sequence.checked_add(1) { @@ -3593,21 +3593,24 @@ impl TrackRun { Poll::Pending } - /// Lite-05/06 name a sequence as soon as this cursor gives it up; whether one went out. + /// Lite-05/06 name a sequence as soon as this cursor gives it up; whether one was taken. /// Before SUBSCRIBE_START there is no owed range yet, so the eventual start accounts - /// for it instead. + /// for it instead, including ranges skipped while finding the first servable group. fn poll_forward_drop( &mut self, writer: &mut Writer, waiter: &kio::Waiter, ) -> Result { - if self.start.is_none() || self.served.is_none() { + if !self.emit_range || (self.start.is_some() && self.served.is_none()) { return Ok(false); } let Poll::Ready(range) = self.track.poll_drop(waiter) else { return Ok(false); }; - self.drop_range(range, writer)?; + match self.start { + None => self.skipped.insert(range), + Some(_) => self.drop_range(range, writer)?, + } Ok(true) } @@ -3684,12 +3687,12 @@ impl TrackRun { writer: &mut Writer, ) -> Result<(), Error> { let mut start = source.map_or(group.sequence, |source| source.min(group.sequence)); - // A skipped group is never served, so it cannot be where the feed starts. This stops - // at the held group at the latest, since it was not skipped. - while self.skipped.contains(&start) { - start += 1; + // A group given up cannot be where the feed starts. This stops at the held group + // at the latest, since it was not skipped. + if let Some(skipped) = self.skipped.get(&start) { + start = skipped.end.min(group.sequence); } - self.skipped.clear(); + self.skipped = RangeSet::new(); self.start = Some(start); // Only the group: the subscriber derives the start frame from its own request // (see `lite::SubscribeStart`). The track is live by now, since its first group @@ -5213,6 +5216,74 @@ mod serve_group_test { assert_eq!(drops(&mut writer, &log, version).await, [(1, 1)]); } + /// A relay's retained stale groups must not change the start it answers downstream. + #[moq_clock::test] + async fn the_start_names_past_the_stale_groups_it_skips() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut starts = Vec::new(); + for warm in [false, true] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(if warm { 0 } else { 2 }).unwrap(); + for sequence in (if warm { 0 } else { 2 })..3 { + write_group(&mut track, sequence, sequence * 10); + } + moq_clock::sim::advance(Duration::from_secs(1)).await; + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + kio::wait(|waiter| writer.poll_flush(&mut waiter.context())) + .await + .unwrap(); + let writes = log.writes.lock().unwrap().clone(); + let (msg, _) = lite::SubscribeResponse::decode_slice(&writes, version).unwrap(); + let lite::SubscribeResponse::Start(start) = msg else { + panic!("expected SUBSCRIBE_START, got {msg:?}"); + }; + starts.push(start.group); + } + assert_eq!(starts, [2, 2], "{version}: cold and warm relay starts differ"); + } + } + + /// Upstream drops received before the start exclude the entire range, even on lite-07. + #[moq_clock::test] + async fn the_start_names_past_the_upstream_dropped_range() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(0).unwrap(); + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + track.drop_groups(0..1_000_000); + write_group(&mut track, 1_000_000, 0); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + kio::wait(|waiter| writer.poll_flush(&mut waiter.context())) + .await + .unwrap(); + let writes = log.writes.lock().unwrap().clone(); + let (msg, _) = lite::SubscribeResponse::decode_slice(&writes, version).unwrap(); + let lite::SubscribeResponse::Start(start) = msg else { + panic!("expected SUBSCRIBE_START, got {msg:?}"); + }; + assert_eq!(start.group, 1_000_000, "{version}: named a dropped group"); + } + } + /// A SUBSCRIBE_UPDATE that lowers the start below SUBSCRIBE_START asks for those groups /// too, so the drops at the end count from there. #[moq_clock::test] From 0901088b7d01af16de9dcc3a4b35e13de9f39c3d Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:21:45 -0700 Subject: [PATCH 3/4] test(net): cover relay starts and maximum missing heads Co-Authored-By: GPT-6.1-Sol --- rs/moq-net/src/lite/publisher.rs | 125 ++++++++++++++++++++++++++++++- 1 file changed, 124 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index c482e33ccb7..451cf59d443 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3543,7 +3543,10 @@ impl TrackRun { // only be reset; the next servable group resolves the start. tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence = group.sequence, "skipping group with a missing head"); if self.emit_range && self.start.is_none() { - self.skipped.insert(group.sequence..group.sequence.saturating_add(1)); + // At u64::MAX no later group can resolve a start past this one. + if let Some(end) = group.sequence.checked_add(1) { + self.skipped.insert(group.sequence..end); + } } else if self.start.is_some() && let Some(end) = group.sequence.checked_add(1) { @@ -5216,6 +5219,126 @@ mod serve_group_test { assert_eq!(drops(&mut writer, &log, version).await, [(1, 1)]); } + /// A missing head at the model's largest sequence has no successor to name. + #[moq_clock::test] + async fn a_missing_head_at_the_maximum_sequence_does_not_panic() { + let track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(60))); + let mut group = track.create_group(group::Info { sequence: u64::MAX }).unwrap(); + group.start_at(1).unwrap(); + group.write_frame(Timestamp::ZERO, b"x".as_slice()).unwrap(); + group.finish().unwrap(); + let (mut run, mut writer, _) = lite_run(Version::Lite07, SinkSession::new(Log::default()), subscriber); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + } + + /// A long-budget reader warms the relay without changing a live reader's resolved start. + #[moq_clock::test] + async fn cold_and_warm_relays_resolve_the_same_start() { + fn drive(mut driver: D) { + drop(moq_clock::sim::spawn(moq_clock::drive(move |now, waiter| { + crate::time::Driver::poll(&mut driver, now, waiter) + }))); + } + + fn origin(hop: u64) -> origin::Producer { + let (origin, driver) = origin::Producer::new(origin::Config::new(crate::Hop::new(hop).unwrap())); + drive(driver); + origin + } + + async fn connect( + version: Version, + publisher: &origin::Producer, + subscriber: &origin::Producer, + ) -> (crate::Session, crate::Session) { + let version = crate::Version::Lite(version); + let config = crate::transport::memory::Config { + protocol: Some(version.alpn().into()), + ..Default::default() + }; + let (client_transport, server_transport) = crate::transport::memory::pair(config); + let client = crate::Client::new() + .with_versions(version.into()) + .with_subscriber(subscriber.clone()); + let server = crate::Server::new() + .with_versions(version.into()) + .with_publisher(publisher); + futures::join!( + async { + let (session, driver) = client.connect(moq_clock::now(), client_transport).await.unwrap(); + drive(driver); + session + }, + async { + let (session, driver) = server.accept(moq_clock::now(), server_transport).await.unwrap(); + drive(driver); + session + } + ) + } + + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut starts = Vec::new(); + for warm in [false, true] { + let publisher = origin(1); + let broadcast = publisher.create_broadcast("room").unwrap(); + let mut track = broadcast.create_track("test", crate::track::Info::timed()).unwrap(); + track.start_at(0).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let relay = origin(2); + let _upstream = connect(version, &publisher, &relay).await; + let remote = relay.consume().routed_broadcast("room").await.unwrap(); + let mut keeper = match warm { + true => Some( + remote + .track("test") + .unwrap() + .subscribe( + track::Subscription::default() + .with_groups(0..) + .with_max_delay(Duration::from_secs(60)), + ) + .await + .unwrap(), + ), + false => None, + }; + for sequence in 0..3 { + write_group(&mut track, sequence, sequence * 10); + if let Some(keeper) = &mut keeper { + assert_eq!(keeper.recv_group().await.unwrap().unwrap().sequence, sequence); + } + } + moq_clock::sim::advance(Duration::from_secs(1)).await; + let subscriber = origin(3); + let _downstream = connect(version, &relay, &subscriber).await; + let remote = subscriber.consume().routed_broadcast("room").await.unwrap(); + let mut live = remote + .track("test") + .unwrap() + .subscribe(track::Subscription::default().with_groups(0..)) + .await + .unwrap(); + assert_eq!(live.recv_group().await.unwrap().unwrap().sequence, 2); + starts.push(kio::wait(|waiter| live.poll_start(waiter)).await); + } + assert_eq!( + starts, + [Some(2), Some(2)], + "{version}: cold and warm relay starts differ" + ); + } + } + /// A relay's retained stale groups must not change the start it answers downstream. #[moq_clock::test] async fn the_start_names_past_the_stale_groups_it_skips() { From 127f92e1dd61424d632e2c53b607ec7017e81e3c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 19 Oct 2026 22:51:48 -0700 Subject: [PATCH 4/4] fix(net): drop ranges skipped above a resolved start Ranges given up before SUBSCRIBE_START but above the resolved start were cleared, so lite-05+ subscribers waited until the end sweep for them. Forward them as SUBSCRIBE_DROP right after the start. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 32 +++++++++++++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 4896b28b463..1b61c27b5fa 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3685,7 +3685,7 @@ impl TrackRun { if let Some(skipped) = self.skipped.get(&start) { start = skipped.end.min(group.sequence); } - self.skipped = RangeSet::new(); + let skipped = std::mem::take(&mut self.skipped); self.start = Some(start); // Only the group: the subscriber derives the start frame from its own request // (see `lite::SubscribeStart`). The track is live by now, since its first group @@ -3700,6 +3700,11 @@ impl TrackRun { group: start, largest, }))?; + // The subscriber now owes itself everything from the start, so a range given up + // above it is named as it would have been after the start. + for range in skipped.iter() { + self.drop_range(range.start.max(start)..range.end, writer)?; + } // SUBSCRIBE_START names where delivery starts. Groups already skipped between // the requested floor and this group are not served. A later group at or above an // explicit floor, still inside the subscriber's max age, is delivered, so the cursor @@ -5412,6 +5417,31 @@ mod serve_group_test { } } + /// A range given up before the start resolves, but above it, is still named promptly. + #[moq_clock::test] + async fn a_range_skipped_above_the_start_is_dropped() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(0).unwrap(); + let subscriber = track.subscribe(track::Subscription::default()); + let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); + // The newer group beats the older ones, so the start resolves at the source's 0. + track.drop_groups(2..4); + write_group(&mut track, 5, 0); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + assert_eq!(run.start, Some(0), "{version}"); + assert_eq!(drops(&mut writer, &log, version).await, [(2, 3)], "{version}"); + } + } + /// A SUBSCRIBE_UPDATE that lowers the start below SUBSCRIBE_START asks for those groups /// too, so the drops at the end count from there. #[moq_clock::test]