diff --git a/quest/m1/README.md b/quest/m1/README.md index b644de46934..6795ed9ad1e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -104,7 +104,6 @@ lands before the cut. - [A widened max age re-offers cached groups](/quest/m1/resolved-demand.md) - a warm relay serves a reader with a wider max age the cached groups a cold relay would, with no wire change - [Drop set](/quest/m1/drop-set.md) - each cursor judges missing groups against its own budget, so a narrow reader drops promptly and a widening re-offers groups already reported dropped on lite-07 - [SUBSCRIBE_OK names where delivery starts](/quest/m1/subscribe-ok-start.md) - a warm relay names the same start group as a cold one instead of groups it skipped as stale -- [Rust resumed reads](/quest/m1/resume-ordered.md) - a Rust reader takes the lowest group across route copies in both orders, as JS does - [Resumed close](/quest/m1/resume-closed.md) - a resumed JS track keeps its declared end across copies, and its `closed` and `largest()` settle on a clean end - [Resumed in-flight groups](/quest/m1/resume-in-flight.md) - a replaced route copy stays while a group below its bound is in flight or it backs a handed-out group late, and no longer than the reader's budget - [Resumed forks](/quest/m1/resume-fork.md) - a fork of a resumed JS subscriber follows route swaps like its parent diff --git a/quest/m1/resume-closed.md b/quest/m1/resume-closed.md index b7073ab9b3c..572abc8f58d 100644 --- a/quest/m1/resume-closed.md +++ b/quest/m1/resume-closed.md @@ -79,7 +79,3 @@ Regression tests: read pending rejects that read with the abort. Public API: behavior only. Wire: none. - -## Related - -- [Rust resumed reads](/quest/m1/resume-ordered.md) - another resume follow-up from #5836 diff --git a/quest/m1/resume-in-flight.md b/quest/m1/resume-in-flight.md index 323459d99d0..45f4e863860 100644 --- a/quest/m1/resume-in-flight.md +++ b/quest/m1/resume-in-flight.md @@ -104,4 +104,3 @@ Public API: behavior only. Wire: none. ## Related - [Resume reorder](/quest/m1/resume-reorder.md) - the gap rule this reuses -- [Rust resumed reads](/quest/m1/resume-ordered.md) - another resume follow-up from #5836 diff --git a/quest/m1/resume-ordered.md b/quest/m1/resume-ordered.md deleted file mode 100644 index 7270a4feb09..00000000000 --- a/quest/m1/resume-ordered.md +++ /dev/null @@ -1,43 +0,0 @@ -# [S] Rust resumed reads take the lowest group across route copies - -## Goal - -A Rust reader of a logical track across a same-epoch route swap takes the -lowest buffered group across every route copy, in both orders. In sequence -order (`poll_next_group`) it never skips a group a replaced copy already holds -because the serving copy offered a later one first. In arrival order -(`poll_recv_group`) it hands out the lowest buffered group first, as a direct -reader does. This matches JS since #5836. - -## Plan - -Decided 2026-10-17 by the maintainer on #5836: JS keeps reading the lowest next -group across copies, and Rust follows rather than JS reverting. A later review -round on #5836 extended the JS rule to arrival reads, so this covers both -orders. - -Today `Reader::poll_group_once` in `rs/moq-net/src/model/resume.rs` polls the -serving copy first in both orders. In sequence order, `track::Subscriber::hand_out` -then forgets every sequence below the group it returns. So a replaced copy's -buffered N+1 is dropped when the serving copy offers N+2 first, a group a direct -ordered reader would have delivered. - -- Mirror JS `Reader.#lowest`: peek each live copy's next deliverable group - without moving its cursor, drop what is not deliverable, and take from the - copy with the lowest one, in either order. Arrival order still delivers a - lower group that truly arrives later. -- The peek needs a non-consuming counterpart of `poll_next_group` and - `poll_recv_group` on `track::Subscriber` that applies the same floor, cap, - and drift budget (JS `hooks.peekGroup`). -- Feed the taken group through `Reader::observe` as the other paths do, so the - deferred bound on replaced copies still applies. -- Regression test, in both orders: a reader across a swap where the old copy - holds group 1 unread and the new copy only has group 2 reads 1 then 2, as the - JS `... takes a replaced copy's buffered group before the new copy's later - one` tests do. - -Public API: behavior only. Wire: none. - -## Related - -- [Resume reorder](/quest/m1/resume-reorder.md) - another resume verdict that reads a passed group too eagerly diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 461dc9185ab..1be9bfb6184 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -1277,7 +1277,10 @@ fn bench_copy_walk(c: &mut Criterion) { /// `(readers, retained, routes)` for [`bench_resume_overlap`]: readers of one track, /// the groups every route holds, and the routes it swapped through. -const RESUME_OVERLAP: [(usize, usize, usize); 12] = [ +const RESUME_OVERLAP: [(usize, usize, usize); 15] = [ + (1, 1, 1), + (1, 64, 1), + (32, 64, 1), (1, 1, 2), (1, 64, 2), (1, 1024, 2), @@ -1298,10 +1301,24 @@ const RESUME_OVERLAP: [(usize, usize, usize); 12] = [ struct ResumeOverlap { _producer: origin::Producer, _generations: Vec<(broadcast::Producer, track::Producer)>, - readers: Vec, + readers: Vec, } -async fn resume_overlap(readers: usize, retained: usize, routes: usize) -> ResumeOverlap { +enum ResumeReader { + Arrival(track::Subscriber), + Sequence(track::Ordered), +} + +impl ResumeReader { + fn poll_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + match self { + Self::Arrival(reader) => reader.poll_recv_group(waiter), + Self::Sequence(reader) => reader.poll_next_group(waiter), + } + } +} + +async fn resume_overlap(readers: usize, retained: usize, routes: usize, ordered: bool) -> ResumeOverlap { let (producer, driver) = origin::Producer::new(origin::Config::default()); tokio::spawn(moq_net::time::run(driver)); let epoch = Epoch::mint(); @@ -1359,10 +1376,17 @@ async fn resume_overlap(readers: usize, retained: usize, routes: usize) -> Resum for subscriber in &mut subscribers { subscriber.set_groups(..); } + let readers = subscribers + .into_iter() + .map(|subscriber| match ordered { + true => ResumeReader::Sequence(subscriber.ordered()), + false => ResumeReader::Arrival(subscriber), + }) + .collect(); ResumeOverlap { _producer: producer, _generations: generations, - readers: subscribers, + readers, } } @@ -1372,31 +1396,33 @@ async fn resume_overlap(readers: usize, retained: usize, routes: usize) -> Resum fn bench_resume_overlap(c: &mut Criterion) { let mut group = c.benchmark_group("origin/resume_overlap"); group.sample_size(10); - for (readers, retained, routes) in RESUME_OVERLAP { - group.throughput(Throughput::Elements((readers * retained) as u64)); - group.bench_function( - BenchmarkId::from_parameter(format!("{readers}r_{retained}g_{routes}c")), - |b| { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); - let waiter = kio::Waiter::noop(); - b.iter_batched_ref( - || runtime.block_on(resume_overlap(readers, retained, routes)), - |rig| { - for reader in &mut rig.readers { - let mut delivered = 0; - while let Poll::Ready(Ok(Some(_))) = reader.poll_recv_group(&waiter) { - delivered += 1; + for (order, ordered) in [("arrival", false), ("sequence", true)] { + for (readers, retained, routes) in RESUME_OVERLAP { + group.throughput(Throughput::Elements((readers * retained) as u64)); + group.bench_function( + BenchmarkId::new(order, format!("{readers}r_{retained}g_{routes}c")), + |b| { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let waiter = kio::Waiter::noop(); + b.iter_batched_ref( + || runtime.block_on(resume_overlap(readers, retained, routes, ordered)), + |rig| { + for reader in &mut rig.readers { + let mut delivered = 0; + while let Poll::Ready(Ok(Some(_))) = reader.poll_group(&waiter) { + delivered += 1; + } + assert_eq!(delivered, retained, "every group once"); } - assert_eq!(delivered, retained, "every group once"); - } - }, - BatchSize::PerIteration, - ); - }, - ); + }, + BatchSize::PerIteration, + ); + }, + ); + } } group.finish(); } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index bbd3762b759..e82a3a24126 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -190,6 +190,7 @@ impl Consumer { datagram: None, groups: (floor, Cap::from(..)), groups_read: false, + peeked: None, }) }); // Subscribed to the serving copy at once, so groups it has now are the reader's @@ -398,6 +399,28 @@ impl Subscriber { self.reader().poll_group(ordered, delivered, waiter) } + /// The lowest next group across copies, without handing it out. + pub(super) fn poll_peek_group( + &mut self, + ordered: bool, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Poll>> { + let mut reader = self.reader(); + reader.peeked = None; + loop { + if let Some(result) = reader.poll_lowest(ordered, delivered, waiter) { + return result.map(|result| { + result.map(|next| { + let sequence = next.as_ref().map(|next| next.sequence); + reader.peeked = next; + sequence + }) + }); + } + } + } + /// Poll for the next datagram in arrival order. `delivered` decides which replaced /// copies still have a group to hand out. pub(super) fn poll_recv_datagram( @@ -536,8 +559,17 @@ impl Subscriber { } } +struct Candidate { + generation: u64, + sequence: u64, + /// A nested reader may sync a route swap before its caller consumes this + /// candidate. Keep its exact copy alive until that commit. + _lease: Arc<()>, +} + /// A [`Subscriber`]'s state. struct Reader { + peeked: Option, this: Weak>, route: Consumer, /// The reader's own preferences, mirrored into every copy's subscription. @@ -816,11 +848,52 @@ impl Reader { delivered: &Delivered, waiter: &kio::Waiter, ) -> Option>>> { + let candidate = match self.peeked.take() { + Some(candidate) => candidate, + None => match self.poll_lowest(ordered, delivered, waiter)? { + Poll::Ready(Ok(Some(next))) => next, + Poll::Ready(Ok(None)) => return Some(Poll::Ready(Ok(None))), + Poll::Ready(Err(err)) => return Some(Poll::Ready(Err(err))), + Poll::Pending => return Some(Poll::Pending), + }, + }; + let index = self + .copies + .iter() + .position(|copy| copy.generation == candidate.generation) + .expect("the candidate's lease retains its copy"); + let Sub::Ready(sub) = &mut self.copies[index].sub else { + unreachable!() + }; + let result = match ordered { + true => sub.poll_next_group(waiter), + false => sub.poll_recv_group(waiter), + }; + // Selection reserves the exact group, so eviction can abort that group but + // cannot substitute a later one and advance the logical reader past a gap. + let Poll::Ready(Ok(Some(group))) = result else { + unreachable!("the peek reserved a group") + }; + assert_eq!(group.sequence, candidate.sequence, "the peek reserved this exact group"); + self.observe(index, group.sequence); + let group = self.hand_out(index, group); + Some(Poll::Ready(Ok(Some(group)))) + } + + /// Choose without consuming a candidate, so a newer copy cannot overtake a lower + /// group still buffered on an older one. A tie favors the newer copy, which bounds + /// the copies it replaced when it hands that group out. + fn poll_lowest( + &mut self, + ordered: bool, + delivered: &Delivered, + waiter: &kio::Waiter, + ) -> Option>>> { self.sync(waiter); self.groups_read = true; let groups = self.groups; - // The serving copy first: it is the one that keeps delivering. - for index in (0..self.copies.len()).rev() { + let mut lowest: Option = None; + for index in 0..self.copies.len() { let copy = &mut self.copies[index]; if copy.done.is_some() { continue; @@ -835,19 +908,28 @@ impl Reader { } Poll::Ready(Ok(sub)) => sub, }; - let res = match ordered { - true => sub.poll_next_group(waiter), - false => sub.poll_recv_group(waiter), - }; + let res = sub.poll_peek_group(ordered, waiter); match res { Poll::Pending => break, - Poll::Ready(Ok(Some(group))) => { - if !Self::deliverable(&self.copies[index], group.sequence, delivered) { + Poll::Ready(Ok(Some(sequence))) => { + if !Self::deliverable(&self.copies[index], sequence, delivered) { + let Sub::Ready(sub) = &mut self.copies[index].sub else { + unreachable!() + }; + let _ = match ordered { + true => sub.poll_next_group(waiter), + false => sub.poll_recv_group(waiter), + }; continue; } - self.observe(index, group.sequence); - let group = self.hand_out(index, group); - return Some(Poll::Ready(Ok(Some(group)))); + if lowest.as_ref().is_none_or(|lowest| sequence <= lowest.sequence) { + lowest = Some(Candidate { + generation: self.copies[index].generation, + sequence, + _lease: self.copies[index].lease.clone(), + }); + } + break; } // Drained: the track ended as the copy did, which is not cleanly when // its session went away short of the declared end. @@ -865,6 +947,9 @@ impl Reader { } } } + if let Some(lowest) = lowest { + return Some(Poll::Ready(Ok(Some(lowest)))); + } self.let_go(delivered); self.poll_end(waiter, |copy| copy.done.clone()) } @@ -1043,6 +1128,7 @@ impl Reader { } fn raise_start_to(&mut self, start: u64) { + self.peeked = None; self.groups.0 = self.groups.0.max(start); for copy in &mut self.copies { if let Sub::Ready(sub) = &mut copy.sub { @@ -1057,6 +1143,7 @@ impl Reader { /// resume point: groups under it were handed out by an earlier route, or never owed. #[cfg_attr(not(feature = "async"), allow(dead_code))] fn start_at(&mut self, start: u64) { + self.peeked = None; self.groups.0 = start; for copy in &mut self.copies { if let Sub::Ready(sub) = &mut copy.sub { @@ -1066,6 +1153,7 @@ impl Reader { } fn end_at(&mut self, end: Cap) { + self.peeked = None; self.groups.1 = end; for copy in &mut self.copies { if let Sub::Ready(sub) = &mut copy.sub { @@ -1916,6 +2004,88 @@ mod test { } } + /// A replaced copy is bounded only once the new copy delivers its first group: a group + /// buffered on it must be read before a later group on the serving copy. + #[test] + fn sequence_reads_take_a_replaced_copys_lower_buffered_group() { + replaced_copy_reads_lowest(true); + } + + #[test] + fn arrival_reads_take_a_replaced_copys_lower_buffered_group() { + replaced_copy_reads_lowest(false); + } + + fn replaced_copy_reads_lowest(ordered: bool) { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 0); + assert_eq!(drain(&mut sub, ordered), [0]); + write(&a, 1); + let b = copy(); + routes.serve(b.consume()); + write(&b, 2); + assert_eq!(drain(&mut sub, ordered), [1, 2]); + } + + #[test] + fn a_reserved_group_survives_eviction_and_nested_route_replacement() { + for ordered in [false, true] { + for nested in [false, true] { + let routes = Producer::new(); + let inner = logical(&routes); + let outer_routes = Producer::new(); + let outer = logical(&outer_routes); + outer_routes.serve(inner.consume()); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(if nested { &outer } else { &inner }, HOUR); + let mut one = a.create_group(group::Info { sequence: 1 }).unwrap(); + one.write_frame(ts(1), b"x".as_ref()).unwrap(); + one.finish().unwrap(); + write(&a, 3); + let waiter = kio::Waiter::noop(); + assert_eq!( + sub.poll_peek_group(ordered, &waiter).map(Result::unwrap), + Poll::Ready(Some(1)) + ); + one.abort(Error::Evicted).unwrap(); + let b = copy(); + routes.serve(b.consume()); + write(&b, 2); + let next = match ordered { + true => sub.poll_next_group(&waiter), + false => sub.poll_recv_group(&waiter), + }; + let Poll::Ready(Ok(Some(group))) = next else { + panic!("the reserved group") + }; + assert_eq!(group.sequence, 1, "ordered={ordered}, nested={nested}"); + assert_eq!(drain(&mut sub, ordered), [2], "ordered={ordered}, nested={nested}"); + } + } + } + + #[test] + fn arrival_reads_still_deliver_a_lower_group_that_arrives_later() { + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = subscribe(&logical, HOUR); + write(&a, 0); + assert_eq!(drain(&mut sub, false), [0]); + let b = copy(); + routes.serve(b.consume()); + write(&b, 2); + assert_eq!(drain(&mut sub, false), [2]); + write(&b, 1); + assert_eq!(drain(&mut sub, false), [1]); + } + /// A replaced copy is bounded only once the new copy delivers its first group: a group /// the old route gets after the swap, which the new route never has, still arrives. #[test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 24d9ea02c76..c91cfc6a5a8 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4858,6 +4858,9 @@ struct Cursor { /// `u64::MAX` and no group can follow. /// Used only by that method to skip late arrivals; does not affect `recv_group`. next_sequence: Option, + /// The exact candidate a route-copy selection inspected, including its arrival + /// position. Consuming it must not look up a different group after eviction. + peeked: Option>, /// Exclusive upper sequence bound for `next_group` and `recv_group`, in the form /// [`Cap::exclusive`] produces. `None` means no cap. Set by `end_at`; /// can be raised, lowered, or unset at any time. Groups at or past the cap stay in @@ -4883,6 +4886,12 @@ struct Cursor { drops: Option>, } +struct Peeked { + group: group::Consumer, + index: Option, + ordered: bool, +} + /// The sequences a [`DropCursor`] reports from: a cursor's span, narrowed to its window. struct Within<'a> { span: ops::Range, @@ -5034,6 +5043,7 @@ impl Cursor { index: 0, datagram_index, next_sequence: Some(0), + peeked: None, end_sequence: None, servable: Servable::ALL, // Replaced by the cursor's own below, before anything is handed out. @@ -5174,8 +5184,9 @@ impl Cursor { self.filter.admitted_at = Some(at); } - /// The lowest cached group still owed, consuming the spans scanned to find it: only - /// one the walk could have read, never fetched backfill or a withheld group. + /// The lowest cached group still owed, consuming the spans scanned below it: only one + /// the walk could have read, never fetched backfill or a withheld group. The group + /// itself stays owed until [`Self::advance_arrival`] takes it, so a peek keeps it. fn next_owed(&mut self) -> Option { if self.filter.owed.is_empty() { return None; @@ -5200,7 +5211,11 @@ impl Cursor { .flatten(); match found { Some(group) => { - self.filter.owed.remove(*run.start()..=group.sequence); + if let Some(below) = group.sequence.checked_sub(1) + && *run.start() <= below + { + self.filter.owed.remove(*run.start()..=below); + } return Some(group.clone()); } None => self.filter.owed.remove(run), @@ -5308,27 +5323,41 @@ impl Cursor { })) } - fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + fn poll_recv_group(&mut self, take: bool, waiter: &kio::Waiter) -> Poll>> { + if let Some(peeked) = self.peeked.take().filter(|peeked| take && !peeked.ordered) { + self.advance_arrival(peeked.group.sequence, peeked.index); + peeked.group.cache_refresh(); + return Poll::Ready(Ok(Some(self.with_expiry(peeked.group)))); + } self.sync(); self.admit(); // One scan for the whole poll, so walking a backlog off stays linear in its size. let drift = ready!(self.poll_drift(&self.servable, waiter))?; - self.poll_recv_with(drift, waiter) + self.poll_recv_with(take, drift, waiter) } - /// [`Self::poll_recv_group`] once synced, judging against `drift`. - fn poll_recv_with(&mut self, mut drift: Drift, waiter: &kio::Waiter) -> Poll>> { + /// [`Self::poll_recv_group`] once synced, judging against `drift`. Without `take`, it + /// peeks: the candidate stays at its cursor position until a taking read. + fn poll_recv_with( + &mut self, + take: bool, + mut drift: Drift, + waiter: &kio::Waiter, + ) -> Poll>> { loop { // Re-offer what a widening newly covers ahead of the arrival walk, lowest first: // it is the oldest thing still owed. A group the walk has yet to reach comes // again from it, and the subscriber's delivered set drops the repeat. - let consumer = match self.next_owed() { + let (consumer, next_index) = match self.next_owed() { Some(producer) => { let consumer = producer.consume(); - // A re-offer is a delivery: stamp it like a fresh hand-out. - consumer.cache_refresh(); - consumer + // A re-offer is a delivery: stamp it like a fresh hand-out. Peeking is + // not an access. + if take { + consumer.cache_refresh(); + } + (consumer, None) } None => { let floor = self.floor(); @@ -5339,9 +5368,10 @@ impl Cursor { }; let consumer = producer.consume(); // Stamp with the track guard released, so delivery never nests the - // group's state lock under the track's. - consumer.cache_refresh(); - self.index = found_index + 1; + // group's state lock under the track's. Peeking is not an access. + if take { + consumer.cache_refresh(); + } // The first group to arrive resolves a relative floor, which may land // above it. @@ -5353,6 +5383,7 @@ impl Cursor { drift = ready!(self.poll_drift(&self.servable, waiter))?; } if consumer.sequence < self.floor() { + self.index = found_index + 1; continue; } } @@ -5361,15 +5392,22 @@ impl Cursor { // widening finds it in the cache again. Keep scanning so an in-range group // that arrived behind it still flows. if !self.admits(consumer.sequence) { + self.index = found_index + 1; continue; } - consumer + (consumer, Some(found_index + 1)) } }; // Drop a group the drift budget has given up on and keep scanning, so one // poll walks a whole backlog off rather than handing it out group by group. - if ready!(self.poll_stale(&consumer, &drift, waiter))? { + let stale = ready!(self.poll_stale(&consumer, &drift, waiter))?; + // A peek discards what cannot be delivered, but keeps its candidate at the + // same cursor position until the reader chooses this copy. + if take || stale { + self.advance_arrival(consumer.sequence, next_index); + } + if stale { self.stale.add(consumer.content()); if let Some(drops) = &mut self.drops && let Some(end) = consumer.sequence.checked_add(1) @@ -5378,7 +5416,25 @@ impl Cursor { } continue; } - return Poll::Ready(Ok(Some(self.with_expiry(consumer)))); + return Poll::Ready(Ok(Some(if take { + self.with_expiry(consumer) + } else { + self.peeked = Some(Box::new(Peeked { + group: consumer.clone(), + index: next_index, + ordered: false, + })); + consumer + }))); + } + } + + /// Move the arrival cursor past a group it read: from the walk, to `index`; from what + /// a widening owes, out of the owed spans. + fn advance_arrival(&mut self, sequence: u64, index: Option) { + match index { + Some(index) => self.index = index, + None => self.filter.owed.remove(sequence..=sequence), } } @@ -5411,7 +5467,10 @@ impl Cursor { /// The lowest servable group past the last one returned, walking off everything the /// drift budget has convicted on the way. - fn poll_next_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + fn poll_next_group(&mut self, take: bool, waiter: &kio::Waiter) -> Poll>> { + if let Some(peeked) = self.peeked.take().filter(|peeked| take && peeked.ordered) { + return Poll::Ready(Ok(Some(self.advance_sequence(peeked.group)))); + } let Some(next) = self.next_sequence else { return Poll::Ready(Ok(None)); }; @@ -5419,12 +5478,14 @@ impl Cursor { let floor = next.max(self.floor()); // One scan for the whole poll, so walking a backlog off stays linear in its size. let drift = ready!(self.poll_drift(&self.servable, waiter))?; - self.poll_next_with(floor, drift, waiter) + self.poll_next_with(take, floor, drift, waiter) } /// [`Self::poll_next_group`] once synced, from `floor`, judging against `drift`. + /// Without `take`, it peeks: the candidate is reserved, not delivered. fn poll_next_with( &mut self, + take: bool, mut floor: u64, mut drift: Drift, waiter: &kio::Waiter, @@ -5470,15 +5531,27 @@ impl Cursor { // A conviction the budget walked back: the group is handed over after all, // so it must never reach the stale count. self.seek_pending.remove(&group.sequence); - break self.with_expiry(group); + break group; }; + Poll::Ready(Ok(Some(if take { + self.advance_sequence(group) + } else { + self.peeked = Some(Box::new(Peeked { + group: group.clone(), + index: None, + ordered: true, + })); + group + }))) + } + + fn advance_sequence(&mut self, group: group::Consumer) -> group::Consumer { self.next_sequence = group.sequence.checked_add(1); // The delivery commits everything the seek stepped over to reach this group. self.commit_seek_stale(self.next_sequence.unwrap_or(u64::MAX)); - // Delivery is a cache access, same as the arrival-order path. group.cache_refresh(); - Poll::Ready(Ok(Some(group))) + self.with_expiry(group) } /// Count the sequence path's convictions below `committed` into [`Self::stale`], @@ -5637,6 +5710,8 @@ impl Subscriber { /// edge and writes the rest off; raise it to read history. [`Self::set_groups`] and /// [`Subscription::ranges`] are filters, not exemptions: backfill needs a budget that /// covers it. [`Consumer::fetch_group`] is the way to ask for one old group outright. + /// Across same-epoch route copies, the lowest next buffered group is delivered + /// first; a lower group that arrives later is still delivered in arrival order. /// The budget remains attached to a returned group: if it stalls while newer data /// advances, its pending frame read ends with [`Error::Old`]. /// @@ -5661,8 +5736,8 @@ impl Subscriber { let res = match &mut self.inner { Inner::Plain(cursor) => loop { let res = match ordered { - true => cursor.poll_next_group(waiter), - false => cursor.poll_recv_group(waiter), + true => cursor.poll_next_group(true, waiter), + false => cursor.poll_recv_group(true, waiter), }; match res { Poll::Ready(Ok(Some(group))) if self.delivered.contains(group.sequence) => continue, @@ -5678,6 +5753,31 @@ impl Subscriber { res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter)))) } + /// Reserve the next group without advancing past it, so the subsequent read + /// takes that exact group even if a producer evicts it meanwhile. + pub(super) fn poll_peek_group(&mut self, ordered: bool, waiter: &kio::Waiter) -> Poll>> { + match &mut self.inner { + Inner::Plain(cursor) => loop { + let group = ready!(match ordered { + true => cursor.poll_next_group(false, waiter), + false => cursor.poll_recv_group(false, waiter), + })?; + let Some(group) = group else { + return Poll::Ready(Ok(None)); + }; + if !self.delivered.contains(group.sequence) { + return Poll::Ready(Ok(Some(group.sequence))); + } + // A replay advances only this copy's cursor, as an ordinary read would. + ready!(match ordered { + true => cursor.poll_next_group(true, waiter), + false => cursor.poll_recv_group(true, waiter), + })?; + }, + Inner::Resume(resume, _) => resume.poll_peek_group(ordered, &self.delivered, waiter), + } + } + /// Record `sequence` as handed out. An ordered read never goes back, and any reader /// stops expecting what is past its budget, so only the gaps within it cost memory. fn hand_out(&mut self, sequence: u64, ordered: bool) { @@ -5987,7 +6087,10 @@ impl Subscriber { // pending: /quest/m1/public-model/serve.md pub(crate) fn start_at(&mut self, sequence: u64) { match &mut self.inner { - Inner::Plain(cursor) => cursor.min_sequence = sequence, + Inner::Plain(cursor) => { + cursor.peeked = None; + cursor.min_sequence = sequence; + } // Assigns, including downward. A front serves every viewer of a path, // and a later SUBSCRIBE_UPDATE can widen the floor; raising only would // leave a finished group below the old floor unread. @@ -6000,7 +6103,10 @@ impl Subscriber { // pending: /quest/m1/public-model/serve.md pub(crate) fn raise_start_to(&mut self, sequence: u64) { match &mut self.inner { - Inner::Plain(cursor) => cursor.min_sequence = cursor.min_sequence.max(sequence), + Inner::Plain(cursor) => { + cursor.peeked = None; + cursor.min_sequence = cursor.min_sequence.max(sequence); + } Inner::Resume(resume, _) => resume.raise_start_to(sequence), } self.largest.bounds_moved(); @@ -6023,6 +6129,7 @@ impl Subscriber { pub(crate) fn end_at(&mut self, end: impl Into) { match &mut self.inner { Inner::Plain(cursor) => { + cursor.peeked = None; cursor.end_sequence = end.into().exclusive(); cursor.update_servable(); } @@ -10058,8 +10165,8 @@ mod test { append_at(&mut producer, 0); append_at(&mut producer, 1000); let res = match ordered { - true => cursor.poll_next_with(0, drift, &waiter), - false => cursor.poll_recv_with(drift, &waiter), + true => cursor.poll_next_with(true, 0, drift, &waiter), + false => cursor.poll_recv_with(true, drift, &waiter), }; let Poll::Ready(Ok(Some(group))) = res else { panic!("group 0 is the newest group its range reads (ordered: {ordered})"); @@ -11272,6 +11379,135 @@ mod test { assert_eq!(group.sequence, 5); } + #[test] + fn changed_bounds_invalidate_a_peeked_group() { + for ordered in [false, true] { + for raise_floor in [false, true] { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay()); + for sequence in [1, 2] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + let waiter = kio::Waiter::noop(); + assert_eq!( + subscriber.poll_peek_group(ordered, &waiter).map(Result::unwrap), + Poll::Ready(Some(1)) + ); + if raise_floor { + subscriber.set_groups(2..); + let Poll::Ready(Ok(Some(group))) = subscriber.poll_group(ordered, &waiter) else { + panic!("group 2 is inside the new floor"); + }; + assert_eq!(group.sequence, 2); + } else { + subscriber.set_groups(..0); + assert!(subscriber.poll_group(ordered, &waiter).is_pending()); + subscriber.set_groups(..); + let Poll::Ready(Ok(Some(group))) = subscriber.poll_group(ordered, &waiter) else { + panic!("group 1 was parked, not consumed"); + }; + assert_eq!(group.sequence, 1); + } + } + } + } + + #[test] + fn taking_a_peeked_group_does_not_substitute_after_eviction() { + for ordered in [false, true] { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay()); + for sequence in [1, 3] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + let waiter = kio::Waiter::noop(); + assert_eq!( + subscriber.poll_peek_group(ordered, &waiter).map(Result::unwrap), + Poll::Ready(Some(1)) + ); + evict(&producer, 1); + let Poll::Ready(Ok(Some(group))) = subscriber.poll_group(ordered, &waiter) else { + panic!("the exact reserved group must be returned"); + }; + assert_eq!( + group.sequence, 1, + "ordered={ordered}: eviction must not substitute group 3" + ); + } + } + + /// A group a widening re-offers stays owed while it is only peeked. + #[test] + fn peeking_keeps_a_re_offered_group_owed() { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay()); + subscriber.set_groups(..0); + for sequence in [1, 2] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + let waiter = kio::Waiter::noop(); + assert!(subscriber.poll_group(false, &waiter).is_pending()); + subscriber.set_groups(..); + for _ in 0..2 { + assert_eq!( + subscriber.poll_peek_group(false, &waiter).map(Result::unwrap), + Poll::Ready(Some(1)) + ); + } + let mut read = Vec::new(); + while let Poll::Ready(Ok(Some(group))) = subscriber.poll_group(false, &waiter) { + read.push(group.sequence); + } + assert_eq!(read, [1, 2]); + } + + #[test] + fn peeking_keeps_each_orders_cursor_and_rechecks_bounds() { + for ordered in [false, true] { + let producer = track_producer("test", Info::timed()); + let mut subscriber = producer.subscribe(replay()); + for sequence in [2, 1] { + producer + .create_group(group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + let waiter = kio::Waiter::noop(); + let first = if ordered { 1 } else { 2 }; + for _ in 0..2 { + assert_eq!( + subscriber.poll_peek_group(ordered, &waiter).map(Result::unwrap), + Poll::Ready(Some(first)) + ); + } + let Poll::Ready(Ok(Some(group))) = subscriber.poll_group(ordered, &waiter) else { + panic!("the peeked group is still available"); + }; + assert_eq!(group.sequence, first); + let second = if ordered { 2 } else { 1 }; + subscriber.set_groups(..0); + assert!(subscriber.poll_peek_group(ordered, &waiter).is_pending()); + subscriber.set_groups(..); + assert_eq!( + subscriber.poll_peek_group(ordered, &waiter).map(Result::unwrap), + Poll::Ready(Some(second)) + ); + } + } + #[test] #[cfg(feature = "async")] fn ordered_and_arrival_cursors_are_independent() {