From 34b492604073f33f401ce19bb724fe6d811a7968 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 23:49:44 -0700 Subject: [PATCH 1/3] fix(net): wait for a track's tail before ending a subscription A group stream can reach the subscriber after the subscription's end, since QUIC does not order streams. moq-lite keeps the subscription registered past the subscribe stream's FIN, and IETF keeps the alias past PUBLISH_DONE, until every owed group is accounted for or a grace gives up on one reset before its header. The IETF publisher reports the real Stream Count and sends END_OF_TRACK, which the subscriber now honors. Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-net.md | 1 + quest/m1/README.md | 2 +- quest/m1/lite-stream-count.md | 10 +- quest/m1/quic/reliable-reset.md | 10 +- quest/m1/rust-track-tail.md | 60 ------- quest/m1/session-death-error.md | 4 - quest/m1/track-tail-interop.md | 34 ++++ rs/moq-net/src/ietf/publisher.rs | 133 ++++++++++++--- rs/moq-net/src/ietf/subscriber.rs | 255 +++++++++++++++++++++++----- rs/moq-net/src/lib.rs | 1 + rs/moq-net/src/lite/subscriber.rs | 106 +++++++++++- rs/moq-net/src/model/track.rs | 69 +++++++- rs/moq-net/src/tail.rs | 128 ++++++++++++++ rs/moq-net/tests/goaway.rs | 4 +- rs/moq-net/tests/support/harness.rs | 9 +- rs/moq-net/tests/support/mock.rs | 47 ++++- rs/moq-net/tests/track_tail.rs | 192 +++++++++++++++++++++ 17 files changed, 910 insertions(+), 155 deletions(-) delete mode 100644 quest/m1/rust-track-tail.md create mode 100644 quest/m1/track-tail-interop.md create mode 100644 rs/moq-net/src/tail.rs create mode 100644 rs/moq-net/tests/track_tail.rs diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 2f56c399b6..5f6cb60673 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -20,6 +20,7 @@ above ([hang](/lib/rs/hang)); relays and CDNs implement only this. - **Patterns** (`Pattern`, `Patterns`) are re-exported from [`moq-pattern`](https://docs.rs/moq-pattern). Literal `Path` stays a coordinate. - **Tracks** carry groups with a priority, a retention window, and a timescale. Subscribers set their own priority and max age and can change them live. - **Groups** are written frame by frame and delivered on independent streams. Old groups are cached for fetch-by-sequence; stale groups are skipped per the subscriber's budget. +- **Track ends**: `finish()` ends a track at its live edge, while `finish_at(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber awaits it with `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF. - **Datagrams** send a single small frame unreliably on moq-lite 05+. - **Routes** record the relay hops and a cost, which is what the relay [cluster](/bin/relay/cluster) routes on. A hop of 0 marks the chain anonymous: `Route::is_anonymous()` is true, and that route ranks below every fully identified one. `Route::source()` says where a delivered route entered: `Source::Local`, or `Source::Peer(hop)` when a handle marked `origin::Producer::peer()` announced it. `origin::Consumer::local()` sees only the local ones. - **Stats** counters per broadcast and session, drained by [`moq-stats`](https://docs.rs/moq-stats). diff --git a/quest/m1/README.md b/quest/m1/README.md index 07e925cd97..879b1f676e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -22,9 +22,9 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [libmoq hidden opt-in](/quest/m1/libmoq-hidden.md) - `moq_origin_announced` takes a `hidden` flag so C callers can list `.`-named broadcasts - [lite-07 stream count](/quest/m1/lite-stream-count.md) - moq-lite-07 replaces SUBSCRIBE_DROP with a group-stream count in SUBSCRIBE_END, like moq-transport - [Announce compression](/quest/m1/announce-compression.md) - a lite-07 announce reuses the path head and hop-chain tail of a live announcement on its stream instead of resending them -- [Rust track tail](/quest/m1/rust-track-tail.md) - a moq-net subscriber accepts groups that arrive after the subscription's end, and PublishDone carries the real stream count - [Session death error](/quest/m1/session-death-error.md) - a dying session ends its tracks with its own error in Rust and JS, never a clean end, `Dropped`, or `Cancel` - [JS bare FIN](/quest/m1/js-bare-fin.md) - a `@moq/net` subscriber aborts a track whose subscribe stream FINs before its declared end, like Rust +- [Track tail interop](/quest/m1/track-tail-interop.md) - a Rust publisher ending a track with a group in flight is read to its end by the JS subscriber, and the reverse, in `just test interop` - [Signal.race cleanup](/quest/m1/signal-race.md) - `Signal.race` releases its signal listeners when its result loses a race - [Origin narrowing](/quest/m1/origin-narrowing.md) - a live origin grant narrows in place and ends the subscriptions it no longer covers, the deafen boundary #2714 asked for - [Auth expiry clock](/quest/m1/auth-expiry-clock.md) - moq-auth and the relay hold one fixed expiry deadline and honour the same skew allowance diff --git a/quest/m1/lite-stream-count.md b/quest/m1/lite-stream-count.md index b866d94686..7bc958d9bc 100644 --- a/quest/m1/lite-stream-count.md +++ b/quest/m1/lite-stream-count.md @@ -28,16 +28,16 @@ before its header arrived is still invisible, so the track-tail grace stays. the upstream count. - Subscribers on lite-07 stop waiting once the count is reached, accepting a late stream below the end within the grace. On lite-05 and -06, the - DROP accounting from JS track tail (#4086) stays as it is. + DROP accounting already in moq-net (`rs/moq-net/src/tail.rs`) and `@moq/net` + (`js/net/src/tail.ts`) stays as it is. - Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group that is never counted, a reset stream, and a count of zero. Add a Rust-JS interop case. -This lands before lite-07 is finalized. Rust has never sent or acted on -SUBSCRIBE_DROP, so [Rust track tail](/quest/m1/rust-track-tail.md) builds its -lite accounting on the count rather than on drops. +This lands before lite-07 is finalized. Published drafts keep SUBSCRIBE_DROP. +lite-07 replaces it with the count in both Rust and JS. Rust still does not +send SUBSCRIBE_DROP; it only accounts one a peer sends. ## Related -- [Rust track tail](/quest/m1/rust-track-tail.md) - builds on this count for moq-lite - [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - makes the count exact by keeping a reset stream's header diff --git a/quest/m1/quic/reliable-reset.md b/quest/m1/quic/reliable-reset.md index ed01ee6620..10746d214f 100644 --- a/quest/m1/quic/reliable-reset.md +++ b/quest/m1/quic/reliable-reset.md @@ -56,9 +56,9 @@ the native interop matrix. Then use it in MoQ. A publisher that resets a group stream uses a Reliable Size covering the group header, so the subscriber can always attribute the reset to its group. Once reliable reset is negotiated, subscribers drop the -grace they wait for missing groups below a track's declared end (see the track -tail quests). Browsers keep the grace until WebTransport exposes -reliable reset. +grace they wait for missing groups below a track's declared end +(`rs/moq-net/src/tail.rs`, `js/net/src/tail.ts`). Browsers keep the grace until +WebTransport exposes reliable reset. Track the unversioned draft during implementation. The planning baseline is draft 10, with transport parameter `0x1d` and frame type `0x24`; do not freeze @@ -66,8 +66,8 @@ provisional codepoints if the document changes before release. ## Related -- [Rust track tail](/quest/m1/rust-track-tail.md) - waits a grace for a group - whose reset lost its header until this lands, as `@moq/net` already does +- moq-net (`rs/moq-net/src/tail.rs`) and `@moq/net` (`js/net/src/tail.ts`) wait + a grace for a group whose reset lost its header until this lands - [qmux on the QUIC stream state machine](/quest/m1/quic/qmux.md) - consumes the same reset state without a parallel implementation - The removed quiche backend was the one stack that had this, so it is the diff --git a/quest/m1/rust-track-tail.md b/quest/m1/rust-track-tail.md deleted file mode 100644 index 2da144893a..0000000000 --- a/quest/m1/rust-track-tail.md +++ /dev/null @@ -1,60 +0,0 @@ -# [M] Rust track tail - -## Goal - -A moq-net subscriber, lite and IETF, accepts every group below a track's -declared end until each is accounted for, even when its stream arrives after -the subscription's end, then ends cleanly. The IETF publisher reports how many -data streams it opened. The rule and the grace match the JS subscriber, so a -reader cannot tell which side it is talking to. - -## Plan - -Rust already records SUBSCRIBE_END with `finish_at`, which is not terminal, -and its publishers drain their group tasks before they end a subscription. The -remaining gap is the subscriber's bookkeeping: - -- Lite: once the subscribe stream FINs, `remove_subscribe` drops the entry, - so a group stream whose header decodes afterwards fails with - `Error::Cancel` in `lite/subscriber.rs`. Keep the entry, with its boundary, - until every group below the boundary is accounted for or the grace expires. -- IETF: retiring the alias on PublishDone makes a late stream fail with - `Error::Cancel` (`ietf/subscriber.rs`, `Alias::Retired`). Retire it only - once the streams are accounted for or the grace expires. -- Grace: a group reset before its header arrived can never be accounted for, - so give up after the same grace as `@moq/net` (`js/net/src/tail.ts`) and end - cleanly, skipping the missing group as stale: the effective `max_age` as a - wall-clock stopgap on moq-lite, 1s when it is zero, and 1s on IETF. The - reliable-reset quest removes both. -- Only stream-delivered groups are waited for; datagrams are never. -- IETF publisher: send the real number of data streams opened in PublishDone - instead of `stream_count: 0`. On receipt, treat the count as a hint: stop - waiting once that many are accounted for, but accept a late stream below the - boundary within the grace, so a published peer's 0 keeps working. -- IETF on drafts 14-22: write the END_OF_TRACK object at the boundary, and - decode it. `@moq/net` already sends it on its own stream at object 0 of - the group `final`, after the group streams drain; moq-net's subscriber - rejects status 0x4 as `Unsupported` today, so it aborts a bogus group - `final` at the end of every JS-published IETF track until this lands. - -`@moq/net` settled the draft reading: on 14-22 Stream Count counts every data -stream opened, fill streams included (20+), with a 2^62-1 or 2^64-1 unknown -sentinel. PUBLISH_DONE carries no end location on any of them, so the end is -the END_OF_TRACK object: at object 0 of group G the track ends at G, -otherwise at G+1. Only TRACK_ENDED (and SUBSCRIPTION_ENDED before 20) ends a -track cleanly; an error status aborts it. - -Reproduce each case before fixing it: a group header decoded after the -subscribe stream's FIN, and a late stream after PublishDone, over the mock -session in `rs/moq-net/tests/support`. Add a Rust-JS interop case to -`just test smoke --all` for a publisher that ends a track with a group still -in flight. - -## Required - -- [lite-07 stream count](/quest/m1/lite-stream-count.md) - the moq-lite accounting this builds on, instead of SUBSCRIBE_DROP - -## Related - -- [Session death error](/quest/m1/session-death-error.md) - tracks ending wrong when the session dies -- [Reliable stream reset](/quest/m1/quic/reliable-reset.md) - removes the grace diff --git a/quest/m1/session-death-error.md b/quest/m1/session-death-error.md index 223bfbaab6..b4112d5c24 100644 --- a/quest/m1/session-death-error.md +++ b/quest/m1/session-death-error.md @@ -53,7 +53,3 @@ controls: a finished track still ends clean while its session lives. ## Closes - [#4061](https://github.com/moq-dev/moq/issues/4061) - close this issue when the quest finishes - -## Related - -- [Rust track tail](/quest/m1/rust-track-tail.md) - the clean-end half: a track that did end is delivered whole diff --git a/quest/m1/track-tail-interop.md b/quest/m1/track-tail-interop.md new file mode 100644 index 0000000000..0078cdae17 --- /dev/null +++ b/quest/m1/track-tail-interop.md @@ -0,0 +1,34 @@ +# [S] Track tail interop + +## Goal + +`just test interop` covers a publisher that ends a track while its last group +is still in flight: a Rust publisher's track is read to its declared end by the +JS subscriber, and a JS publisher's by the Rust subscriber, through the relay. +Each reader gets every group below the end and then a clean end, never an +error or a stall. + +## Plan + +Both moq-net and `@moq/net` wait for a track's tail once the publisher ends a +subscription, and each is covered by its own in-process tests. Nothing checks +that the two agree across a relay on real QUIC. + +What stood in the way when the Rust half landed: + +- `moq import` exits the moment stdin ends, closing its session with the tail + still in flight, so a finite CLI publisher cannot end a track cleanly. The + fix is a publisher that waits for its subscriptions to drain before it + closes; `moq export`'s linger ([Export linger](/quest/m1/export-linger.md)) is + the reader-side cousin. +- The native JS subscriber (`test/interop/clients/js-native`) returns on the + first frame. It needs a mode that reads a track to its end and reports how it + ended and which groups it saw. + +QUIC on localhost rarely reorders, so this is a smoke check that the end is +delivered and clean. The ordering race itself stays in the unit tests. + +## Related + +- [Session death error](/quest/m1/session-death-error.md) - the other way a + track ends wrong diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 1b4d7f3af6..ee5676d165 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3,6 +3,10 @@ use crate::{frame, group, origin, track}; use std::{ collections::HashMap, ops::Bound, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, task::{Poll, ready}, time::Duration, }; @@ -587,35 +591,52 @@ where // Run the track, cancelling on reader close (Unsubscribe or stream close). // The fill (when one was requested) runs alongside on its own fetch stream; // its failures reset that stream and never touch the subscription. - let res = { - let mut track_serve = - TrackServe::new(self.session.clone(), track, request_id, self.version, range, timescale); + let mut track_serve = + TrackServe::new(self.session.clone(), track, request_id, self.version, range, timescale); + let served = { let serve = async { match fill { Some((fill, cache, timescale)) => { let fill = self.run_fill(request_id, priority, fill, cache, timescale); let track = kio::wait(|waiter| track_serve.poll(waiter)); - let (res, ()) = futures::join!(track, fill); - res + let (res, filled) = futures::join!(track, fill); + (res, filled) } - None => kio::wait(|waiter| track_serve.poll(waiter)).await, + None => (kio::wait(|waiter| track_serve.poll(waiter)).await, false), } }; let mut serve = std::pin::pin!(serve); let mut closed_session = self.session.clone(); kio::wait(|waiter| { - if let Poll::Ready(res) = waiter.poll_future(serve.as_mut()) { - return Poll::Ready(res); + if let Poll::Ready(served) = waiter.poll_future(serve.as_mut()) { + return Poll::Ready(Some(served)); } let mut cx = std::task::Context::from_waker(waiter.waker()); if stream.reader.poll_closed(&mut cx).is_ready() || closed_session.poll_closed(&mut cx).is_ready() { - return Poll::Ready(Ok(())); + return Poll::Ready(None); } Poll::Pending }) .await }; + // Every data stream this subscription opened is closed by now, which PUBLISH_DONE + // requires, so the count it reports is final. + let (res, filled) = served.unwrap_or((Ok(()), false)); + let mut streams = track_serve.opened() + u64::from(filled); + + // Draft-14 on carries no end location in PUBLISH_DONE: an END_OF_TRACK object is + // what tells the subscriber where the track ended. + if res.is_ok() + && let Some(end) = track_serve.end() + { + match track_serve.write_end_of_track(end, priority).await { + Ok(()) => streams += 1, + // A failure only costs the subscriber the early boundary. + Err(err) => tracing::debug!(%err, id = %request_id, "end of track failed"), + } + } + // Send PublishDone let (status, reason) = match &res { Ok(()) => (ietf::PublishDoneStatus::TrackEnded, "track ended"), @@ -630,7 +651,7 @@ where _ => None, }, status_code: status.code(self.version), - stream_count: 0, + stream_count: streams, reason_phrase: reason.into(), }) .await; @@ -716,6 +737,8 @@ where /// A fill is a promise once requested. An empty range opens no stream, but a range we /// cannot serve still opens one and resets it right after the FETCH_HEADER, the /// draft's fill-failure signal. Nothing here touches the subscription either way. + /// + /// Returns whether it opened a stream, which PUBLISH_DONE's Stream Count includes. async fn run_fill( &self, request_id: RequestId, @@ -723,9 +746,9 @@ where fill: FillServe, track: track::Consumer, timescale: Option, - ) { + ) -> bool { if matches!(fill, FillServe::Empty) { - return; + return false; } let mut session = self.session.clone(); @@ -733,7 +756,7 @@ where Ok(stream) => stream, Err(err) => { tracing::debug!(err = %Error::from_transport(err), fill = %request_id, "fill stream failed to open"); - return; + return false; } }; let mut stream = Writer::new(stream, self.version); @@ -775,6 +798,7 @@ where stream.abort(&err); } } + true } /// Write one group's frames in the negotiated draft's FETCH object layout. @@ -1929,6 +1953,10 @@ struct TrackServe { children: kio::Tasks>, /// The track finished: the in-flight group machines drain, then FIN. draining: bool, + /// Group streams opened, shared with the group machines. + opened: Arc, + /// The track's exclusive end, once its groups ran out because it finished. + end: Option, } impl TrackServe { @@ -1959,9 +1987,51 @@ impl TrackServe { timescale, children: kio::Tasks::new(), draining: false, + opened: Default::default(), + end: None, } } + /// Group streams opened so far. + fn opened(&self) -> u64 { + self.opened.load(Ordering::Relaxed) + } + + /// Where the track ends, when the subscription ran to that end rather than stopping at + /// its own range first. + fn end(&self) -> Option { + let end = self.end?; + let reached = self.range.end.is_none_or(|last| last.group.saturating_add(1) >= end); + reached.then_some(end) + } + + /// Mark the track's end with an END_OF_TRACK object on its own stream, at object 0 of + /// the group that will never exist. + /// + /// The last group's stream has usually finished before the track ends, so the marker + /// cannot ride on it. The stream counts toward PUBLISH_DONE once it is open. + async fn write_end_of_track(&mut self, end: u64, priority: u8) -> Result<(), Error> { + let mut stream = std::future::poll_fn(|cx| self.session.poll_open_uni(cx)) + .await + .map_err(Error::from_transport)?; + stream.set_priority(priority); + + let mut writer = Writer::new(stream, self.version); + writer.buffer(&ietf::GroupHeader { + track_alias: self.request_id.0, + group_id: end, + sub_group_id: 0, + publisher_priority: super::priority::to_wire(self.track.info().priority), + flags: ietf::GroupFlags::default(), + })?; + // Object ID delta 0, then an empty object whose status is END_OF_TRACK. + writer.buffer(&0u64)?; + writer.buffer(&0u64)?; + writer.encode(&END_OF_TRACK).await?; + // PUBLISH_DONE follows once this closes, like every other data stream. + writer.close().await + } + fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { if self.draining { return self.children.poll(waiter).map(Ok); @@ -2006,18 +2076,24 @@ impl TrackServe { }, }; - self.children.push(GroupServe::new( - self.session.clone(), - msg, - self.track.subscription().priority, - group, - self.timescale, - self.version, - slice, - )); + self.children.push( + GroupServe::new( + self.session.clone(), + msg, + self.track.subscription().priority, + group, + self.timescale, + self.version, + slice, + ) + .counted(self.opened.clone()), + ); } Poll::Ready(Ok(None)) => { self.draining = true; + if let Poll::Ready(Ok(end)) = self.track.poll_finished(waiter) { + self.end = Some(end); + } return self.children.poll(waiter).map(Ok); } Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), @@ -2034,6 +2110,8 @@ impl TrackServe { /// subgroup format. struct GroupServe { session: S, + /// The subscription's count of opened streams, bumped once this one opens. + opened: Arc, msg: ietf::GroupHeader, priority: u8, group: group::Consumer, @@ -2090,6 +2168,7 @@ impl GroupServe { let object_delta = group.index(); Self { session, + opened: Default::default(), msg, priority, group, @@ -2100,6 +2179,12 @@ impl GroupServe { } } + /// Count this group's stream into `opened` once it opens. + fn counted(mut self, opened: Arc) -> Self { + self.opened = opened; + self + } + fn poll_serve(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { @@ -2116,6 +2201,7 @@ impl GroupServe { return Poll::Ready(Err(Error::from_transport(err))); } }; + self.opened.fetch_add(1, Ordering::Relaxed); let mut stream = stream; stream.set_priority(self.priority); @@ -2272,6 +2358,9 @@ impl GroupServe { } } +/// Object status: no object at or past this location exists. +const END_OF_TRACK: u64 = 0x4; + /// Buffer one object's header and prefix: the id delta, optional extension /// headers carrying the timestamp, the size, and (for an empty object) the status. fn buffer_object( diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 6185af5443..b497317dde 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -7,7 +7,7 @@ use std::{ use crate::{ Error, Path, PathOwned, SessionError, Timescale, broadcast, - coding::{Reader, Stream}, + coding::{Decode, DecodeError, Reader, Stream}, frame, group, ietf::{self, Control, FetchType, Filter, GroupOrder, RequestId}, origin, track, @@ -15,6 +15,7 @@ use crate::{ }; use super::{Message, Version, cluster, error::request, peer}; +use crate::tail::{self, Settle, Tail}; use kio::Lock; @@ -322,6 +323,9 @@ struct TrackState { // A pre-draft-20 joining FETCH, which reuses the fill rendezvous. joining: Option, + + // The data streams read so far, which PUBLISH_DONE's Stream Count is checked against. + tail: kio::Producer, } impl TrackState { @@ -348,6 +352,21 @@ impl TrackState { largest: None, fetch_id: None, joining, + tail: Default::default(), + } + } +} + +/// Counts a data stream toward its subscription's [`Tail`] once its handler is done with it. +/// +/// Counting at the end rather than at the header means a Stream Count that is reached never +/// races the stream's own work, such as the END_OF_TRACK that declares where the track ends. +struct Counted(kio::Producer); + +impl Drop for Counted { + fn drop(&mut self) { + if let Ok(mut tail) = self.0.write() { + tail.stream(); } } } @@ -1753,7 +1772,7 @@ where // does not disturb subscriptions already in flight. enum End { Unused, - Done(Result<(), Error>), + Done(Result), } let mut fetch_done = fetching.is_none(); @@ -1784,8 +1803,34 @@ where }, End::Done(res) => { match res { - Ok(()) => { + Ok(count) => { tracing::info!(broadcast = %self.origin.absolute(&broadcast_path), track = %track_name, "subscribe complete"); + // The publisher sends PUBLISH_DONE once every data stream it opened + // is closed, but QUIC does not order them, so some can still be on + // their way. Wait until Stream Count of them are read, or a bounded + // grace for any reset before its header (the draft says to use a + // timeout). The count is a hint: a published peer sends 0, which + // waits out the grace. + let tail = self + .state + .lock() + .subscribes + .get(&request_id) + .map(|held| held.tail.consume()); + if let Some(tail) = tail { + let mut settle = Settle::new(&self.runtime, tail, tail::GRACE); + kio::wait(|waiter| { + if !fetch_done + && let Some(fut) = fetching.as_mut() + && waiter.poll_future(fut.as_mut()).is_ready() + { + fetch_done = true; + } + settle.poll(waiter, |tail| count > 0 && tail.streams() >= count) + }) + .await; + } + // A no-op once an END_OF_TRACK declared the end. let _ = track.finish(); } Err(err) => { @@ -1812,11 +1857,12 @@ where } } - /// Read the PUBLISH_DONE that ends an Established subscription, as the end it reports. + /// Read the PUBLISH_DONE that ends an Established subscription, as the end it reports + /// and, for a clean end, its Stream Count. /// /// The publisher must send it before its FIN (draft-19 section 3.3.2), so a FIN /// without one is a failed request, not a clean end. - async fn read_publish_done(reader: &mut Reader, version: Version) -> Result<(), Error> { + async fn read_publish_done(reader: &mut Reader, version: Version) -> Result { match reader.decode_maybe::().await? { Some(ietf::PublishDone::ID) => {} Some(_) => return Err(Error::UnexpectedMessage), @@ -1824,7 +1870,8 @@ where } let msg: ietf::PublishDone = reader.decode().await?; tracing::debug!(message = ?msg, "received publish done"); - msg.end(version) + msg.end(version)?; + Ok(msg.stream_count) } /// Tell the publisher to stop serving a subscription we are walking away from. @@ -2099,13 +2146,16 @@ where } }; - let (track, timescale, fill) = { + let (mut track, timescale, fill, _counted) = { let state = self.state.lock(); let track = state.subscribes.get(&request_id).ok_or(Error::NotFound)?; ( track.producer.clone().ok_or(Error::NotFound)?, track.timescale, track.fill.clone(), + // Every data stream counts toward PUBLISH_DONE's Stream Count, even one + // dropped below. + Counted(track.tail.clone()), ) }; @@ -2140,9 +2190,9 @@ where // The peek inside blocks until the publisher produces the group's first object, so // race it against the subscription going away the same way the group read below is. // Otherwise dropping the local subscriber cannot end this handler. - let (producer, start) = { + let opened = { let mut opening = track.clone(); - let mut open = std::pin::pin!(self.open_group(stream, &mut opening, &fill, group.group_id)); + let mut open = std::pin::pin!(self.open_group(stream, &mut opening, &fill, &group)); kio::wait(|waiter| { if let Poll::Ready(err) = track.poll_closed(waiter) { return Poll::Ready(Err(err)); @@ -2151,6 +2201,11 @@ where }) .await? }; + let (producer, start) = match opened { + Opened::Group(producer, start) => (producer, start), + // No object at or past object 0 of this group exists, so neither does the group. + Opened::EndOfTrack => return end_track(&mut track, group.group_id), + }; // Guarded: this handler can be dropped at any await below, and a group producer // that dies without a terminal leaves its consumer waiting on nothing. @@ -2179,8 +2234,13 @@ where tracing::debug!(%err, group = %producer.sequence, "group error"); let _ = producer.abort(err); } - _ => { + Ok(Ended::Group) => { + let _ = producer.finish(); + } + // No object past this group's last one exists, so the track ends after it. + Ok(Ended::Track) => { let _ = producer.finish(); + return end_track(&mut track, group.group_id.saturating_add(1)); } } @@ -2188,35 +2248,107 @@ where } } +/// Mark where the track ends, as an END_OF_TRACK object said. +/// +/// Draft-14 on carries no end location in PUBLISH_DONE, so this is what lets a subscriber +/// learn the end before the live edge reaches it. A boundary at or below a group already +/// received is the publisher breaking its own end, which no later group can repair. +fn end_track(track: &mut track::Producer, end: u64) -> Result<(), Error> { + if track.final_sequence() == Some(end) { + return Ok(()); + } + if let Err(err) = track.finish_at(end) { + tracing::warn!(%err, end, "invalid END_OF_TRACK"); + let _ = track.clone().abort(Error::ProtocolViolation); + return Err(Error::ProtocolViolation); + } + Ok(()) +} + +/// How [`Subscriber::open_group`] resolved a subgroup stream. +enum Opened { + /// The group producer the stream writes into, and the Object ID it starts at. + Group(group::Producer, u64), + /// The stream's first object is an END_OF_TRACK at object 0: the group does not exist. + EndOfTrack, +} + +/// How a subgroup stream ended cleanly. +#[derive(Debug, PartialEq, Eq)] +enum Ended { + /// The stream finished, or carried an explicit end of group. + Group, + /// It carried an END_OF_TRACK after the group's last object. + Track, +} + +/// Object status: no object at or past this location exists (every implemented draft). +const END_OF_TRACK: u64 = 0x4; + +/// The start of a subgroup stream's first object, peeked before its group is created. +#[derive(Debug, Clone, Copy)] +struct FirstObject { + /// The Object ID, which the first object's delta is. + id: u64, + end_of_track: bool, +} + +/// [`FirstObject`] for a stream whose objects carry extensions, or don't. +#[derive(Debug)] +struct PeekFirst(FirstObject); + +impl Decode for PeekFirst { + fn decode(buf: &mut B, version: Version) -> Result { + let id = u64::decode(buf, version)?; + if EXTENSIONS { + let size = usize::decode(buf, version)?; + if buf.remaining() < size { + return Err(DecodeError::Short); + } + buf.advance(size); + } + let size = u64::decode(buf, version)?; + let end_of_track = size == 0 && u64::decode(buf, version)? == END_OF_TRACK; + Ok(Self(FirstObject { id, end_of_track })) + } +} + impl Subscriber where S: crate::transport::poll::Boxable, { /// The group producer this subgroup stream writes into, and the Object ID it starts at. /// - /// Normally the stream starts the group. While a fill is outstanding it may instead be - /// the tail of the group the fill fetch stream began, which the first Object ID decides, - /// so the stream is peeked before any producer exists. A subscription with no fill skips - /// the peek: creating the group up front is what it has always done, and waiting for the - /// first object would hold the group back for as long as the publisher takes to produce - /// it. + /// The first object is peeked before any producer exists, since an END_OF_TRACK at + /// object 0 means the group does not exist at all. Normally the stream then starts the + /// group. While a fill is outstanding it may instead be the tail of the group the fill + /// fetch stream began, which the first Object ID decides. async fn open_group( &self, stream: &mut Reader, track: &mut track::Producer, fill: &kio::Producer, - sequence: u64, - ) -> Result<(group::Producer, u64), Error> { + header: &ietf::GroupHeader, + ) -> Result { + let sequence = header.group_id; // Stats (groups/frames/bytes) are counted in the model as the group is written, // through the tagged `track::Producer`. let create = |track: &mut track::Producer| track.create_group(group::Info { sequence }); + let first = match header.flags.has_extensions { + true => stream.decode_peek_maybe::>().await?.map(|peek| peek.0), + false => stream.decode_peek_maybe::>().await?.map(|peek| peek.0), + }; + if first.is_some_and(|first| first.id == 0 && first.end_of_track) { + return Ok(Opened::EndOfTrack); + } + if !fill.read().outstanding() { - return Ok((create(track)?, 0)); + return Ok(Opened::Group(create(track)?, 0)); } // The first object's ID delta is its absolute Object ID (see `next_object_id`). - match stream.decode_peek_maybe::().await? { + match first.map(|first| first.id) { // A group delivered from its start stands alone, unless the fill already // headed this very sequence: the publisher then served those objects twice, // and the model has one producer per group. Publish the head as the prefix it @@ -2231,13 +2363,13 @@ where return Err(Error::Unsupported); } - Ok((create(track)?, 0)) + Ok(Opened::Group(create(track)?, 0)) } // A group starting partway through is the tail of one the fill began, and // without that head it has a hole at the front. Some(start) => match self.claim_fill(fill, track, sequence, Some(start)).await? { - Some(producer) => Ok((producer, start)), + Some(producer) => Ok(Opened::Group(producer, start)), None => { tracing::warn!(sequence, start, "no fill to stitch a mid-group stream onto"); Err(Error::Unsupported) @@ -2247,8 +2379,8 @@ where // A stream that ends without an object: the group is over and had nothing // outside the fill's range, so the head it delivered is the whole group. None => match self.claim_fill(fill, track, sequence, None).await? { - Some(producer) => Ok((producer, 0)), - None => Ok((create(track)?, 0)), + Some(producer) => Ok(Opened::Group(producer, 0)), + None => Ok(Opened::Group(create(track)?, 0)), }, } } @@ -2319,8 +2451,8 @@ enum IngestPhase { Status { timestamp: Option }, /// Streaming the object payload. Payload { frame: frame::ProducerOwned }, - /// An explicit end-of-group status arrived. - Finished, + /// An explicit end-of-group or end-of-track status arrived. + Finished(Ended), } impl GroupIngest { @@ -2361,17 +2493,17 @@ where let _: u64 = stream.decode().await?; let header: ietf::FetchHeader = stream.decode().await?; - let (subscribe_id, fill, joining, largest) = { + let (subscribe_id, fill, joining, largest, _counted) = { let state = self.state.lock(); // A draft-20 fill is named by the SUBSCRIBE's request id. A pre-draft-20 joining // FETCH has its own id, which `fetches` maps back to that subscription. - let subscribe_id = state - .fetches - .get(&header.request_id) - .copied() - .unwrap_or(header.request_id); + let joined = state.fetches.get(&header.request_id).copied(); + let subscribe_id = joined.unwrap_or(header.request_id); let track = state.subscribes.get(&subscribe_id).ok_or(Error::NotFound)?; - (subscribe_id, track.fill.clone(), track.joining, track.largest) + // A fill is one of the subscription's own data streams, so PUBLISH_DONE counts + // it. A joining FETCH is a request of its own. + let counted = joined.is_none().then(|| Counted(track.tail.clone())); + (subscribe_id, track.fill.clone(), track.joining, track.largest, counted) }; // SUBSCRIBE_OK declares the units these object timestamps are in, and this stream can @@ -2740,21 +2872,21 @@ fn settle_join_live(fill: &kio::Producer) { } impl GroupIngest { - /// `Ready(Ok(()))` once the stream FINs on an object boundary (or an explicit - /// end-of-group status arrives). The caller finishes or aborts the group; an - /// object cut short mid-payload was already aborted here with the reason. + /// `Ready(Ok(_))` once the stream FINs on an object boundary, or an explicit + /// end-of-group or end-of-track status arrives. The caller finishes or aborts the + /// group; an object cut short mid-payload was already aborted here with the reason. fn poll( &mut self, reader: &mut Reader, group: &mut group::Producer, waiter: &kio::Waiter, - ) -> Poll> { + ) -> Poll> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { match &mut self.phase { IngestPhase::Delta => { let Some(id_delta) = ready!(reader.poll_decode_maybe::(&mut cx))? else { - return Poll::Ready(Ok(())); + return Poll::Ready(Ok(Ended::Group)); }; self.prior_object = Some(next_object_id(self.prior_object, id_delta, self.start)?); self.phase = match self.has_extensions { @@ -2799,7 +2931,11 @@ impl GroupIngest { frame.finish()?; self.phase = IngestPhase::Delta; } else if status == 3 && !self.has_end { - self.phase = IngestPhase::Finished; + self.phase = IngestPhase::Finished(Ended::Group); + } else if status == END_OF_TRACK { + // Defined on every implemented draft, whether or not the header marks + // the group's end. + self.phase = IngestPhase::Finished(Ended::Track); } else { return Poll::Ready(Err(Error::Unsupported)); } @@ -2820,7 +2956,10 @@ impl GroupIngest { } } } - IngestPhase::Finished => return Poll::Ready(Ok(())), + IngestPhase::Finished(ended) => { + let ended = std::mem::replace(ended, Ended::Group); + return Poll::Ready(Ok(ended)); + } } } } @@ -5738,6 +5877,42 @@ mod stitch_tests { assert!(matches!(*h.fill.read(), Fill::Done), "the head was claimed"); } + /// Append an END_OF_TRACK object: delta 0, an empty payload, then its status. + fn end_of_track(mut stream: Vec) -> Vec { + for value in [0u64, 0, END_OF_TRACK] { + value.encode(&mut stream, VERSION).unwrap(); + } + stream + } + + /// END_OF_TRACK after a group's last object ends the track right after that group. + #[tokio::test] + async fn an_end_of_track_after_a_group_ends_the_track_after_it() { + let h = Harness::new(Fill::Done, vec![end_of_track(tail_stream(SEQUENCE, 0, &[b"last"]))]); + let mut consumer = h.track.subscribe(None); + let mut stream = h.stream().await; + + h.subscriber.clone().recv_group(&mut stream).await.unwrap(); + assert_eq!(h.track.final_sequence(), Some(SEQUENCE + 1)); + + let mut group = consumer.recv_group().await.unwrap().expect("the group arrives"); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload.as_ref(), b"last"); + assert!(group.read_frame().await.unwrap().is_none(), "the group is finished"); + assert!(consumer.recv_group().await.unwrap().is_none(), "then the track ends"); + } + + /// END_OF_TRACK at object 0 says the group does not exist, so the track ends before it + /// and no group is created for it. + #[tokio::test] + async fn an_end_of_track_at_object_zero_creates_no_group() { + let h = Harness::new(Fill::Done, vec![end_of_track(tail_stream(SEQUENCE, 0, &[]))]); + let mut stream = h.stream().await; + + h.subscriber.clone().recv_group(&mut stream).await.unwrap(); + assert_eq!(h.track.final_sequence(), Some(SEQUENCE)); + assert_eq!(h.track.latest(), None, "no group was created"); + } + /// The group ended exactly where we joined it, so the subscription's stream carries no /// objects at all. That still ends the group, which is what publishes the head. #[tokio::test] diff --git a/rs/moq-net/src/lib.rs b/rs/moq-net/src/lib.rs index bb07231e07..0ec7ba1348 100644 --- a/rs/moq-net/src/lib.rs +++ b/rs/moq-net/src/lib.rs @@ -86,6 +86,7 @@ mod model; pub mod path; mod recv; mod setup; +mod tail; mod util; mod version; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 9013f381cc..46b90b0d5c 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -15,6 +15,7 @@ use crate::{ }; use super::Version; +use crate::tail::{self, Settle, Tail}; use kio::Lock; @@ -87,6 +88,8 @@ struct TrackEntry { /// Timestamp scale from this track's TRACK_INFO, known before the SUBSCRIBE is /// even opened, so group streams decode frames without blocking. timescale: Option, + /// The groups received so far, so the subscription's end can wait for the ones owed. + tail: kio::Producer, } impl Subscriber { @@ -438,6 +441,10 @@ impl Subscriber { let timestamp = Timestamp::new(dg.timestamp, scale).map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; + // A datagram is never owed a stream, so it never holds the subscription's end open. + if let Ok(mut tail) = entry.tail.write() { + tail.account(dg.sequence..dg.sequence.saturating_add(1)); + } entry.producer.insert_datagram(dg.sequence, timestamp, dg.payload)?; Ok(()) } @@ -719,6 +726,9 @@ impl GroupRecv { let (group, track, timescale) = { let mut subs = self.subscriber.subscribes.lock(); let entry = subs.get_mut(&hdr.subscribe).ok_or(Error::Cancel)?; + if let Ok(mut tail) = entry.tail.write() { + tail.open(hdr.sequence); + } let group_info = group::Info { sequence: hdr.sequence }; // Stats (groups/frames/bytes) are counted in the model as the group @@ -1365,6 +1375,7 @@ mod tests { TrackEntry { producer, timescale: Some(Timescale::default()), + tail: Default::default(), }, ); @@ -2485,6 +2496,24 @@ struct SubStream { /// start sits elsewhere the request-tracked floor stands instead, and demand /// returning here makes the declaration valid again. requested: Option, + /// The groups received for this subscription, shared with its [`TrackEntry`]. + tail: kio::Producer, + /// The first group the publisher serves (SUBSCRIBE_START), once declared. + served: Option, + /// The track's exclusive end (SUBSCRIBE_END), once declared. + end: Option, +} + +impl SubStream { + /// The groups the publisher still owes once it has ended the subscription. + /// + /// `None` when nothing says which: drafts before SUBSCRIBE_END only have the FIN. + /// Without a SUBSCRIBE_START the publisher served no group at all. + fn owed(&self, requested_end: Option) -> Option> { + let end = self.end?; + let end = requested_end.map_or(end, |requested| requested.min(end)); + Some(self.served.unwrap_or(end)..end) + } } enum Sub { @@ -2787,11 +2816,13 @@ impl TrackServe { tracing::info!(id, broadcast = %self.subscriber.log_path(&self.path), track = %self.name, "subscribe started"); + let tail = kio::Producer::new(Tail::default()); self.subscriber.subscribes.lock().insert( id, TrackEntry { producer: producer.clone(), timescale, + tail: tail.clone(), }, ); @@ -2802,6 +2833,7 @@ impl TrackServe { session, id, subscription, + tail, state: EstablishState::Open, } } @@ -2901,6 +2933,7 @@ struct Establish { closed: S, id: u64, subscription: Subscription, + tail: kio::Producer, state: EstablishState, } @@ -2985,6 +3018,9 @@ impl Establish { start: self.subscription.start, priority: self.subscription.priority, requested: self.subscription.start, + tail: self.tail.clone(), + served: None, + end: None, } } } @@ -3197,6 +3233,13 @@ enum ServeMode { /// Driving an upstream SUBSCRIBE open. The demand arms wait meanwhile, /// exactly like the old inline await. Establish(Establish), + /// The upstream FIN'd, so the track is over, but QUIC does not order streams: keep + /// the subscription routable until every group it owes is accounted for (a stream's + /// header or a SUBSCRIBE_DROP), or the grace gives up on one reset before its header. + Tail { + settle: Settle, + owed: Option>, + }, } impl ServeLoop { @@ -3242,6 +3285,23 @@ impl ServeLoop { } } } + ServeMode::Tail { settle, owed } => { + let _ = self.fetches.poll(waiter); + if settle + .poll(waiter, |tail| owed.clone().is_some_and(|owed| tail.covers(owed))) + .is_ready() + { + return Poll::Ready(ServeEnd::Finished); + } + if self.fetches.is_empty() && self.serving.poll_unused(waiter).is_ready() { + return Poll::Ready(ServeEnd::Idle); + } + let mut cx = std::task::Context::from_waker(waiter.waker()); + if self.closed.poll_closed(&mut cx).is_ready() { + return Poll::Ready(ServeEnd::GiveBack(Error::Dropped)); + } + return Poll::Pending; + } ServeMode::Select => { let mut cx = std::task::Context::from_waker(waiter.waker()); @@ -3341,6 +3401,7 @@ impl ServeLoop { if let Err(err) = self.serving.finish_at(end.group) { tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end"); } + active.end = Some(end.group); } // SUBSCRIBE_START names the first group this feed serves: // the publisher skipped everything below it (e.g. it could @@ -3357,21 +3418,50 @@ impl ServeLoop { if active.start == active.requested { let _ = self.serving.start_at(start.group); } + active.served = Some(start.group); + } + // The publisher will never send these groups, so they + // are accounted for without a stream. + lite::SubscribeResponse::Drop(dropped) => { + if let Ok(mut tail) = active.tail.write() { + tail.account(dropped.start..dropped.end.saturating_add(1)); + } + } + // OK just resolves the range (the producer already orders + // groups). + lite::SubscribeResponse::Ok(_) => { + tracing::debug!(track = %serve.name, ?msg, "subscribe response") } - // OK/DROP just resolve the range (the producer already - // orders groups). - _ => tracing::debug!(track = %serve.name, ?msg, "subscribe response"), } continue; } Ok(None) => { tracing::info!(broadcast = %serve.subscriber.log_path(&serve.path), track = %serve.name, "subscribe complete"); // Upstream FIN'd the subscription: the publisher only FINs - // once the track's final sequence is known and delivered, so - // the logical track is over for good (bounded downstream - // demand alone never FINs; the publisher parks, since a cap - // can be raised). - return Poll::Ready(ServeEnd::Finished); + // once the track's final sequence is known and every group + // stream finished, so the logical track is over for good + // (bounded downstream demand alone never FINs; the publisher + // parks, since a cap can be raised). Those streams can still + // be in flight, so wait for the tail before finishing. + let subscription = self.serving.subscription(); + let requested_end = subscription.as_ref().and_then(|sub| sub.end).map(|end| match end + .frame + { + 0 => end.group, + _ => end.group.saturating_add(1), + }); + // The effective max age is the stopgap grace: the wrong clock + // (it bounds presentation-time drift), but it is how long the + // subscriber was willing to wait for a late group anyway. + let grace = subscription + .map(|sub| sub.max_age) + .filter(|max_age| !max_age.is_zero()) + .unwrap_or(tail::GRACE); + self.mode = ServeMode::Tail { + settle: Settle::new(&serve.subscriber.runtime, active.tail.consume(), grace), + owed: active.owed(requested_end), + }; + continue; } Err(err) => { tracing::warn!(broadcast = %serve.subscriber.log_path(&serve.path), track = %serve.name, %err, "subscribe error"); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ba83c78728..6cfda585dc 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -218,6 +218,10 @@ pub(crate) struct TrackState { // The sequence number at which the track was finalized. final_sequence: Option, + // The last producer dropped after the boundary was declared, so a group still missing + // below it will never be produced. + sealed: bool, + // The first sequence the live feed serves, once the publisher declared one // (the wire's SUBSCRIBE_START). Lower groups never arrive on their own; a // fetch can still create them. @@ -437,9 +441,11 @@ impl TrackState { } // `final_sequence` is one past the last possible sequence. If our // floor is already at/past it, nothing else can land in range. - if let Some(fin) = self.final_sequence - && next_sequence >= fin - { + // A sealed track produces nothing more either: the last producer + // dropped with the boundary already declared, so a gap below it is + // the end, not a wait. Cached in-range groups were returned above. + // This is the cursor the ordered and spliced readers use. + if self.sealed || self.final_sequence.is_some_and(|fin| next_sequence >= fin) { return Poll::Ready(Ok(None)); } Poll::Pending @@ -1068,11 +1074,12 @@ impl TrackState { /// Whether the track has reached its end: the final boundary is set and the live /// edge has caught up to it, so no further group can arrive. A future boundary /// (declared via [`Producer::finish_at`] ahead of the live edge) stays incomplete - /// until the remaining groups are produced. Drives the end-of-stream signal from + /// until the remaining groups are produced, or until the last producer drops without + /// them. Drives the end-of-stream signal from /// the read methods (`recv_group` / `next_group` / `read_frame` return `None`). fn is_complete(&self) -> bool { self.final_sequence - .is_some_and(|fin| self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin) + .is_some_and(|fin| self.sealed || self.max_sequence.map_or(0, |max| max.saturating_add(1)) >= fin) } /// Where a replacement route should pick this track up: one past the last frame @@ -1902,7 +1909,13 @@ impl Drop for Alive { // leaves it open with `final_sequence` set, so inspect both outcomes. match self.state.write() { Ok(mut state) => { - if state.final_sequence.is_some() || state.abort.is_some() { + if state.final_sequence.is_some() { + // Groups still missing below the boundary can no longer arrive, so a + // reader waiting on one ends cleanly instead of with `Dropped`. + state.sealed = true; + return; + } + if state.abort.is_some() { return; } tracing::warn!( @@ -6721,6 +6734,50 @@ mod test { assert!(done.is_none(), "consumer should drain then see clean finish"); } + /// A boundary declared ahead of the live edge, then the last producer dropping, means + /// the missing groups will never come: the reader ends cleanly rather than with + /// `Dropped`, as it would for a track that never declared its end. + #[tokio::test] + async fn drop_short_of_the_boundary_ends_cleanly() { + let mut producer = track_producer("test", None); + producer.append_group().unwrap(); + producer.finish_at(3).unwrap(); + + let mut consumer = producer.subscribe(None); + assert_eq!(consumer.assert_group().sequence, 0); + assert!( + consumer.recv_group().now_or_never().is_none(), + "groups 1 and 2 are owed" + ); + + drop(producer); + let done = consumer.recv_group().now_or_never().expect("should not block").unwrap(); + assert!( + done.is_none(), + "the track ends at its boundary without the missing groups" + ); + } + + /// The sequence cursor ends the same way. A gap below the boundary is skipped, + /// a later cached group is still delivered, and the read finishes cleanly. + #[tokio::test] + async fn ordered_ends_cleanly_when_sealed_short_of_the_boundary() { + let mut producer = track_producer("test", None); + producer.create_group(group::Info { sequence: 0 }).unwrap(); + producer.create_group(group::Info { sequence: 2 }).unwrap(); + producer.finish_at(4).unwrap(); + + let mut ordered = producer.subscribe(None).ordered(); + drop(producer); + + let first = ordered.next_group().now_or_never().expect("group 0").unwrap().unwrap(); + assert_eq!(first.sequence, 0); + let second = ordered.next_group().now_or_never().expect("group 2").unwrap().unwrap(); + assert_eq!(second.sequence, 2); + let done = ordered.next_group().now_or_never().expect("end").unwrap(); + assert!(done.is_none(), "missing groups below the boundary are not an error"); + } + #[tokio::test] async fn cached_groups_preserve_arrival_order() { let producer = track_producer("test", None); diff --git a/rs/moq-net/src/tail.rs b/rs/moq-net/src/tail.rs new file mode 100644 index 0000000000..be30f10857 --- /dev/null +++ b/rs/moq-net/src/tail.rs @@ -0,0 +1,128 @@ +//! A subscription's tail: the group streams still owed once the publisher has ended it. +//! +//! A publisher ends a subscription only after every group stream it opened has finished, +//! but QUIC does not order streams, so one opened before the end can still reach us after +//! it. The subscriber keeps the subscription routable until each owed group is accounted +//! for, or a grace expires for a group whose stream was reset before its header arrived. +//! A reset that keeps the header (reliable reset) would make the grace unnecessary. + +use std::{ops::Range, task::Poll, time::Duration}; + +use crate::runtime::Deadline; + +/// How long a subscriber waits for a group stream it cannot account for once the +/// publisher has ended the subscription. +/// +/// Bounds the wait on IETF, and on moq-lite when the subscription has no max age to bound +/// it with. Matches `@moq/net`, so a reader cannot tell which side it is talking to. +pub(crate) const GRACE: Duration = Duration::from_secs(1); + +/// The data streams a subscription has received, and the groups they account for. +#[derive(Default, Debug)] +pub(crate) struct Tail { + // Disjoint, sorted, non-adjacent ranges of accounted sequences, so this grows with the + // number of gaps rather than the number of groups. + accounted: Vec>, + streams: u64, +} + +impl Tail { + /// Record a group stream whose header arrived. + pub fn open(&mut self, sequence: u64) { + self.stream(); + self.account(sequence..sequence.saturating_add(1)); + } + + /// Record a data stream that names no single group, such as a fill. + pub fn stream(&mut self) { + self.streams += 1; + } + + /// Record groups as accounted for without a stream: dropped, or sent as a datagram. + pub fn account(&mut self, groups: Range) { + if groups.is_empty() { + return; + } + + // Every range the insert overlaps or touches merges with it into one. + let first = self.accounted.partition_point(|range| range.end < groups.start); + let last = self.accounted.partition_point(|range| range.start <= groups.end); + let merged = match &self.accounted[first..last] { + [] => groups, + [head, .., tail] => head.start.min(groups.start)..tail.end.max(groups.end), + [only] => only.start.min(groups.start)..only.end.max(groups.end), + }; + self.accounted.splice(first..last, [merged]); + } + + /// Whether every group in `groups` is accounted for. + pub fn covers(&self, groups: Range) -> bool { + // Ranges merge on insert, so one range covers the span or none does. + groups.is_empty() + || self + .accounted + .iter() + .any(|range| range.start <= groups.start && groups.end <= range.end) + } + + /// Data streams received, whether they finished or were reset. + pub fn streams(&self) -> u64 { + self.streams + } +} + +/// Waits out a subscription's tail: until the owed streams are accounted for, or the grace. +pub(crate) struct Settle { + tail: kio::Consumer, + grace: Deadline, +} + +impl Settle { + /// Start waiting on `tail`, giving up on missing streams after `grace`. + pub fn new(runtime: &crate::time::Clock, tail: kio::Consumer, grace: Duration) -> Self { + Self { + tail, + grace: Deadline::after(runtime, grace), + } + } + + /// Ready once `complete` holds, the grace expires, or the subscription is gone. + pub fn poll(&mut self, waiter: &kio::Waiter, mut complete: impl FnMut(&Tail) -> bool) -> Poll<()> { + if self.grace.poll(waiter).is_ready() { + return Poll::Ready(()); + } + self.tail + .poll(waiter, |tail| match complete(tail) { + true => Poll::Ready(()), + false => Poll::Pending, + }) + .map(|_| ()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ranges_merge_across_gaps() { + let mut tail = Tail::default(); + tail.open(0); + tail.open(2); + assert!(!tail.covers(0..3), "group 1 is missing"); + assert_eq!(tail.accounted, [0..1, 2..3]); + + tail.account(1..2); + assert!(tail.covers(0..3)); + assert_eq!(tail.accounted, [0..3], "adjacent ranges merge"); + + tail.account(5..7); + tail.account(9..10); + tail.account(4..9); + assert_eq!(tail.accounted, [0..3, 4..10], "one insert swallows several ranges"); + assert!(tail.covers(4..10)); + assert!(!tail.covers(2..5)); + assert!(tail.covers(7..7), "an empty range is always covered"); + assert_eq!(tail.streams(), 2, "only streams count, not drops"); + } +} diff --git a/rs/moq-net/tests/goaway.rs b/rs/moq-net/tests/goaway.rs index 2eefac0f0a..c85713fa82 100644 --- a/rs/moq-net/tests/goaway.rs +++ b/rs/moq-net/tests/goaway.rs @@ -338,7 +338,7 @@ async fn goaway_gates_new_subscribes_moq_lite_04() { let mut opts = MockConnectOptions::new(version); opts.server_publish = Some(pub_origin.clone()); opts.client_subscribe = Some(sub_origin.clone()); - let MockPair { client, server } = connect_mock(opts).await; + let MockPair { client, server, .. } = connect_mock(opts).await; // Subscribe BEFORE the GOAWAY and receive a first group. let sub = sub_origin.consume(); @@ -425,7 +425,7 @@ async fn goaway_drains_routes(version: Version) { let mut opts = MockConnectOptions::new(version); opts.server_publish = Some(pub_origin.clone()); opts.client_subscribe = Some(sub_origin.clone()); - let MockPair { client, server } = connect_mock(opts).await; + let MockPair { client, server, .. } = connect_mock(opts).await; let sub = sub_origin.consume(); let route = sub.routed("test").await.expect("route announced"); diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index 3eba29225a..701d83fb32 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -9,7 +9,7 @@ use moq_net::{Client, Server, Session, Version, origin}; -use super::mock::create_mock_session_pair; +use super::mock::{MockSession, create_mock_session_pair}; pub use moq_net::time::run; @@ -49,6 +49,10 @@ impl MockConnectOptions { pub struct MockPair { pub client: Session, pub server: Session, + /// The client's end of the mock transport, for steering delivery. + pub client_transport: MockSession, + /// The server's end of the mock transport, for steering delivery. + pub server_transport: MockSession, } /// Run the MoQ handshake over the mock transport, returning connected sessions. @@ -63,6 +67,7 @@ pub struct MockPair { pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { let protocol = opts.version.alpn(); let (client_transport, server_transport) = create_mock_session_pair(Some(protocol)); + let transports = (client_transport.clone(), server_transport.clone()); let mut client = Client::new().with_versions(opts.version.into()); if let Some(publish) = &opts.client_publish { @@ -105,5 +110,7 @@ pub async fn connect_mock(opts: MockConnectOptions) -> MockPair { MockPair { client: client_session, server: server_session, + client_transport: transports.0, + server_transport: transports.1, } } diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index c30665b066..81fbd49f1c 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -97,6 +97,9 @@ pub struct MockSendStream { tx: Option>, closed: Arc, park: kio::Park, + /// Acknowledge the FIN as soon as it is sent, for a stream the peer's transport holds + /// back from its application (see [`MockSession::hold_unis`]). + ack_fin: bool, } impl poll::SendStream for MockSendStream { @@ -117,6 +120,9 @@ impl poll::SendStream for MockSendStream { fn finish(&mut self) -> Result<(), Self::Error> { if let Some(tx) = self.tx.take() { let _ = tx.try_push(StreamChunk::Fin); + if self.ack_fin { + self.closed.set(Ok(())); + } } Ok(()) } @@ -255,6 +261,7 @@ fn new_stream_pair() -> (MockSendStream, MockRecvStream) { tx: Some(queue.clone()), closed: closed.clone(), park: kio::Park::default(), + ack_fin: false, }; let recv = MockRecvStream { rx: queue, @@ -298,6 +305,8 @@ struct SessionSide { protocol: Option<&'static str>, /// Connection-level close state shared with the peer. conn: Arc, + /// Uni streams this side opened that the peer has not accepted yet, while held. + held: Mutex>>, } /// An in-memory mock WebTransport session. @@ -372,7 +381,13 @@ impl poll::Session for MockSession { } fn poll_open_uni(&mut self, _cx: &mut Context<'_>) -> Poll> { - let (our_send, peer_recv) = new_stream_pair(); + let (mut our_send, peer_recv) = new_stream_pair(); + + if let Some(held) = self.side.held.lock().unwrap().as_mut() { + our_send.ack_fin = true; + held.push(peer_recv); + return Poll::Ready(Ok(our_send)); + } // Deliver peer_recv to the peer's accept_uni. match self.side.peer_uni.try_push(peer_recv) { @@ -440,6 +455,34 @@ impl poll::Session for MockSession { } } +// Only some test binaries steer delivery. +#[allow(dead_code)] +impl MockSession { + /// Hold back the uni streams this side opens from now on. + /// + /// The peer's transport has them, so a FIN is acknowledged at once, but its application + /// does not see them until [`Self::release_unis`]. That is QUIC delivering streams out of + /// order: a publisher can see a group stream acknowledged and end the subscription + /// before the subscriber has read the group's header. + pub fn hold_unis(&self) { + self.side.held.lock().unwrap().get_or_insert_default(); + } + + /// Deliver the held uni streams to the peer in the order they were opened, and stop + /// holding. + pub fn release_unis(&self) { + for stream in self.side.held.lock().unwrap().take().unwrap_or_default() { + let _ = self.side.peer_uni.try_push(stream); + } + } + + /// Lose the held uni streams, as if each were reset before its header arrived, and + /// stop holding. + pub fn drop_unis(&self) { + self.side.held.lock().unwrap().take(); + } +} + impl MockSession { fn close_error(&self) -> MockError { self.side @@ -484,6 +527,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, peer_datagrams: c2s_datagrams.clone(), protocol, conn: conn.clone(), + held: Mutex::default(), }); let server_side = Arc::new(SessionSide { @@ -495,6 +539,7 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, peer_datagrams: s2c_datagrams, protocol, conn, + held: Mutex::default(), }); let new = |side| MockSession { diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs new file mode 100644 index 0000000000..e177e8deac --- /dev/null +++ b/rs/moq-net/tests/track_tail.rs @@ -0,0 +1,192 @@ +//! A track's tail: a group stream that reaches the subscriber after the publisher has +//! ended the subscription still belongs to the track. +//! +//! Publishers end a subscription only once every group stream they opened is finished, +//! but QUIC does not order streams, so the subscriber can read the end (moq-lite's +//! subscribe stream FIN, IETF's PUBLISH_DONE) before a group's header. The mock holds the +//! publisher's group streams back from the subscriber to make that ordering +//! deterministic, while acknowledging them to the publisher like a real transport would. +//! +//! Time is paused, so the one-second grace for a group that never arrives is free. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Timestamp, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); +const PAYLOAD: &[u8] = b"frame"; + +/// How long a subscriber waits for a group it cannot account for, with no max age set. +const GRACE: Duration = Duration::from_secs(1); + +/// moq-lite drafts with and without SUBSCRIBE_END, and IETF drafts over the control stream +/// adapter (14), on their own streams (17), and with subscription fills (20+). +const VERSIONS: &[&str] = &[ + "moq-lite-03", + "moq-lite-05", + "moq-lite-07", + "moq-transport-14", + "moq-transport-17", + "moq-transport-20", + "moq-transport-22", +]; + +/// What becomes of the group streams held back past the subscription's end. +#[derive(Clone, Copy, Debug)] +enum Late { + /// They arrive after the end. + Delivered, + /// They never arrive, like a stream reset before its header. + Lost, +} + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +struct Outcome { + frames: Vec>, + err: Option, + /// How long after the held streams were released (or lost) the track ended. + elapsed: Duration, +} + +/// Publish a one-group track, end it while its group stream is held back, then deliver or +/// lose that stream, and return what the subscriber read. +async fn round(version: &str, late: Late) -> Outcome { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.clone()); + options.client_subscribe = Some(subscriber.clone()); + let pair = connect_mock(options).await; + + let consumer = subscriber.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let reader = tokio::spawn(async move { + let subscription = moq_net::track::Subscription::default().with_start(moq_net::track::Position::group(0)); + let mut sub = remote + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let mut frames = Vec::new(); + let err = loop { + let mut group = match sub.recv_group().await { + Ok(Some(group)) => group, + Ok(None) => break None, + Err(err) => break Some(err), + }; + loop { + match group.read_frame().await { + Ok(Some(frame)) => frames.push(frame.payload.to_vec()), + Ok(None) => break, + Err(err) => panic!("group failed: {err}"), + } + } + }; + (frames, err, tokio::time::Instant::now()) + }); + + tokio::time::timeout(TIMEOUT, track.used()) + .await + .expect("no subscriber appeared") + .unwrap(); + + pair.server_transport.hold_unis(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, PAYLOAD).unwrap(); + group.finish().unwrap(); + track.finish().unwrap(); + drop(track); + + // Paused time only advances once every task is idle, so this runs the publisher to its + // end of the subscription and the subscriber through reading it. + tokio::time::sleep(GRACE / 10).await; + assert!( + !reader.is_finished(), + "{version}: the track ended before its group arrived" + ); + + let released = tokio::time::Instant::now(); + match late { + Late::Delivered => pair.server_transport.release_unis(), + Late::Lost => pair.server_transport.drop_unis(), + } + + let (frames, err, ended) = tokio::time::timeout(TIMEOUT, reader) + .await + .expect("the subscription never ended") + .expect("reader panicked"); + drop((pair, broadcast, publisher, subscriber)); + Outcome { + frames, + err, + elapsed: ended - released, + } +} + +/// A group whose header arrives after the subscription's end is delivered, then the track +/// ends cleanly. +#[tokio::test] +async fn a_group_after_the_end_is_delivered() { + tokio::time::pause(); + for version in VERSIONS { + let outcome = round(version, Late::Delivered).await; + assert!( + outcome.err.is_none() && outcome.frames == [PAYLOAD], + "{version}: got {} frame(s), err={:?}", + outcome.frames.len(), + outcome.err, + ); + + // Drafts that say where the track ends, or how many streams the publisher opened, + // end as soon as the last group arrives rather than waiting out the grace. + if *version != "moq-lite-03" { + assert!( + outcome.elapsed < GRACE / 10, + "{version}: ended after {:?}", + outcome.elapsed + ); + } + } +} + +/// A group that never arrives is given up on after the grace, and the track still ends +/// cleanly without it. +#[tokio::test] +async fn a_lost_group_ends_the_track_after_the_grace() { + tokio::time::pause(); + for version in VERSIONS { + let outcome = round(version, Late::Lost).await; + assert!( + outcome.err.is_none() && outcome.frames.is_empty(), + "{version}: got {} frame(s), err={:?}", + outcome.frames.len(), + outcome.err, + ); + assert!( + outcome.elapsed >= GRACE / 2, + "{version}: ended after {:?}", + outcome.elapsed + ); + } +} From 3f28ad20a06eea279fee11aeba62af8398e82d44 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 23:53:50 -0700 Subject: [PATCH 2/3] fix(net): leave a tail ended early alone, and skip END_OF_TRACK when cancelled Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 7 +++++-- rs/moq-net/src/ietf/subscriber.rs | 6 ++++-- rs/moq-net/src/tail.rs | 7 +++---- 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index ee5676d165..6177770669 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -622,12 +622,15 @@ where // Every data stream this subscription opened is closed by now, which PUBLISH_DONE // requires, so the count it reports is final. + let completed = served.is_some(); let (res, filled) = served.unwrap_or((Ok(()), false)); let mut streams = track_serve.opened() + u64::from(filled); // Draft-14 on carries no end location in PUBLISH_DONE: an END_OF_TRACK object is - // what tells the subscriber where the track ended. - if res.is_ok() + // what tells the subscriber where the track ended. A cancelled subscription is + // owed nothing more. + if completed + && res.is_ok() && let Some(end) = track_serve.end() { match track_serve.write_end_of_track(end, priority).await { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index b497317dde..342cf6e054 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2252,9 +2252,11 @@ where /// /// Draft-14 on carries no end location in PUBLISH_DONE, so this is what lets a subscriber /// learn the end before the live edge reaches it. A boundary at or below a group already -/// received is the publisher breaking its own end, which no later group can repair. +/// received is the publisher breaking its own end, which no later group can repair. A +/// marker that lands after the subscription already ended (its grace expired) changes +/// nothing. fn end_track(track: &mut track::Producer, end: u64) -> Result<(), Error> { - if track.final_sequence() == Some(end) { + if track.final_sequence().is_some() { return Ok(()); } if let Err(err) = track.finish_at(end) { diff --git a/rs/moq-net/src/tail.rs b/rs/moq-net/src/tail.rs index be30f10857..34e2ff6a29 100644 --- a/rs/moq-net/src/tail.rs +++ b/rs/moq-net/src/tail.rs @@ -110,16 +110,15 @@ mod tests { tail.open(0); tail.open(2); assert!(!tail.covers(0..3), "group 1 is missing"); - assert_eq!(tail.accounted, [0..1, 2..3]); + assert_eq!(tail.accounted, vec![0..1, 2..3]); tail.account(1..2); - assert!(tail.covers(0..3)); - assert_eq!(tail.accounted, [0..3], "adjacent ranges merge"); + assert!(tail.covers(0..3) && tail.accounted.len() == 1, "adjacent ranges merge"); tail.account(5..7); tail.account(9..10); tail.account(4..9); - assert_eq!(tail.accounted, [0..3, 4..10], "one insert swallows several ranges"); + assert_eq!(tail.accounted, vec![0..3, 4..10], "one insert swallows several ranges"); assert!(tail.covers(4..10)); assert!(!tail.covers(2..5)); assert!(tail.covers(7..7), "an empty range is always covered"); From be312bc14dae6a7767b6eb36d85ae8b3b9e80207 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 09:51:05 -0700 Subject: [PATCH 3/3] fix(net): test the tail on moq-lite-07-wip moq-lite-07 no longer parses; the draft is opt-in as moq-lite-07-wip. Point the lite-07 stream count quest at the accounting that landed instead of the deleted rust tail quest. --- quest/m1/lite-stream-count.md | 6 +++--- rs/moq-net/tests/track_tail.rs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/quest/m1/lite-stream-count.md b/quest/m1/lite-stream-count.md index 7bc958d9bc..053984fc56 100644 --- a/quest/m1/lite-stream-count.md +++ b/quest/m1/lite-stream-count.md @@ -34,9 +34,9 @@ before its header arrived is still invisible, so the track-tail grace stays. that is never counted, a reset stream, and a count of zero. Add a Rust-JS interop case. -This lands before lite-07 is finalized. Published drafts keep SUBSCRIBE_DROP. -lite-07 replaces it with the count in both Rust and JS. Rust still does not -send SUBSCRIBE_DROP; it only accounts one a peer sends. +This lands before lite-07 is finalized. Published drafts keep SUBSCRIBE_DROP, +which moq-net and `@moq/net` already account. lite-07 replaces it with the +count in both. Rust still does not send SUBSCRIBE_DROP. ## Related diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs index e177e8deac..28ef3d957f 100644 --- a/rs/moq-net/tests/track_tail.rs +++ b/rs/moq-net/tests/track_tail.rs @@ -27,7 +27,7 @@ const GRACE: Duration = Duration::from_secs(1); const VERSIONS: &[&str] = &[ "moq-lite-03", "moq-lite-05", - "moq-lite-07", + "moq-lite-07-wip", "moq-transport-14", "moq-transport-17", "moq-transport-20",