From 4dbc350f91b0ab5c03551a5c919bb6fcd2d70b4e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 22:49:08 -0700 Subject: [PATCH 1/4] chore(quest): claim quest/m1/subscribe-ranges/lite.md Quest-Claim: 231fdacf563a76a2 From b7f573d755ae53e6f2bdb94cc480ae47fd6ba254 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 31 Oct 2026 23:41:34 -0700 Subject: [PATCH 2/4] wip(net): fill a lite-07 upstream through the subscription's ranges A lite-07 copy no longer sends FETCH for a fill: each held fill's next owed run joins the upstream SUBSCRIBE ranges (or opens a fills-only bounded SUBSCRIBE when no demand is open), and the fill answers itself from the cache as the groups arrive. A fill is judged against the subscription that asked for it (SUBSCRIBE_DROP, a range start past it, or a covered FIN), never the track's arrivals, so a group an earlier subscription cut is filled whole. An asked run stays on the wire until its streams finish. Work in progress: on-demand publishers, datagram groups, missing groups under an unbounded budget, and multi-hop cached fetches still regress; see the PR's Decisions needed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 486 +++++++++++++++++-- rs/moq-net/tests/integration/fetch_fanout.rs | 13 +- rs/moq-net/tests/integration/main.rs | 1 + rs/moq-net/tests/integration/range_fill.rs | 238 +++++++++ 4 files changed, 699 insertions(+), 39 deletions(-) create mode 100644 rs/moq-net/tests/integration/range_fill.rs diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 0ce4ed44bb7..35570402ac7 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2142,6 +2142,11 @@ mod tests { priority: 0, requested: None, ranges: vec![lite::Range::OLDEST], + count: 1, + demand: None, + fills: Default::default(), + requested_count: 1, + requested_fills: Default::default(), tail: tail.clone(), served: None, starts: Vec::new(), @@ -2317,6 +2322,11 @@ mod tests { priority: 0, requested: Some(floor), ranges: vec![lite::Range::span(Some(floor), 0, None, None)], + count: 1, + demand: None, + fills: Default::default(), + requested_count: 1, + requested_fills: Default::default(), tail: Default::default(), served: None, starts: Vec::new(), @@ -2332,7 +2342,9 @@ mod tests { .await .unwrap(); } else { - let begin = h.serve.begin_subscription(&mut track, &mut Sub::None, lowered, true); + let begin = h + .serve + .begin_subscription(&mut track, &mut Sub::None, lowered, &FillAsk::default(), true); assert!(matches!(begin, Ok(Begin::Establish(_)))); } @@ -2773,6 +2785,11 @@ mod tests { priority: 0, requested: Some(2), ranges: vec![lite::Range::span(Some(2), 0, None, None)], + count: 1, + demand: None, + fills: Default::default(), + requested_count: 1, + requested_fills: Default::default(), tail: Default::default(), served: None, starts: Vec::new(), @@ -4568,8 +4585,13 @@ mod tests { let broadcast = crate::broadcast::Info::new().produce(); let request = broadcast.reserve_track("audio").unwrap(); let mut serving = ServeLoop::new(&serve, request, Default::default()); - let establish = - serve.prepare_establish(&mut serving.serving, Subscription::default(), vec![lite::Range::OLDEST]); + let wire = Wire { + subscription: Subscription::default(), + demand: Some(Subscription::default()), + ranges: vec![lite::Range::OLDEST], + count: 1, + }; + let establish = serve.prepare_establish(&mut serving.serving, wire, &Default::default()); serving.mode = ServeMode::Establish(establish); let gate_run = serve.gate(); let mut running = TrackServeRun { @@ -6552,8 +6574,17 @@ struct SubStream { /// The lowest absolute start among the ranges, or `None` when one counts back from /// the latest group: the floor below which nothing is asked for. start: Option, - /// The ranges on the wire, never empty. + /// The ranges on the wire, never empty: the demand's `count`, then the fills'. ranges: Vec, + count: usize, + /// The demand the ranges ask for, `None` when they ask only for fills. + demand: Option, + /// The groups the ranges ask for fills. + fills: rangemap::RangeSet, + /// `count` and `fills` as the SUBSCRIBE itself carried them, which its SUBSCRIBE_START + /// answers. + requested_count: usize, + requested_fills: rangemap::RangeSet, priority: i8, order: crate::track::Order, /// The start the SUBSCRIBE itself carried, fixed for the stream's life. A @@ -6616,7 +6647,8 @@ impl SubStream { /// a range is open, or the ranges changed with nothing served: the publisher covered /// what it asked before an update it may not have read. fn covered(&self) -> Option>> { - let Some(served) = self.served else { + // A fill's range starts too, though only the demand's set where the feed is served. + let Some(served) = self.served.or_else(|| self.starts.iter().flatten().min().copied()) else { return (!self.updated).then(Vec::new); }; self.ranges @@ -7007,6 +7039,7 @@ impl TrackServe { producer: &mut track::Producer, sub: &mut Sub, pref: Option, + fills: &FillAsk, supports_update: bool, ) -> Result { // An empty half-open range asks for nothing, and the wire cannot say that: its @@ -7017,6 +7050,18 @@ impl TrackServe { // stands in for. let pref = pref.filter(|sub| sub.end().is_none_or(|end| end > sub.start().unwrap_or_default())); let ranged = self.subscriber.version.has_ranges(); + // Lite-07 asks the held fills' groups through the same subscription. Demand leaving + // while they are out keeps what it asked until they are answered, rather than retire + // the live feed's ranges under a subscription that still runs. + let none = FillAsk::default(); + let fills = match ranged { + true => fills, + false => &none, + }; + let pref = match (pref, &*sub) { + (None, Sub::Active(active)) if !fills.is_empty() => active.demand.clone(), + (pref, _) => pref, + }; let request = match pref { Some(subscription) => { // Lite-07 carries every range asked for; an older wire one span covering them. @@ -7042,12 +7087,48 @@ impl TrackServe { } None => None, }; + // What goes on the wire: the demand's ranges, then the fills', under parameters that + // serve both. The demand alone sets the floor. + let request = match request { + Some((demand, mut ranges)) => { + let count = ranges.len(); + ranges.extend(fills.wire()); + Some(Wire { + subscription: fills.merge(demand.clone()), + demand: Some(demand), + ranges, + count, + }) + } + None if fills.is_empty() => None, + None => Some(Wire { + subscription: fills.subscription(), + demand: None, + ranges: fills.wire().collect(), + count: 0, + }), + }; match request { - Some((subscription, ranges)) => { + Some(wire) => { + let Wire { + subscription, + demand, + ranges, + count, + } = wire; match sub { // Open an upstream SUBSCRIBE for the first subscriber. - Sub::None => Ok(Begin::Establish(self.prepare_establish(producer, subscription, ranges))), + Sub::None => Ok(Begin::Establish(self.prepare_establish( + producer, + Wire { + subscription, + demand, + ranges, + count, + }, + &fills.groups, + ))), Sub::Active(active) => { // Downstream preferences changed: forward them upstream as a // SUBSCRIBE_UPDATE (Lite03+ only; older peers can't carry one). @@ -7063,14 +7144,24 @@ impl TrackServe { start..floor.unwrap_or(u64::MAX) }); if let Ok(mut tail) = active.tail.write() { + let now = self.subscriber.runtime.now(); tail.set_grace(tail::grace(subscription.budget())); if let Some(groups) = lowered { - tail.demand(groups, self.subscriber.runtime.now()); + tail.demand(groups, now); + } + // So do the groups a fill newly asks for. + for groups in fills.groups.iter() { + for groups in active.fills.gaps(groups) { + tail.demand(groups, now); + } } } active.start = subscription.start(); active.updated |= active.ranges != ranges; active.ranges = ranges; + active.count = count; + active.demand = demand; + active.fills = fills.groups.clone(); if supports_update { // The floor follows the requested start, in both directions: // moving below a declared SUBSCRIBE_START reopens those groups @@ -7098,9 +7189,12 @@ impl TrackServe { self.subscriber.remove_subscribe(active.id); let _ = active.stream.writer.finish(); tracing::debug!(track = %self.name, "subscribe canceled (idle)"); + // The copy is still held: what it cached goes stale from here, unless + // the subscription only filled it and never fed it live. + if active.demand.is_some() { + producer.set_idle(); + } *sub = Sub::None; - // The copy is still held: what it cached goes stale from here. - producer.set_idle(); } Ok(Begin::None) } @@ -7120,8 +7214,8 @@ impl TrackServe { fn prepare_establish( &self, producer: &mut track::Producer, - subscription: Subscription, - ranges: Vec, + wire: Wire, + fills: &rangemap::RangeSet, ) -> Establish { let id = self.subscriber.next_id.fetch_add(1, atomic::Ordering::Relaxed); @@ -7130,11 +7224,20 @@ impl TrackServe { // previous subscription's declaration must not outlive its demand, and // live-edge demand (None) starts with no floor at all. Lite-06+ only resolves // the start with its SUBSCRIBE_START, so until then the floor is just a request. - let floor = subscription.start(); - let _ = match self.subscriber.version.resolves_start() { - true => producer.request_start(floor), - false => producer.start_at(floor), - }; + // One asking only for fills is no live feed, so it leaves the floor alone. + let Wire { + subscription, + demand, + ranges, + count, + } = wire; + if demand.is_some() { + let floor = subscription.start(); + let _ = match self.subscriber.version.resolves_start() { + true => producer.request_start(floor), + false => producer.start_at(floor), + }; + } tracing::debug!(id, broadcast = %self.subscriber.log_path(&self.path), track = %self.name, "subscribe started"); @@ -7158,7 +7261,10 @@ impl TrackServe { session, id, subscription, + demand, ranges, + count, + fills: fills.clone(), tail, state: EstablishState::Start, } @@ -7177,7 +7283,13 @@ impl TrackServe { false => wire_span(subscription), }; let ranges = wire_ranges(&subscription, producer.info().timescale)?.expect("a non-empty subscription"); - let mut est = Box::new(self.prepare_establish(producer, subscription, ranges)); + let wire = Wire { + subscription: subscription.clone(), + demand: Some(subscription), + count: ranges.len(), + ranges, + }; + let mut est = Box::new(self.prepare_establish(producer, wire, &Default::default())); let id = est.id; match kio::wait(move |waiter| est.poll(waiter)).await { Ok(active) => { @@ -7201,7 +7313,7 @@ impl TrackServe { pref: Option, supports_update: bool, ) -> Result<(), Error> { - match self.begin_subscription(producer, sub, pref, supports_update)? { + match self.begin_subscription(producer, sub, pref, &FillAsk::default(), supports_update)? { Begin::Establish(est) => { let mut est = Box::new(est); let id = est.id; @@ -7264,13 +7376,28 @@ struct Establish { // A dedicated close-watch handle for the SUBSCRIBE_OK wait. closed: Transport, id: u64, + /// The parameters the SUBSCRIBE carries. subscription: Subscription, - /// The ranges the SUBSCRIBE carries for `subscription`. + /// The demand it asks for, `None` when it asks only for fills. + demand: Option, + /// The ranges the SUBSCRIBE carries: the demand's `count`, then the fills'. ranges: Vec, + count: usize, + /// The groups it asks for fills. + fills: rangemap::RangeSet, tail: kio::Producer, state: EstablishState, } +/// What a subscription puts on the wire: its parameters, the demand it asks for, and the +/// ranges, the demand's `count` first, then the held fills'. +struct Wire { + subscription: Subscription, + demand: Option, + ranges: Vec, + count: usize, +} + #[kio::machine(self.state: EstablishState)] impl Establish { fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { @@ -7335,6 +7462,11 @@ impl Establish { requested_max_delay: wire_max_delay(self.subscription.max_delay), start: self.subscription.start(), ranges: self.ranges.clone(), + count: self.count, + demand: self.demand.clone(), + fills: self.fills.clone(), + requested_count: self.count, + requested_fills: self.fills.clone(), priority: self.subscription.priority, order: self.subscription.order, requested: self.subscription.start(), @@ -7678,6 +7810,8 @@ struct ServeLoop { fetches: kio::Tasks, /// Fetches a pre-lite-05 upstream has no FETCH for, held for the live feed. feed_fetches: FeedFetches, + /// Fills a lite-07 upstream serves through the subscription. + range_fills: RangeFills, // A dedicated close-watch handle for the session-died arm. closed: Transport, // SUBSCRIBE_UPDATE only exists on Lite03+, so older peers can't carry a @@ -7691,8 +7825,9 @@ struct ServeLoop { /// The TRACK stream that answered, until a SUBSCRIBE has a response or nobody holds /// the copy. held: Option, - /// The ranges of a bounded lite-07 subscription the publisher covered, while the - /// demand still asks for just them: subscribing again would only resend them. + /// The demand's ranges of a bounded lite-07 subscription the publisher covered, empty + /// when it asked only for fills, while the demand still asks for just them and no fill + /// is held: subscribing again would only resend them. covered: Option>, /// Our side of the covered subscription, finished and held until the publisher reads /// the FIN, since dropping it would reset it. @@ -7919,6 +8054,216 @@ impl FeedFetches { } } +/// What a lite-07 copy's held fills ask of its upstream subscription, beside the demand. +#[derive(Clone, Debug, PartialEq)] +struct FillAsk { + /// The groups to ask for, empty when no fill is held. + groups: rangemap::RangeSet, + /// The most urgent priority a held fill asks with. + priority: i8, + /// The most tolerant budget a held fill asks with, `None` for the publisher's max age. + max_delay: Option, + order: track::Order, +} + +impl Default for FillAsk { + fn default() -> Self { + Self { + groups: Default::default(), + priority: i8::MIN, + max_delay: Some(Duration::ZERO), + order: track::Order::Ascending, + } + } +} + +impl FillAsk { + fn is_empty(&self) -> bool { + self.groups.is_empty() + } + + /// Count `fill`'s preferences in. + fn add(&mut self, fill: &track::Fill) { + self.priority = self.priority.max(fill.priority()); + self.max_delay = match (self.max_delay, fill.max_delay()) { + (Some(ours), Some(theirs)) => Some(ours.max(theirs)), + _ => None, + }; + if fill.order() == track::Order::Descending { + self.order = track::Order::Descending; + } + } + + /// The bounded ranges the groups take on the wire, each ending before a group's first + /// frame as [`wire_ranges`] ends an absolute one. + fn wire(&self) -> impl Iterator + '_ { + self.groups.iter().map(|groups| lite::Range { + start: Some(lite::Anchor::Absolute(Position::group(groups.start))), + end: Some(lite::Anchor::Absolute(Position::group(groups.end))), + }) + } + + /// `demand` loosened to serve the fills too: one budget caps every range, so the most + /// tolerant one stands, as merging subscribers does. Its ranges and order stay. + fn merge(&self, demand: Subscription) -> Subscription { + let max_delay = match (demand.max_delay, self.max_delay) { + (Some(ours), Some(theirs)) => Some(ours.max(theirs)), + _ => None, + }; + let priority = demand.priority.max(self.priority); + demand.with_priority(priority).with_max_delay(max_delay) + } + + /// A subscription asking for the fills alone: no range of its own, so no floor. + fn subscription(&self) -> Subscription { + Subscription::default() + .with_ranges([]) + .with_priority(self.priority) + .with_max_delay(self.max_delay) + .with_order(self.order) + } +} + +/// Fills a lite-07 upstream serves through the subscription rather than FETCH: each held +/// fill's next owed run joins the upstream ranges until its groups arrive or the publisher +/// names them dropped, and the fill answers itself from the cache. +/// +/// A fill is judged against this subscription alone. The track's arrivals still account for +/// a group an earlier subscription cut, which this one may still deliver, so only what this +/// subscription says it will not deliver (a SUBSCRIBE_DROP, a start past the range, or a +/// covered FIN) refuses a fill. +/// +/// A group stays asked until its stream finishes, since narrowing the subscription under a +/// stream still arriving would reset it. +#[derive(Default)] +struct RangeFills { + held: Vec, + /// What the upstream subscription asks for the fills now. + ask: FillAsk, + /// Asked groups no fill owes any more whose stream is still being read. + reading: BTreeMap, + /// Asked groups no fill owes any more, already searched for streams still being read. + draining: rangemap::RangeSet, + /// Groups the subscription said it will not deliver, and why. + refused: rangemap::RangeMap, +} + +impl RangeFills { + /// Whether no fill is held or asked for. + fn is_empty(&self) -> bool { + self.held.is_empty() && self.ask.is_empty() + } + + fn push(&mut self, fill: track::Fill) { + self.held.push(fill); + } + + /// The subscription will not deliver `groups`: a fill owing one is refused with `err`. + fn refuse(&mut self, groups: std::ops::Range, err: crate::StreamError) { + // Only what a fill asked for: a fill asks for nothing the subscription dropped + // before it. + let asked: Vec<_> = self + .ask + .groups + .overlapping(&groups) + .map(|run| run.start.max(groups.start)..run.end.min(groups.end)) + .collect(); + for groups in asked { + self.refused.insert(groups, err.clone()); + } + } + + /// A fresh subscription may deliver what the last one refused. + fn forget_refused(&mut self) { + self.refused = Default::default(); + } + + /// Settle each held fill against what the subscription refused, and recompute what to ask. + /// Returns whether the ask changed. + fn poll(&mut self, track: &track::Consumer, waiter: &kio::Waiter) -> bool { + let mut ask = FillAsk::default(); + let refused = &self.refused; + self.held.retain_mut(|fill| { + if fill.poll_withdrawn(waiter).is_ready() { + return false; + } + // Arms every input of the owed groups: readers, answers, and the cache. + let _ = fill.poll_wanted_changed(waiter); + loop { + let Some(run) = fill.next_run() else { + return false; + }; + // `next_run` extends from the group owed first, which the subscription may have + // refused already. Refusing the run answers it, so the loop moves on. + let first = match fill.order() { + track::Order::Ascending => run.start, + track::Order::Descending => run.end - 1, + }; + if let Some((groups, err)) = refused.get_key_value(&first) { + let groups = run.start.max(groups.start)..run.end.min(groups.end); + refuse_fill(fill, groups, err.clone()); + continue; + } + ask.groups.insert(run); + ask.add(fill); + return true; + } + }); + + // A run stays asked whole while a fill still owes any of it, so the subscription + // changes once as a run is asked and once as it is done, and the streams of its + // answered groups keep arriving. One no fill owes stays while any of its streams is + // still being read, each group searched for once. + self.reading.retain(|_, group| group.poll_finished(waiter).is_pending()); + let mut groups = ask.groups.clone(); + let mut draining = rangemap::RangeSet::new(); + for run in self.ask.groups.iter() { + if ask.groups.overlaps(run) { + groups.insert(run.clone()); + continue; + } + for gap in self.draining.gaps(run) { + for sequence in gap { + if let Some(mut group) = track.peek_group(sequence) + && group.poll_finished(waiter).is_pending() + { + self.reading.insert(sequence, group); + } + } + } + if self.reading.range(run.clone()).next().is_some() { + groups.insert(run.clone()); + draining.insert(run.clone()); + } + } + self.reading.retain(|sequence, _| draining.contains(sequence)); + self.draining = draining; + // What finishes a stream still asks with what its fills asked. + if self.held.is_empty() && !groups.is_empty() { + ask.priority = self.ask.priority; + ask.max_delay = self.ask.max_delay; + ask.order = self.ask.order; + } + ask.groups = groups; + if ask == self.ask { + return false; + } + self.ask = ask; + true + } +} + +/// Answer `groups` of `fill` as the subscription that asked for them will not deliver +/// them, with why: a miss names them dropped, as a FETCH's miss does. +fn refuse_fill(fill: &track::Fill, groups: std::ops::Range, err: crate::StreamError) { + match err { + crate::StreamError::NotFound => fill.drop_groups(groups), + crate::StreamError::Old => fill.reject_groups(groups, Error::Old), + crate::StreamError::NotFetchable => fill.reject_groups(groups, Error::NotFetchable), + err => fill.refuse_groups(groups, Error::Stream(err)), + } +} + impl ServeLoop { fn new(serve: &TrackServe, request: track::Request, info: track::Info) -> Self { // Register the fetch handler before accepting: `accept` releases the @@ -7950,6 +8295,7 @@ impl ServeLoop { sub: Sub::None, fetches: kio::Tasks::new(), feed_fetches: FeedFetches::new(&serve.subscriber.runtime), + range_fills: RangeFills::default(), closed: serve.subscriber.session.clone(), supports_update: !matches!(serve.subscriber.version, Version::Lite01 | Version::Lite02), supports_fetch: serve.subscriber.version.has_track_stream(), @@ -7969,16 +8315,29 @@ impl ServeLoop { Some(pref) => wire_ranges(pref, self.serving.info().timescale)?, None => None, }; - if ranges.as_ref() == Some(covered) { + // A held fill asks for what the covered ranges did not name. + if self.range_fills.ask.is_empty() && ranges.unwrap_or_default() == *covered { return Ok(()); } // What the copy cached may go stale before the next subscription says where - // the live feed is. + // the live feed is, unless the covered one only filled it. + if !covered.is_empty() { + self.serving.set_idle(); + } self.covered = None; - self.serving.set_idle(); } - match serve.begin_subscription(&mut self.serving, &mut self.sub, pref, self.supports_update)? { - Begin::Establish(est) => self.mode = ServeMode::Establish(est), + match serve.begin_subscription( + &mut self.serving, + &mut self.sub, + pref, + &self.range_fills.ask, + self.supports_update, + )? { + Begin::Establish(est) => { + // A fresh subscription may deliver what the last one would not. + self.range_fills.forget_refused(); + self.mode = ServeMode::Establish(est); + } Begin::None => {} } Ok(()) @@ -8040,16 +8399,30 @@ impl ServeLoop { } // Unchanged ranges are what the publisher covered, even if a datagram // went missing. Changed ones settled by the grace instead may have - // been covered before an update it never read. - self.covered = (whole || !active.updated).then_some(active.ranges); + // been covered before an update it never read. A fill still owing + // a group of a covered range gets nothing more from it. + if whole || !active.updated { + for groups in active.fills.iter() { + self.range_fills.refuse(groups.clone(), crate::StreamError::NotFound); + } + let mut ranges = active.ranges; + ranges.truncate(active.count); + self.covered = Some(ranges); + } else { + self.covered = None; + } } + self.range_fills.poll(&self.edge, waiter); // Whatever the covered ranges leave out of the demand is asked again. if let Err(err) = self.begin(serve, self.serving.subscription()) { return Poll::Ready(ServeEnd::GiveBack(err)); } continue; } - if self.fetches.is_empty() && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() { + if self.fetches.is_empty() + && self.range_fills.is_empty() + && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() + { return Poll::Ready(ServeEnd::Idle); } // The publisher already ended the track, so a session that dies now @@ -8107,7 +8480,10 @@ impl ServeLoop { // subscription churn. match self.dynamic.poll_requested_fill(waiter) { Poll::Ready(Ok(fill)) => { - if self.supports_fetch { + if serve.subscriber.version.has_ranges() { + // Lite-07 asks for the fill's groups through the subscription. + self.range_fills.push(fill); + } else if self.supports_fetch { self.fetches .push(FillServeRun::new(serve.clone(), fill, self.serving.info().clone())); } else { @@ -8160,12 +8536,22 @@ impl ServeLoop { let _ = self.fetches.poll(waiter); self.feed_fetches .poll(&self.sub, serve.subscriber.runtime.now(), waiter); + // A fill asking for other groups changes the subscription. + if self.range_fills.poll(&self.edge, waiter) { + if let Err(err) = self.begin(serve, self.serving.subscription()) { + return Poll::Ready(ServeEnd::GiveBack(err)); + } + continue; + } // (3) Nobody holds this copy anymore. Its upstream subscription already // went with the last subscriber, so it lingers, cache and all, for a // reader or fetch that asks again soon, then is dropped. In-flight // fetches keep it alive: work already accepted still gets finished. - if self.fetches.is_empty() && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() { + if self.fetches.is_empty() + && self.range_fills.is_empty() + && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() + { self.held = None; if self.linger.deadline().is_none() { let now = serve.subscriber.runtime.now(); @@ -8238,11 +8624,36 @@ impl ServeLoop { tracing::warn!(track = %serve.name, starts = start.starts.len(), ranges = active.ranges.len(), "subscribe start resolves other ranges"); return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); } + // The fills' ranges follow the demand's. A range's start + // past where it asked skips groups it will not deliver. + // After an update the starts may answer other ranges, so + // only the demand's lowest counts, as before fills. + let demand = match active.updated { + false => { + let count = active.requested_count.min(start.starts.len()); + let (demand, fills) = start.starts.split_at(count); + let now = serve.subscriber.runtime.now(); + for (groups, start) in active.requested_fills.iter().zip(fills) { + let skipped = groups.start + ..start.unwrap_or(groups.end).clamp(groups.start, groups.end); + if let Ok(mut tail) = active.tail.write() { + tail.account(skipped.clone(), now); + } + self.range_fills.refuse(skipped, crate::StreamError::NotFound); + } + demand + } + true if active.requested_count == 0 => &[][..], + true => &start.starts[..], + }; // No range starting anywhere leaves nothing owed. - let served = start.starts.iter().flatten().min().copied(); + let served = demand.iter().flatten().min().copied(); // Where the live feed is, on versions whose answer says. if serve.subscriber.version.has_largest() { - self.serving.set_live(start.largest, served); + // One asking only for fills feeds nothing live. + if active.requested_count > 0 { + self.serving.set_live(start.largest, served); + } if let Some(largest) = start.largest { self.serving.report_largest(largest); } @@ -8257,7 +8668,8 @@ impl ServeLoop { // that moved the start makes it stale, and one that // moved back restores it); elsewhere the // request-tracked floor stands rather than a guess. - if active.start == active.requested { + // One asking only for fills declares no feed. + if active.start == active.requested && active.requested_count > 0 { let _ = match (serve.subscriber.version.resolves_start(), served) { (true, Some(served)) => { self.serving.start_resolved(served, active.requested_max_delay) @@ -8290,6 +8702,10 @@ impl ServeLoop { dropped.start..dropped.end.saturating_add(1), dropped.error(serve.subscriber.version), ); + self.range_fills.refuse( + dropped.start..dropped.end.saturating_add(1), + dropped.error(serve.subscriber.version), + ); } // OK just resolves the range (the producer already orders // groups). diff --git a/rs/moq-net/tests/integration/fetch_fanout.rs b/rs/moq-net/tests/integration/fetch_fanout.rs index ff257cadfcf..377157abf1d 100644 --- a/rs/moq-net/tests/integration/fetch_fanout.rs +++ b/rs/moq-net/tests/integration/fetch_fanout.rs @@ -1,7 +1,8 @@ #![cfg(feature = "async")] //! A relay fetches each group's payload upstream once, independently of how many -//! viewers ask for it, and keeps serving the cached copy after a reader leaves. +//! viewers ask for it, and keeps serving the cached copy after a reader leaves. Over +//! moq-lite-07 it asks for the group with a SUBSCRIBE to its range instead of a FETCH. use crate::support; use moq_net::{Hop, Version, origin, track}; @@ -18,6 +19,11 @@ fn origin(hop: u64) -> origin::Producer { async fn sequential_viewers_share_the_relays_fetched_groups() { for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"] { let version: Version = version.parse().unwrap(); + // FETCH's lite stream type is 3 and SUBSCRIBE's 2, each varint one byte. + let (asks, other) = match version.as_str() { + "moq-lite-07-wip" => (2, 3), + _ => (3, 2), + }; let publisher = origin(1); let broadcast = publisher.publish("room", Default::default()).unwrap(); let track = broadcast @@ -66,12 +72,11 @@ async fn sequential_viewers_share_the_relays_fetched_groups() { .unwrap(); assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"payload"); assert!(group.read_frame().await.unwrap().is_none()); - // FETCH's lite stream type is 3, and its varint occupies one byte. let fetches = upstream .client_transport .bidi_types() .into_iter() - .filter(|kind| *kind == 3) + .filter(|kind| *kind == asks) .count(); assert_eq!(fetches, round as usize + 1, "{version}: viewer {i}, group {sequence}"); } @@ -83,7 +88,7 @@ async fn sequential_viewers_share_the_relays_fetched_groups() { 1, "metadata is learned once" ); - assert!(!types.contains(&2), "a fetch-only relay must never subscribe"); + assert!(!types.contains(&other), "{version}: asked the other way"); assert!(track.subscription().is_none()); } } diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index 9f756677753..9e10c8504af 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -43,6 +43,7 @@ mod multi_cdn; mod partial_group; mod quiet_catalog; mod range_anchors; +mod range_fill; mod received_rewind; mod rejoin; mod relay_upstream_close; diff --git a/rs/moq-net/tests/integration/range_fill.rs b/rs/moq-net/tests/integration/range_fill.rs new file mode 100644 index 00000000000..ba7f72f6be2 --- /dev/null +++ b/rs/moq-net/tests/integration/range_fill.rs @@ -0,0 +1,238 @@ +#![cfg(feature = "async")] + +//! A copy fills a group its cache misses over moq-lite-07 by subscribing to the missing +//! range, never with FETCH, which older versions still use. + +use crate::support; + +use std::time::Duration; + +use moq_net::stats::{Message, Registry, Tier}; +use moq_net::track::{Info, Range, Subscription}; +use moq_net::{Timescale, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +const LITE07: &str = "moq-lite-07-wip"; + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = + moq_net::origin::Producer::new(moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +fn write_group(track: &moq_net::track::Producer, sequence: u64) { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group + .write_frame( + Timestamp::from_millis(sequence * 1000).unwrap(), + format!("{sequence}").into_bytes(), + ) + .unwrap(); + group.finish().unwrap(); +} + +struct Fixture { + track: moq_net::track::Producer, + remote: moq_net::broadcast::Consumer, + registry: Registry, + tier: Tier, + _broadcast: moq_net::origin::Publication, + _pair: support::harness::MockPair, + _origins: (moq_net::origin::Producer, moq_net::origin::Producer), +} + +impl Fixture { + /// A publisher on `version` holding `groups`, and the subscriber's view of it, whose + /// session counts its control bytes. + async fn new(version: &str, groups: &[u64]) -> Self { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast + .create_track("video", Info::default().with_timescale(Timescale::MILLI)) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + for sequence in groups { + write_group(&track, *sequence); + } + + let subscriber = produce_origin(2); + let registry = Registry::new(Default::default()); + let tier = Tier::new(moq_net::Path::from_static("internal")); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.client_publish = Some(publisher.consume()); + options.server_subscribe = Some(subscriber.clone()); + options.server_stats = registry.tier(tier.clone()).session(moq_net::Path::from_static("")); + let pair = connect_mock(options).await; + let remote = moq_clock::timeout(TIMEOUT, subscriber.consume().follow("bcast", None).unwrap().available()) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + Self { + track, + remote, + registry, + tier, + _broadcast: broadcast, + _pair: pair, + _origins: (publisher, subscriber), + } + } + + /// The control bytes this session sent for `class`. + fn sent(&self, class: Message) -> u64 { + self.registry + .snapshot() + .control() + .iter() + .filter(|(tier, row, _)| *tier == self.tier && *row == class) + .map(|(_, _, bytes)| bytes.tx) + .sum() + } + + /// Fetch group `sequence` and read its one frame. + async fn fetch(&self, sequence: u64) -> moq_net::Result { + let fetching = self.remote.track("video").unwrap().fetch_group(sequence, None); + let mut group = moq_clock::timeout(TIMEOUT, fetching).await.expect("fetch timed out")?; + let frame = moq_clock::timeout(TIMEOUT, group.read_frame()) + .await + .expect("read timed out")? + .expect("a frame"); + Ok(String::from_utf8(frame.payload.to_vec()).unwrap()) + } +} + +/// With no subscription open, a fetch of an uncached group subscribes to just that group, +/// and the publisher ends the subscription once it covers it. Lite-06 fetches it instead. +#[moq_clock::test] +async fn a_miss_subscribes_to_its_group() { + for version in ["moq-lite-06", LITE07] { + let fixture = Fixture::new(version, &[0, 1, 2]).await; + assert_eq!(fixture.fetch(1).await.expect("fetched"), "1", "{version}"); + + let (fetch, subscribe) = (fixture.sent(Message::Fetch), fixture.sent(Message::Subscribe)); + if version != LITE07 { + assert!(fetch > 0, "{version}: fetched"); + continue; + } + assert_eq!(fetch, 0, "{version}: no FETCH"); + assert!(subscribe > 0, "{version}: subscribed"); + moq_clock::timeout(TIMEOUT, support::unwanted(fixture.track.demand())) + .await + .expect("the covered subscription ends") + .unwrap(); + + // Cached now, so a second fetch asks nothing more. + assert_eq!(fixture.fetch(1).await.expect("cached"), "1"); + assert_eq!(fixture.sent(Message::Subscribe), subscribe, "served from the cache"); + } +} + +/// A miss beside a live subscription rides it as one more range, and the live feed goes on. +#[moq_clock::test] +async fn a_miss_rides_the_live_subscription() { + let fixture = Fixture::new(LITE07, &[0, 1, 2]).await; + let mut live = fixture + .remote + .track("video") + .unwrap() + .subscribe(Subscription::default().with_ranges([Range::live()])) + .await + .expect("subscribe"); + let first = moq_clock::timeout(TIMEOUT, live.recv_group()) + .await + .expect("group timeout") + .unwrap() + .expect("a group"); + assert_eq!(first.sequence, 2); + + assert_eq!(fixture.fetch(0).await.expect("fetched"), "0"); + assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); + + write_group(&fixture.track, 3); + let next = moq_clock::timeout(TIMEOUT, live.recv_group()) + .await + .expect("group timeout") + .unwrap() + .expect("a group"); + assert_eq!(next.sequence, 3, "the live feed goes on"); + let subscription = fixture.track.subscription().expect("still subscribed"); + assert_eq!(subscription.ranges, [Range::live()], "the filled range is retired"); +} + +/// A group the publisher never had is a miss, named by the subscription it was asked on +/// once the publisher's feed gives up on it, past the budget's grace. +#[moq_clock::test] +async fn a_group_the_publisher_never_had_is_a_miss() { + let fixture = Fixture::new(LITE07, &[0, 2, 3]).await; + let fetching = fixture.remote.track("video").unwrap().fetch_group( + 1, + moq_net::group::Fetch::default().with_max_delay(Duration::from_secs(1)), + ); + let err = match moq_clock::timeout(Duration::from_secs(120), fetching) + .await + .expect("fetch timed out") + { + Ok(_) => panic!("group 1 was served"), + Err(err) => err, + }; + assert!(matches!(err, moq_net::Error::NotFound), "{err}"); + assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); +} + +/// A group whose stream an earlier subscription cut is filled whole by the next one, +/// rather than judged missing from what the earlier one left behind. +#[moq_clock::test] +async fn a_cut_group_is_filled_whole() { + let fixture = Fixture::new(LITE07, &[]).await; + let mut open = fixture + .track + .create_group(moq_net::group::Info { sequence: 0 }) + .unwrap(); + open.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); + + let mut live = fixture + .remote + .track("video") + .unwrap() + .subscribe(Subscription::default().with_ranges([Range::live()])) + .await + .expect("subscribe"); + let mut group = moq_clock::timeout(TIMEOUT, live.recv_group()) + .await + .expect("group timeout") + .unwrap() + .expect("a group"); + let frame = moq_clock::timeout(TIMEOUT, group.read_frame()) + .await + .expect("read timed out") + .unwrap() + .expect("a frame"); + assert_eq!(frame.payload.as_ref(), b"head"); + // The only reader leaves mid-group, so the subscription goes and cuts the copy. + drop((group, live)); + moq_clock::timeout(TIMEOUT, support::unwanted(fixture.track.demand())) + .await + .expect("the subscription ends") + .unwrap(); + + open.write_frame(Timestamp::from_millis(1).unwrap(), b"tail".as_slice()) + .unwrap(); + open.finish().unwrap(); + let mut fetched = moq_clock::timeout(TIMEOUT, fixture.remote.track("video").unwrap().fetch_group(0, None)) + .await + .expect("fetch timed out") + .expect("fetched"); + let mut frames = Vec::new(); + while let Some(frame) = moq_clock::timeout(TIMEOUT, fetched.read_frame()) + .await + .expect("read timed out") + .expect("read") + { + frames.push(frame.payload.to_vec()); + } + assert_eq!(frames, [b"head".to_vec(), b"tail".to_vec()]); + assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); +} From 288e4a9e1486d574c78cc043aeca36bab380379b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 10:17:14 -0800 Subject: [PATCH 3/4] revert(net): move the lite-07 range fill out until fetch-range lands The range fill through the subscription (b7f573d755) fails nine integration tests until Fetch as a range's miss judging and cached backfill land. Keep it in the quest instead, so this PR lands as a checkpoint. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/subscriber.rs | 486 ++----------------- rs/moq-net/tests/integration/fetch_fanout.rs | 13 +- rs/moq-net/tests/integration/main.rs | 1 - rs/moq-net/tests/integration/range_fill.rs | 238 --------- 4 files changed, 39 insertions(+), 699 deletions(-) delete mode 100644 rs/moq-net/tests/integration/range_fill.rs diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 86379baffb0..d57c5bf54c5 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2142,11 +2142,6 @@ mod tests { priority: 0, requested: None, ranges: vec![lite::Range::OLDEST], - count: 1, - demand: None, - fills: Default::default(), - requested_count: 1, - requested_fills: Default::default(), tail: tail.clone(), served: None, starts: Vec::new(), @@ -2323,11 +2318,6 @@ mod tests { priority: 0, requested: Some(floor), ranges: vec![lite::Range::span(Some(floor), 0, None, None)], - count: 1, - demand: None, - fills: Default::default(), - requested_count: 1, - requested_fills: Default::default(), tail: Default::default(), served: None, starts: Vec::new(), @@ -2344,9 +2334,7 @@ mod tests { .await .unwrap(); } else { - let begin = h - .serve - .begin_subscription(&mut track, &mut Sub::None, lowered, &FillAsk::default(), true); + let begin = h.serve.begin_subscription(&mut track, &mut Sub::None, lowered, true); assert!(matches!(begin, Ok(Begin::Establish(_)))); } @@ -2787,11 +2775,6 @@ mod tests { priority: 0, requested: Some(2), ranges: vec![lite::Range::span(Some(2), 0, None, None)], - count: 1, - demand: None, - fills: Default::default(), - requested_count: 1, - requested_fills: Default::default(), tail: Default::default(), served: None, starts: Vec::new(), @@ -4588,13 +4571,8 @@ mod tests { let broadcast = crate::broadcast::Info::new().produce(); let request = broadcast.reserve_track("audio").unwrap(); let mut serving = ServeLoop::new(&serve, request, Default::default()); - let wire = Wire { - subscription: Subscription::default(), - demand: Some(Subscription::default()), - ranges: vec![lite::Range::OLDEST], - count: 1, - }; - let establish = serve.prepare_establish(&mut serving.serving, wire, &Default::default()); + let establish = + serve.prepare_establish(&mut serving.serving, Subscription::default(), vec![lite::Range::OLDEST]); serving.mode = ServeMode::Establish(establish); let gate_run = serve.gate(); let mut running = TrackServeRun { @@ -6577,17 +6555,8 @@ struct SubStream { /// The lowest absolute start among the ranges, or `None` when one counts back from /// the latest group: the floor below which nothing is asked for. start: Option, - /// The ranges on the wire, never empty: the demand's `count`, then the fills'. + /// The ranges on the wire, never empty. ranges: Vec, - count: usize, - /// The demand the ranges ask for, `None` when they ask only for fills. - demand: Option, - /// The groups the ranges ask for fills. - fills: rangemap::RangeSet, - /// `count` and `fills` as the SUBSCRIBE itself carried them, which its SUBSCRIBE_START - /// answers. - requested_count: usize, - requested_fills: rangemap::RangeSet, priority: i8, order: crate::track::Order, /// The start the SUBSCRIBE itself carried, fixed for the stream's life. A @@ -6652,8 +6621,7 @@ impl SubStream { /// a range is open, or the ranges changed with nothing served: the publisher covered /// what it asked before an update it may not have read. fn covered(&self) -> Option>> { - // A fill's range starts too, though only the demand's set where the feed is served. - let Some(served) = self.served.or_else(|| self.starts.iter().flatten().min().copied()) else { + let Some(served) = self.served else { return (!self.updated).then(Vec::new); }; self.ranges @@ -7044,7 +7012,6 @@ impl TrackServe { producer: &mut track::Producer, sub: &mut Sub, pref: Option, - fills: &FillAsk, supports_update: bool, ) -> Result { // An empty half-open range asks for nothing, and the wire cannot say that: its @@ -7055,18 +7022,6 @@ impl TrackServe { // stands in for. let pref = pref.filter(|sub| sub.end().is_none_or(|end| end > sub.start().unwrap_or_default())); let ranged = self.subscriber.version.has_ranges(); - // Lite-07 asks the held fills' groups through the same subscription. Demand leaving - // while they are out keeps what it asked until they are answered, rather than retire - // the live feed's ranges under a subscription that still runs. - let none = FillAsk::default(); - let fills = match ranged { - true => fills, - false => &none, - }; - let pref = match (pref, &*sub) { - (None, Sub::Active(active)) if !fills.is_empty() => active.demand.clone(), - (pref, _) => pref, - }; let request = match pref { Some(subscription) => { // Lite-07 carries every range asked for; an older wire one span covering them. @@ -7092,48 +7047,12 @@ impl TrackServe { } None => None, }; - // What goes on the wire: the demand's ranges, then the fills', under parameters that - // serve both. The demand alone sets the floor. - let request = match request { - Some((demand, mut ranges)) => { - let count = ranges.len(); - ranges.extend(fills.wire()); - Some(Wire { - subscription: fills.merge(demand.clone()), - demand: Some(demand), - ranges, - count, - }) - } - None if fills.is_empty() => None, - None => Some(Wire { - subscription: fills.subscription(), - demand: None, - ranges: fills.wire().collect(), - count: 0, - }), - }; match request { - Some(wire) => { - let Wire { - subscription, - demand, - ranges, - count, - } = wire; + Some((subscription, ranges)) => { match sub { // Open an upstream SUBSCRIBE for the first subscriber. - Sub::None => Ok(Begin::Establish(self.prepare_establish( - producer, - Wire { - subscription, - demand, - ranges, - count, - }, - &fills.groups, - ))), + Sub::None => Ok(Begin::Establish(self.prepare_establish(producer, subscription, ranges))), Sub::Active(active) => { // Downstream preferences changed: forward them upstream as a // SUBSCRIBE_UPDATE (Lite03+ only; older peers can't carry one). @@ -7149,24 +7068,14 @@ impl TrackServe { start..floor.unwrap_or(u64::MAX) }); if let Ok(mut tail) = active.tail.write() { - let now = self.subscriber.runtime.now(); tail.set_grace(tail::grace(subscription.budget())); if let Some(groups) = lowered { - tail.demand(groups, now); - } - // So do the groups a fill newly asks for. - for groups in fills.groups.iter() { - for groups in active.fills.gaps(groups) { - tail.demand(groups, now); - } + tail.demand(groups, self.subscriber.runtime.now()); } } active.start = subscription.start(); active.updated |= active.ranges != ranges; active.ranges = ranges; - active.count = count; - active.demand = demand; - active.fills = fills.groups.clone(); if supports_update { // The floor follows the requested start, in both directions: // moving below a declared SUBSCRIBE_START reopens those groups @@ -7194,12 +7103,9 @@ impl TrackServe { self.subscriber.remove_subscribe(active.id); let _ = active.stream.writer.finish(); tracing::debug!(track = %self.name, "subscribe canceled (idle)"); - // The copy is still held: what it cached goes stale from here, unless - // the subscription only filled it and never fed it live. - if active.demand.is_some() { - producer.set_idle(); - } *sub = Sub::None; + // The copy is still held: what it cached goes stale from here. + producer.set_idle(); } Ok(Begin::None) } @@ -7219,8 +7125,8 @@ impl TrackServe { fn prepare_establish( &self, producer: &mut track::Producer, - wire: Wire, - fills: &rangemap::RangeSet, + subscription: Subscription, + ranges: Vec, ) -> Establish { let id = self.subscriber.next_id.fetch_add(1, atomic::Ordering::Relaxed); @@ -7229,20 +7135,11 @@ impl TrackServe { // previous subscription's declaration must not outlive its demand, and // live-edge demand (None) starts with no floor at all. Lite-06+ only resolves // the start with its SUBSCRIBE_START, so until then the floor is just a request. - // One asking only for fills is no live feed, so it leaves the floor alone. - let Wire { - subscription, - demand, - ranges, - count, - } = wire; - if demand.is_some() { - let floor = subscription.start(); - let _ = match self.subscriber.version.resolves_start() { - true => producer.request_start(floor), - false => producer.start_at(floor), - }; - } + let floor = subscription.start(); + let _ = match self.subscriber.version.resolves_start() { + true => producer.request_start(floor), + false => producer.start_at(floor), + }; tracing::debug!(id, broadcast = %self.subscriber.log_path(&self.path), track = %self.name, "subscribe started"); @@ -7267,10 +7164,7 @@ impl TrackServe { session, id, subscription, - demand, ranges, - count, - fills: fills.clone(), tail, state: EstablishState::Start, } @@ -7289,13 +7183,7 @@ impl TrackServe { false => wire_span(subscription), }; let ranges = wire_ranges(&subscription, producer.info().timescale)?.expect("a non-empty subscription"); - let wire = Wire { - subscription: subscription.clone(), - demand: Some(subscription), - count: ranges.len(), - ranges, - }; - let mut est = Box::new(self.prepare_establish(producer, wire, &Default::default())); + let mut est = Box::new(self.prepare_establish(producer, subscription, ranges)); let id = est.id; match kio::wait(move |waiter| est.poll(waiter)).await { Ok(active) => { @@ -7319,7 +7207,7 @@ impl TrackServe { pref: Option, supports_update: bool, ) -> Result<(), Error> { - match self.begin_subscription(producer, sub, pref, &FillAsk::default(), supports_update)? { + match self.begin_subscription(producer, sub, pref, supports_update)? { Begin::Establish(est) => { let mut est = Box::new(est); let id = est.id; @@ -7382,30 +7270,15 @@ struct Establish { // A dedicated close-watch handle for the SUBSCRIBE_OK wait. closed: Transport, id: u64, - /// The parameters the SUBSCRIBE carries. subscription: Subscription, - /// The demand it asks for, `None` when it asks only for fills. - demand: Option, - /// The ranges the SUBSCRIBE carries: the demand's `count`, then the fills'. + /// The ranges the SUBSCRIBE carries for `subscription`. ranges: Vec, - count: usize, - /// The groups it asks for fills. - fills: rangemap::RangeSet, tail: kio::Producer, /// Holds the session busy while the SUBSCRIBE opens. _working: crate::session::Working, state: EstablishState, } -/// What a subscription puts on the wire: its parameters, the demand it asks for, and the -/// ranges, the demand's `count` first, then the held fills'. -struct Wire { - subscription: Subscription, - demand: Option, - ranges: Vec, - count: usize, -} - #[kio::machine(self.state: EstablishState)] impl Establish { fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { @@ -7470,11 +7343,6 @@ impl Establish { requested_max_delay: wire_max_delay(self.subscription.max_delay), start: self.subscription.start(), ranges: self.ranges.clone(), - count: self.count, - demand: self.demand.clone(), - fills: self.fills.clone(), - requested_count: self.count, - requested_fills: self.fills.clone(), priority: self.subscription.priority, order: self.subscription.order, requested: self.subscription.start(), @@ -7829,8 +7697,6 @@ struct ServeLoop { fetches: kio::Tasks, /// Fetches a pre-lite-05 upstream has no FETCH for, held for the live feed. feed_fetches: FeedFetches, - /// Fills a lite-07 upstream serves through the subscription. - range_fills: RangeFills, // A dedicated close-watch handle for the session-died arm. closed: Transport, // SUBSCRIBE_UPDATE only exists on Lite03+, so older peers can't carry a @@ -7844,9 +7710,8 @@ struct ServeLoop { /// The TRACK stream that answered, until a SUBSCRIBE has a response or nobody holds /// the copy. held: Option, - /// The demand's ranges of a bounded lite-07 subscription the publisher covered, empty - /// when it asked only for fills, while the demand still asks for just them and no fill - /// is held: subscribing again would only resend them. + /// The ranges of a bounded lite-07 subscription the publisher covered, while the + /// demand still asks for just them: subscribing again would only resend them. covered: Option>, /// Our side of the covered subscription, finished and held until the publisher reads /// the FIN, since dropping it would reset it. The subscription keeps the session busy @@ -8077,216 +7942,6 @@ impl FeedFetches { } } -/// What a lite-07 copy's held fills ask of its upstream subscription, beside the demand. -#[derive(Clone, Debug, PartialEq)] -struct FillAsk { - /// The groups to ask for, empty when no fill is held. - groups: rangemap::RangeSet, - /// The most urgent priority a held fill asks with. - priority: i8, - /// The most tolerant budget a held fill asks with, `None` for the publisher's max age. - max_delay: Option, - order: track::Order, -} - -impl Default for FillAsk { - fn default() -> Self { - Self { - groups: Default::default(), - priority: i8::MIN, - max_delay: Some(Duration::ZERO), - order: track::Order::Ascending, - } - } -} - -impl FillAsk { - fn is_empty(&self) -> bool { - self.groups.is_empty() - } - - /// Count `fill`'s preferences in. - fn add(&mut self, fill: &track::Fill) { - self.priority = self.priority.max(fill.priority()); - self.max_delay = match (self.max_delay, fill.max_delay()) { - (Some(ours), Some(theirs)) => Some(ours.max(theirs)), - _ => None, - }; - if fill.order() == track::Order::Descending { - self.order = track::Order::Descending; - } - } - - /// The bounded ranges the groups take on the wire, each ending before a group's first - /// frame as [`wire_ranges`] ends an absolute one. - fn wire(&self) -> impl Iterator + '_ { - self.groups.iter().map(|groups| lite::Range { - start: Some(lite::Anchor::Absolute(Position::group(groups.start))), - end: Some(lite::Anchor::Absolute(Position::group(groups.end))), - }) - } - - /// `demand` loosened to serve the fills too: one budget caps every range, so the most - /// tolerant one stands, as merging subscribers does. Its ranges and order stay. - fn merge(&self, demand: Subscription) -> Subscription { - let max_delay = match (demand.max_delay, self.max_delay) { - (Some(ours), Some(theirs)) => Some(ours.max(theirs)), - _ => None, - }; - let priority = demand.priority.max(self.priority); - demand.with_priority(priority).with_max_delay(max_delay) - } - - /// A subscription asking for the fills alone: no range of its own, so no floor. - fn subscription(&self) -> Subscription { - Subscription::default() - .with_ranges([]) - .with_priority(self.priority) - .with_max_delay(self.max_delay) - .with_order(self.order) - } -} - -/// Fills a lite-07 upstream serves through the subscription rather than FETCH: each held -/// fill's next owed run joins the upstream ranges until its groups arrive or the publisher -/// names them dropped, and the fill answers itself from the cache. -/// -/// A fill is judged against this subscription alone. The track's arrivals still account for -/// a group an earlier subscription cut, which this one may still deliver, so only what this -/// subscription says it will not deliver (a SUBSCRIBE_DROP, a start past the range, or a -/// covered FIN) refuses a fill. -/// -/// A group stays asked until its stream finishes, since narrowing the subscription under a -/// stream still arriving would reset it. -#[derive(Default)] -struct RangeFills { - held: Vec, - /// What the upstream subscription asks for the fills now. - ask: FillAsk, - /// Asked groups no fill owes any more whose stream is still being read. - reading: BTreeMap, - /// Asked groups no fill owes any more, already searched for streams still being read. - draining: rangemap::RangeSet, - /// Groups the subscription said it will not deliver, and why. - refused: rangemap::RangeMap, -} - -impl RangeFills { - /// Whether no fill is held or asked for. - fn is_empty(&self) -> bool { - self.held.is_empty() && self.ask.is_empty() - } - - fn push(&mut self, fill: track::Fill) { - self.held.push(fill); - } - - /// The subscription will not deliver `groups`: a fill owing one is refused with `err`. - fn refuse(&mut self, groups: std::ops::Range, err: crate::StreamError) { - // Only what a fill asked for: a fill asks for nothing the subscription dropped - // before it. - let asked: Vec<_> = self - .ask - .groups - .overlapping(&groups) - .map(|run| run.start.max(groups.start)..run.end.min(groups.end)) - .collect(); - for groups in asked { - self.refused.insert(groups, err.clone()); - } - } - - /// A fresh subscription may deliver what the last one refused. - fn forget_refused(&mut self) { - self.refused = Default::default(); - } - - /// Settle each held fill against what the subscription refused, and recompute what to ask. - /// Returns whether the ask changed. - fn poll(&mut self, track: &track::Consumer, waiter: &kio::Waiter) -> bool { - let mut ask = FillAsk::default(); - let refused = &self.refused; - self.held.retain_mut(|fill| { - if fill.poll_withdrawn(waiter).is_ready() { - return false; - } - // Arms every input of the owed groups: readers, answers, and the cache. - let _ = fill.poll_wanted_changed(waiter); - loop { - let Some(run) = fill.next_run() else { - return false; - }; - // `next_run` extends from the group owed first, which the subscription may have - // refused already. Refusing the run answers it, so the loop moves on. - let first = match fill.order() { - track::Order::Ascending => run.start, - track::Order::Descending => run.end - 1, - }; - if let Some((groups, err)) = refused.get_key_value(&first) { - let groups = run.start.max(groups.start)..run.end.min(groups.end); - refuse_fill(fill, groups, err.clone()); - continue; - } - ask.groups.insert(run); - ask.add(fill); - return true; - } - }); - - // A run stays asked whole while a fill still owes any of it, so the subscription - // changes once as a run is asked and once as it is done, and the streams of its - // answered groups keep arriving. One no fill owes stays while any of its streams is - // still being read, each group searched for once. - self.reading.retain(|_, group| group.poll_finished(waiter).is_pending()); - let mut groups = ask.groups.clone(); - let mut draining = rangemap::RangeSet::new(); - for run in self.ask.groups.iter() { - if ask.groups.overlaps(run) { - groups.insert(run.clone()); - continue; - } - for gap in self.draining.gaps(run) { - for sequence in gap { - if let Some(mut group) = track.peek_group(sequence) - && group.poll_finished(waiter).is_pending() - { - self.reading.insert(sequence, group); - } - } - } - if self.reading.range(run.clone()).next().is_some() { - groups.insert(run.clone()); - draining.insert(run.clone()); - } - } - self.reading.retain(|sequence, _| draining.contains(sequence)); - self.draining = draining; - // What finishes a stream still asks with what its fills asked. - if self.held.is_empty() && !groups.is_empty() { - ask.priority = self.ask.priority; - ask.max_delay = self.ask.max_delay; - ask.order = self.ask.order; - } - ask.groups = groups; - if ask == self.ask { - return false; - } - self.ask = ask; - true - } -} - -/// Answer `groups` of `fill` as the subscription that asked for them will not deliver -/// them, with why: a miss names them dropped, as a FETCH's miss does. -fn refuse_fill(fill: &track::Fill, groups: std::ops::Range, err: crate::StreamError) { - match err { - crate::StreamError::NotFound => fill.drop_groups(groups), - crate::StreamError::Old => fill.reject_groups(groups, Error::Old), - crate::StreamError::NotFetchable => fill.reject_groups(groups, Error::NotFetchable), - err => fill.refuse_groups(groups, Error::Stream(err)), - } -} - impl ServeLoop { fn new(serve: &TrackServe, request: track::Request, info: track::Info) -> Self { // Register the fetch handler before accepting: `accept` releases the @@ -8318,7 +7973,6 @@ impl ServeLoop { sub: Sub::None, fetches: kio::Tasks::new(), feed_fetches: FeedFetches::new(&serve.subscriber.runtime), - range_fills: RangeFills::default(), closed: serve.subscriber.session.clone(), supports_update: !matches!(serve.subscriber.version, Version::Lite01 | Version::Lite02), supports_fetch: serve.subscriber.version.has_track_stream(), @@ -8338,29 +7992,16 @@ impl ServeLoop { Some(pref) => wire_ranges(pref, self.serving.info().timescale)?, None => None, }; - // A held fill asks for what the covered ranges did not name. - if self.range_fills.ask.is_empty() && ranges.unwrap_or_default() == *covered { + if ranges.as_ref() == Some(covered) { return Ok(()); } // What the copy cached may go stale before the next subscription says where - // the live feed is, unless the covered one only filled it. - if !covered.is_empty() { - self.serving.set_idle(); - } + // the live feed is. self.covered = None; + self.serving.set_idle(); } - match serve.begin_subscription( - &mut self.serving, - &mut self.sub, - pref, - &self.range_fills.ask, - self.supports_update, - )? { - Begin::Establish(est) => { - // A fresh subscription may deliver what the last one would not. - self.range_fills.forget_refused(); - self.mode = ServeMode::Establish(est); - } + match serve.begin_subscription(&mut self.serving, &mut self.sub, pref, self.supports_update)? { + Begin::Establish(est) => self.mode = ServeMode::Establish(est), Begin::None => {} } Ok(()) @@ -8422,30 +8063,16 @@ impl ServeLoop { } // Unchanged ranges are what the publisher covered, even if a datagram // went missing. Changed ones settled by the grace instead may have - // been covered before an update it never read. A fill still owing - // a group of a covered range gets nothing more from it. - if whole || !active.updated { - for groups in active.fills.iter() { - self.range_fills.refuse(groups.clone(), crate::StreamError::NotFound); - } - let mut ranges = active.ranges; - ranges.truncate(active.count); - self.covered = Some(ranges); - } else { - self.covered = None; - } + // been covered before an update it never read. + self.covered = (whole || !active.updated).then_some(active.ranges); } - self.range_fills.poll(&self.edge, waiter); // Whatever the covered ranges leave out of the demand is asked again. if let Err(err) = self.begin(serve, self.serving.subscription()) { return Poll::Ready(ServeEnd::GiveBack(err)); } continue; } - if self.fetches.is_empty() - && self.range_fills.is_empty() - && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() - { + if self.fetches.is_empty() && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() { return Poll::Ready(ServeEnd::Idle); } // The publisher already ended the track, so a session that dies now @@ -8503,10 +8130,7 @@ impl ServeLoop { // subscription churn. match self.dynamic.poll_requested_fill(waiter) { Poll::Ready(Ok(fill)) => { - if serve.subscriber.version.has_ranges() { - // Lite-07 asks for the fill's groups through the subscription. - self.range_fills.push(fill); - } else if self.supports_fetch { + if self.supports_fetch { self.fetches .push(FillServeRun::new(serve.clone(), fill, self.serving.info().clone())); } else { @@ -8559,22 +8183,12 @@ impl ServeLoop { let _ = self.fetches.poll(waiter); self.feed_fetches .poll(&self.sub, serve.subscriber.runtime.now(), waiter); - // A fill asking for other groups changes the subscription. - if self.range_fills.poll(&self.edge, waiter) { - if let Err(err) = self.begin(serve, self.serving.subscription()) { - return Poll::Ready(ServeEnd::GiveBack(err)); - } - continue; - } // (3) Nobody holds this copy anymore. Its upstream subscription already // went with the last subscriber, so it lingers, cache and all, for a // reader or fetch that asks again soon, then is dropped. In-flight // fetches keep it alive: work already accepted still gets finished. - if self.fetches.is_empty() - && self.range_fills.is_empty() - && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() - { + if self.fetches.is_empty() && crate::util::poll_unwanted(&mut self.demand, waiter).is_ready() { self.held = None; if self.linger.deadline().is_none() { let now = serve.subscriber.runtime.now(); @@ -8647,36 +8261,11 @@ impl ServeLoop { tracing::warn!(track = %serve.name, starts = start.starts.len(), ranges = active.ranges.len(), "subscribe start resolves other ranges"); return Poll::Ready(ServeEnd::GiveBack(Error::ProtocolViolation)); } - // The fills' ranges follow the demand's. A range's start - // past where it asked skips groups it will not deliver. - // After an update the starts may answer other ranges, so - // only the demand's lowest counts, as before fills. - let demand = match active.updated { - false => { - let count = active.requested_count.min(start.starts.len()); - let (demand, fills) = start.starts.split_at(count); - let now = serve.subscriber.runtime.now(); - for (groups, start) in active.requested_fills.iter().zip(fills) { - let skipped = groups.start - ..start.unwrap_or(groups.end).clamp(groups.start, groups.end); - if let Ok(mut tail) = active.tail.write() { - tail.account(skipped.clone(), now); - } - self.range_fills.refuse(skipped, crate::StreamError::NotFound); - } - demand - } - true if active.requested_count == 0 => &[][..], - true => &start.starts[..], - }; // No range starting anywhere leaves nothing owed. - let served = demand.iter().flatten().min().copied(); + let served = start.starts.iter().flatten().min().copied(); // Where the live feed is, on versions whose answer says. if serve.subscriber.version.has_largest() { - // One asking only for fills feeds nothing live. - if active.requested_count > 0 { - self.serving.set_live(start.largest, served); - } + self.serving.set_live(start.largest, served); if let Some(largest) = start.largest { self.serving.report_largest(largest); } @@ -8691,8 +8280,7 @@ impl ServeLoop { // that moved the start makes it stale, and one that // moved back restores it); elsewhere the // request-tracked floor stands rather than a guess. - // One asking only for fills declares no feed. - if active.start == active.requested && active.requested_count > 0 { + if active.start == active.requested { let _ = match (serve.subscriber.version.resolves_start(), served) { (true, Some(served)) => { self.serving.start_resolved(served, active.requested_max_delay) @@ -8725,10 +8313,6 @@ impl ServeLoop { dropped.start..dropped.end.saturating_add(1), dropped.error(serve.subscriber.version), ); - self.range_fills.refuse( - dropped.start..dropped.end.saturating_add(1), - dropped.error(serve.subscriber.version), - ); } // OK just resolves the range (the producer already orders // groups). diff --git a/rs/moq-net/tests/integration/fetch_fanout.rs b/rs/moq-net/tests/integration/fetch_fanout.rs index 377157abf1d..ff257cadfcf 100644 --- a/rs/moq-net/tests/integration/fetch_fanout.rs +++ b/rs/moq-net/tests/integration/fetch_fanout.rs @@ -1,8 +1,7 @@ #![cfg(feature = "async")] //! A relay fetches each group's payload upstream once, independently of how many -//! viewers ask for it, and keeps serving the cached copy after a reader leaves. Over -//! moq-lite-07 it asks for the group with a SUBSCRIBE to its range instead of a FETCH. +//! viewers ask for it, and keeps serving the cached copy after a reader leaves. use crate::support; use moq_net::{Hop, Version, origin, track}; @@ -19,11 +18,6 @@ fn origin(hop: u64) -> origin::Producer { async fn sequential_viewers_share_the_relays_fetched_groups() { for version in ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"] { let version: Version = version.parse().unwrap(); - // FETCH's lite stream type is 3 and SUBSCRIBE's 2, each varint one byte. - let (asks, other) = match version.as_str() { - "moq-lite-07-wip" => (2, 3), - _ => (3, 2), - }; let publisher = origin(1); let broadcast = publisher.publish("room", Default::default()).unwrap(); let track = broadcast @@ -72,11 +66,12 @@ async fn sequential_viewers_share_the_relays_fetched_groups() { .unwrap(); assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"payload"); assert!(group.read_frame().await.unwrap().is_none()); + // FETCH's lite stream type is 3, and its varint occupies one byte. let fetches = upstream .client_transport .bidi_types() .into_iter() - .filter(|kind| *kind == asks) + .filter(|kind| *kind == 3) .count(); assert_eq!(fetches, round as usize + 1, "{version}: viewer {i}, group {sequence}"); } @@ -88,7 +83,7 @@ async fn sequential_viewers_share_the_relays_fetched_groups() { 1, "metadata is learned once" ); - assert!(!types.contains(&other), "{version}: asked the other way"); + assert!(!types.contains(&2), "a fetch-only relay must never subscribe"); assert!(track.subscription().is_none()); } } diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index e302381b001..9f7f1589974 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -43,7 +43,6 @@ mod multi_cdn; mod partial_group; mod quiet_catalog; mod range_anchors; -mod range_fill; mod received_rewind; mod rejoin; mod relay_upstream_close; diff --git a/rs/moq-net/tests/integration/range_fill.rs b/rs/moq-net/tests/integration/range_fill.rs deleted file mode 100644 index ba7f72f6be2..00000000000 --- a/rs/moq-net/tests/integration/range_fill.rs +++ /dev/null @@ -1,238 +0,0 @@ -#![cfg(feature = "async")] - -//! A copy fills a group its cache misses over moq-lite-07 by subscribing to the missing -//! range, never with FETCH, which older versions still use. - -use crate::support; - -use std::time::Duration; - -use moq_net::stats::{Message, Registry, Tier}; -use moq_net::track::{Info, Range, Subscription}; -use moq_net::{Timescale, Timestamp, Version}; -use support::harness::{MockConnectOptions, connect_mock}; - -const TIMEOUT: Duration = Duration::from_secs(10); - -const LITE07: &str = "moq-lite-07-wip"; - -fn produce_origin(hop: u64) -> moq_net::origin::Producer { - let (producer, driver) = - moq_net::origin::Producer::new(moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap())); - support::harness::spawn(driver); - producer -} - -fn write_group(track: &moq_net::track::Producer, sequence: u64) { - let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); - group - .write_frame( - Timestamp::from_millis(sequence * 1000).unwrap(), - format!("{sequence}").into_bytes(), - ) - .unwrap(); - group.finish().unwrap(); -} - -struct Fixture { - track: moq_net::track::Producer, - remote: moq_net::broadcast::Consumer, - registry: Registry, - tier: Tier, - _broadcast: moq_net::origin::Publication, - _pair: support::harness::MockPair, - _origins: (moq_net::origin::Producer, moq_net::origin::Producer), -} - -impl Fixture { - /// A publisher on `version` holding `groups`, and the subscriber's view of it, whose - /// session counts its control bytes. - async fn new(version: &str, groups: &[u64]) -> Self { - let publisher = produce_origin(1); - let broadcast = publisher.create_broadcast("bcast").unwrap(); - let track = broadcast - .create_track("video", Info::default().with_timescale(Timescale::MILLI)) - .unwrap(); - broadcast.announce(Default::default()).unwrap(); - for sequence in groups { - write_group(&track, *sequence); - } - - let subscriber = produce_origin(2); - let registry = Registry::new(Default::default()); - let tier = Tier::new(moq_net::Path::from_static("internal")); - let mut options = MockConnectOptions::new(version.parse::().unwrap()); - options.client_publish = Some(publisher.consume()); - options.server_subscribe = Some(subscriber.clone()); - options.server_stats = registry.tier(tier.clone()).session(moq_net::Path::from_static("")); - let pair = connect_mock(options).await; - let remote = moq_clock::timeout(TIMEOUT, subscriber.consume().follow("bcast", None).unwrap().available()) - .await - .expect("resolve timeout") - .expect("broadcast resolves"); - Self { - track, - remote, - registry, - tier, - _broadcast: broadcast, - _pair: pair, - _origins: (publisher, subscriber), - } - } - - /// The control bytes this session sent for `class`. - fn sent(&self, class: Message) -> u64 { - self.registry - .snapshot() - .control() - .iter() - .filter(|(tier, row, _)| *tier == self.tier && *row == class) - .map(|(_, _, bytes)| bytes.tx) - .sum() - } - - /// Fetch group `sequence` and read its one frame. - async fn fetch(&self, sequence: u64) -> moq_net::Result { - let fetching = self.remote.track("video").unwrap().fetch_group(sequence, None); - let mut group = moq_clock::timeout(TIMEOUT, fetching).await.expect("fetch timed out")?; - let frame = moq_clock::timeout(TIMEOUT, group.read_frame()) - .await - .expect("read timed out")? - .expect("a frame"); - Ok(String::from_utf8(frame.payload.to_vec()).unwrap()) - } -} - -/// With no subscription open, a fetch of an uncached group subscribes to just that group, -/// and the publisher ends the subscription once it covers it. Lite-06 fetches it instead. -#[moq_clock::test] -async fn a_miss_subscribes_to_its_group() { - for version in ["moq-lite-06", LITE07] { - let fixture = Fixture::new(version, &[0, 1, 2]).await; - assert_eq!(fixture.fetch(1).await.expect("fetched"), "1", "{version}"); - - let (fetch, subscribe) = (fixture.sent(Message::Fetch), fixture.sent(Message::Subscribe)); - if version != LITE07 { - assert!(fetch > 0, "{version}: fetched"); - continue; - } - assert_eq!(fetch, 0, "{version}: no FETCH"); - assert!(subscribe > 0, "{version}: subscribed"); - moq_clock::timeout(TIMEOUT, support::unwanted(fixture.track.demand())) - .await - .expect("the covered subscription ends") - .unwrap(); - - // Cached now, so a second fetch asks nothing more. - assert_eq!(fixture.fetch(1).await.expect("cached"), "1"); - assert_eq!(fixture.sent(Message::Subscribe), subscribe, "served from the cache"); - } -} - -/// A miss beside a live subscription rides it as one more range, and the live feed goes on. -#[moq_clock::test] -async fn a_miss_rides_the_live_subscription() { - let fixture = Fixture::new(LITE07, &[0, 1, 2]).await; - let mut live = fixture - .remote - .track("video") - .unwrap() - .subscribe(Subscription::default().with_ranges([Range::live()])) - .await - .expect("subscribe"); - let first = moq_clock::timeout(TIMEOUT, live.recv_group()) - .await - .expect("group timeout") - .unwrap() - .expect("a group"); - assert_eq!(first.sequence, 2); - - assert_eq!(fixture.fetch(0).await.expect("fetched"), "0"); - assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); - - write_group(&fixture.track, 3); - let next = moq_clock::timeout(TIMEOUT, live.recv_group()) - .await - .expect("group timeout") - .unwrap() - .expect("a group"); - assert_eq!(next.sequence, 3, "the live feed goes on"); - let subscription = fixture.track.subscription().expect("still subscribed"); - assert_eq!(subscription.ranges, [Range::live()], "the filled range is retired"); -} - -/// A group the publisher never had is a miss, named by the subscription it was asked on -/// once the publisher's feed gives up on it, past the budget's grace. -#[moq_clock::test] -async fn a_group_the_publisher_never_had_is_a_miss() { - let fixture = Fixture::new(LITE07, &[0, 2, 3]).await; - let fetching = fixture.remote.track("video").unwrap().fetch_group( - 1, - moq_net::group::Fetch::default().with_max_delay(Duration::from_secs(1)), - ); - let err = match moq_clock::timeout(Duration::from_secs(120), fetching) - .await - .expect("fetch timed out") - { - Ok(_) => panic!("group 1 was served"), - Err(err) => err, - }; - assert!(matches!(err, moq_net::Error::NotFound), "{err}"); - assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); -} - -/// A group whose stream an earlier subscription cut is filled whole by the next one, -/// rather than judged missing from what the earlier one left behind. -#[moq_clock::test] -async fn a_cut_group_is_filled_whole() { - let fixture = Fixture::new(LITE07, &[]).await; - let mut open = fixture - .track - .create_group(moq_net::group::Info { sequence: 0 }) - .unwrap(); - open.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); - - let mut live = fixture - .remote - .track("video") - .unwrap() - .subscribe(Subscription::default().with_ranges([Range::live()])) - .await - .expect("subscribe"); - let mut group = moq_clock::timeout(TIMEOUT, live.recv_group()) - .await - .expect("group timeout") - .unwrap() - .expect("a group"); - let frame = moq_clock::timeout(TIMEOUT, group.read_frame()) - .await - .expect("read timed out") - .unwrap() - .expect("a frame"); - assert_eq!(frame.payload.as_ref(), b"head"); - // The only reader leaves mid-group, so the subscription goes and cuts the copy. - drop((group, live)); - moq_clock::timeout(TIMEOUT, support::unwanted(fixture.track.demand())) - .await - .expect("the subscription ends") - .unwrap(); - - open.write_frame(Timestamp::from_millis(1).unwrap(), b"tail".as_slice()) - .unwrap(); - open.finish().unwrap(); - let mut fetched = moq_clock::timeout(TIMEOUT, fixture.remote.track("video").unwrap().fetch_group(0, None)) - .await - .expect("fetch timed out") - .expect("fetched"); - let mut frames = Vec::new(); - while let Some(frame) = moq_clock::timeout(TIMEOUT, fetched.read_frame()) - .await - .expect("read timed out") - .expect("read") - { - frames.push(frame.payload.to_vec()); - } - assert_eq!(frames, [b"head".to_vec(), b"tail".to_vec()]); - assert_eq!(fixture.sent(Message::Fetch), 0, "no FETCH"); -} From 1720721f8ca99686d52c794634d0a8dfac2fe1a6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 10:18:39 -0800 Subject: [PATCH 4/4] quest(m1): lite-07 ranges require fetch-range's miss judging Record the maintainer's 2026-11-01 decision that the range fill waits for Fetch as a range's miss judging and cached backfill, keep the first attempt's design and its failing tests in the Plan, carry the datagram, partial-group, and budget questions forward, and point the covered-FIN end at fetch-range's explicit end signal. Co-Authored-By: Claude Opus 5.5 --- quest/m1/subscribe-ranges/lite.md | 122 ++++++++++++++++++++++++------ 1 file changed, 97 insertions(+), 25 deletions(-) diff --git a/quest/m1/subscribe-ranges/lite.md b/quest/m1/subscribe-ranges/lite.md index 8ed65d2d990..b528dc5d1d9 100644 --- a/quest/m1/subscribe-ranges/lite.md +++ b/quest/m1/subscribe-ranges/lite.md @@ -15,6 +15,13 @@ moves out of this quest (to its own quest since 2026-10-27), since JS still requ cross-language capability. This quest keeps FETCH on lite-07 and makes the range fill work without it. +Decided 2026-11-01 (maintainer, on +[moq-dev/moq#7487](https://github.com/moq-dev/moq/pull/7487)): this quest +waits for [Fetch as a range](/quest/m1/subscribe-ranges/fetch-range.md)'s miss +judging (a miss with no feed frontier, and holes below where a live feed +started) and its cached backfill rule. Without them a range fill is a +functional regression from FETCH, which answers a miss at once. + ### Range fill through the subscription Lite-07 SUBSCRIBE and SUBSCRIBE_UPDATE carry the range list (`lite::Range`, @@ -23,35 +30,100 @@ the codec, and a merged request goes upstream as one span, from group 0 when its starts mix (`wire_span`). A lite-07 publisher FINs a covered bounded subscription without SUBSCRIBE_END; the Rust copy settles its tail, keeps the cached copy, and subscribes again only for demand the covered ranges leave -out. - -An attempt that routed lite-07 fills through `FeedFetches`, adding each held -fill's groups to the upstream ranges, broke fetches of an open or previously -cut group (`catalog_handover` fetch tests, `restart::retired_fetch_lite07`, -`fetch_cancel`, `untimed`, `rejoin`): - -- A held fill is judged by the track's arrivals ledger, which still holds the - cut group from the subscription that left. It settles `NotFound` before the - new subscription delivers the group. Judge it against the subscription that - asked for it. -- Dropping a fill's range once it is answered shrinks or closes the upstream - subscription while its group stream is still arriving, which resets it. - Keep the range until that stream finishes. -- A relay serving the fill from its own cut copy has the same problem one hop - up. +out. The maintainer has since directed an explicit end signal (SUBSCRIBE_END +or similar) in place of that FIN, since a FIN prevents later updates (on +[moq-dev/moq#7545](https://github.com/moq-dev/moq/pull/7545)). +[Fetch as a range](/quest/m1/subscribe-ranges/fetch-range.md) owns that +design; build the fill on whatever it settles, not on the FIN. + +The first attempt +([moq-dev/moq#7487](https://github.com/moq-dev/moq/pull/7487), WIP commit +[b7f573d](https://github.com/moq-dev/moq/commit/b7f573d755ae53e6f2bdb94cc480ae47fd6ba254)) built this in +`rs/moq-net/src/lite/subscriber.rs` and was moved back out until the +requirement lands. Reuse its design: + +- A `RangeFills` set holds each fill and asks for its next owed run + (`Fill::next_run`), added as an absolute bounded range after the demand's + ranges on the upstream SUBSCRIBE or SUBSCRIBE_UPDATE. With no demand open, + a bounded SUBSCRIBE asks for the fills alone. A fill answers itself from the + cache as its groups arrive. +- A fill is judged only against the subscription that asked for it, not the + track's arrivals ledger, which still holds a group an earlier subscription + cut. A SUBSCRIBE_DROP over its groups, a SUBSCRIBE_START past its range's + start, or a covered end refuses it, so a cut group is filled whole. +- A run stays on the wire while a fill owes part of it, then until each of its + streams still being read finishes, so narrowing never resets a group in + flight. +- Only the demand's ranges set the floor and the live state. A subscription + carrying only fills sets neither, and canceling it does not idle the copy. + If the demand leaves while fills are out, the subscription keeps the + demand's ranges until they are answered. +- A NotFound drop or a covered end names the fill's groups dropped + (`Fill::drop_groups`), as a FETCH miss does; Old and NOT_FETCHABLE reject + them, and any other error refuses only that request. + +Its integration tests (`range_fill.rs`: a miss subscribes and sends no FETCH, +a miss rides the live subscription then retires its range, a group the +publisher never had becomes a miss after the grace, a cut group is filled +whole) passed. These failed, and must pass once the requirement lands: + +- On-demand `requests()` tracks with no live feed, such as moq-archive's, never + ask a fill and park (`untimed::a_fetched_untimed_frame_arrives_untimed`, + `fetch_cancel::a_refetch_within_the_linger_stays_on_the_relay_lite07`, + `subscription_end_integrity::a_session_close_cancels_a_pending_fetch_from_either_side`). + Fetch as a range's miss with no feed frontier covers them. +- A missing or expired group fetched with no budget parks where FETCH answered + NotFound at once (`rejoin::{rejoined,taken_over}_quiet_catalog_survives_cache_expiry`). + Fetch as a range's holes below where a live feed started cover them. +- A relay re-asks upstream for groups it cached, since its covered memory + leaves with the downstream demand + (`fetch_fanout::sequential_viewers_share_the_relays_fetched_groups`, two + upstream SUBSCRIBEs where FETCH needed one). Fetch as a range's cached + backfill rule covers it: a relay leaves cached groups out of a bounded + upstream range. On lite-07 the test counts SUBSCRIBE streams, not FETCH. + +Open questions the attempt raised, each with the attempt's recommendation: + +- **Datagram sequences.** A range fill of a sequence served only as a datagram + never completes: datagrams resolve no SUBSCRIBE_START, so a FIN or end + overtaking them leaves the receiver no owed range to wait on + (`datagram::a_fetched_datagram_is_refused_per_version`). Recommended: a + lite-07 publisher sends SUBSCRIBE_START for a bounded past range, then + SUBSCRIBE_DROP with NOT_FETCHABLE for each sequence it holds only as a + datagram; the subscriber already maps that drop onto the fill. This is a + wire change. The alternative keeps FETCH for a fill the first answer shows is + a datagram, at a round trip and a second path. +- **Fills from partway through a group.** Route recovery (`resume::Recover`) + fills from frame N, which a whole-group range cannot say. Recommended: keep + FETCH on lite-07 for a fill whose first group starts past frame 0, matching + Fetch as a range's crate-private fetch for that case. The alternative asks + for the whole group and drops the frames the reader already has. +- **One budget caps every range.** Merging the fills into the upstream + Subscriber Max Age takes the most tolerant budget, which loosens the live + feed's cap while a fill is out + (`route_change::resume_delivers_the_outage_tail_within_the_budget` delivers + a stale group 5). Recommended: settle the partial-group question first, + since the failing fill is a recovery fill, and rerun the test. The + alternatives fill under the demand's budget, refusing older groups as Old, + or add a per-range budget to the wire. +- `auth::lite_07_narrowing_resets_a_fetch_in_flight` read `Stream(Cancel)` + instead of `Unauthorized`/`Unroutable` when the grant narrowed mid-group; + the refusal must carry the grant's error. The Rust copy sends a relative or time end as open, since nothing names a resolved end and it resolves the end itself (`wire_ranges`), so only absolute bounded ranges complete upstream. A hole on a live track counts as dropped -only once the feed gives up on it, past the arrivals grace; -[Fetch as a range](/quest/m1/subscribe-ranges/fetch-range.md) owns judging -such a miss sooner. A range served only by datagrams never completes: they -resolve no SUBSCRIBE_START, so a FIN overtaking them leaves the receiver no -owed range to wait on. Datagram fills need the receiver told what is owed -first. - -Update `drafts/draft-lcurley-moq-lite.md` with each wire change here. Run -`just drafts check` and `just test interop --all`. +only once the feed gives up on it, past the arrivals grace, until Fetch as a +range judges such a miss sooner. + +Update `drafts/draft-lcurley-moq-lite.md` with each wire change here, +including that a subscriber fills missing groups through its ranges. Run +`just drafts check` and `just test interop --all`, and the `moq-net-tests` +`restart::retired_fetch_lite07` and `moq-relay` suites. + +## Required + +- [Fetch as a range](/quest/m1/subscribe-ranges/fetch-range.md) - its miss judging and cached backfill rule, which a range fill needs to answer a miss as FETCH does ## Related