diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index de0190a75f..0fbdb8140e 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -925,3 +925,27 @@ test.each([ subscriber.close(); }); + +// A reader leaving partway through a fetched group cancels the FETCH: the truncated group +// must not end clean, as a FIN would make it read whole. +test("the last reader leaving mid-response cancels the fetch", async () => { + const { quic, streams } = fakeSession(); + const subscriber = new Subscriber(quic, Version.DRAFT_05, HopSchema.parse(1n)); + + const fetch = subscriber.fetchGroup(Path.from("room"), "video", 0); + await drainUntil(() => streams.length === 1); + await answerTrackInfo(streams[0]); + await drainUntil(() => streams.length === 2); + await streams[1].reading; + + // One frame of the group: a zero timestamp delta, then the sized payload. + streams[1].inbound.enqueue(new Uint8Array([0, 4, ...new TextEncoder().encode("head")])); + const group = await fetch; + expect(new TextDecoder().decode((await group.readFrame())?.payload)).toBe("head"); + group.close(); + + const err = fromTransport(await streams[1].aborted) as StreamError; + expect(err.code).toBe(StreamCode.Cancel); + + subscriber.close(); +}); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 3ac633580c..ba81f25468 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -853,11 +853,13 @@ export class Subscriber { pending ??= stream.reader.decodeMaybe(decode); const next = await race([pending, group.closed, unused]); if (next === idle) { - if (!group.isClosed && group.used.peek()) { + if (group.isClosed) break; + if (group.used.peek()) { unused = group.unused().then((): typeof idle => idle); continue; } - break; + // Abandoned mid-group: the truncated group must never end clean. + throw new StreamError(StreamCode.Cancel, { message: "cancel" }); } pending = undefined; if (!next || next instanceof Error) break; diff --git a/quest/m1/README.md b/quest/m1/README.md index 928f1394a2..21130343c5 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -48,7 +48,6 @@ QUIC studies there on that rule. - [Reader end parity](/quest/m1/reader-end-parity.md) - JS readers see a track's end once the newest group reaches the declared end, as Rust readers do - [JS cache window](/quest/m1/js-prune-cache-window.md) - js/net ages idle groups on a cache window and keeps max_age as media-time staleness, as Rust does - [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - bursty small-group tracks cross two relays without lost groups, unanswered FETCHes, or stalls -- [FETCH abandonment](/quest/m1/fetch-abandonment.md) - Rust and JS lite subscribers close an upstream FETCH once demand leaves, so an abandoned FETCH is released across relays - [Watch and publish under CSP](/quest/m1/csp-assets.md) - blob workers stay the default; strict-CSP apps host the files and set a base URL - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [A week of nightly interop after #4529](/quest/m1/interop-week.md) - the nightly traces the stall diagnosis reads diff --git a/quest/m1/fetch-abandonment.md b/quest/m1/fetch-abandonment.md deleted file mode 100644 index 107ac99d70..0000000000 --- a/quest/m1/fetch-abandonment.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] An abandoned FETCH is released along the mesh - -## Goal - -When nobody wants a FETCH's group any more, the lite subscriber in Rust and in -`@moq/net` closes its upstream FETCH stream, so a relay stops pinning the -request and the publisher behind it stops serving. With -[#4531](https://github.com/moq-dev/moq/pull/4531)'s publisher side, an -abandoned FETCH is then released across every hop. - -## Plan - -#4531 made the lite publisher stop serving a FETCH once the requester FINs or -resets, but the Rust subscriber's `FetchServeRun` never watches demand, so a -relay holds its upstream FETCH until it is answered. Watch demand for the -whole wait and close the stream when it leaves, the same way the subscriber -already drops an unwanted SUBSCRIBE. Check `@moq/net`'s subscriber for the same -gap; #4531 found it FINs an abandoned FETCH, so confirm it does so on demand -loss and not only on close. - -Decided with the maintainer: both subscribers, proven end to end. A relay-chain -test (publisher, relay, subscriber) abandons a FETCH mid-wait and asserts the -upstream publisher's serve ends, which #4531 could only test in-crate. - -Public API: none. Wire: none. - -## Related - -- [Cross-relay bursts](/quest/m1/cross-relay-bursts.md) - unanswered FETCHes across relays diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 1af5cf08a7..41bf3ce820 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -4133,12 +4133,25 @@ enum FetchRunState { Ingest { stream: Stream, producer: group::Producer, + // The accepted request's demand: fetches that joined it but have not yet + // picked the group up from the cache. + joined: kio::Producer, // Boxed so the other states stay small; one allocation per fetch. ingest: Box, }, Done, } +impl FetchRunState { + /// The downstream request, while it waits on the publisher's answer. + fn request(&self) -> Option<&group::Request> { + match self { + Self::Open { request } | Self::Send { request, .. } | Self::Answer { request, .. } => request.as_ref(), + Self::Ingest { .. } | Self::Done => None, + } + } +} + impl FetchServeRun { fn new(serve: TrackServe, request: group::Request, timescale: Option) -> Self { let session = serve.subscriber.session.clone(); @@ -4157,6 +4170,20 @@ impl kio::Task for FetchServeRun { fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { + // A fetch nobody waits on any more is cancelled upstream, so the publisher stops + // serving it (and a relay there releases its own FETCH). Ingest has its own check. + if let Some(request) = self.state.request() + && request.poll_unused(waiter).is_ready() + { + tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); + if let FetchRunState::Send { stream, .. } | FetchRunState::Answer { stream, .. } = + std::mem::replace(&mut self.state, FetchRunState::Done) + { + stream.writer.abort(&Error::Cancel); + } + return Poll::Ready(()); + } + match &mut self.state { FetchRunState::Open { request } => { tracing::info!(broadcast = %self.serve.subscriber.log_path(&self.serve.path), track = %self.serve.name, group = self.group, "fetch started"); @@ -4278,6 +4305,9 @@ impl kio::Task for FetchServeRun { let group_info = track::Info::default() .with_timescale(self.timescale.unwrap_or_default()) .with_max_age(self.serve.subscriber.origin.default_max_age()); + // The joined fetches pick the group up from the cache only when next + // polled, so their demand outlives the request until then. + let joined = request.result.clone(); let mut producer = match request.accept(group_info) { Ok(producer) => producer, Err(err) => { @@ -4299,15 +4329,36 @@ impl kio::Task for FetchServeRun { self.state = FetchRunState::Ingest { stream, producer, + joined, ingest: Box::new(FrameIngest::new(&self.serve.subscriber, self.timescale)), }; } FetchRunState::Ingest { stream, producer, + joined, ingest, } => { - let res = ready!(ingest.poll(&mut stream.reader, producer, waiter)); + let Poll::Ready(res) = ingest.poll(&mut stream.reader, producer, waiter) else { + // Still short of its end, so completion always wins. Once nobody + // wants the rest (no joined fetch left to pick it up, no reader), + // cancel upstream and abort the truncated group, never caching it + // as complete. The abort is atomic with a new reader arriving. + if joined.poll_unused(waiter).is_pending() || producer.poll_unused(waiter).is_pending() { + return Poll::Pending; + } + if !producer.abort_unused(Error::Cancel) { + continue; + } + tracing::debug!(track = %self.serve.name, group = self.group, "fetch abandoned"); + let FetchRunState::Ingest { stream, .. } = + std::mem::replace(&mut self.state, FetchRunState::Done) + else { + unreachable!() + }; + stream.writer.abort(&Error::Cancel); + return Poll::Ready(()); + }; let FetchRunState::Ingest { producer, .. } = std::mem::replace(&mut self.state, FetchRunState::Done) else { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index bb725dcba1..e199da5f88 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -783,13 +783,30 @@ impl Producer { self.close_aborted(err) } + /// Abort with `err` only while nothing consumes the group, returning whether it is + /// closed. Consumer creation and the check share a lock, so a reader arriving after + /// [`poll_unused`](Self::poll_unused) keeps the group alive instead of reading the abort. + pub(crate) fn abort_unused(&self, err: Error) -> bool { + match self.state.write_unused() { + kio::Unused::Idle(guard) => { + self.commit_abort(guard, err); + true + } + kio::Unused::Closed => true, + kio::Unused::Used => false, + } + } + fn close_aborted(&self, err: Error) -> Result<()> { - let mut guard = modify(&self.state)?; + self.commit_abort(modify(&self.state)?, err); + Ok(()) + } + + fn commit_abort(&self, mut guard: kio::Mut<'_, GroupState>, err: Error) { guard.abort = Some(err); self.alive.aborted.store(true, Ordering::Release); guard.release(); guard.close(); - Ok(()) } /// Abort a write that would grow the group past its budget, holding the lock already @@ -913,11 +930,22 @@ impl Producer { /// Create a new consumer for the group. pub fn consume(&self) -> Consumer { + self.consumer(self.state.consume()) + } + + /// Create a consumer, or `None` once the group is aborted. Paired with + /// [`abort_unused`](Self::abort_unused): the closed check and the count share its + /// lock, so a consumer either exists in time to decline the abort or is never made. + pub(crate) fn try_consume(&self) -> Option { + self.state.weak().try_consume().map(|state| self.consumer(state)) + } + + fn consumer(&self, state: kio::Consumer) -> Consumer { Consumer { info: self.info, track: self.track.clone(), inner: ConsumerKind::Plain(Plain { - state: self.state.consume(), + state, index: 0, end: None, prefetch: Prefetch::default(), @@ -970,6 +998,11 @@ impl Producer { self.state.unused().await.map_err(|_| self.abort_reason()) } + /// Poll for the group becoming unused (every consumer dropped). + pub(crate) fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> { + self.state.poll_unused(waiter).map(|_| ()) + } + /// The recorded abort reason, or [`Error::Dropped`] if the group closed without one. fn abort_reason(&self) -> Error { self.state.read().abort.clone().unwrap_or(Error::Dropped) @@ -1969,6 +2002,21 @@ mod test { assert!(matches!(result, Err(crate::Error::Cancel))); } + #[test] + fn abort_unused_pairs_with_try_consume() { + let producer = Info { sequence: 0 }.produce(); + + let consumer = producer.try_consume().expect("open"); + assert!( + !producer.abort_unused(crate::Error::Cancel), + "a reader declines the abort" + ); + drop(consumer); + + assert!(producer.abort_unused(crate::Error::Cancel)); + assert!(producer.try_consume().is_none(), "an aborted group mints no reader"); + } + #[test] fn abort_clears_cached_frames() { let mut producer = Info { sequence: 0 }.produce(); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ee4d90299b..d526934dff 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -635,12 +635,16 @@ impl TrackState { /// end-of-stream. The handler side (a rejection, or no [`Dynamic`] at all) lives /// in [`FetchState`]; [`Fetching`] polls both. fn poll_fetch_cached(&self, sequence: u64, frame_start: u64) -> Poll> { - if let Some(group) = self.covering_group(sequence, frame_start) { + // A group aborted between the lookup and the consume (an abandoned fetch cut + // short) is a miss, not a hit that fails on its first read. + if let Some(group) = self.covering_group(sequence, frame_start) + && let Some(consumer) = group.try_consume() + { // A cache hit refreshes the group: it resets both its age (expiry keys // off the last access) and its standing against the pool-wide average, // so the eviction walk keeps it over never-read groups. group.cache_refresh(); - return Poll::Ready(Ok(group.consume())); + return Poll::Ready(Ok(consumer)); } if let Some(err) = &self.abort { @@ -1226,7 +1230,11 @@ impl TrackState { // An evicted sequence can be re-fetched; a live one is a duplicate. self.claim_sequence(sequence, frame_start)?; - let group = group::Producer::new(group::Info { sequence }, info, self.cache.clone()); + let mut group = group::Producer::new(group::Info { sequence }, info, self.cache.clone()); + // Start where the request did before the group is visible: a fetch looking it up + // in between would otherwise see it begin at 0 and get a reader that later skips + // the head it asked for. + group.start_at(frame_start)?; // A backfill exists because someone is fetching it right now: stamp that // access so the eviction walk can't kill it before the fetch resolves. // It is also invisible to arrival-order subscribers: fetched on demand, @@ -2824,13 +2832,20 @@ impl Consumer { // Queue a request only when the group isn't already resolvable from the track // (cached, aborted, or past-final all resolve through `Fetching::poll` without // a queue entry). - let (fetch, unresolved) = { + let (fetch, cached) = { let state = state.read(); ( state.fetch.clone(), - state.poll_fetch_cached(sequence, options.frame_start).is_pending(), + state.poll_fetch_cached(sequence, options.frame_start), ) }; + let unresolved = cached.is_pending(); + // Hold a hit until the caller polls: the held consumer keeps the group wanted, + // so an abandoned fetch filling it can't abort it in between. + let hit = match cached { + Poll::Ready(Ok(group)) => Some(Box::new(group)), + _ => None, + }; if unresolved { let mut fetch = fetch.lock(); @@ -2871,6 +2886,7 @@ impl Consumer { sequence, frame_start: options.frame_start, result, + hit, }, stats: self.stats.clone(), }) @@ -3077,8 +3093,10 @@ impl group::Request { /// The first frame of the group the consumer wants; 0 is the whole group. /// - /// A handler serving this must [`start_at`](group::Producer::start_at) it, so the frames - /// it writes carry the indices they have in the group rather than restarting at 0. + /// The group [`accept`](Self::accept) returns already starts here, so the frames a + /// handler writes carry the indices they have in the group rather than restarting at + /// 0. A handler serving from elsewhere moves it with + /// [`start_at`](group::Producer::start_at) before the first frame. /// /// There is no end: the handler fetches through the end of the group so the result /// is cacheable for anyone (see [`group::Fetch::frame_start`]). @@ -3101,6 +3119,12 @@ impl group::Request { let res = TrackState::modify(&self.state) .and_then(|mut state| state.insert_group_request(self.sequence, self.frame_start, info.into())); self.remove(); + // A joined fetch the cached group can't cover (a wider start that joined after + // the range went on the wire) fails rather than waits. Closing the channel says + // the same, but a handler tracking the joined demand keeps it open. + if let Ok(mut outcome) = self.result.write() { + outcome.rejected = Some(Error::NotFound); + } res } @@ -3125,6 +3149,25 @@ impl group::Request { } } + /// Poll for the request becoming unused (every waiting [`Consumer::fetch_group`] + /// dropped), so a handler can stop serving and drop the request. + /// + /// Once ready, a later fetch of the group starts a fresh request rather than + /// joining this abandoned one. + pub fn poll_unused(&self, waiter: &kio::Waiter) -> Poll<()> { + loop { + ready!(self.result.poll_unused(waiter)); + // Fetches join under the fetch lock, so re-checking under it makes the + // withdrawal atomic with a join racing the last one leaving. + let mut fetch = self.fetch.lock(); + if self.result.is_used() { + continue; + } + fetch.remove_if(&self.sequence, |pending| pending.result.same_channel(&self.result)); + return Poll::Ready(()); + } + } + /// Remove this attempt from the fetch state, unless a newer attempt for the same /// sequence has already replaced it. fn remove(&self) { @@ -3166,8 +3209,12 @@ enum FetchingKind { // This caller's own start, so a cached group that begins above it is a miss // rather than a short answer. frame_start: u64, - // The joined attempt's result channel; `None` when no handler existed to queue on. + // The joined attempt's result channel; `None` on a cache hit, or when no handler + // existed to queue on. result: Option>, + // The group already cached when the fetch was made, held so it stays wanted. + // Boxed: a group consumer dwarfs the rest of the handle. + hit: Option>, }, /// A spliced track's fetch: waits for a segment, then fetches from it. Spliced(kio::Pending), @@ -3177,14 +3224,15 @@ impl kio::Pollable for Fetching { type Output = Result; fn poll(&self, waiter: &kio::Waiter) -> Poll { - let (state, fetch, sequence, frame_start, result) = match &self.inner { + let (state, fetch, sequence, frame_start, result, hit) = match &self.inner { FetchingKind::Plain { state, fetch, sequence, frame_start, result, - } => (state, fetch, *sequence, *frame_start, result.as_ref()), + hit, + } => (state, fetch, *sequence, *frame_start, result.as_ref(), hit), FetchingKind::Spliced(spliced) => { // A fetched group is metered here (once), at the tagged handle: the // spliced source track it comes from is the origin's own, untagged. @@ -3193,22 +3241,26 @@ impl kio::Pollable for Fetching { } }; + // Hand back a consumer sitting where the caller asked rather than at the group's + // own start. Coverage was checked first, so this can only skip frames they excluded. + let resolve = |mut group: group::Consumer| { + group.start_at(frame_start); + group.with_meter(self.stats.meter()) + }; + if let Some(group) = hit { + return Poll::Ready(Ok(resolve((**group).clone()))); + } + // Track side: the cached group, the abort error, or past-final. The outer // error is the channel closing without any of those. - match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) { - Poll::Ready(Ok(res)) => { - return Poll::Ready(res.map(|mut group| { - // Hand back a consumer sitting where the caller asked rather than at - // the group's own start. Coverage was checked above, so this can only - // skip frames they excluded. - group.start_at(frame_start); - group.with_meter(self.stats.meter()) - })); - } - Poll::Ready(Err(closed)) => { - return Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))); - } - Poll::Pending => {} + let cached = + |waiter: &kio::Waiter| match state.poll(waiter, |state| state.poll_fetch_cached(sequence, frame_start)) { + Poll::Ready(Ok(res)) => Poll::Ready(res.map(resolve)), + Poll::Ready(Err(closed)) => Poll::Ready(Err(closed.abort.clone().unwrap_or(Error::Dropped))), + Poll::Pending => Poll::Pending, + }; + if let Poll::Ready(res) = cached(waiter) { + return Poll::Ready(res); } // Handler side. @@ -3226,13 +3278,19 @@ impl kio::Pollable for Fetching { // A written rejection fails every joined fetch. The channel closing without // one means the attempt was dropped unserved (its handlers went away). - match result.poll(waiter, |outcome| match &outcome.rejected { + let err = match result.poll(waiter, |outcome| match &outcome.rejected { Some(err) => Poll::Ready(err.clone()), None => Poll::Pending, }) { - Poll::Ready(Ok(err)) => Poll::Ready(Err(err)), - Poll::Ready(Err(_closed)) => Poll::Ready(Err(Error::NotFound)), - Poll::Pending => Poll::Pending, + Poll::Ready(Ok(err)) => err, + Poll::Ready(Err(_closed)) => Error::NotFound, + Poll::Pending => return Poll::Pending, + }; + // An accept caches the group before it answers the channel, which may land + // between the track check above and this one: look again before failing. + match cached(waiter) { + Poll::Ready(res) => Poll::Ready(res), + Poll::Pending => Poll::Ready(Err(err)), } } } @@ -8009,6 +8067,120 @@ mod test { assert!(matches!(pending.await, Err(Error::Dropped))); } + /// A request stays wanted while any joined fetch waits, and once the last leaves it + /// is withdrawn: a later fetch starts a fresh request instead of joining it. + #[tokio::test] + async fn fetch_request_unused_once_every_fetch_leaves() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let second = consumer.fetch_group(5, None); + + drop(first); + assert!(req.poll_unused(&kio::Waiter::noop()).is_pending()); + drop(second); + assert!(req.poll_unused(&kio::Waiter::noop()).is_ready()); + + let retry = consumer.fetch_group(5, None); + let fresh = dynamic + .requested_group() + .now_or_never() + .expect("the retry queues a fresh request") + .unwrap(); + drop(req); + assert!(fresh.poll_unused(&kio::Waiter::noop()).is_pending()); + fresh.accept(None).unwrap().finish().unwrap(); + assert_eq!(retry.await.unwrap().sequence, 5); + } + + /// A fetch that hits the cache holds the group until polled, so a handler that aborts + /// the group once nobody wants it (an abandoned upstream fetch) leaves it alone. + #[tokio::test] + async fn fetch_hit_keeps_the_group_wanted() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let mut group = req.accept(None).unwrap(); + group + .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) + .unwrap(); + drop(first.await.unwrap()); + + let hit = consumer.fetch_group(5, None); + assert!(!group.abort_unused(Error::Cancel), "the pending hit wants the group"); + let mut fetched = hit.await.unwrap(); + assert_eq!(&fetched.read_frame().await.unwrap().unwrap().payload[..], b"head"); + } + + /// An accepted group starts where its request did before anyone can see it, so a + /// wider fetch arriving before the handler writes misses instead of being handed a + /// reader that would skip the head it asked for. + #[tokio::test] + async fn accepted_group_starts_at_the_request() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let _tail = consumer.fetch_group(5, group::Fetch::default().with_frame_start(3)); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let _group = req.accept(None).unwrap(); + + let whole = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("the wider fetch misses and queues its own request") + .unwrap(); + assert_eq!(req.frame_start(), 0); + assert!(kio::Pollable::poll(&*whole, &kio::Waiter::noop()).is_pending()); + } + + /// Once a group nobody wanted is aborted, a later fetch misses rather than reading + /// the abort, and queues a fresh request. + #[tokio::test] + async fn fetch_misses_an_abandoned_group() { + let producer = track_producer("test", None); + let dynamic = producer.dynamic(); + let consumer = producer.consume(); + + let first = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("should not block") + .unwrap(); + let group = req.accept(None).unwrap(); + drop(first.await.unwrap()); + assert!(group.abort_unused(Error::Cancel)); + + let retry = consumer.fetch_group(5, None); + let req = dynamic + .requested_group() + .now_or_never() + .expect("the retry queues a fresh request") + .unwrap(); + req.accept(None).unwrap().finish().unwrap(); + assert!(retry.await.unwrap().read_frame().await.unwrap().is_none()); + } + #[tokio::test] async fn fetch_reject_does_not_poison_retry() { let producer = track_producer("test", None); @@ -8115,6 +8287,9 @@ mod test { assert!(futures::poll!(widest.as_mut()).is_pending()); assert_eq!(request.frame_start(), 2, "the in-flight range is already on the wire"); + // A handler may keep tracking the joined demand past the accept, as the lite + // subscriber does, which holds the result channel open. + let _joined = request.result.clone(); let mut group = request.accept(None).unwrap(); // A handler numbers the frames from where it was asked to start, as `serve_fetch` // does; that offset is what makes the cached group too narrow for `widest`. @@ -8122,13 +8297,14 @@ mod test { group .write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"from2")) .unwrap(); - group.finish().unwrap(); // The two callers it covers resolve; the one it doesn't fails cleanly rather - // than being handed a group that starts above what it asked for. + // than being handed a group that starts above what it asked for, even while + // the group is still being written. assert_eq!(narrow.await.unwrap().index(), 5); assert_eq!(wider.await.unwrap().index(), 2); assert!(matches!(widest.await, Err(Error::NotFound))); + group.finish().unwrap(); } #[tokio::test] diff --git a/rs/moq-net/tests/fetch_cancel.rs b/rs/moq-net/tests/fetch_cancel.rs new file mode 100644 index 0000000000..5e2d45ff7c --- /dev/null +++ b/rs/moq-net/tests/fetch_cancel.rs @@ -0,0 +1,154 @@ +//! A FETCH nobody wants anymore is cancelled all the way to the publisher. +//! +//! Each relay on the path holds a FETCH stream and the group behind it. When the +//! reader gives up, every hop must let go, whether the publisher has yet to answer +//! (lite has no FETCH_OK, so one that never answers would pin them for good) or is +//! partway through the group. A group cut short is aborted, never cached as whole. + +mod support; + +use std::time::Duration; + +use bytes::Bytes; +use moq_net::{Hop, Timestamp, Version, origin, track}; +use support::harness::peer; + +fn produce_origin(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Where the reader gives up. +#[derive(Clone, Copy, Debug)] +enum Stage { + /// Before the publisher answers the FETCH. + Unanswered, + /// After the first frame, with the rest of the group still to come. + MidResponse, + /// Once every frame is read, as the FIN arrives. + Drained, +} + +/// `relays` relays between the publisher and the reader. +async fn abandoned_fetch_reaches_the_publisher(version: &str, relays: u64, stage: Stage) { + let version: Version = version.parse().unwrap(); + let ctx = format!("{version:?} over {relays} relays, {stage:?}"); + let nodes: Vec<_> = (1..=relays + 1).map(produce_origin).collect(); + let mut _pairs = Vec::new(); + for pair in nodes.windows(2) { + _pairs.push(peer(version, &pair[0], &pair[1]).await); + } + + let broadcast = nodes[0].create_broadcast("room").unwrap(); + let mut dynamic = broadcast.dynamic(); + broadcast.announce(Default::default()).unwrap(); + tokio::time::sleep(Duration::from_secs(1)).await; + + let consumer = nodes[relays as usize] + .consume() + .request_broadcast("room") + .await + .unwrap(); + let track = consumer.track("video").unwrap(); + let mut waiting = Box::pin(track.fetch_group(0, None)); + + // Answer the TRACK_INFO, then take the FETCH. + let request = tokio::select! { + request = dynamic.requested_track() => request.expect("the track request reaches the publisher"), + _ = &mut waiting => panic!("nothing answered the track"), + }; + let groups = request.dynamic(); + let _track = request.accept(track::Info::default()); + let request = tokio::select! { + request = groups.requested_group() => request.expect("the fetch reaches the publisher"), + _ = &mut waiting => panic!("nothing answered the fetch"), + }; + + match stage { + Stage::Unanswered => { + drop(waiting); + tokio::time::timeout(Duration::from_secs(5), kio::wait(|waiter| request.poll_unused(waiter))) + .await + .unwrap_or_else(|_| panic!("{ctx}: the publisher's fetch is still wanted")); + } + Stage::MidResponse => { + let mut group = request.accept(None).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + + let mut fetched = waiting.await.unwrap(); + let frame = fetched.read_frame().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], b"head", "{ctx}"); + drop(fetched); + + // The publisher stops serving the group: nobody reads it any more. + tokio::time::timeout(Duration::from_secs(5), group.unused()) + .await + .unwrap_or_else(|_| panic!("{ctx}: the publisher still serves the fetch")) + .unwrap(); + + // No hop cached the head as the whole group: a fresh fetch reads it all. + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"tail")).unwrap(); + group.finish().unwrap(); + let mut fetched = tokio::time::timeout(Duration::from_secs(5), track.fetch_group(0, None)) + .await + .unwrap_or_else(|_| panic!("{ctx}: the refetch stalled")) + .unwrap(); + let mut payloads = Vec::new(); + while let Some(frame) = fetched.read_frame().await.unwrap() { + payloads.push(frame.payload); + } + assert_eq!(payloads, [&b"head"[..], &b"tail"[..]], "{ctx}"); + } + Stage::Drained => { + let mut group = request.accept(None).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"head")).unwrap(); + group.write_frame(Timestamp::ZERO, Bytes::from_static(b"tail")).unwrap(); + + let mut fetched = waiting.await.unwrap(); + for expected in [&b"head"[..], b"tail"] { + let frame = fetched.read_frame().await.unwrap().unwrap(); + assert_eq!(&frame.payload[..], expected, "{ctx}"); + } + + // The FIN reaches the reader's node at the instant the reader leaves: the + // group is whole, so it must be cached as such rather than cancelled. + const LATENCY: Duration = Duration::from_millis(100); + _pairs.last().unwrap().client_transport.set_latency(LATENCY); + group.finish().unwrap(); + tokio::time::sleep(LATENCY).await; + drop(fetched); + tokio::time::sleep(Duration::from_secs(1)).await; + + // The publisher can no longer serve it, so only a cached copy answers. + group.abort(moq_net::Error::Cancel).unwrap(); + let mut fetched = tokio::time::timeout(Duration::from_secs(5), track.fetch_group(0, None)) + .await + .unwrap_or_else(|_| panic!("{ctx}: the refetch stalled")) + .unwrap_or_else(|err| panic!("{ctx}: the whole group was not cached: {err}")); + let mut payloads = Vec::new(); + while let Some(frame) = fetched.read_frame().await.unwrap() { + payloads.push(frame.payload); + } + assert_eq!(payloads, [&b"head"[..], &b"tail"[..]], "{ctx}"); + } + } +} + +async fn abandoned_fetches_reach_the_publisher(version: &str) { + for stage in [Stage::Unanswered, Stage::MidResponse, Stage::Drained] { + for relays in [1, 2] { + abandoned_fetch_reaches_the_publisher(version, relays, stage).await; + } + } +} + +#[tokio::test(start_paused = true)] +async fn abandoned_fetch_reaches_the_publisher_lite05() { + abandoned_fetches_reach_the_publisher("moq-lite-05").await; +} + +#[tokio::test(start_paused = true)] +async fn abandoned_fetch_reaches_the_publisher_lite06() { + abandoned_fetches_reach_the_publisher("moq-lite-06").await; +}