From 5bf274ac879e7a76868df94ccf6975f88f0a5dc2 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:03:41 -0800 Subject: [PATCH 1/3] chore(quest): claim quest/m2/subscribe-ranges/start-bookkeeping.md Quest-Claim: c94f4f6d69f9f561 From 0a797a17a77ab6d17fa8813697717abc0ffaa153 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 12 Nov 2026 19:54:40 -0800 Subject: [PATCH 2/3] fix(moq-net): one lite start state, so a start past the live edge names its held drops Co-Authored-By: Claude Opus 5.5 --- quest/m2/lite07-finalize.md | 1 - .../m2/subscribe-ranges/start-bookkeeping.md | 61 ---- quest/m2/time-addressing/model.md | 6 +- rs/moq-net/src/lite/publisher.rs | 295 ++++++++++++------ 4 files changed, 194 insertions(+), 169 deletions(-) delete mode 100644 quest/m2/subscribe-ranges/start-bookkeeping.md diff --git a/quest/m2/lite07-finalize.md b/quest/m2/lite07-finalize.md index f4942d04950..949dd746c34 100644 --- a/quest/m2/lite07-finalize.md +++ b/quest/m2/lite07-finalize.md @@ -130,7 +130,6 @@ transcript freezes at the cut and a flaw in it can only be fixed on the wire. - [Lite-07 by start](/quest/m2/time-addressing/lite07.md) - lite-07 carries group starts, previous-start links, GROUP_END, and time bounds instead of sequence numbers - [@moq/net by start](/quest/m2/time-addressing/js.md) - JS speaks the same lite-07 wire - [A group a subscriber gives up on is always reported](/quest/m2/group-drops.md) - may add SUBSCRIBE_DROP for a stopped group; leaves this list if it takes the no-wire path -- [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md) - may change START per update on lite-07 - [Prove cache-aware routing](/quest/m2/cache-routing-proof.md) - rewrites lite-07's one-hop warm routes and `Serves` for routes that propagate ## Related diff --git a/quest/m2/subscribe-ranges/start-bookkeeping.md b/quest/m2/subscribe-ranges/start-bookkeeping.md deleted file mode 100644 index 04c7000872e..00000000000 --- a/quest/m2/subscribe-ranges/start-bookkeeping.md +++ /dev/null @@ -1,61 +0,0 @@ -# [S] Delete the old start bookkeeping - -## Goal - -Where a lite subscription starts lives in one place per version: lite -`TrackRun`'s `start` and `first`, the pre-06 re-pin branch, and -the per-version start branches collapse into one start state that only -translates into SUBSCRIBE_START per version, and on lite-07 the -`requested == start` gate is gone, since a fresh START per update makes it -unnecessary. What a cursor is owed or was served is out of scope: it stays -with Drop set ([moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476)). - -## Plan - -Decided 2026-11-05 (maintainer, /quest-plan): narrow this quest to the start state and wait for #7670, as the -agent that declined it on 2026-11-05 recommended. #7670 rewrites the same -publisher state, Drop set already promises to delete the served-groups -record (`skipped`, `served`, `Delivered`), and -[moq-dev/moq#7872](https://github.com/moq-dev/moq/pull/7872)'s -SUBSCRIBE_STARTED makes the lite-07 start gate removable. Not chosen: -folding it into Drop set, or doing it now on main. - -Decided 2026-11-05 (maintainer, /quest-audit): ✅ the wait on -[moq-dev/moq#7670](https://github.com/moq-dev/moq/pull/7670), Drop set's own -PR, becomes a `Required` link to Drop set ([moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476)), since -#7670 rewrites the same lite publisher state (about 1,250 lines of -`lite/publisher.rs`). - - -Moved out of the [Subscribe ranges](/quest/m2/subscribe-ranges/README.md) -README's own work, where the 2026-10-24 pre-cut audit put it. Decided -2026-10-27 (maintainer): a child quest of its own, since -[Rust groups by start](/quest/m2/time-addressing/model.md) requires it before -the cut moves this state onto times, while the rest of the README waits on -every child. - -Lite `TrackRun`'s fixes cluster on where a subscription starts and what it -is owed or dropped (e713c30991, 36f82ac450, 35fa44f6bf, e1b5d2741f, -f235fcfcbd, df0bd5c36f, 342445f251, 06765f4c1a, 4927167904, f52848d634, -0513c07369, 5ad3652871). That state spans `TrackRun`'s `floor`, `cap`, -`start`, `first`, `skipped`, and `served` (frame bounds moved to `Frames`; -`start_frame` remains only on lite FETCH, `lite/fetch.rs`), plus -`cursor_floor` and the pre-06 re-pin branch in `TrackRun::update`, encoding -three wire generations of "start". - -- Lite-07 is unpublished, so sending a fresh START per update can remove the - `active.start == active.requested` gate (`lite/subscriber.rs`, - `apply_response`). - -- The start state must make it impossible to resolve without settling what - was given up before the start. Today there are two paths. `resolve()` - drains `skipped`, while the shortcut that answers a start past the live - edge at once (the "start past everything" branch in `TrackRun`'s poll) only - sets `self.start`. So any drop collected while unresolved is never sent, - and the subscriber waits out its grace for those groups. Reproduce this - first. Found by the 2026-11-07 audit of implicit state machines. - Recommended: `enum Start { Legacy, Unresolved { skipped, first }, - Resolved { start, served } }`, with `skipped` moved out at the transition. - -Public API: none expected. Wire: lite-07 (unpublished) may send a START per -update; published versions are unchanged. diff --git a/quest/m2/time-addressing/model.md b/quest/m2/time-addressing/model.md index 0fb6e0db4f4..43a8d3bb0f1 100644 --- a/quest/m2/time-addressing/model.md +++ b/quest/m2/time-addressing/model.md @@ -56,7 +56,7 @@ a producer would need per-track memory of its last number. subscribe-ranges children that put sequence ranges into the model and the wire, Lite-07 ranges ([moq-dev/moq#7872](https://github.com/moq-dev/moq/pull/7872), merged 2026-11-05) and - [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md), + Delete the old start bookkeeping (finished in [moq-dev/moq#8613](https://github.com/moq-dev/moq/pull/8613)), not the whole epic. Drop set ([moq-dev/moq#8476](https://github.com/moq-dev/moq/pull/8476)) lands first too: its runs keyed by group sequence convert mechanically to starts. - **Old relays** (decided 2026-10-26, maintainer): a released relay fills a @@ -216,7 +216,3 @@ timestamp, untimed tracks become scale-less, `Timestamp::now` is gone, and `Timed` is moq-json's, and `Anchor` is `{Oldest, Relative, Time}` with inclusive ends. Wire: no format change; legacy and lite-07-wip group IDs become start ticks. - -## Required - -- [Delete the old start bookkeeping](/quest/m2/subscribe-ranges/start-bookkeeping.md) - the lite cursor state lives in the model before the cut rekeys it diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 5887a4ffeca..b70e06b7909 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -5092,6 +5092,50 @@ impl Subscription { /// A subscription's run loop: one subscriber cursor serving groups and SUBSCRIBE_UPDATE /// messages, with a machine per group it serves, one at a time /// waiting for stream credit. +/// Where a subscription's delivery starts, which SUBSCRIBE_START reports once. +enum Start { + /// Lite-03/04 send no SUBSCRIBE_START, and so no drops either. + Legacy, + /// Not yet reported. + Unresolved { + /// The first servable group, held until the source resolves where its feed starts. + first: Option, + /// Groups given up meanwhile, with why, named once the start is. + skipped: RangeMap, + }, + /// Reported: delivery starts at this group. + Resolved(u64), +} + +impl Start { + /// Whether the Subscribe Stream reports the range (SUBSCRIBE_START, DROP, FINISH). + fn emits_range(&self) -> bool { + !matches!(self, Self::Legacy) + } + + fn is_unresolved(&self) -> bool { + matches!(self, Self::Unresolved { .. }) + } + + /// The resolved start, if any. + #[cfg(test)] + fn resolved(&self) -> Option { + match self { + Self::Resolved(start) => Some(*start), + _ => None, + } + } + + /// The groups held back for the start. + #[cfg(test)] + fn skipped(&self) -> Option<&RangeMap> { + match self { + Self::Unresolved { skipped, .. } => Some(skipped), + _ => None, + } + } +} + struct TrackRun { ctx: Subscription, track: track::Subscriber, @@ -5108,16 +5152,8 @@ struct TrackRun { // A SUBSCRIBE_UPDATE this track cannot serve, such as a time anchor on a track the // peer was told is untimed, which ends the subscription. refused: Option, - // Lite05+ resolves the range on the Subscribe Stream itself: SUBSCRIBE_START - // once the first group is known, SUBSCRIBE_FINISH as soon as the track declares its - // exclusive final sequence (which may be ahead of the live edge). - emit_range: bool, - // Where SUBSCRIBE_START resolved the feed, once sent. - start: Option, - // The first servable group, held until the source resolves where its feed starts. - first: Option, - // Groups given up before the start resolved, which it must not name, with why. - skipped: RangeMap, + // Where delivery starts, which SUBSCRIBE_START reports. + start: Start, // The number of the SUBSCRIBE (0) or latest SUBSCRIBE_UPDATE, which a drop echoes. update: u64, // The expired runs an update owes a drop again, sent as ranges on the next poll. Lite-07 @@ -5176,7 +5212,13 @@ impl TrackRun { let track = track.with_range((Bound::Included(floor), cap)).parked(); // Lite03 and Lite04 declare no end, so neither side could settle on a drop. - let emit_range = ctx.version.has_track_stream(); + let start = match ctx.version.has_track_stream() { + true => Start::Unresolved { + first: None, + skipped: RangeMap::new(), + }, + false => Start::Legacy, + }; let bounded = bounds.ends(ctx.version); let initial = bounds.ranges.len() as u64; @@ -5193,10 +5235,7 @@ impl TrackRun { floor, cap, steer_tx, - emit_range, - start: None, - first: None, - skipped: RangeMap::new(), + start, update: 0, redrop: RangeSet::new(), reoffered: Default::default(), @@ -5310,11 +5349,11 @@ impl TrackRun { if let Some(start) = floor { self.floor = start; // Before SUBSCRIBE_START nothing is owed yet; the start accounts for it. - if let Some(resolved) = self.start { + if let Start::Resolved(resolved) = &mut self.start { // A lowered floor owes the groups it newly asks for, so the end sweep covers them. // A raised one sends no early drop: the subscriber already stops owing below its // floor, and a later update may lower it again to reopen those groups. - self.start = Some(resolved.min(start)); + *resolved = (*resolved).min(start); } } self.cap = end.map_or(Bound::Unbounded, Bound::Excluded); @@ -5402,7 +5441,7 @@ impl TrackRun { finished: bool, writer: &mut Writer, ) -> Result { - let Some(from) = self.start else { + let Start::Resolved(from) = self.start else { return Ok(false); }; if self.unstarted.is_empty() { @@ -5567,25 +5606,23 @@ impl TrackRun { } // The first group waits for the source to resolve where its feed starts. - if let Some(group) = self.first.take() { - match self.track.poll_start(waiter) { - Poll::Ready(source) => { - self.start(group, source, writer)?; - return Poll::Ready(Ok(ControlFlow::Continue(()))); - } - Poll::Pending => { - self.first = Some(group); - return Poll::Pending; - } - } + if let Start::Unresolved { + first: first @ Some(_), .. + } = &mut self.start + { + let Poll::Ready(source) = self.track.poll_start(waiter) else { + return Poll::Pending; + }; + let group = first.take().expect("matched"); + self.start(group, source, writer)?; + return Poll::Ready(Ok(ControlFlow::Continue(()))); } // A start past everything the live feed has (a subscriber resuming just after what it // holds) is answered at once with the largest position, on versions that carry it: // a quiet track may not reach that start for a while, and the subscriber judges // what it holds against the answer. - if self.emit_range - && self.start.is_none() + if self.start.is_unresolved() && self.ctx.version.has_largest() && let Some(start) = self.named && let Poll::Ready(Some(largest)) = self.track.poll_live(waiter) @@ -5594,11 +5631,7 @@ impl TrackRun { frame: self.frames.start(start), }) > largest { - self.start = Some(start); - writer.buffer(&lite::SubscribeResponse::Start(lite::SubscribeStart { - largest: Some(largest), - starts: self.starts(start), - }))?; + self.resolve(start, Some(largest), writer)?; self.raise_floor(start); return Poll::Ready(Ok(ControlFlow::Continue(()))); } @@ -5639,11 +5672,11 @@ impl TrackRun { let sequence = group.sequence; // The pool folded it while it waited, which is final for this reader: tell the wire // it is gone, since an update re-offered it. - if self.emit_range && self.track.record_mut().folded(sequence) { + if self.start.emits_range() && self.track.record_mut().folded(sequence) { writer.buffer(&self.hole(sequence..sequence + 1, &Hole::drop(StreamError::Evicted)))?; return Poll::Ready(Ok(ControlFlow::Continue(()))); } - if self.emit_range && self.track.record_mut().told(sequence) { + if self.start.emits_range() && self.track.record_mut().told(sequence) { return Poll::Ready(Ok(ControlFlow::Continue(()))); } match position_group(group, &self.frames) { @@ -5656,7 +5689,7 @@ impl TrackRun { } // One cursor drives the whole subscription: poll the cap-aware arrival-order group. - let emit_boundary = self.emit_range && !self.end_sent; + let emit_boundary = self.start.emits_range() && !self.end_sent; // A continuation waiting for credit ranks below every stream it could take, so it must // not hold back a newer group that outranks one of them: the cursor hands one more out. let groups = self.opening.is_none() || (self.continuing && self.stashed.is_none()); @@ -5673,9 +5706,9 @@ impl TrackRun { // delivers: the update that reopens the drop would never re-offer it. let sequence = group.sequence; let record = self.track.record_mut(); - if self.emit_range && record.told_stale(sequence) { + if self.start.emits_range() && record.told_stale(sequence) { self.replace_stale(sequence..sequence + 1); - } else if self.emit_range && record.told(sequence) { + } else if self.start.emits_range() && record.told(sequence) { return Poll::Ready(Ok(ControlFlow::Continue(()))); } let Some(group) = position_group(group, &self.frames) else { @@ -5683,18 +5716,22 @@ impl TrackRun { // partial group. Skip it rather than open a stream that can // only be reset; the next servable group resolves the start. tracing::debug!(subscribe = self.ctx.id, track = %self.ctx.track_name, sequence, "skipping group with a missing head"); - if self.emit_range && self.start.is_none() { - self.skipped - .insert(sequence..sequence + 1, Hole::drop(StreamError::NotFound)); - } else if self.start.is_some() { - // Lite-05+ names it now; a no-op on other versions. - self.drop_range(sequence..sequence + 1, &Hole::drop(StreamError::NotFound), writer)?; + let hole = Hole::drop(StreamError::NotFound); + match &mut self.start { + Start::Unresolved { skipped, .. } => skipped.insert(sequence..sequence + 1, hole), + Start::Resolved(_) => { + self.drop_range(sequence..sequence + 1, &hole, writer)?; + } + Start::Legacy => {} } return Poll::Ready(Ok(ControlFlow::Continue(()))); }; - match self.emit_range && self.start.is_none() { - true => self.first = Some(group), - false => self.serve(group), + match &mut self.start { + Start::Unresolved { first, .. } => { + debug_assert!(first.is_none(), "a held group is resolved before the next"); + *first = Some(group); + } + _ => self.serve(group), } } Recv::Boundary(end) => { @@ -5714,7 +5751,7 @@ impl TrackRun { // and the drops below then account for every group. let mut answered = false; if self.ctx.version.keeps_subscribe_open() - && self.emit_range && self.start.is_none() + && self.start.is_unresolved() && let Poll::Ready(Ok(fin)) = self.track.poll_finished(&kio::Waiter::noop()) && final_sequence(fin) > 0 { @@ -5756,11 +5793,10 @@ impl TrackRun { /// Whether a bounded subscription still unanswered has given up every group its ranges /// name. fn unservable(&mut self) -> bool { - self.bounded - && self.start.is_none() - && self.first.is_none() - && self.opening.is_none() - && self.track.covered(self.floor, &self.skipped) + let Start::Unresolved { first: None, skipped } = &self.start else { + return false; + }; + self.bounded && self.opening.is_none() && self.track.covered(self.floor, skipped) } /// Lite-05+ names a sequence as soon as this cursor gives it up; whether one was taken. @@ -5774,7 +5810,7 @@ impl TrackRun { writer: &mut Writer, waiter: &kio::Waiter, ) -> Result { - if !self.emit_range { + if !self.start.emits_range() { return Ok(false); } let polled = match probe { @@ -5789,9 +5825,9 @@ impl TrackRun { } return Ok(false); }; - match self.start { - None => self.skipped.insert(range, hole), - Some(_) => { + match &mut self.start { + Start::Unresolved { skipped, .. } => skipped.insert(range, hole), + _ => { self.forward_range(range, &hole, writer)?; } } @@ -5802,7 +5838,7 @@ impl TrackRun { /// re-offers a group it still holds and drops again, under the new update, one it cannot /// serve. Earlier versions' drops are final, so their cursor takes nothing back. fn owe(&mut self, reopened: Vec>) { - if self.emit_range { + if self.start.emits_range() { let record = self.track.record_mut(); for run in reopened { record.untell(run); @@ -5812,7 +5848,7 @@ impl TrackRun { /// Record `sequence` as served, for [`Self::drop_unserved`]. fn mark_served(&mut self, sequence: u64) { - if self.emit_range { + if self.start.emits_range() { self.track.record_mut().tell(sequence..sequence + 1, Report::Settled); } } @@ -5827,7 +5863,7 @@ impl TrackRun { hole: &Hole, writer: &mut Writer, ) -> Result { - if !self.emit_range || range.is_empty() { + if !self.start.emits_range() || range.is_empty() { return Ok(false); } self.redrop.remove(range.clone()); @@ -5886,7 +5922,7 @@ impl TrackRun { hole: &Hole, writer: &mut Writer, ) -> Result { - if !self.emit_range || range.is_empty() { + if !self.start.emits_range() || range.is_empty() { return Ok(false); } let gaps = self.track.record_mut().untold(range); @@ -5920,7 +5956,7 @@ impl TrackRun { &mut self, writer: &mut Writer, ) -> Result { - let Some(start) = self.start else { + let Start::Resolved(start) = self.start else { // Without a SUBSCRIBE_START the subscriber owes itself no group (see // `SubStream::owed`), so there is nothing to drop. return Ok(false); @@ -5979,12 +6015,14 @@ impl TrackRun { // A group given up cannot be where the feed starts. This stops at the held group // at the latest, since it was not skipped. // Adjacent runs given up with different errors stay separate, so walk them all. - while start < group.sequence - && let Some((skipped, _)) = self.skipped.get_key_value(&start) - { - start = skipped.end.min(group.sequence); + if let Start::Unresolved { skipped, .. } = &self.start { + while start < group.sequence + && let Some((given_up, _)) = skipped.get_key_value(&start) + { + start = given_up.end.min(group.sequence); + } } - self.resolve(start, writer)?; + self.resolve(start, None, 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 @@ -6010,25 +6048,32 @@ impl TrackRun { writer: &mut Writer, ) -> Result<(), Error> { let start = self.starts(self.floor).into_iter().flatten().min().unwrap_or(none); - self.resolve(start, writer) + self.resolve(start, None, writer) } /// Send SUBSCRIBE_START resolving delivery at `start`, then name the groups given up - /// above it while it was held back. + /// above it while it was held back. The only way out of [`Start::Unresolved`], so no + /// start leaves its held drops unsent. `live` is the source's largest, if already polled. fn resolve( &mut self, start: u64, + live: Option, writer: &mut Writer, ) -> Result<(), Error> { - let skipped = std::mem::take(&mut self.skipped); - self.start = Some(start); + let Start::Unresolved { first, skipped } = std::mem::replace(&mut self.start, Start::Resolved(start)) else { + unreachable!("resolved once"); + }; + debug_assert!(first.is_none(), "a held group is taken before it resolves"); // Only groups: the subscriber derives the start frame from its own request // (see `lite::SubscribeStart`). A relay can hold just the first group of a backlog // its source already answered past, so the source's largest says where the live // feed is. - let live = match self.track.poll_live(&kio::Waiter::noop()) { - Poll::Ready(largest) => largest, - Poll::Pending => None, + let live = match live { + Some(live) => Some(live), + None => match self.track.poll_live(&kio::Waiter::noop()) { + Poll::Ready(largest) => largest, + Poll::Pending => None, + }, }; let largest = live.max(self.track.peek_largest()); writer.buffer(&lite::SubscribeResponse::Start(lite::SubscribeStart { @@ -6217,8 +6262,12 @@ impl CoverageBench { this.track.drop_groups(groups, err); while this.step().is_ready() {} } - assert!(this.run.start.is_none(), "no group resolves a start"); - assert_eq!(this.run.skipped.iter().count(), skipped, "each drop is its own entry"); + assert!(this.run.start.resolved().is_none(), "no group resolves a start"); + assert_eq!( + this.run.start.skipped().expect("unresolved").iter().count(), + skipped, + "each drop is its own entry" + ); this } @@ -7608,6 +7657,21 @@ mod serve_group_test { (TrackRun::new(ctx, track, bounds, 0, steer), writer, log) } + /// [`lite_run`] with the SUBSCRIBE naming group `start`. + fn lite_run_from( + version: Version, + track: track::Subscriber, + start: u64, + ) -> (TrackRun, Writer, Log) { + let (run, writer, log) = lite_run(version, SinkSession::new(Log::default()), track); + let bounds = Bounds::new(&[lite::Range::span(Some(start), 0, None, None)], None).unwrap(); + ( + TrackRun::new(run.ctx, run.track, bounds, start, run.steer_tx), + writer, + log, + ) + } + #[kio::test] async fn wanted_updates_ignore_stale_groups_and_recheck_the_ratio() { let track = track::Producer::new( @@ -7971,7 +8035,7 @@ mod serve_group_test { .with_range(0..1), ); let (mut run, mut writer, log) = lite07_run(SinkSession::new(Log::default()), subscriber); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } @@ -8023,7 +8087,7 @@ mod serve_group_test { .with_range(0..1), ); let (mut run, mut writer, _log) = lite07_run(SinkSession::new(Log::default()), subscriber); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } @@ -8067,7 +8131,7 @@ mod serve_group_test { .with_range(first..first + 1), ); let (mut run, mut writer, _log) = lite07_run(SinkSession::new(Log::default()), subscriber); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } let time = lite::Range { @@ -8131,7 +8195,7 @@ mod serve_group_test { run.update(lite07_add(&[(0, first)], 1, &[added])); run.update(lite07_add(&[(0, first)], 2, &[])); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some()); + assert!(run.start.resolved().is_some()); assert!( !responses(&mut writer, &log) .await @@ -8208,7 +8272,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some(), "{version}: started"); + assert!(run.start.resolved().is_some(), "{version}: started"); track.give_up(1..3, Hole::drop(StreamError::Cancel)); write_group(&mut track, 3, 3); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} @@ -8286,7 +8350,7 @@ mod serve_group_test { let subscriber = track.subscribe(None); let (mut run, mut writer, log) = lite07_run(SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } kio::wait(|waiter| writer.poll_flush(waiter)).await.unwrap(); @@ -8361,7 +8425,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some(), "{version}: started"); + assert!(run.start.resolved().is_some(), "{version}: started"); let stale = Hole::stale(Duration::from_secs(1)); forward(&mut run, 1..3, &stale, &mut writer); forward(&mut run, 1..2, &Hole::drop(StreamError::Cancel), &mut writer); @@ -8929,7 +8993,7 @@ mod serve_group_test { let subscriber = track.subscribe(None); let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } @@ -8961,7 +9025,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some()); + assert!(run.start.resolved().is_some()); let group = track.create_group(group::Info { sequence: 1 }).unwrap(); group.abort(Error::App(1)).unwrap(); @@ -8989,7 +9053,7 @@ mod serve_group_test { let subscriber = track.subscribe(None); let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } @@ -9022,7 +9086,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 5, 5); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert_eq!(run.start, Some(5)); + assert_eq!(run.start.resolved(), Some(5)); track.drop_groups(2..3, crate::StreamError::NotFound); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} @@ -9081,7 +9145,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some()); + assert!(run.start.resolved().is_some()); let group = track.create_group(group::Info { sequence: 1 }).unwrap(); group.abort(Error::App(1)).unwrap(); @@ -9120,7 +9184,7 @@ mod serve_group_test { write_group(&mut track, 0, 0); write_group(&mut track, 2, 2); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some()); + assert!(run.start.resolved().is_some()); assert_eq!(drops(&mut writer, &log, version).await, []); // Its successor arrived a full budget ago, so group 1 is stale on the wall clock. @@ -9151,7 +9215,7 @@ mod serve_group_test { let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); while run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready() {} - assert!(run.start.is_some()); + assert!(run.start.resolved().is_some()); let mut headless = track.create_group(group::Info { sequence: 1 }).unwrap(); headless.start_at(1).unwrap(); @@ -9374,11 +9438,38 @@ mod serve_group_test { .now_or_never() .is_none() ); - assert_eq!(run.start, Some(0), "{version}"); + assert_eq!(run.start.resolved(), Some(0), "{version}"); assert_eq!(drops(&mut writer, &log, version).await, [(2, 3)], "{version}"); } } + /// A start past the live edge is answered at once, and a group given up while it was held + /// is still named: the subscriber owes itself every group from the start. + #[kio::test] + async fn a_start_past_the_live_edge_names_what_it_held() { + // Only lite-07 answers a start past the live edge at once. + { + let version = Version::Lite07; + let mut track = track::Producer::new( + Arc::new(broadcast::Info::default()), + "test", + crate::track::Info::timed(), + ); + track.start_at(0).unwrap(); + write_group(&mut track, 5, 0); + track.drop_groups(11..12, crate::StreamError::NotFound); + let subscriber = track.subscribe(track::Subscription::default().with_start(10)); + let (mut run, mut writer, log) = lite_run_from(version, subscriber, 10); + assert!( + kio::wait(|waiter| run.poll(&mut writer, waiter)) + .now_or_never() + .is_none() + ); + assert_eq!(run.start.resolved(), Some(10), "{version}"); + assert_eq!(drops(&mut writer, &log, version).await, [(11, 11)], "{version}"); + } + } + /// A bounded lite-07 subscription is answered once its source gives up every range, even /// with no group to resolve its start, and not while one range is still open. The stream /// stays open for an update. @@ -9388,7 +9479,7 @@ mod serve_group_test { for (mut bench, last) in [(CoverageBench::new(3, 2), 4), (CoverageBench::adjacent(4), 4)] { bench.wake(); bench.wake(); - assert!(bench.run.start.is_none(), "a range is still open"); + assert!(bench.run.start.resolved().is_none(), "a range is still open"); bench.track.drop_groups(last..last + 1, StreamError::NotFound); loop { match bench.run.poll_step(&mut bench.writer, &kio::Waiter::noop()) { @@ -9397,8 +9488,8 @@ mod serve_group_test { other => panic!("the subscription stays open, got {other:?}"), } } - assert_eq!(bench.run.start, Some(0), "answered from the floor"); - assert!(bench.run.skipped.is_empty(), "every held drop went out"); + assert_eq!(bench.run.start.resolved(), Some(0), "answered from the floor"); + assert!(bench.run.start.skipped().is_none(), "every held drop went out"); } } @@ -9421,11 +9512,11 @@ mod serve_group_test { write_group(&mut track, 5, 5); track.finish_at(track::Position::group(7), None).unwrap(); // Serve group 5, which resolves SUBSCRIBE_START. - while run.start.is_none() { + while run.start.resolved().is_none() { let step = run.poll_step(&mut writer, &kio::Waiter::noop()); assert!(step.is_ready(), "the start never resolved"); } - assert_eq!(run.start, Some(5)); + assert_eq!(run.start.resolved(), Some(5)); run.replace(lite::SubscribeUpdate { wanted_group: None, @@ -9463,7 +9554,7 @@ mod serve_group_test { let subscriber = track.subscribe(subscription); let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 5, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { let step = run.poll_step(&mut writer, &kio::Waiter::noop()); assert!(step.is_ready(), "the start never resolved"); } @@ -9511,7 +9602,7 @@ mod serve_group_test { let subscriber = track.subscribe(subscription); let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { let step = run.poll_step(&mut writer, &kio::Waiter::noop()); assert!(step.is_ready(), "the start never resolved"); } @@ -9559,7 +9650,7 @@ mod serve_group_test { let subscriber = track.subscribe(subscription); let (mut run, mut writer, log) = lite_run(version, SinkSession::new(Log::default()), subscriber); write_group(&mut track, 0, 0); - while run.start.is_none() { + while run.start.resolved().is_none() { assert!(run.poll_step(&mut writer, &kio::Waiter::noop()).is_ready()); } From 8ae5d7ea646d030873ff8fb65aec9ad21319f311 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Thu, 12 Nov 2026 20:57:03 -0800 Subject: [PATCH 3/3] fix(moq-net): keep TrackRun's doc on TrackRun, and Start's test accessors in fuzz builds Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 968f727d385..a68e87eeb67 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -5091,9 +5091,6 @@ impl Subscription { } } -/// A subscription's run loop: one subscriber cursor serving groups and SUBSCRIBE_UPDATE -/// messages, with a machine per group it serves, one at a time -/// waiting for stream credit. /// Where a subscription's delivery starts, which SUBSCRIBE_START reports once. enum Start { /// Lite-03/04 send no SUBSCRIBE_START, and so no drops either. @@ -5120,7 +5117,7 @@ impl Start { } /// The resolved start, if any. - #[cfg(test)] + #[cfg(any(test, feature = "fuzz"))] fn resolved(&self) -> Option { match self { Self::Resolved(start) => Some(*start), @@ -5129,7 +5126,7 @@ impl Start { } /// The groups held back for the start. - #[cfg(test)] + #[cfg(any(test, feature = "fuzz"))] fn skipped(&self) -> Option<&RangeMap> { match self { Self::Unresolved { skipped, .. } => Some(skipped), @@ -5138,6 +5135,9 @@ impl Start { } } +/// A subscription's run loop: one subscriber cursor serving groups and SUBSCRIBE_UPDATE +/// messages, with a machine per group it serves, one at a time +/// waiting for stream credit. struct TrackRun { ctx: Subscription, track: track::Subscriber,