diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 81f7773f9b..5134660d5a 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -146,8 +146,10 @@ scope member. The Rust consumer is a `Stream` and the TypeScript one an async iterable. The Rust consumer also yields one `announce::Event::Live` once the routes live at subscribe time have all been delivered, including those a peer session was still sending: moq-lite-05+ counts them in `ANNOUNCE_OK`, -moq-lite-01/02 send them in `ANNOUNCE_INIT`, and older or IETF sessions wait -for the stream to go quiet. +moq-lite-01/02 send them in `ANNOUNCE_INIT`, IETF draft-16+ sessions count them +in `REQUEST_OK` when both sides speak +[active-count](/draft/moq-active-count), and anything else waits for the +stream to go quiet. Announcements are hints; requests are the authority. When a subscriber asks for a covered path the advertiser will not serve, the advertiser refuses that diff --git a/doc/concept/standard.md b/doc/concept/standard.md index f2c7e9929e..7305af1566 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -44,7 +44,9 @@ tracks waiting for a group that never arrives. Several project drafts extend the IETF wire without breaking it, since `SETUP` ignores unknown parameters: [cluster](/draft/moq-cluster) routing hop lists, [solicit](/draft/moq-solicit) to make announcements opt-in, -[hidden](/draft/moq-hidden) to keep `.`-named namespaces out of discovery, and +[hidden](/draft/moq-hidden) to keep `.`-named namespaces out of discovery, +[active-count](/draft/moq-active-count) to count the `NAMESPACE` messages +before a `SUBSCRIBE_NAMESPACE` is caught up, and [probe](/draft/moq-probe) for bandwidth estimation. [moq-e2ee](/draft/moq-e2ee) is not a transport extension: it encrypts application payloads so relays still forward named tracks they cannot read. diff --git a/drafts/draft-lcurley-moq-active-count.md b/drafts/draft-lcurley-moq-active-count.md new file mode 100644 index 0000000000..3be4d75db7 --- /dev/null +++ b/drafts/draft-lcurley-moq-active-count.md @@ -0,0 +1,122 @@ +--- +title: "MoQ Active Count Extension" +abbrev: "moq-active-count" +category: info + +docname: draft-lcurley-moq-active-count-latest +submissiontype: IETF # also: "independent", "editorial", "IAB", or "IRTF" +number: +date: +v: 3 +area: wit +workgroup: moq + +author: + - + fullname: Luke Curley + email: kixelated@gmail.com + +normative: + moqt: I-D.ietf-moq-transport + +informative: + +--- abstract + +This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many NAMESPACE messages answering its SUBSCRIBE_NAMESPACE come before the first change. +The subscriber then knows when it has caught up, instead of guessing from a quiet stream. + +--- note_Note_to_Readers + +This document was generated by an AI model from the implementation at [github.com/moq-dev/moq](https://github.com/moq-dev/moq) and is maintained alongside it. +Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github.com/moq-dev/moq/pulls) if this spec sucks and you want to fix anything. + +--- middle + +# Conventions and Definitions +{::boilerplate bcp14-tagged} + +A suffix is **active** on a SUBSCRIBE_NAMESPACE stream from a NAMESPACE message carrying it until a NAMESPACE_DONE message carrying it. +A subscriber is **caught up** on a SUBSCRIBE_NAMESPACE once it has read Count ({{count}}) NAMESPACE messages after REQUEST_OK, immediately on a Count of 0. + + +# Introduction +A publisher answers SUBSCRIBE_NAMESPACE with a NAMESPACE message for each suffix it already advertises under the subscription's prefix, then keeps the stream open to report changes. +Nothing on the wire separates the two, so a subscriber cannot tell "there are no broadcasts" from "they have not arrived yet". +The best it can do is wait for the stream to go quiet, which is slow when there is nothing to send and wrong when the network stalls partway. + +This extension has the publisher put on its REQUEST_OK how many NAMESPACE messages come before the first change. + + +# Setup Negotiation + +An endpoint declares this extension with the following Setup Option ({{moqt}} Section 10.3): + +~~~ +ACTIVE_COUNT Setup Option { + Option Key (vi64) = 0x40B64 + Option Value (vi64) = 1 +} +~~~ + +A receiver MUST ignore the value. +The extension is negotiated when both endpoints declare the option, and applies to every SUBSCRIBE_NAMESPACE on the session in both directions. +An endpoint MUST NOT declare the option on a version of {{moqt}} without the NAMESPACE message. + + +# Counting {#count} + +When the extension is negotiated, a publisher MUST include the following parameter in the REQUEST_OK answering a SUBSCRIBE_NAMESPACE: + +~~~ +ACTIVE_COUNT Parameter { + Type (vi64) = 0x40B66 + Value (vi64) = Count +} +~~~ + +The publisher MUST send exactly Count NAMESPACE messages immediately after REQUEST_OK, before any other message on the stream, making active each suffix it advertises to the subscriber when it answers, one message per suffix; later changes follow as usual. +A suffix the publisher does not advertise to this subscriber, such as one whose path loops back through it, is neither sent nor counted. + +A subscriber counts every NAMESPACE message toward Count, including any it discards on receipt. + +A subscriber MUST close the session with a PROTOCOL_VIOLATION if the REQUEST_OK answering a SUBSCRIBE_NAMESPACE omits the parameter when the extension is negotiated, or carries it when it is not. +An endpoint MUST close the session with a PROTOCOL_VIOLATION if a REQUEST_OK answering any other request carries the parameter. + + +# Security Considerations + +The count reveals no more than the NAMESPACE messages that follow it. +A publisher that promises more than it sends can only delay the subscriber's notion of having caught up, which it could equally do by withholding NAMESPACE messages. + + +# IANA Considerations + +This document requests the following registrations. +High, distinctive values are requested to avoid the low ranges reserved by {{moqt}} and to minimize collisions with provisional registrations by other extensions. + +## MOQT Setup Options + +This document requests one registration in the "MOQT Setup Options" registry ({{moqt}} Section 15.4), whose policy is Specification Required. + +| Value | Name | Reference | +|:--------|:----------------|:--------------| +| 0x40B64 | ACTIVE_COUNT | This Document | + +## MOQT Message Parameters + +This document requests one registration in the "MOQT Message Parameters" registry ({{moqt}} Section 15.7). + +| Value | Name | Carried In | Reference | +|:--------|:----------------|:-----------|:--------------| +| 0x40B66 | ACTIVE_COUNT | REQUEST_OK | This Document | + +Both values are even, so each is a bare varint. + + +--- back + +# Acknowledgments +{:numbered="false"} + +This document was drafted with the assistance of Claude, an AI assistant by Anthropic. diff --git a/quest/m1/README.md b/quest/m1/README.md index b682f06727..e01b985d3e 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -31,7 +31,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Close codes](/quest/m1/close-codes.md) - a client sees the peer's application close code over WebSocket and raw QUIC, like WebTransport - [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, moq-c, and every wrapper yield the same flat announce event, `Live` included -- [IETF announce count](/quest/m1/ietf-announce-count.md) - an opt-in moq-transport extension carries the replay count, so IETF announce consumers go live without a timer +- [JS active count](/quest/m1/js-active-count.md) - @moq/net speaks MoQ Active Count, so its IETF announce consumers go live without a timer - [Publish delay](/quest/m1/publish-delay.md) - js/publish encoders advertise `delay` behind the earliest rendition, like moq-mux - [Data jitter](/quest/m1/data-jitter.md) - JSON and binary tracks with a capture time advertise a detected `delay` and `jitter` diff --git a/quest/m1/ietf-announce-count.md b/quest/m1/ietf-announce-count.md deleted file mode 100644 index c077ad10bc..0000000000 --- a/quest/m1/ietf-announce-count.md +++ /dev/null @@ -1,20 +0,0 @@ -# [M] An IETF namespace subscription says how many routes it replays - -## Goal - -A moq-transport session that negotiates a new extension learns the size of -the initial set a SUBSCRIBE_NAMESPACE replays, the way `AnnounceOk.active` -tells a moq-lite session, so the announce `Live` marker fires on the count -instead of the settle timer. Without the extension the timer stays. - -## Plan - -- Specify the extension in a `drafts/` document (a new one, or an existing - lcurley extension draft if one fits), with its setup parameter and where - the count rides, and validate with `just drafts check`. -- Implement it in `rs/moq-net`'s IETF publisher and subscriber; mirror in - `js/net` if its IETF path carries announces. -- Test: a negotiated session clears on the count with N routes and with none; - an un-negotiated one still clears on the timer. - -Public API: none. Wire: a new opt-in extension. diff --git a/quest/m1/js-active-count.md b/quest/m1/js-active-count.md new file mode 100644 index 0000000000..9f9ae15ce7 --- /dev/null +++ b/quest/m1/js-active-count.md @@ -0,0 +1,27 @@ +# [S] @moq/net counts an IETF namespace subscription until it is caught up + +## Goal + +`@moq/net` speaks the MoQ Active Count extension +(`drafts/draft-lcurley-moq-active-count.md`) on draft-16+, as `rs/moq-net` +does: it declares the ACTIVE_COUNT option, its publisher puts on the REQUEST_OK +answering SUBSCRIBE_NAMESPACE how many NAMESPACE messages come before the +subscription is caught up, and its subscriber lands that stream's source on +the count instead of the quiet timer. + +## Plan + +- Mirror `rs/moq-net/src/ietf/active_count.rs` in `js/net/src/ietf/`, like + `hidden.ts`: declare on draft-16+ only, and read the peer's declaration. +- Publisher: count what it advertises before writing REQUEST_OK, then send + exactly those NAMESPACE messages first. +- Subscriber: a missing count when negotiated, one when not, or one on any + other REQUEST_OK is a protocol violation, as in Rust. +- Extend the IETF cases of the JS caught-up tests to show the counted versions + land without waiting out the quiet gap. + +Public API: none. Wire: the extension, already specified. + +## Required + +- [JS caught up](/quest/m1/js-announce-caught-up.md) - the per-source guard the count lands diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index e7292ed723..c1c69edcdd 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -343,6 +343,7 @@ impl Client { } ietf::solicit::into_setup(&mut parameters, ietf_encoding); ietf::hidden::into_setup(&mut parameters, ietf_encoding); + ietf::active_count::into_setup(&mut parameters, ietf_encoding); let parameters = parameters.encode_bytes(ietf_encoding)?; let client = setup::Client { @@ -393,6 +394,7 @@ impl Client { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶meters, v)?, hidden: ietf::hidden::from_setup(¶meters, v), + active_count: ietf::active_count::from_setup(¶meters, v), ..Default::default() }; diff --git a/rs/moq-net/src/ietf/active_count.rs b/rs/moq-net/src/ietf/active_count.rs new file mode 100644 index 0000000000..37138f4fb6 --- /dev/null +++ b/rs/moq-net/src/ietf/active_count.rs @@ -0,0 +1,52 @@ +//! The MoQ Active Count extension (draft-lcurley-moq-active-count-00). +//! +//! A publisher answers SUBSCRIBE_NAMESPACE with a NAMESPACE for every suffix it already +//! advertises, then reports changes on the same stream. Nothing in moq-transport says +//! where the first part ends, so a subscriber can only guess from the stream going +//! quiet. When both endpoints declare the ACTIVE_COUNT option, the publisher puts on its +//! REQUEST_OK how many NAMESPACE messages come before the subscription is caught up. + +use super::Version; + +/// ACTIVE_COUNT Setup Option: the sender counts the NAMESPACE messages before a +/// subscription is caught up, and reads the peer's count. Even, so the value is a bare varint. +pub const ACTIVE_COUNT: u64 = 0x40B64; + +/// ACTIVE_COUNT REQUEST_OK parameter: how many NAMESPACE messages immediately follow +/// before the subscription is caught up. Only answers SUBSCRIBE_NAMESPACE. +/// Even, so the value is a bare varint. +pub const ACTIVE_COUNT_PARAM: u64 = 0x40B66; + +/// Whether this version answers SUBSCRIBE_NAMESPACE with NAMESPACE messages. Draft-14/15 +/// answer with PUBLISH_NAMESPACE requests instead, which the stream cannot count. +fn supported(version: Version) -> bool { + !matches!(version, Version::Draft14 | Version::Draft15) +} + +/// Whether the peer declared the option. We always declare it, so this alone decides +/// whether the extension is negotiated. +pub fn from_setup(params: &super::Parameters, version: Version) -> bool { + supported(version) && params.get_varint(super::ParameterVarInt::ActiveCount).is_some() +} + +/// Declare that we count and read the peer's count. +pub fn into_setup(params: &mut super::Parameters, version: Version) { + if supported(version) { + params.set_varint(super::ParameterVarInt::ActiveCount, 1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_versions_with_namespace_negotiate() { + for version in [Version::Draft14, Version::Draft15, Version::Draft16, Version::Draft22] { + let mut params = super::super::Parameters::default(); + assert!(!from_setup(¶ms, version)); + into_setup(&mut params, version); + assert_eq!(from_setup(¶ms, version), supported(version), "{version:?}"); + } + } +} diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 8ae7dac1ec..6e77a8bc17 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -6,6 +6,7 @@ #[macro_use] mod parameters; +pub mod active_count; mod adapter; pub mod cluster; mod control; diff --git a/rs/moq-net/src/ietf/parameters.rs b/rs/moq-net/src/ietf/parameters.rs index 7b3b426f88..5fb8ad044a 100644 --- a/rs/moq-net/src/ietf/parameters.rs +++ b/rs/moq-net/src/ietf/parameters.rs @@ -28,6 +28,8 @@ pub enum ParameterVarInt { Solicit = super::solicit::SOLICIT, /// HIDDEN, from the MoQ Hidden extension. Hidden = super::hidden::HIDDEN, + /// ACTIVE_COUNT, from the MoQ Active Count extension. + ActiveCount = super::active_count::ACTIVE_COUNT, #[num_enum(catch_all)] Unknown(u64), } diff --git a/rs/moq-net/src/ietf/peer.rs b/rs/moq-net/src/ietf/peer.rs index 20f793ed12..ae4d164915 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -19,6 +19,10 @@ pub(crate) struct Peer { /// MoQ Hidden: whether the peer understands the HIDDEN parameter on /// SUBSCRIBE_NAMESPACE, so we may send it. pub hidden: bool, + + /// MoQ Active Count: whether the REQUEST_OK answering SUBSCRIBE_NAMESPACE counts the + /// NAMESPACE messages before the subscription is caught up, in both directions. + pub active_count: bool, } /// Shared slot for [`Peer`], filled when the peer's SETUP is read. @@ -78,6 +82,7 @@ mod tests { }, solicit: None, hidden: false, + active_count: false, }; let slot = PeerSetup::default(); @@ -89,6 +94,7 @@ mod tests { }, solicit: Some(true), hidden: true, + active_count: true, }); assert_eq!(slot.get().await, first); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 8fa8c7ad11..e7dd065aaa 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1496,6 +1496,16 @@ where } (_, ietf::RequestOk::ID) => { let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; + // ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count). Closed + // here because the solicited path runs this inside a request task, whose + // errors end only that stream. + if msg.active.is_some() { + self.session.clone().close( + crate::SessionError::ProtocolViolation.to_code(), + "ACTIVE_COUNT on a PUBLISH_NAMESPACE answer", + ); + return Err(Error::ProtocolViolation); + } tracing::debug!(message = ?msg, "publish namespace ok"); } (_, ietf::RequestError::ID) => { @@ -1574,6 +1584,15 @@ where match type_id { ietf::RequestOk::ID => { let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; + // ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count); closed + // here for the same reason as in `advertise_namespace`. + if msg.active.is_some() { + self.session.clone().close( + crate::SessionError::ProtocolViolation.to_code(), + "ACTIVE_COUNT on a REQUEST_UPDATE answer", + ); + return Err(Error::ProtocolViolation); + } tracing::debug!(message = ?msg, "publish_namespace update ok"); Ok(Refused::No) } @@ -1736,7 +1755,8 @@ where let origin = self.excluding(&peer); let ns = Namespaces::new(peer, Target::Requests(None)); - self.run_namespaces(origin, crate::Path::empty().to_owned(), ns).await + self.run_namespaces(origin.announced(), crate::Path::empty().to_owned(), ns, Vec::new()) + .await } /// Handle a SUBSCRIBE_NAMESPACE on its bidi stream. @@ -1764,6 +1784,50 @@ where .unwrap_or_default(); let origin = self.origin.scope("", &scope).unwrap_or_else(|_| self.origin.empty()); + // The extension changes what an advertisement carries, so nothing can be + // sent until the peer's SETUP says whether it speaks it. The same SETUP says + // whether the OK counts what it sends first (MoQ Active Count). + let peer = self.peer().await; + let declared = self.peer_setup.get().await; + // Register the split-horizon peer on the announce cursor too. The origin + // model uses this exposure to park a reflected copy before it can replace + // the source we are currently advertising to that peer. + let origin = origin.excluding(self.exclude(&peer)); + + // Hidden namespaces are left out unless the peer opted in (MoQ Hidden). A publish + // origin that already opted in (the caller's choice for this peer) keeps them. + let origin = match msg.hidden { + true => origin.with_hidden(true), + false => origin, + }; + + // Unless the peer asked to be told only on request, it has already heard what an + // unsolicited PUBLISH_NAMESPACE can say. Repeating it here would leave it holding + // two sources for one namespace, so this stream carries only what that loop hid + // from the empty prefix and this request may see, and otherwise simply stays open + // until the peer is done with it. + let origin = match declared.solicit.unwrap_or(false) { + true => origin, + false if self.origin.includes_hidden() => origin.empty(), + false => origin.beyond(&self.origin), + }; + + let mut announced = origin.announced(); + + // MoQ Active Count: take what is advertised now, so the OK can say how many + // NAMESPACE messages carry it. They go out first, ahead of any change. + let (initial, active) = match declared.active_count { + true => { + let initial = Self::snapshot(&mut announced); + let count = initial + .iter() + .filter(|update| self.select(&update.route, &peer).wanted()) + .count(); + (initial, Some(count as u64)) + } + false => (Vec::new(), None), + }; + // Send OK response match self.version { Version::Draft14 => { @@ -1781,23 +1845,22 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(msg.request_id), + active, }) .await?; } _ => { stream.writer.encode(&ietf::RequestOk::ID).await?; - stream.writer.encode(&ietf::RequestOk { request_id: None }).await?; + stream + .writer + .encode(&ietf::RequestOk { + request_id: None, + active, + }) + .await?; } } - // The extension changes what an advertisement carries, so nothing can be - // sent until the peer's SETUP says whether it speaks it. - let peer = self.peer().await; - // Register the split-horizon peer on the announce cursor too. The origin - // model uses this exposure to park a reflected copy before it can replace - // the source we are currently advertising to that peer. - let origin = origin.excluding(self.exclude(&peer)); - // Draft-14/15 predate NAMESPACE, so they answer with their own PUBLISH_NAMESPACE // requests and keep this stream open for the subscription's lifetime. let target = match self.version { @@ -1805,26 +1868,29 @@ where _ => Target::Inline(stream), }; - // Hidden namespaces are left out unless the peer opted in (MoQ Hidden). A publish - // origin that already opted in (the caller's choice for this peer) keeps them. - let origin = match msg.hidden { - true => origin.with_hidden(true), - false => origin, - }; - - // Unless the peer asked to be told only on request, it has already heard what an - // unsolicited PUBLISH_NAMESPACE can say. Repeating it here would leave it holding - // two sources for one namespace, so this stream carries only what that loop hid - // from the empty prefix and this request may see, and otherwise simply stays open - // until the peer is done with it. - let origin = match self.requires_solicitation().await { - true => origin, - false if self.origin.includes_hidden() => origin.empty(), - false => origin.beyond(&self.origin), - }; - let ns = Namespaces::new(peer, target); - self.run_namespaces(origin, prefix, ns).await + self.run_namespaces(announced, prefix, ns, initial).await + } + + /// The routes an announce cursor holds right now, without waiting for more. + /// + /// A route announced and retracted within the snapshot is left out, and a repeat + /// keeps its latest metadata, so each path appears once. The origin's live marker is + /// skipped: it says when the origin caught up, not what the peer should hear. + fn snapshot(announced: &mut crate::announce::Consumer) -> Vec { + let mut initial = std::collections::BTreeMap::new(); + while let Some(event) = announced.try_next() { + match event { + crate::announce::Event::Announced(update) | crate::announce::Event::Updated(update) => { + initial.insert(update.prefix.clone(), update); + } + crate::announce::Event::Retracted(update) => { + initial.remove(&update.prefix); + } + crate::announce::Event::Live => {} + } + } + initial.into_values().collect() } /// Forward origin (un)announces to the peer until the loop ends. @@ -1834,11 +1900,14 @@ where /// for a subset). async fn run_namespaces( &self, - origin: origin::Consumer, + mut announced: crate::announce::Consumer, prefix: crate::PathOwned, mut ns: Namespaces, + initial: Vec, ) -> Result<(), Error> { - let mut announced = origin.announced(); + for update in initial { + self.apply_update(&mut ns, &prefix, update, true).await?; + } // When to re-offer whatever the peer should hold and doesn't, and how long to wait // the next time that fails. Jittered so a relay's namespaces don't all come back on @@ -1923,31 +1992,7 @@ where return stream.writer.closed().await; } NamespaceEvent::Update(Some((update, active))) => { - let path = update.prefix; - let suffix = path - .strip_prefix(&prefix) - .expect("origin returned invalid prefix") - .to_owned(); - - if active { - // A repeat for a live suffix is a metadata update: keep the - // peer's refusal state and re-run the selection. - match ns.watched.get_mut(&suffix) { - Some(watch) => watch.route = update.route, - None => { - ns.watched.insert(suffix.clone(), Watched::new(update.route)); - } - } - self.sync_namespace(&mut ns, &suffix, &path).await?; - } else { - // Only close out namespaces the peer actually saw. - let held = ns.watched.remove(&suffix).is_some_and(|watch| watch.sent.wanted()); - if held { - tracing::debug!(route = %self.origin.absolute(&path), "namespace_done"); - self.withdraw_namespace(&mut ns.target, &mut ns.requests, suffix) - .await?; - } - } + self.apply_update(&mut ns, &prefix, update, active).await?; } } }; @@ -1957,6 +2002,42 @@ where res } + + /// Reconcile what the peer holds with one route (un)announce. + async fn apply_update( + &self, + ns: &mut Namespaces, + prefix: &crate::PathOwned, + update: crate::announce::Announce, + active: bool, + ) -> Result<(), Error> { + let path = update.prefix; + let suffix = path + .strip_prefix(prefix) + .expect("origin returned invalid prefix") + .to_owned(); + + if active { + // A repeat for a live suffix is a metadata update: keep the + // peer's refusal state and re-run the selection. + match ns.watched.get_mut(&suffix) { + Some(watch) => watch.route = update.route, + None => { + ns.watched.insert(suffix.clone(), Watched::new(update.route)); + } + } + self.sync_namespace(ns, &suffix, &path).await + } else { + // Only close out namespaces the peer actually saw. + let held = ns.watched.remove(&suffix).is_some_and(|watch| watch.sent.wanted()); + if held { + tracing::debug!(route = %self.origin.absolute(&path), "namespace_done"); + self.withdraw_namespace(&mut ns.target, &mut ns.requests, suffix) + .await?; + } + Ok(()) + } + } } /// Serves a track's groups, one machine per group, with unlimited concurrency. @@ -3831,6 +3912,80 @@ mod tests { announced.assert_next_wait(); } + /// MoQ Active Count: the OK counts exactly the NAMESPACE messages that follow it, + /// so a route this peer is never told about is not counted either. Counting one would + /// leave the peer waiting on a message that never comes. + #[tokio::test] + async fn the_ok_counts_only_what_is_advertised() { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + + let session = SinkSession::gated_bi(kio::Producer::new(true).consume()); + let log = session.log.clone(); + let setup = peer::PeerSetup::default(); + setup.set(peer::Peer { + cluster: cluster::Peer { + hop: Some(crate::Hop::new(9).unwrap()), + cost: None, + }, + solicit: Some(true), + active_count: true, + ..Default::default() + }); + let publisher = Publisher::new( + crate::time::Clock::tokio(), + session.clone(), + origin.consume(), + Control::new(None, false), + None, + setup, + Version::Draft17, + ); + + let mut clean = crate::Hops::new(); + clean.push(crate::Hop::new(778).unwrap()).unwrap(); + let _clean = origin + .announce("cam-clean", crate::origin::Route::default().with_hops(clean)) + .unwrap(); + // A chain with no room left for our own hop cannot be forwarded. + let mut full = crate::Hops::new(); + while full.push(crate::Hop::new(100 + full.len() as u64).unwrap()).is_ok() {} + let _full = origin + .announce("cam-full", crate::origin::Route::default().with_hops(full)) + .unwrap(); + settle().await; + + let stream = Stream::open(&mut session.clone(), Version::Draft17).await.unwrap(); + let msg = ietf::SubscribeNamespace { + request_id: RequestId(1), + namespace: crate::Path::new(""), + hidden: false, + }; + let mut run = std::pin::pin!(publisher.run_subscribe_namespace_stream(stream, msg)); + assert!(futures::poll!(run.as_mut()).is_pending()); + + assert_eq!(occurrences(&log, b"cam-clean"), 1); + assert_eq!(occurrences(&log, b"cam-full"), 0); + + let expected = crate::lite::test_transport::Log::default(); + let mut writer = crate::coding::Writer::new( + crate::lite::test_transport::SinkSend::new(expected.clone()), + Version::Draft17, + ); + writer.encode(&ietf::RequestOk::ID).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: Some(1), + }) + .await + .unwrap(); + let expected = expected.writes.lock().unwrap().clone(); + assert!( + log.writes.lock().unwrap().starts_with(&expected), + "the OK counts one NAMESPACE" + ); + } + /// A same-path source can splice into (or detach from) an existing broadcast /// without an origin-level (un)announce, silently flipping `advertisable`. /// Namespace forwarding must follow: advertise when a clean route appears, @@ -4001,6 +4156,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: Some(RequestId(1)), + active: None, }) .await .unwrap(); @@ -4008,7 +4164,13 @@ mod tests { // Draft-17+ dropped the request id: the response rides the request's stream. _ => { writer.encode(&ietf::RequestOk::ID).await.unwrap(); - writer.encode(&ietf::RequestOk { request_id: None }).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: None, + }) + .await + .unwrap(); } } @@ -4150,6 +4312,54 @@ mod tests { assert_eq!(log.bi_opens(), 1, "one request stream"); } + /// ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count), so one on the OK + /// to a PUBLISH_NAMESPACE is the peer breaking the extension, negotiated or not. + #[tokio::test] + async fn a_counted_publish_namespace_ok_is_a_violation() { + const VERSION: Version = Version::Draft17; + + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _local = origin.announce("local-cam", crate::origin::Route::default()).unwrap(); + settle().await; + + let ok = crate::lite::test_transport::Log::default(); + let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(ok.clone()), VERSION); + writer.encode(&ietf::RequestOk::ID).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: Some(0), + }) + .await + .unwrap(); + let ok = ok.writes.lock().unwrap().clone(); + let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![ok]); + let log = session.log.clone(); + + let peer_setup = peer::PeerSetup::default(); + peer_setup.set(peer::Peer { + active_count: true, + ..Default::default() + }); + let publisher = Publisher::new( + crate::time::Clock::tokio(), + session, + origin.consume(), + Control::new(None, false), + None, + peer_setup, + VERSION, + ); + + let res = tokio::time::timeout(Duration::from_secs(5), publisher.run_publish_namespaces()) + .await + .expect("the violation ends the loop"); + assert!(matches!(res, Err(Error::ProtocolViolation)), "{res:?}"); + // Closed at the source, since the solicited path's request task only logs errors. + let code = crate::SessionError::ProtocolViolation.to_code(); + assert!(log.closes().iter().any(|(c, _)| *c == code), "{:?}", log.closes()); + } + /// Drive both announce loops at once against a peer that declared `solicit`, /// returning how many times the namespace hit the wire and how many bidi streams /// were opened. One stream means the entry rode the subscription inline; two means @@ -4446,6 +4656,7 @@ mod tests { }, solicit, hidden: false, + active_count: false, }); slot } @@ -4527,6 +4738,69 @@ mod tests { assert_eq!(log.bi_opens(), 1, "the update rode the request's own stream"); } + /// ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count), so one on the OK + /// to a REQUEST_UPDATE is the peer breaking the extension too. + #[tokio::test] + async fn a_counted_update_ok_is_a_violation() { + const VERSION: Version = Version::Draft19; + + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let _cold = origin + .announce("cam", crate::origin::Route::default().with_cost(4)) + .unwrap(); + settle().await; + + // The PUBLISH_NAMESPACE is answered cleanly; its update's OK carries a count. + let counted = crate::lite::test_transport::Log::default(); + let mut writer = + crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(counted.clone()), VERSION); + writer.encode(&ietf::RequestOk::ID).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: Some(0), + }) + .await + .unwrap(); + let counted = counted.writes.lock().unwrap().clone(); + let ok = publish_namespace_ok(VERSION).await; + let session = crate::lite::test_transport::ScriptedSession::per_stream(vec![[ok, counted].concat()]); + let log = session.log.clone(); + + let publisher = Publisher::new( + crate::time::Clock::tokio(), + session, + origin.consume(), + Control::new(None, false), + None, + clustered(Some(false)), + VERSION, + ); + + let mut run = std::pin::pin!(publisher.run_publish_namespaces()); + for _ in 0..100 { + assert!(futures::poll!(run.as_mut()).is_pending()); + if occurrences(&log, b"cam") >= 1 { + break; + } + settle().await; + } + assert_eq!(occurrences(&log, b"cam"), 1, "the advertisement never went out"); + + // Repricing sends the REQUEST_UPDATE whose OK is counted. + let _warm = origin + .announce("cam", crate::origin::Route::default().with_cost(0)) + .unwrap(); + + let res = tokio::time::timeout(Duration::from_secs(5), run) + .await + .expect("the violation ends the loop"); + assert!(matches!(res, Err(Error::ProtocolViolation)), "{res:?}"); + // Closed at the source, since the solicited path's request task only logs errors. + let code = crate::SessionError::ProtocolViolation.to_code(); + assert!(log.closes().iter().any(|(c, _)| *c == code), "{:?}", log.closes()); + } + /// A route from a different original publisher is not an update: its content is not /// continuous with what the peer holds, so the draft has the advertisement withdrawn /// and made again rather than repriced in place. diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index 1b783556d8..dc0cfb2889 100644 --- a/rs/moq-net/src/ietf/request.rs +++ b/rs/moq-net/src/ietf/request.rs @@ -3,6 +3,7 @@ use std::borrow::Cow; use crate::coding::{Decode, DecodeError, Encode, EncodeError}; use super::Message; +use super::active_count::ACTIVE_COUNT_PARAM; use super::Version; @@ -85,6 +86,9 @@ impl Message for RequestsBlocked { #[derive(Clone, Debug)] pub struct RequestOk { pub request_id: Option, + /// MoQ Active Count: how many NAMESPACE messages follow before the subscription is + /// caught up. Only on the answer to a SUBSCRIBE_NAMESPACE, when negotiated. + pub active: Option, } impl Message for RequestOk { @@ -98,7 +102,7 @@ impl Message for RequestOk { } else { assert!(self.request_id.is_none(), "request_id must be None for draft17+"); } - encode_params!(w, version,); + encode_params!(w, version, ACTIVE_COUNT_PARAM => self.active); Ok(()) } @@ -111,8 +115,9 @@ impl Message for RequestOk { // A REQUEST_UPDATE_OK may refresh EXPIRES, which is ignored like SUBSCRIBE_OK's. decode_params!(r, version, 0x08 => _expires: Option, + ACTIVE_COUNT_PARAM => active: Option, ); - Ok(Self { request_id }) + Ok(Self { request_id, active }) } } @@ -188,6 +193,7 @@ mod tests { fn test_request_ok_round_trip() { let msg = RequestOk { request_id: Some(RequestId(42)), + active: None, }; let encoded = encode_message(&msg, Version::Draft15); @@ -205,6 +211,20 @@ mod tests { assert_eq!(decoded.request_id, Some(RequestId(7))); } + #[test] + fn test_request_ok_active_round_trips() { + for (version, request_id) in [(Version::Draft16, Some(RequestId(3))), (Version::Draft22, None)] { + for count in [None, Some(0), Some(7)] { + let msg = RequestOk { + request_id, + active: count, + }; + let decoded: RequestOk = decode_message(&encode_message(&msg, version), version).unwrap(); + assert_eq!(decoded.active, count, "{version:?}"); + } + } + } + #[test] fn test_request_error_round_trip() { let msg = RequestError { @@ -243,7 +263,10 @@ mod tests { #[test] fn test_request_ok_v17_round_trip() { - let msg = RequestOk { request_id: None }; + let msg = RequestOk { + request_id: None, + active: None, + }; let encoded = encode_message(&msg, Version::Draft17); let decoded: RequestOk = decode_message(&encoded, Version::Draft17).unwrap(); @@ -271,7 +294,10 @@ mod tests { #[test] fn test_request_ok_v18_round_trip() { - let msg = RequestOk { request_id: None }; + let msg = RequestOk { + request_id: None, + active: None, + }; let encoded = encode_message(&msg, Version::Draft18); let decoded: RequestOk = decode_message(&encoded, Version::Draft18).unwrap(); @@ -283,7 +309,10 @@ mod tests { /// treated as Draft14-16 and panic in the encoder. #[test] fn test_request_ok_v18_wire_matches_v17() { - let msg = RequestOk { request_id: None }; + let msg = RequestOk { + request_id: None, + active: None, + }; let v17 = encode_message(&msg, Version::Draft17); let v18 = encode_message(&msg, Version::Draft18); assert_eq!(v17, v18); diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 090355fc68..849d2f5b46 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -8,7 +8,7 @@ use crate::{ }; use super::{ - Control, Message, Publisher, Subscriber, Version, + Control, Message, Publisher, Subscriber, Version, active_count, adapter::ControlStreamAdapter, cluster, hidden, peer, solicit, subscriber::{is_protocol_violation, subscribe_prefixes}, @@ -512,6 +512,7 @@ fn peer_from_params(params: &ietf::Parameters, version: Version) -> Result( cluster::peer_into_setup(&mut parameters, self_origin, cost, version); solicit::into_setup(&mut parameters, version); hidden::into_setup(&mut parameters, version); + active_count::into_setup(&mut parameters, version); let parameters = parameters.encode_bytes(version)?; writer.encode(&setup::Setup { parameters }).await?; @@ -900,7 +902,13 @@ mod tests { let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); writer.encode(&ietf::RequestOk::ID).await.unwrap(); - writer.encode(&ietf::RequestOk { request_id: None }).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: None, + }) + .await + .unwrap(); writer.encode(&ietf::Namespace::ID).await.unwrap(); writer .encode(&ietf::Namespace { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1ec9428b85..876e25770b 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -467,6 +467,25 @@ pub(super) fn subscribe_prefixes(origin: &origin::Producer) -> Vec<(PathOwned, c .collect() } +/// How a SUBSCRIBE_NAMESPACE stream's initial set ends. +enum Landing { + /// MoQ Active Count: after this many more NAMESPACE messages. + Count(u64), + /// Once the stream goes quiet, which is all the base protocol offers. + Quiet(crate::model::Quiet), +} + +impl Landing { + /// Ready once the initial set has landed. + fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { + match self { + Self::Count(0) => Poll::Ready(()), + Self::Count(_) => Poll::Pending, + Self::Quiet(quiet) => quiet.poll(waiter), + } + } +} + /// Resolve the subscription a data stream belongs to. /// /// SUBSCRIBE_OK can be reordered behind the stream it describes, so an alias we have not @@ -711,7 +730,8 @@ where // peer into the origin and each local reader opts in on its own. The parameter // fails decoding at a peer that doesn't know it, so it waits on the peer's SETUP // to say whether it does (MoQ Hidden). - let hidden = self.peer_setup.get().await.hidden; + let declared = self.peer_setup.get().await; + let hidden = declared.hidden; let request_id = self.control.next_request_id(&self.runtime).await?; @@ -746,13 +766,12 @@ where let size: u16 = stream.reader.decode().await?; let mut data = stream.reader.read_exact(size as usize).await?; - match type_id { + let count = match type_id { ietf::SubscribeNamespaceOk::ID if self.version == Version::Draft14 => { - let _msg = ietf::SubscribeNamespaceOk::decode_msg(&mut data, self.version)?; - } - ietf::RequestOk::ID => { - let _msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; + ietf::SubscribeNamespaceOk::decode_msg(&mut data, self.version)?; + None } + ietf::RequestOk::ID => ietf::RequestOk::decode_msg(&mut data, self.version)?.active, ietf::SubscribeNamespaceError::ID if self.version == Version::Draft14 => { let msg = ietf::SubscribeNamespaceError::decode_msg(&mut data, self.version)?; let err = request::from_code(msg.error_code, request::Kind::SubscribeNamespace, self.version); @@ -766,13 +785,20 @@ where return Err(err); } _ => return Err(Error::UnexpectedMessage), - } + }; - tracing::debug!(%prefix, "subscribe_namespace ok"); + tracing::debug!(%prefix, ?count, "subscribe_namespace ok"); - // Nothing on this wire says where the initial set ends, so it has landed once - // the stream goes quiet. - let mut landing = Some((replaying, crate::model::Quiet::new(&self.runtime))); + // MoQ Active Count says how many NAMESPACE messages come before we are caught up. + // Without it nothing on this wire marks the end, so the set has landed once the + // stream goes quiet. A count the negotiation did not promise, or a missing one it + // did, is the peer breaking the extension. + let landing = match (declared.active_count, count) { + (true, Some(count)) => Landing::Count(count), + (false, None) => Landing::Quiet(crate::model::Quiet::new(&self.runtime)), + _ => return Err(Error::ProtocolViolation), + }; + let mut landing = Some((replaying, landing)); // The extension changes the NAMESPACE encoding, so we can't parse one until // the peer's SETUP says whether it negotiated. @@ -808,15 +834,17 @@ where prefix: &PathOwned, peer: &cluster::Peer, live: &mut std::collections::HashSet, - // The replay guard and its quiet timer, until the initial set has landed. - landing: &mut Option<(crate::model::Replaying, crate::model::Quiet)>, + // The replay guard and how it lands, until the initial set has. + landing: &mut Option<(crate::model::Replaying, Landing)>, ) -> Result<(), Error> { loop { let next = { let mut decode = std::pin::pin!(stream.reader.decode_maybe::()); kio::wait(|waiter| { - if let Some((_, quiet)) = landing - && quiet.poll(waiter).is_ready() + // Land before decoding past the boundary, so no live update enters the + // origin ahead of the marker. + if let Some((_, how)) = landing + && how.poll(waiter).is_ready() { *landing = None; } @@ -828,7 +856,7 @@ where Some(id) => id, None => break, // Stream closed }; - if let Some((_, quiet)) = landing { + if let Some((_, Landing::Quiet(quiet))) = landing { quiet.heard(); } let size: u16 = stream.reader.decode().await?; @@ -843,6 +871,12 @@ where if !data.is_empty() { return Err(Error::WrongSize); } + // Counted whether or not we keep it: the publisher cannot know we will + // drop a reflection. The guard only drops at the top of the loop, after + // this one is in the origin. + if let Some((_, Landing::Count(remaining))) = landing { + *remaining = remaining.saturating_sub(1); + } let path = prefix.join(&msg.suffix); let Some(advert) = self.route(msg.cluster.as_ref(), peer) else { // Looped back through us: forwarding it would extend the loop and @@ -1234,12 +1268,19 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(request_id), + active: None, }) .await?; } _ => { stream.writer.encode(&ietf::RequestOk::ID).await?; - stream.writer.encode(&ietf::RequestOk { request_id: None }).await?; + stream + .writer + .encode(&ietf::RequestOk { + request_id: None, + active: None, + }) + .await?; } } Ok(()) @@ -3285,7 +3326,13 @@ mod tests { let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), version); writer.encode(&ietf::RequestOk::ID).await.unwrap(); - writer.encode(&ietf::RequestOk { request_id: None }).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: None, + }) + .await + .unwrap(); writer.encode(&ietf::Namespace::ID).await.unwrap(); writer .encode(&ietf::Namespace { @@ -3359,6 +3406,60 @@ mod tests { ); } + /// MoQ Active Count is negotiated, so a REQUEST_OK that breaks the negotiation + /// either way is the peer's fault: a count we cannot rely on, or one missing where we + /// would otherwise wait on it forever. + #[tokio::test(start_paused = true)] + async fn a_count_the_negotiation_did_not_promise_is_a_violation() { + const VERSION: Version = Version::Draft18; + + for (negotiated, count) in [(true, None), (false, Some(0))] { + let log = crate::lite::test_transport::Log::default(); + let mut writer = + crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), VERSION); + writer.encode(&ietf::RequestOk::ID).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: count, + }) + .await + .unwrap(); + let response = log.writes.lock().unwrap().clone(); + + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let session = crate::lite::test_transport::ScriptedSession::new(response); + let (tasks, _task_set) = crate::util::TaskSet::new(); + let peer_setup = peer::PeerSetup::default(); + peer_setup.set(peer::Peer { + active_count: negotiated, + ..Default::default() + }); + let mut subscriber = Subscriber::new( + crate::time::Clock::tokio(), + session.clone(), + origin, + Control::new(None, false), + None, + peer_setup, + crate::Hop::new(1).unwrap(), + None, + VERSION, + tasks, + Default::default(), + ); + + let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); + let res = subscriber + .run_subscribe_namespace(stream, crate::Path::new("").to_owned(), subscriber.origin.replaying("")) + .await; + assert!( + matches!(res, Err(Error::ProtocolViolation)), + "negotiated {negotiated}, count {count:?}: {res:?}" + ); + } + } + #[test] fn retiring_old_track_does_not_retire_reused_alias() { let aliases = TrackAliases::default(); @@ -4470,7 +4571,13 @@ mod tests { let mut writer = crate::coding::Writer::new(crate::lite::test_transport::SinkSend::new(log.clone()), VERSION); writer.encode(&ietf::RequestOk::ID).await.unwrap(); - writer.encode(&ietf::RequestOk { request_id: None }).await.unwrap(); + writer + .encode(&ietf::RequestOk { + request_id: None, + active: None, + }) + .await + .unwrap(); for cost in [4, 0] { writer.encode(&ietf::Namespace::ID).await.unwrap(); writer diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index fd8ec6a325..846005b79a 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3073,7 +3073,8 @@ impl Drop for Replaying { } /// When a [`Replaying`] source has landed, for wires that never say where the -/// initial set ends (lite-03/04, moq-transport): once its stream goes quiet. +/// initial set ends (lite-03/04, moq-transport without MoQ Active Count): once +/// its stream goes quiet. /// /// A peer writes its whole set back to back, so the first announcement gets a /// round trip's grace and each one after it only has to beat its siblings. diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index e26f5e632d..53a9e487dd 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -336,6 +336,7 @@ impl Server { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶ms, v)?, hidden: ietf::hidden::from_setup(¶ms, v), + active_count: ietf::active_count::from_setup(¶ms, v), ..Default::default() }; (path, request_id_max, peer_declared) @@ -560,6 +561,7 @@ where parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); ietf::solicit::into_setup(&mut parameters, v); ietf::hidden::into_setup(&mut parameters, v); + ietf::active_count::into_setup(&mut parameters, v); parameters.encode_bytes(v)? } Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?, diff --git a/rs/moq-net/tests/caught_up.rs b/rs/moq-net/tests/caught_up.rs index cdadddbbc5..ded65f45c2 100644 --- a/rs/moq-net/tests/caught_up.rs +++ b/rs/moq-net/tests/caught_up.rs @@ -14,16 +14,27 @@ fn produce_origin(hop: Hop) -> origin::Producer { producer } -/// Versions whose wire says where the initial set ends: ANNOUNCE_INIT or -/// ANNOUNCE_OK's count. -const COUNTED: &[&str] = &["moq-lite-01", "moq-lite-02", "moq-lite-05", "moq-lite-06"]; +/// Versions whose wire says where the initial set ends: ANNOUNCE_INIT, ANNOUNCE_OK's +/// count, or REQUEST_OK's (MoQ Active Count, negotiated from draft-16). +const COUNTED: &[&str] = &[ + "moq-lite-01", + "moq-lite-02", + "moq-lite-05", + "moq-lite-06", + "moq-transport-16", + "moq-transport-19", +]; /// Versions that land the initial set once the stream goes quiet instead. -const QUIET: &[&str] = &["moq-lite-03", "moq-lite-04", "moq-transport-14", "moq-transport-19"]; +const QUIET: &[&str] = &["moq-lite-03", "moq-lite-04", "moq-transport-14", "moq-transport-15"]; + +/// The shortest a quiet stream can take to land: the gap after its last announcement. +const QUIET_GAP: Duration = Duration::from_millis(30); /// Connect a subscriber to a peer publishing `paths`, then read its cursor up to -/// the marker, returning the paths delivered before it. -async fn caught_up(version: &str, paths: &[&str]) -> Vec { +/// the marker, returning the paths delivered before it and how long the marker took +/// after the session was up. +async fn caught_up(version: &str, paths: &[&str]) -> (Vec, Duration) { let version: Version = version.parse().unwrap(); let published = produce_origin(Hop::new(1).unwrap()); let broadcasts: Vec<_> = paths @@ -40,6 +51,7 @@ async fn caught_up(version: &str, paths: &[&str]) -> Vec { options.server_publish = Some(published.consume()); options.client_subscribe = Some(subscribed.clone()); let _pair = connect_mock(options).await; + let start = tokio::time::Instant::now(); let mut announced = subscribed.consume().announced(); let mut live = Vec::new(); @@ -56,21 +68,37 @@ async fn caught_up(version: &str, paths: &[&str]) -> Vec { .await .unwrap_or_else(|_| panic!("{version}: never caught up")); + let elapsed = start.elapsed(); + drop(broadcasts); live.sort(); - live + (live, elapsed) } -#[tokio::test] +// Paused, so time only moves when every task waits on a timer: a counted set lands +// without one, and a quiet one has to wait out the gap. +#[tokio::test(start_paused = true)] async fn the_marker_follows_the_whole_initial_set() { for version in COUNTED.iter().chain(QUIET) { - assert_eq!(caught_up(version, &["a", "b", "c"]).await, ["a", "b", "c"], "{version}"); + let (live, elapsed) = caught_up(version, &["a", "b", "c"]).await; + assert_eq!(live, ["a", "b", "c"], "{version}"); + assert_eq!( + QUIET.contains(version), + elapsed >= QUIET_GAP, + "{version}: took {elapsed:?}" + ); } } -#[tokio::test] +#[tokio::test(start_paused = true)] async fn an_empty_peer_is_caught_up() { for version in COUNTED.iter().chain(QUIET) { - assert!(caught_up(version, &[]).await.is_empty(), "{version}"); + let (live, elapsed) = caught_up(version, &[]).await; + assert!(live.is_empty(), "{version}"); + assert_eq!( + QUIET.contains(version), + elapsed >= QUIET_GAP, + "{version}: took {elapsed:?}" + ); } }