From 5cf352bdf398a20b70e0c44c6cbc9456c7433e04 Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:27:52 -0700 Subject: [PATCH 1/5] chore: claim resume-ordered quest Co-Authored-By: GPT-6.1-Sol From 88cc8c4d7cb01b436eed5ec9115d92a80d2f334c Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:37:31 -0700 Subject: [PATCH 2/5] fix(net): read lower buffered groups across route copies Co-Authored-By: GPT-6.1-Sol --- quest/m1/README.md | 1 - quest/m1/resume-closed.md | 4 - quest/m1/resume-in-flight.md | 1 - quest/m1/resume-ordered.md | 43 ----------- rs/moq-net/benches/origin.rs | 77 ++++++++++++------- rs/moq-net/src/model/resume.rs | 127 +++++++++++++++++++++++++++++--- rs/moq-net/src/model/track.rs | 131 ++++++++++++++++++++++++++------- 7 files changed, 271 insertions(+), 113 deletions(-) delete mode 100644 quest/m1/resume-ordered.md diff --git a/quest/m1/README.md b/quest/m1/README.md index cd43f0178f7..29aaf8fdb47 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -106,7 +106,6 @@ lands before the cut. - [Timed-out requests clean up](/quest/m1/legacy-request-timeout-cleanup.md) - on drafts 14-16, Rust and JS send PUBLISH_NAMESPACE_DONE or UNSUBSCRIBE_NAMESPACE for a timed-out request before its local reset - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails -- [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 df226650408..de4d5577a89 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -1298,10 +1298,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 +1373,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 +1393,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 f325a0f6d59..fcd7f691190 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -371,6 +371,21 @@ 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(); + loop { + if let Some(result) = reader.poll_lowest(ordered, delivered, waiter) { + return result.map(|result| result.map(|next| next.map(|next| 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( @@ -484,6 +499,11 @@ impl Subscriber { } } +struct Candidate { + index: usize, + sequence: u64, +} + /// A [`Subscriber`]'s state. struct Reader { this: Weak>, @@ -676,11 +696,44 @@ impl Reader { delivered: &Delivered, waiter: &kio::Waiter, ) -> Option>>> { + let index = match self.poll_lowest(ordered, delivered, waiter)? { + Poll::Ready(Ok(Some(next))) => next.index, + 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 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), + }; + if let Poll::Ready(Ok(Some(group))) = result + && Self::deliverable(&self.copies[index], group.sequence, delivered) + { + self.observe(index, group.sequence); + let group = self.hand_out(index, group); + return Some(Poll::Ready(Ok(Some(group)))); + } + // A producer may evict or replace the candidate between peeking and taking it. + None + } + + /// 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; @@ -695,19 +748,24 @@ 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 { index, sequence }); + } + break; } // Drained: the track ended as the copy did, which is not cleanly when // its session went away short of the declared end. @@ -725,6 +783,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()) } @@ -1734,6 +1795,50 @@ 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 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 a7946100ea6..9d2a3724179 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4901,7 +4901,7 @@ impl Cursor { aborts.register(waiter); } - fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + fn poll_recv_group(&mut self, take: bool, waiter: &kio::Waiter) -> Poll>> { self.resolve_floor(); if let Some(aborts) = &self.aborts { // A sequence may name both a live group and its fetched head. Only the @@ -4931,16 +4931,13 @@ impl Cursor { loop { // Re-offer the lowest parked group back inside the cap once it rises, // ahead of the arrival cursor: it is the oldest thing still owed. - let consumer = match self.parked.keys().next().copied() { + let (consumer, next_index) = match self.parked.keys().next().copied() { Some(sequence) if super::subscription::before_end(sequence, self.end_sequence) => { - let group = self.parked.remove(&sequence).expect("just looked it up"); - self.aborts - .as_ref() - .expect("parked groups are watched") - .remove(sequence); - // A re-offer is a delivery: stamp it like a fresh hand-out. - group.cache_refresh(); - group + let group = self.parked.get(&sequence).expect("just looked it up").clone(); + if take { + group.cache_refresh(); + } + (group, None) } _ => { let Some((producer, found_index)) = @@ -4955,30 +4952,47 @@ 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. if self.floor_pending && self.resolve_floor() && consumer.sequence < self.min_sequence { + self.index = found_index + 1; continue; } // Park a group beyond the cap instead of dropping it, and keep scanning // so an in-range group that arrived behind it still flows. if !super::subscription::before_end(consumer.sequence, self.end_sequence) { + self.index = found_index + 1; self.park(consumer, waiter); 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 { + if let Some(index) = next_index { + self.index = index; + } else { + self.parked.remove(&consumer.sequence); + self.aborts + .as_ref() + .expect("parked groups are watched") + .remove(consumer.sequence); + } + } + if stale { self.stale.add(consumer.content()); if let Some(drops) = &mut self.drops && let Some(end) = consumer.sequence.checked_add(1) @@ -4987,7 +5001,7 @@ impl Cursor { } continue; } - return Poll::Ready(Ok(Some(self.with_expiry(consumer)))); + return Poll::Ready(Ok(Some(if take { self.with_expiry(consumer) } else { consumer }))); } } @@ -5023,7 +5037,7 @@ 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>> { let Some(next) = self.next_sequence else { return Poll::Ready(Ok(None)); }; @@ -5064,15 +5078,19 @@ 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; }; - 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))) + Poll::Ready(Ok(Some(if take { + 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(); + self.with_expiry(group) + } else { + group + }))) } /// Count the sequence path's convictions below `committed` into [`Self::stale`], @@ -5185,6 +5203,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`]. /// @@ -5209,8 +5229,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, @@ -5226,6 +5246,30 @@ impl Subscriber { res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter)))) } + /// The next group's sequence without handing it out or advancing past it. + 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) { @@ -10414,6 +10458,41 @@ mod test { assert_eq!(group.sequence, 5); } + #[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() { From c2d3b0fe48593cacb9f49f2dfe517cb363062abb Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:47:35 -0700 Subject: [PATCH 3/5] fix(net): reserve exact groups while selecting route copies Co-Authored-By: GPT-6.1-Sol --- rs/moq-net/benches/origin.rs | 5 +- rs/moq-net/src/model/resume.rs | 99 ++++++++++++++++++---- rs/moq-net/src/model/track.rs | 150 ++++++++++++++++++++++++++++----- 3 files changed, 217 insertions(+), 37 deletions(-) diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index de4d5577a89..99abce9b1cc 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), diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index fcd7f691190..b1caa0c0e8b 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -179,6 +179,7 @@ impl Consumer { datagram: None, groups: (0, Cap::from(..)), groups_read: false, + peeked: None, }) }); // Subscribed to the serving copy at once, so groups it has now are the reader's @@ -379,9 +380,16 @@ impl Subscriber { 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| next.map(|next| next.sequence))); + return result.map(|result| { + result.map(|next| { + let sequence = next.as_ref().map(|next| next.sequence); + reader.peeked = next; + sequence + }) + }); } } } @@ -500,12 +508,16 @@ impl Subscriber { } struct Candidate { - index: usize, + 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. @@ -696,12 +708,20 @@ impl Reader { delivered: &Delivered, waiter: &kio::Waiter, ) -> Option>>> { - let index = match self.poll_lowest(ordered, delivered, waiter)? { - Poll::Ready(Ok(Some(next))) => next.index, - 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 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!() }; @@ -709,15 +729,15 @@ impl Reader { true => sub.poll_next_group(waiter), false => sub.poll_recv_group(waiter), }; - if let Poll::Ready(Ok(Some(group))) = result - && Self::deliverable(&self.copies[index], group.sequence, delivered) - { - self.observe(index, group.sequence); - let group = self.hand_out(index, group); - return Some(Poll::Ready(Ok(Some(group)))); - } - // A producer may evict or replace the candidate between peeking and taking it. - None + // 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 @@ -763,7 +783,11 @@ impl Reader { continue; } if lowest.as_ref().is_none_or(|lowest| sequence <= lowest.sequence) { - lowest = Some(Candidate { index, sequence }); + lowest = Some(Candidate { + generation: self.copies[index].generation, + sequence, + _lease: self.copies[index].lease.clone(), + }); } break; } @@ -964,6 +988,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 { @@ -978,6 +1003,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 { @@ -987,6 +1013,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 { @@ -1822,6 +1849,44 @@ mod test { 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(); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 9d2a3724179..3c1225a851d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4587,6 +4587,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 @@ -4620,6 +4623,12 @@ struct Cursor { drops: Option>, } +struct Peeked { + group: group::Consumer, + index: Option, + ordered: bool, +} + /// What a wire publisher's cursor gave up, found as it changes rather than by rescanning /// the cache on every poll. struct DropCursor { @@ -4728,6 +4737,7 @@ impl Cursor { index: 0, datagram_index, next_sequence: Some(0), + peeked: None, end_sequence: None, parked: BTreeMap::new(), aborts: None, @@ -4902,6 +4912,11 @@ impl Cursor { } 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.resolve_floor(); if let Some(aborts) = &self.aborts { // A sequence may name both a live group and its fetched head. Only the @@ -4982,15 +4997,7 @@ impl Cursor { // 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 { - if let Some(index) = next_index { - self.index = index; - } else { - self.parked.remove(&consumer.sequence); - self.aborts - .as_ref() - .expect("parked groups are watched") - .remove(consumer.sequence); - } + self.advance_arrival(consumer.sequence, next_index); } if stale { self.stale.add(consumer.content()); @@ -5001,7 +5008,28 @@ impl Cursor { } continue; } - return Poll::Ready(Ok(Some(if take { self.with_expiry(consumer) } else { 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 + }))); + } + } + + fn advance_arrival(&mut self, sequence: u64, index: Option) { + if let Some(index) = index { + self.index = index; + } else { + self.parked.remove(&sequence); + self.aborts + .as_ref() + .expect("parked groups are watched") + .remove(sequence); } } @@ -5038,6 +5066,9 @@ 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, 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)); }; @@ -5082,17 +5113,25 @@ impl Cursor { }; Poll::Ready(Ok(Some(if take { - 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(); - self.with_expiry(group) + 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)); + group.cache_refresh(); + self.with_expiry(group) + } + /// Count the sequence path's convictions below `committed` into [`Self::stale`], /// exactly once each. /// @@ -5246,7 +5285,8 @@ impl Subscriber { res.map(|res| res.map(|group| group.map(|group| group.with_meter(meter)))) } - /// The next group's sequence without handing it out or advancing past it. + /// 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 { @@ -5534,7 +5574,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. @@ -5546,7 +5589,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), } } @@ -5568,6 +5614,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_drift_cap(); } @@ -10458,6 +10505,71 @@ 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" + ); + } + } + #[test] fn peeking_keeps_each_orders_cursor_and_rechecks_bounds() { for ordered in [false, true] { From 5f5399c8117716725b880ca1a50cbcf4637d67dc Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:27:31 -0700 Subject: [PATCH 4/5] docs(quest): retain the resumed arrival ordering decision Co-Authored-By: GPT-6.1-Sol --- quest/m1/README.md | 1 + quest/m1/resume-closed.md | 4 ++++ quest/m1/resume-in-flight.md | 1 + quest/m1/resume-ordered.md | 43 ++++++++++++++++++++++++++++++++++++ 4 files changed, 49 insertions(+) create mode 100644 quest/m1/resume-ordered.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 29aaf8fdb47..cd43f0178f7 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -106,6 +106,7 @@ lands before the cut. - [Timed-out requests clean up](/quest/m1/legacy-request-timeout-cleanup.md) - on drafts 14-16, Rust and JS send PUBLISH_NAMESPACE_DONE or UNSUBSCRIBE_NAMESPACE for a timed-out request before its local reset - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails +- [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 572abc8f58d..b7073ab9b3c 100644 --- a/quest/m1/resume-closed.md +++ b/quest/m1/resume-closed.md @@ -79,3 +79,7 @@ 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 45f4e863860..323459d99d0 100644 --- a/quest/m1/resume-in-flight.md +++ b/quest/m1/resume-in-flight.md @@ -104,3 +104,4 @@ 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 new file mode 100644 index 00000000000..7270a4feb09 --- /dev/null +++ b/quest/m1/resume-ordered.md @@ -0,0 +1,43 @@ +# [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 From 8a1e506d5a5e7b1c61ba4037a79de92d0ab299ca Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 19 Oct 2026 22:28:23 -0700 Subject: [PATCH 5/5] chore(quest): finish resume-ordered with the lowest-head rule Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/resume-closed.md | 4 ---- quest/m1/resume-in-flight.md | 1 - quest/m1/resume-ordered.md | 43 ------------------------------------ 4 files changed, 49 deletions(-) delete mode 100644 quest/m1/resume-ordered.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 66954fab446..ad984f87715 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