From e7b34430ec2c6617146cb30bc11950a768e8339c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 10:03:25 -0700 Subject: [PATCH 1/6] quest: claim ietf-announce-count From 13e15500882015a1ee7e4d94b6496e1bd4da0032 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 10:26:34 -0700 Subject: [PATCH 2/6] feat(net): count an IETF namespace subscription's initial set Adds the MoQ Namespace Count extension (draft-lcurley-moq-namespace-count): when both sides declare the NAMESPACE_COUNT Setup Option on draft-16+, the REQUEST_OK answering SUBSCRIBE_NAMESPACE says how many NAMESPACE messages replay the initial set, so the announce Live marker fires on the count instead of the quiet timer. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 6 +- doc/concept/standard.md | 4 +- drafts/draft-lcurley-moq-namespace-count.md | 123 ++++++++++ quest/m1/README.md | 2 +- quest/m1/ietf-announce-count.md | 20 -- quest/m1/js-ietf-announce-count.md | 26 ++ quest/m1/js-origin-scope.md | 2 +- rs/moq-net/src/client.rs | 2 + rs/moq-net/src/ietf/mod.rs | 1 + rs/moq-net/src/ietf/namespace_count.rs | 51 ++++ rs/moq-net/src/ietf/parameters.rs | 2 + rs/moq-net/src/ietf/peer.rs | 6 + rs/moq-net/src/ietf/publisher.rs | 258 +++++++++++++++----- rs/moq-net/src/ietf/request.rs | 42 +++- rs/moq-net/src/ietf/session.rs | 12 +- rs/moq-net/src/ietf/subscriber.rs | 145 +++++++++-- rs/moq-net/src/model/origin.rs | 3 +- rs/moq-net/src/server.rs | 2 + rs/moq-net/tests/caught_up.rs | 50 +++- 19 files changed, 637 insertions(+), 120 deletions(-) create mode 100644 drafts/draft-lcurley-moq-namespace-count.md delete mode 100644 quest/m1/ietf-announce-count.md create mode 100644 quest/m1/js-ietf-announce-count.md create mode 100644 rs/moq-net/src/ietf/namespace_count.rs diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 7de0e24bc8..fe71dd81bc 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -143,8 +143,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 +[namespace-count](/draft/moq-namespace-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..0d6a4220f5 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, +[namespace-count](/draft/moq-namespace-count) to say where a +`SUBSCRIBE_NAMESPACE`'s initial set ends, 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-namespace-count.md b/drafts/draft-lcurley-moq-namespace-count.md new file mode 100644 index 0000000000..413c705d79 --- /dev/null +++ b/drafts/draft-lcurley-moq-namespace-count.md @@ -0,0 +1,123 @@ +--- +title: "MoQ Namespace Count Extension" +abbrev: "moq-namespace-count" +category: info + +docname: draft-lcurley-moq-namespace-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 where the initial set of namespaces answering its SUBSCRIBE_NAMESPACE ends. +The publisher counts the NAMESPACE messages that replay what already exists, so the subscriber 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} + +The **initial set** of a SUBSCRIBE_NAMESPACE is the namespaces matching its prefix that the publisher advertises at the time it answers. + + +# Introduction +A publisher answers SUBSCRIBE_NAMESPACE with one NAMESPACE per matching namespace, 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 the set is empty and wrong when the network stalls mid-replay. + +This extension has the publisher count the initial set on its REQUEST_OK. +The subscriber has caught up once that many NAMESPACE messages have arrived. + + +# Setup Negotiation + +An endpoint declares this extension with the following Setup Option ({{moqt}} Section 10.3): + +~~~ +NAMESPACE_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 NAMESPACE, where the initial set does not ride the SUBSCRIBE_NAMESPACE stream. + + +# Counting {#count} + +When the extension is negotiated, a publisher MUST include the following parameter in the REQUEST_OK answering a SUBSCRIBE_NAMESPACE: + +~~~ +NAMESPACE_COUNT Parameter { + Type (vi64) = 0x40B66 + Value (vi64) = Count +} +~~~ + +Count is the number of NAMESPACE messages carrying the initial set. +The publisher MUST send exactly that many NAMESPACE messages on the stream immediately after REQUEST_OK, before any other message; later changes follow them as usual. +A namespace the publisher would not advertise to this subscriber, such as one that loops back through it, is not counted. + +A subscriber has received the initial set once it has read Count NAMESPACE messages, immediately on a Count of 0. +It counts every one of them, including any it discards on receipt. + +A subscriber MUST close the session with a PROTOCOL_VIOLATION if REQUEST_OK omits the parameter when the extension is negotiated, or carries it when it is not. + + +# 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 | NAMESPACE_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 | NAMESPACE_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 f7c6309748..8403bc142a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -33,7 +33,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack - [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, libmoq, 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 IETF announce count](/quest/m1/js-ietf-announce-count.md) - @moq/net speaks MoQ Namespace 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 - [Import discontinuity](/quest/m1/import-discontinuity.md) - a seek or pause resets the flush jitter baseline, from moqsink, libmoq, and moq-ffi 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-ietf-announce-count.md b/quest/m1/js-ietf-announce-count.md new file mode 100644 index 0000000000..c045c7130b --- /dev/null +++ b/quest/m1/js-ietf-announce-count.md @@ -0,0 +1,26 @@ +# [S] @moq/net counts an IETF namespace subscription's initial set + +## Goal + +`@moq/net` speaks the MoQ Namespace Count extension +(`drafts/draft-lcurley-moq-namespace-count.md`) on draft-16+, as `rs/moq-net` +does: it declares the NAMESPACE_COUNT option, its publisher counts the initial +set on the REQUEST_OK answering SUBSCRIBE_NAMESPACE, and its subscriber lands +that stream's source on the count instead of the quiet timer. + +## Plan + +- Mirror `rs/moq-net/src/ietf/namespace_count.rs` in `js/net/src/ietf/`, like + `hidden.ts`: declare on draft-16+ only, and read the peer's declaration. +- Publisher: count the first advertised snapshot before writing REQUEST_OK, + then send exactly those entries first. +- Subscriber: a missing count when negotiated, or one when not, 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/quest/m1/js-origin-scope.md b/quest/m1/js-origin-scope.md index 9bc8530c60..a858cd1419 100644 --- a/quest/m1/js-origin-scope.md +++ b/quest/m1/js-origin-scope.md @@ -24,4 +24,4 @@ origin's scope. No new page. ## Related - [JS caught up](/quest/m1/js-announce-caught-up.md) - when the replay of those prefixes has landed -- [IETF announce count](/quest/m1/ietf-announce-count.md) - how an IETF consumer knows that replay is finished +- [JS IETF announce count](/quest/m1/js-ietf-announce-count.md) - how an IETF consumer knows that replay is finished diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index e7292ed723..24d0777b23 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::namespace_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), + namespace_count: ietf::namespace_count::from_setup(¶meters, v), ..Default::default() }; diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 8ae7dac1ec..4f30476054 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -18,6 +18,7 @@ pub mod hidden; mod location; pub mod message; mod namespace; +pub mod namespace_count; pub mod peer; mod priority; mod properties; diff --git a/rs/moq-net/src/ietf/namespace_count.rs b/rs/moq-net/src/ietf/namespace_count.rs new file mode 100644 index 0000000000..362643d763 --- /dev/null +++ b/rs/moq-net/src/ietf/namespace_count.rs @@ -0,0 +1,51 @@ +//! The MoQ Namespace Count extension (draft-lcurley-moq-namespace-count-00). +//! +//! Nothing in moq-transport says where the NAMESPACE messages answering a +//! SUBSCRIBE_NAMESPACE stop replaying what already existed and start reporting changes, +//! so a subscriber can only guess from the stream going quiet. When both endpoints +//! declare the NAMESPACE_COUNT option, the publisher puts the size of that initial set on +//! its REQUEST_OK, the way moq-lite's ANNOUNCE_OK does. + +use super::Version; + +/// NAMESPACE_COUNT Setup Option: the sender counts its initial NAMESPACE messages and +/// reads the peer's count. Even, so the value is a bare varint. +pub const NAMESPACE_COUNT: u64 = 0x40B64; + +/// NAMESPACE_COUNT REQUEST_OK parameter: how many NAMESPACE messages immediately follow. +/// Even, so the value is a bare varint. +pub const NAMESPACE_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::NamespaceCount).is_some() +} + +/// Declare that we count our initial set and read the peer's. +pub fn into_setup(params: &mut super::Parameters, version: Version) { + if supported(version) { + params.set_varint(super::ParameterVarInt::NamespaceCount, 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/parameters.rs b/rs/moq-net/src/ietf/parameters.rs index 7b3b426f88..e38003eb3e 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, + /// NAMESPACE_COUNT, from the MoQ Namespace Count extension. + NamespaceCount = super::namespace_count::NAMESPACE_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..0a4c895a6e 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 Namespace Count: whether the peer counts the initial set on the REQUEST_OK + /// answering SUBSCRIBE_NAMESPACE, and reads ours. + pub namespace_count: bool, } /// Shared slot for [`Peer`], filled when the peer's SETUP is read. @@ -78,6 +82,7 @@ mod tests { }, solicit: None, hidden: false, + namespace_count: false, }; let slot = PeerSetup::default(); @@ -89,6 +94,7 @@ mod tests { }, solicit: Some(true), hidden: true, + namespace_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..67a55b5f81 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1736,7 +1736,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 +1765,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 the initial set (MoQ Namespace 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 Namespace Count: take what exists now, so the OK can say how many + // NAMESPACE messages replay it. They go out first, ahead of any change. + let (initial, namespace_count) = match declared.namespace_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 +1826,22 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(msg.request_id), + namespace_count, }) .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, + namespace_count, + }) + .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 +1849,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 +1881,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 +1973,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 +1983,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 +3893,80 @@ mod tests { announced.assert_next_wait(); } + /// MoQ Namespace 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), + namespace_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, + namespace_count: 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 +4137,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: Some(RequestId(1)), + namespace_count: None, }) .await .unwrap(); @@ -4008,7 +4145,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, + namespace_count: None, + }) + .await + .unwrap(); } } @@ -4446,6 +4589,7 @@ mod tests { }, solicit, hidden: false, + namespace_count: false, }); slot } diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index 1b783556d8..357977753e 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::namespace_count::NAMESPACE_COUNT_PARAM; use super::Version; @@ -85,6 +86,9 @@ impl Message for RequestsBlocked { #[derive(Clone, Debug)] pub struct RequestOk { pub request_id: Option, + /// MoQ Namespace Count: how many NAMESPACE messages immediately follow, when this + /// answers a SUBSCRIBE_NAMESPACE on a session that negotiated the extension. + pub namespace_count: 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, NAMESPACE_COUNT_PARAM => self.namespace_count); Ok(()) } @@ -111,8 +115,12 @@ 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, + NAMESPACE_COUNT_PARAM => namespace_count: Option, ); - Ok(Self { request_id }) + Ok(Self { + request_id, + namespace_count, + }) } } @@ -188,6 +196,7 @@ mod tests { fn test_request_ok_round_trip() { let msg = RequestOk { request_id: Some(RequestId(42)), + namespace_count: None, }; let encoded = encode_message(&msg, Version::Draft15); @@ -205,6 +214,20 @@ mod tests { assert_eq!(decoded.request_id, Some(RequestId(7))); } + #[test] + fn test_request_ok_namespace_count_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, + namespace_count: count, + }; + let decoded: RequestOk = decode_message(&encode_message(&msg, version), version).unwrap(); + assert_eq!(decoded.namespace_count, count, "{version:?}"); + } + } + } + #[test] fn test_request_error_round_trip() { let msg = RequestError { @@ -243,7 +266,10 @@ mod tests { #[test] fn test_request_ok_v17_round_trip() { - let msg = RequestOk { request_id: None }; + let msg = RequestOk { + request_id: None, + namespace_count: None, + }; let encoded = encode_message(&msg, Version::Draft17); let decoded: RequestOk = decode_message(&encoded, Version::Draft17).unwrap(); @@ -271,7 +297,10 @@ mod tests { #[test] fn test_request_ok_v18_round_trip() { - let msg = RequestOk { request_id: None }; + let msg = RequestOk { + request_id: None, + namespace_count: None, + }; let encoded = encode_message(&msg, Version::Draft18); let decoded: RequestOk = decode_message(&encoded, Version::Draft18).unwrap(); @@ -283,7 +312,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, + namespace_count: 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..41d5528b15 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -10,7 +10,7 @@ use crate::{ use super::{ Control, Message, Publisher, Subscriber, Version, adapter::ControlStreamAdapter, - cluster, hidden, peer, solicit, + cluster, hidden, namespace_count, 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); + namespace_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, + namespace_count: 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 794e83f2c9..fa4e95f3ae 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -486,6 +486,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 Namespace 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 @@ -730,7 +749,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?; @@ -765,13 +785,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)?.namespace_count, 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); @@ -785,13 +804,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 Namespace Count says how many NAMESPACE messages replay the initial set. + // 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.namespace_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. @@ -830,15 +856,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; } @@ -850,7 +878,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?; @@ -865,6 +893,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 @@ -1264,12 +1298,19 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(request_id), + namespace_count: 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, + namespace_count: None, + }) + .await?; } } Ok(()) @@ -3276,7 +3317,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, + namespace_count: None, + }) + .await + .unwrap(); writer.encode(&ietf::Namespace::ID).await.unwrap(); writer .encode(&ietf::Namespace { @@ -3350,6 +3397,60 @@ mod tests { ); } + /// MoQ Namespace 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, + namespace_count: 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 { + namespace_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(); @@ -4461,7 +4562,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, + namespace_count: 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 e485a1e6ab..73b556e8bb 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3076,7 +3076,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 Namespace 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..7559643815 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), + namespace_count: ietf::namespace_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::namespace_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..632c84975f 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 Namespace 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:?}" + ); } } From 5f951f0b2b396870356548e1c93c3d3903ccac40 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 11:32:43 -0700 Subject: [PATCH 3/6] refactor(net): rename the extension to MoQ Active Count It mirrors moq-lite's AnnounceOk.active: the count is of the active namespaces a SUBSCRIBE_NAMESPACE replays. Wire values are unchanged. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 2 +- doc/concept/standard.md | 4 +- ...t.md => draft-lcurley-moq-active-count.md} | 39 ++++++++++--------- quest/m1/README.md | 2 +- quest/m1/js-active-count.md | 26 +++++++++++++ quest/m1/js-ietf-announce-count.md | 26 ------------- quest/m1/js-origin-scope.md | 2 +- rs/moq-net/src/client.rs | 4 +- .../{namespace_count.rs => active_count.rs} | 27 ++++++------- rs/moq-net/src/ietf/mod.rs | 2 +- rs/moq-net/src/ietf/parameters.rs | 4 +- rs/moq-net/src/ietf/peer.rs | 8 ++-- rs/moq-net/src/ietf/publisher.rs | 22 +++++------ rs/moq-net/src/ietf/request.rs | 31 +++++++-------- rs/moq-net/src/ietf/session.rs | 10 ++--- rs/moq-net/src/ietf/subscriber.rs | 22 +++++------ rs/moq-net/src/model/origin.rs | 2 +- rs/moq-net/src/server.rs | 4 +- rs/moq-net/tests/caught_up.rs | 2 +- 19 files changed, 119 insertions(+), 120 deletions(-) rename drafts/{draft-lcurley-moq-namespace-count.md => draft-lcurley-moq-active-count.md} (64%) create mode 100644 quest/m1/js-active-count.md delete mode 100644 quest/m1/js-ietf-announce-count.md rename rs/moq-net/src/ietf/{namespace_count.rs => active_count.rs} (55%) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index fe71dd81bc..fc22b8e85f 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -145,7 +145,7 @@ 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`, IETF draft-16+ sessions count them in `REQUEST_OK` when both sides speak -[namespace-count](/draft/moq-namespace-count), and anything else waits for the +[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 diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 0d6a4220f5..6ca6463ce6 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -45,8 +45,8 @@ 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, -[namespace-count](/draft/moq-namespace-count) to say where a -`SUBSCRIBE_NAMESPACE`'s initial set ends, and +[active-count](/draft/moq-active-count) to count the active namespaces a +`SUBSCRIBE_NAMESPACE` replays, 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-namespace-count.md b/drafts/draft-lcurley-moq-active-count.md similarity index 64% rename from drafts/draft-lcurley-moq-namespace-count.md rename to drafts/draft-lcurley-moq-active-count.md index 413c705d79..6fe85a895d 100644 --- a/drafts/draft-lcurley-moq-namespace-count.md +++ b/drafts/draft-lcurley-moq-active-count.md @@ -1,9 +1,9 @@ --- -title: "MoQ Namespace Count Extension" -abbrev: "moq-namespace-count" +title: "MoQ Active Count Extension" +abbrev: "moq-active-count" category: info -docname: draft-lcurley-moq-namespace-count-latest +docname: draft-lcurley-moq-active-count-latest submissiontype: IETF # also: "independent", "editorial", "IAB", or "IRTF" number: date: @@ -23,8 +23,8 @@ informative: --- abstract -This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber where the initial set of namespaces answering its SUBSCRIBE_NAMESPACE ends. -The publisher counts the NAMESPACE messages that replay what already exists, so the subscriber knows when it has caught up instead of guessing from a quiet stream. +This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many active namespaces a publisher replays in answer to its SUBSCRIBE_NAMESPACE. +The subscriber knows when it has caught up once that many have arrived, instead of guessing from a quiet stream. --- note_Note_to_Readers @@ -36,16 +36,17 @@ Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github. # Conventions and Definitions {::boilerplate bcp14-tagged} -The **initial set** of a SUBSCRIBE_NAMESPACE is the namespaces matching its prefix that the publisher advertises at the time it answers. +A namespace is **active** on a SUBSCRIBE_NAMESPACE stream from the NAMESPACE that advertises it until the NAMESPACE_DONE that ends it. +The **replay** is the NAMESPACE messages a publisher sends for the namespaces it already advertises under the subscription's prefix when it answers, one per namespace. # Introduction -A publisher answers SUBSCRIBE_NAMESPACE with one NAMESPACE per matching namespace, then keeps the stream open to report changes. +A publisher answers SUBSCRIBE_NAMESPACE with the replay, 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 the set is empty and wrong when the network stalls mid-replay. +The best it can do is wait for the stream to go quiet, which is slow when there is nothing to replay and wrong when the network stalls mid-replay. -This extension has the publisher count the initial set on its REQUEST_OK. -The subscriber has caught up once that many NAMESPACE messages have arrived. +This extension has the publisher count the replay on its REQUEST_OK, as moq-lite's ANNOUNCE_OK does. +The subscriber has caught up once that many namespaces are active. # Setup Negotiation @@ -53,7 +54,7 @@ The subscriber has caught up once that many NAMESPACE messages have arrived. An endpoint declares this extension with the following Setup Option ({{moqt}} Section 10.3): ~~~ -NAMESPACE_COUNT Setup Option { +ACTIVE_COUNT Setup Option { Option Key (vi64) = 0x40B64 Option Value (vi64) = 1 } @@ -61,7 +62,7 @@ NAMESPACE_COUNT Setup Option { 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 NAMESPACE, where the initial set does not ride the SUBSCRIBE_NAMESPACE stream. +An endpoint MUST NOT declare the option on a version of {{moqt}} without NAMESPACE, where the replay does not ride the SUBSCRIBE_NAMESPACE stream. # Counting {#count} @@ -69,17 +70,17 @@ An endpoint MUST NOT declare the option on a version of {{moqt}} without NAMESPA When the extension is negotiated, a publisher MUST include the following parameter in the REQUEST_OK answering a SUBSCRIBE_NAMESPACE: ~~~ -NAMESPACE_COUNT Parameter { +ACTIVE_COUNT Parameter { Type (vi64) = 0x40B66 Value (vi64) = Count } ~~~ -Count is the number of NAMESPACE messages carrying the initial set. -The publisher MUST send exactly that many NAMESPACE messages on the stream immediately after REQUEST_OK, before any other message; later changes follow them as usual. -A namespace the publisher would not advertise to this subscriber, such as one that loops back through it, is not counted. +Count is the number of namespaces in the replay. +The publisher MUST send the replay on the stream immediately after REQUEST_OK, exactly Count NAMESPACE messages for distinct namespaces and nothing else in between; later changes follow as usual. +A namespace the publisher does not advertise to this subscriber, such as one that loops back through it, is neither replayed nor counted. -A subscriber has received the initial set once it has read Count NAMESPACE messages, immediately on a Count of 0. +A subscriber has caught up once it has read Count NAMESPACE messages, immediately on a Count of 0. It counts every one of them, including any it discards on receipt. A subscriber MUST close the session with a PROTOCOL_VIOLATION if REQUEST_OK omits the parameter when the extension is negotiated, or carries it when it is not. @@ -102,7 +103,7 @@ This document requests one registration in the "MOQT Setup Options" registry ({{ | Value | Name | Reference | |:--------|:----------------|:--------------| -| 0x40B64 | NAMESPACE_COUNT | This Document | +| 0x40B64 | ACTIVE_COUNT | This Document | ## MOQT Message Parameters @@ -110,7 +111,7 @@ This document requests one registration in the "MOQT Message Parameters" registr | Value | Name | Carried In | Reference | |:--------|:----------------|:-----------|:--------------| -| 0x40B66 | NAMESPACE_COUNT | REQUEST_OK | This Document | +| 0x40B66 | ACTIVE_COUNT | REQUEST_OK | This Document | Both values are even, so each is a bare varint. diff --git a/quest/m1/README.md b/quest/m1/README.md index 8403bc142a..a90c22ce82 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -33,7 +33,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack - [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, libmoq, and every wrapper yield the same flat announce event, `Live` included -- [JS IETF announce count](/quest/m1/js-ietf-announce-count.md) - @moq/net speaks MoQ Namespace Count, so its 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 - [Import discontinuity](/quest/m1/import-discontinuity.md) - a seek or pause resets the flush jitter baseline, from moqsink, libmoq, and moq-ffi diff --git a/quest/m1/js-active-count.md b/quest/m1/js-active-count.md new file mode 100644 index 0000000000..5ec36c32c6 --- /dev/null +++ b/quest/m1/js-active-count.md @@ -0,0 +1,26 @@ +# [S] @moq/net counts the active namespaces an IETF subscription replays + +## 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 counts the active +namespaces it replays on the REQUEST_OK answering SUBSCRIBE_NAMESPACE, 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 the active namespaces before writing REQUEST_OK, then + replay exactly those first. +- Subscriber: a missing count when negotiated, or one when not, 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/quest/m1/js-ietf-announce-count.md b/quest/m1/js-ietf-announce-count.md deleted file mode 100644 index c045c7130b..0000000000 --- a/quest/m1/js-ietf-announce-count.md +++ /dev/null @@ -1,26 +0,0 @@ -# [S] @moq/net counts an IETF namespace subscription's initial set - -## Goal - -`@moq/net` speaks the MoQ Namespace Count extension -(`drafts/draft-lcurley-moq-namespace-count.md`) on draft-16+, as `rs/moq-net` -does: it declares the NAMESPACE_COUNT option, its publisher counts the initial -set on the REQUEST_OK answering SUBSCRIBE_NAMESPACE, and its subscriber lands -that stream's source on the count instead of the quiet timer. - -## Plan - -- Mirror `rs/moq-net/src/ietf/namespace_count.rs` in `js/net/src/ietf/`, like - `hidden.ts`: declare on draft-16+ only, and read the peer's declaration. -- Publisher: count the first advertised snapshot before writing REQUEST_OK, - then send exactly those entries first. -- Subscriber: a missing count when negotiated, or one when not, 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/quest/m1/js-origin-scope.md b/quest/m1/js-origin-scope.md index a858cd1419..2055d25695 100644 --- a/quest/m1/js-origin-scope.md +++ b/quest/m1/js-origin-scope.md @@ -24,4 +24,4 @@ origin's scope. No new page. ## Related - [JS caught up](/quest/m1/js-announce-caught-up.md) - when the replay of those prefixes has landed -- [JS IETF announce count](/quest/m1/js-ietf-announce-count.md) - how an IETF consumer knows that replay is finished +- [JS active count](/quest/m1/js-active-count.md) - how an IETF consumer knows that replay is finished diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 24d0777b23..c1c69edcdd 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -343,7 +343,7 @@ impl Client { } ietf::solicit::into_setup(&mut parameters, ietf_encoding); ietf::hidden::into_setup(&mut parameters, ietf_encoding); - ietf::namespace_count::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 { @@ -394,7 +394,7 @@ impl Client { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶meters, v)?, hidden: ietf::hidden::from_setup(¶meters, v), - namespace_count: ietf::namespace_count::from_setup(¶meters, v), + active_count: ietf::active_count::from_setup(¶meters, v), ..Default::default() }; diff --git a/rs/moq-net/src/ietf/namespace_count.rs b/rs/moq-net/src/ietf/active_count.rs similarity index 55% rename from rs/moq-net/src/ietf/namespace_count.rs rename to rs/moq-net/src/ietf/active_count.rs index 362643d763..2c985412b2 100644 --- a/rs/moq-net/src/ietf/namespace_count.rs +++ b/rs/moq-net/src/ietf/active_count.rs @@ -1,20 +1,21 @@ -//! The MoQ Namespace Count extension (draft-lcurley-moq-namespace-count-00). +//! The MoQ Active Count extension (draft-lcurley-moq-active-count-00). //! -//! Nothing in moq-transport says where the NAMESPACE messages answering a -//! SUBSCRIBE_NAMESPACE stop replaying what already existed and start reporting changes, -//! so a subscriber can only guess from the stream going quiet. When both endpoints -//! declare the NAMESPACE_COUNT option, the publisher puts the size of that initial set on -//! its REQUEST_OK, the way moq-lite's ANNOUNCE_OK does. +//! A publisher answers SUBSCRIBE_NAMESPACE by replaying every namespace it has active, +//! one NAMESPACE each, then reports changes on the same stream. Nothing in moq-transport +//! says where the replay ends, so a subscriber can only guess from the stream going +//! quiet. When both endpoints declare the ACTIVE_COUNT option, the publisher puts the +//! number it replays on its REQUEST_OK, as moq-lite's `AnnounceOk::active` does. use super::Version; -/// NAMESPACE_COUNT Setup Option: the sender counts its initial NAMESPACE messages and +/// ACTIVE_COUNT Setup Option: the sender counts the active namespaces it replays and /// reads the peer's count. Even, so the value is a bare varint. -pub const NAMESPACE_COUNT: u64 = 0x40B64; +pub const ACTIVE_COUNT: u64 = 0x40B64; -/// NAMESPACE_COUNT REQUEST_OK parameter: how many NAMESPACE messages immediately follow. +/// ACTIVE_COUNT REQUEST_OK parameter: how many active namespaces the NAMESPACE messages +/// immediately following it replay. /// Even, so the value is a bare varint. -pub const NAMESPACE_COUNT_PARAM: u64 = 0x40B66; +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. @@ -25,13 +26,13 @@ fn supported(version: Version) -> bool { /// 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::NamespaceCount).is_some() + supported(version) && params.get_varint(super::ParameterVarInt::ActiveCount).is_some() } -/// Declare that we count our initial set and read the peer's. +/// Declare that we count our replay and read the peer's. pub fn into_setup(params: &mut super::Parameters, version: Version) { if supported(version) { - params.set_varint(super::ParameterVarInt::NamespaceCount, 1); + params.set_varint(super::ParameterVarInt::ActiveCount, 1); } } diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 4f30476054..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; @@ -18,7 +19,6 @@ pub mod hidden; mod location; pub mod message; mod namespace; -pub mod namespace_count; pub mod peer; mod priority; mod properties; diff --git a/rs/moq-net/src/ietf/parameters.rs b/rs/moq-net/src/ietf/parameters.rs index e38003eb3e..5fb8ad044a 100644 --- a/rs/moq-net/src/ietf/parameters.rs +++ b/rs/moq-net/src/ietf/parameters.rs @@ -28,8 +28,8 @@ pub enum ParameterVarInt { Solicit = super::solicit::SOLICIT, /// HIDDEN, from the MoQ Hidden extension. Hidden = super::hidden::HIDDEN, - /// NAMESPACE_COUNT, from the MoQ Namespace Count extension. - NamespaceCount = super::namespace_count::NAMESPACE_COUNT, + /// 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 0a4c895a6e..8c80abff4e 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -20,9 +20,9 @@ pub(crate) struct Peer { /// SUBSCRIBE_NAMESPACE, so we may send it. pub hidden: bool, - /// MoQ Namespace Count: whether the peer counts the initial set on the REQUEST_OK + /// MoQ Active Count: whether the peer counts the active namespaces it replays on the REQUEST_OK /// answering SUBSCRIBE_NAMESPACE, and reads ours. - pub namespace_count: bool, + pub active_count: bool, } /// Shared slot for [`Peer`], filled when the peer's SETUP is read. @@ -82,7 +82,7 @@ mod tests { }, solicit: None, hidden: false, - namespace_count: false, + active_count: false, }; let slot = PeerSetup::default(); @@ -94,7 +94,7 @@ mod tests { }, solicit: Some(true), hidden: true, - namespace_count: 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 67a55b5f81..836a1fde66 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1767,7 +1767,7 @@ where // 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 the initial set (MoQ Namespace Count). + // whether the OK counts the active namespaces it replays (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 @@ -1795,9 +1795,9 @@ where let mut announced = origin.announced(); - // MoQ Namespace Count: take what exists now, so the OK can say how many + // MoQ Active Count: take what is active now, so the OK can say how many // NAMESPACE messages replay it. They go out first, ahead of any change. - let (initial, namespace_count) = match declared.namespace_count { + let (initial, active) = match declared.active_count { true => { let initial = Self::snapshot(&mut announced); let count = initial @@ -1826,7 +1826,7 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(msg.request_id), - namespace_count, + active, }) .await?; } @@ -1836,7 +1836,7 @@ where .writer .encode(&ietf::RequestOk { request_id: None, - namespace_count, + active, }) .await?; } @@ -3893,7 +3893,7 @@ mod tests { announced.assert_next_wait(); } - /// MoQ Namespace Count: the OK counts exactly the NAMESPACE messages that follow it, + /// 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] @@ -3909,7 +3909,7 @@ mod tests { cost: None, }, solicit: Some(true), - namespace_count: true, + active_count: true, ..Default::default() }); let publisher = Publisher::new( @@ -3956,7 +3956,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: Some(1), + active: Some(1), }) .await .unwrap(); @@ -4137,7 +4137,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: Some(RequestId(1)), - namespace_count: None, + active: None, }) .await .unwrap(); @@ -4148,7 +4148,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: None, + active: None, }) .await .unwrap(); @@ -4589,7 +4589,7 @@ mod tests { }, solicit, hidden: false, - namespace_count: false, + active_count: false, }); slot } diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index 357977753e..5fa7f4e78c 100644 --- a/rs/moq-net/src/ietf/request.rs +++ b/rs/moq-net/src/ietf/request.rs @@ -3,7 +3,7 @@ use std::borrow::Cow; use crate::coding::{Decode, DecodeError, Encode, EncodeError}; use super::Message; -use super::namespace_count::NAMESPACE_COUNT_PARAM; +use super::active_count::ACTIVE_COUNT_PARAM; use super::Version; @@ -86,9 +86,9 @@ impl Message for RequestsBlocked { #[derive(Clone, Debug)] pub struct RequestOk { pub request_id: Option, - /// MoQ Namespace Count: how many NAMESPACE messages immediately follow, when this - /// answers a SUBSCRIBE_NAMESPACE on a session that negotiated the extension. - pub namespace_count: Option, + /// MoQ Active Count: how many active namespaces the NAMESPACE messages right after this + /// replay, when it answers a SUBSCRIBE_NAMESPACE on a session that negotiated it. + pub active: Option, } impl Message for RequestOk { @@ -102,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, NAMESPACE_COUNT_PARAM => self.namespace_count); + encode_params!(w, version, ACTIVE_COUNT_PARAM => self.active); Ok(()) } @@ -115,12 +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, - NAMESPACE_COUNT_PARAM => namespace_count: Option, + ACTIVE_COUNT_PARAM => active: Option, ); - Ok(Self { - request_id, - namespace_count, - }) + Ok(Self { request_id, active }) } } @@ -196,7 +193,7 @@ mod tests { fn test_request_ok_round_trip() { let msg = RequestOk { request_id: Some(RequestId(42)), - namespace_count: None, + active: None, }; let encoded = encode_message(&msg, Version::Draft15); @@ -215,15 +212,15 @@ mod tests { } #[test] - fn test_request_ok_namespace_count_round_trips() { + 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, - namespace_count: count, + active: count, }; let decoded: RequestOk = decode_message(&encode_message(&msg, version), version).unwrap(); - assert_eq!(decoded.namespace_count, count, "{version:?}"); + assert_eq!(decoded.active, count, "{version:?}"); } } } @@ -268,7 +265,7 @@ mod tests { fn test_request_ok_v17_round_trip() { let msg = RequestOk { request_id: None, - namespace_count: None, + active: None, }; let encoded = encode_message(&msg, Version::Draft17); @@ -299,7 +296,7 @@ mod tests { fn test_request_ok_v18_round_trip() { let msg = RequestOk { request_id: None, - namespace_count: None, + active: None, }; let encoded = encode_message(&msg, Version::Draft18); @@ -314,7 +311,7 @@ mod tests { fn test_request_ok_v18_wire_matches_v17() { let msg = RequestOk { request_id: None, - namespace_count: None, + active: None, }; let v17 = encode_message(&msg, Version::Draft17); let v18 = encode_message(&msg, Version::Draft18); diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 41d5528b15..849d2f5b46 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -8,9 +8,9 @@ use crate::{ }; use super::{ - Control, Message, Publisher, Subscriber, Version, + Control, Message, Publisher, Subscriber, Version, active_count, adapter::ControlStreamAdapter, - cluster, hidden, namespace_count, peer, solicit, + cluster, hidden, peer, solicit, subscriber::{is_protocol_violation, subscribe_prefixes}, }; @@ -512,7 +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); - namespace_count::into_setup(&mut parameters, version); + active_count::into_setup(&mut parameters, version); let parameters = parameters.encode_bytes(version)?; writer.encode(&setup::Setup { parameters }).await?; @@ -905,7 +905,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: None, + active: None, }) .await .unwrap(); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index fa4e95f3ae..bd55b884ea 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -488,7 +488,7 @@ pub(super) fn subscribe_prefixes(origin: &origin::Producer) -> Vec<(PathOwned, c /// How a SUBSCRIBE_NAMESPACE stream's initial set ends. enum Landing { - /// MoQ Namespace Count: after this many more NAMESPACE messages. + /// 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), @@ -790,7 +790,7 @@ where ietf::SubscribeNamespaceOk::decode_msg(&mut data, self.version)?; None } - ietf::RequestOk::ID => ietf::RequestOk::decode_msg(&mut data, self.version)?.namespace_count, + 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); @@ -808,11 +808,11 @@ where tracing::debug!(%prefix, ?count, "subscribe_namespace ok"); - // MoQ Namespace Count says how many NAMESPACE messages replay the initial set. + // MoQ Active Count says how many active namespaces the NAMESPACE messages replay. // 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.namespace_count, count) { + 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), @@ -1298,7 +1298,7 @@ where .writer .encode(&ietf::RequestOk { request_id: Some(request_id), - namespace_count: None, + active: None, }) .await?; } @@ -1308,7 +1308,7 @@ where .writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: None, + active: None, }) .await?; } @@ -3320,7 +3320,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: None, + active: None, }) .await .unwrap(); @@ -3397,7 +3397,7 @@ mod tests { ); } - /// MoQ Namespace Count is negotiated, so a REQUEST_OK that breaks the negotiation + /// 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)] @@ -3412,7 +3412,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: count, + active: count, }) .await .unwrap(); @@ -3423,7 +3423,7 @@ mod tests { let (tasks, _task_set) = crate::util::TaskSet::new(); let peer_setup = peer::PeerSetup::default(); peer_setup.set(peer::Peer { - namespace_count: negotiated, + active_count: negotiated, ..Default::default() }); let mut subscriber = Subscriber::new( @@ -4565,7 +4565,7 @@ mod tests { writer .encode(&ietf::RequestOk { request_id: None, - namespace_count: None, + active: None, }) .await .unwrap(); diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 73b556e8bb..76765f2cc7 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -3076,7 +3076,7 @@ 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 without MoQ Namespace Count): once +/// 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 diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 7559643815..53a9e487dd 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -336,7 +336,7 @@ impl Server { let peer_declared = ietf::peer::Peer { solicit: ietf::solicit::from_setup(¶ms, v)?, hidden: ietf::hidden::from_setup(¶ms, v), - namespace_count: ietf::namespace_count::from_setup(¶ms, v), + active_count: ietf::active_count::from_setup(¶ms, v), ..Default::default() }; (path, request_id_max, peer_declared) @@ -561,7 +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::namespace_count::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 632c84975f..ded65f45c2 100644 --- a/rs/moq-net/tests/caught_up.rs +++ b/rs/moq-net/tests/caught_up.rs @@ -15,7 +15,7 @@ fn produce_origin(hop: Hop) -> origin::Producer { } /// Versions whose wire says where the initial set ends: ANNOUNCE_INIT, ANNOUNCE_OK's -/// count, or REQUEST_OK's (MoQ Namespace Count, negotiated from draft-16). +/// count, or REQUEST_OK's (MoQ Active Count, negotiated from draft-16). const COUNTED: &[&str] = &[ "moq-lite-01", "moq-lite-02", From 1f3222bca86e62fd38389701aedeb80560c573b5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 12:39:40 -0700 Subject: [PATCH 4/6] fix(net): frame Active Count on NAMESPACE messages and refuse it elsewhere The draft counts the NAMESPACE messages before a subscription is caught up rather than namespaces, and no longer leans on moq-lite. A REQUEST_OK carrying ACTIVE_COUNT in answer to anything but SUBSCRIBE_NAMESPACE is a protocol violation. Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 4 +- drafts/draft-lcurley-moq-active-count.md | 28 ++++++------ quest/m1/js-active-count.md | 17 +++---- rs/moq-net/src/ietf/active_count.rs | 20 ++++---- rs/moq-net/src/ietf/peer.rs | 4 +- rs/moq-net/src/ietf/publisher.rs | 58 ++++++++++++++++++++++-- rs/moq-net/src/ietf/request.rs | 4 +- rs/moq-net/src/ietf/subscriber.rs | 2 +- 8 files changed, 95 insertions(+), 42 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 6ca6463ce6..7305af1566 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -45,8 +45,8 @@ 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, -[active-count](/draft/moq-active-count) to count the active namespaces a -`SUBSCRIBE_NAMESPACE` replays, and +[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 index 6fe85a895d..611d845887 100644 --- a/drafts/draft-lcurley-moq-active-count.md +++ b/drafts/draft-lcurley-moq-active-count.md @@ -23,8 +23,8 @@ informative: --- abstract -This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many active namespaces a publisher replays in answer to its SUBSCRIBE_NAMESPACE. -The subscriber knows when it has caught up once that many have arrived, instead of guessing from a quiet stream. +This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many NAMESPACE messages a publisher sends in answer to its SUBSCRIBE_NAMESPACE before the subscription is caught up. +The subscriber then knows when that is, instead of guessing from a quiet stream. --- note_Note_to_Readers @@ -36,17 +36,16 @@ Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github. # Conventions and Definitions {::boilerplate bcp14-tagged} -A namespace is **active** on a SUBSCRIBE_NAMESPACE stream from the NAMESPACE that advertises it until the NAMESPACE_DONE that ends it. -The **replay** is the NAMESPACE messages a publisher sends for the namespaces it already advertises under the subscription's prefix when it answers, one per namespace. +A suffix is **active** on a SUBSCRIBE_NAMESPACE stream from a NAMESPACE message carrying it until a NAMESPACE_DONE message carrying it. +A subscription is **caught up** once every suffix the publisher advertised to it when it sent REQUEST_OK is active. # Introduction -A publisher answers SUBSCRIBE_NAMESPACE with the replay, then keeps the stream open to report changes. +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 replay and wrong when the network stalls mid-replay. +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 count the replay on its REQUEST_OK, as moq-lite's ANNOUNCE_OK does. -The subscriber has caught up once that many namespaces are active. +This extension has the publisher put the number of NAMESPACE messages that precede the subscription being caught up on its REQUEST_OK. # Setup Negotiation @@ -62,7 +61,7 @@ ACTIVE_COUNT Setup Option { 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 NAMESPACE, where the replay does not ride the SUBSCRIBE_NAMESPACE stream. +An endpoint MUST NOT declare the option on a version of {{moqt}} without the NAMESPACE message. # Counting {#count} @@ -76,14 +75,15 @@ ACTIVE_COUNT Parameter { } ~~~ -Count is the number of namespaces in the replay. -The publisher MUST send the replay on the stream immediately after REQUEST_OK, exactly Count NAMESPACE messages for distinct namespaces and nothing else in between; later changes follow as usual. -A namespace the publisher does not advertise to this subscriber, such as one that loops back through it, is neither replayed nor counted. +Count is the number of suffixes the publisher makes active before the subscription is caught up, one NAMESPACE message each. +The publisher MUST send exactly Count NAMESPACE messages, each with a distinct suffix, immediately after REQUEST_OK and before any other message on the stream; 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 has caught up once it has read Count NAMESPACE messages, immediately on a Count of 0. +A subscriber is caught up once it has read Count NAMESPACE messages, immediately on a Count of 0. It counts every one of them, including any it discards on receipt. -A subscriber MUST close the session with a PROTOCOL_VIOLATION if REQUEST_OK omits the parameter when the extension is negotiated, or carries it when it is not. +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 diff --git a/quest/m1/js-active-count.md b/quest/m1/js-active-count.md index 5ec36c32c6..9f9ae15ce7 100644 --- a/quest/m1/js-active-count.md +++ b/quest/m1/js-active-count.md @@ -1,21 +1,22 @@ -# [S] @moq/net counts the active namespaces an IETF subscription replays +# [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 counts the active -namespaces it replays on the REQUEST_OK answering SUBSCRIBE_NAMESPACE, and its subscriber lands -that stream's source on the count instead of the quiet timer. +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 the active namespaces before writing REQUEST_OK, then - replay exactly those first. -- Subscriber: a missing count when negotiated, or one when not, is a protocol - violation, as in Rust. +- 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. diff --git a/rs/moq-net/src/ietf/active_count.rs b/rs/moq-net/src/ietf/active_count.rs index 2c985412b2..37138f4fb6 100644 --- a/rs/moq-net/src/ietf/active_count.rs +++ b/rs/moq-net/src/ietf/active_count.rs @@ -1,19 +1,19 @@ //! The MoQ Active Count extension (draft-lcurley-moq-active-count-00). //! -//! A publisher answers SUBSCRIBE_NAMESPACE by replaying every namespace it has active, -//! one NAMESPACE each, then reports changes on the same stream. Nothing in moq-transport -//! says where the replay ends, so a subscriber can only guess from the stream going -//! quiet. When both endpoints declare the ACTIVE_COUNT option, the publisher puts the -//! number it replays on its REQUEST_OK, as moq-lite's `AnnounceOk::active` does. +//! 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 active namespaces it replays and -/// reads the peer's count. Even, so the value is a bare varint. +/// 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 active namespaces the NAMESPACE messages -/// immediately following it replay. +/// 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; @@ -29,7 +29,7 @@ pub fn from_setup(params: &super::Parameters, version: Version) -> bool { supported(version) && params.get_varint(super::ParameterVarInt::ActiveCount).is_some() } -/// Declare that we count our replay and read the peer's. +/// 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); diff --git a/rs/moq-net/src/ietf/peer.rs b/rs/moq-net/src/ietf/peer.rs index 8c80abff4e..ae4d164915 100644 --- a/rs/moq-net/src/ietf/peer.rs +++ b/rs/moq-net/src/ietf/peer.rs @@ -20,8 +20,8 @@ pub(crate) struct Peer { /// SUBSCRIBE_NAMESPACE, so we may send it. pub hidden: bool, - /// MoQ Active Count: whether the peer counts the active namespaces it replays on the REQUEST_OK - /// answering SUBSCRIBE_NAMESPACE, and reads ours. + /// 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, } diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 836a1fde66..e3949c23eb 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1496,6 +1496,10 @@ where } (_, ietf::RequestOk::ID) => { let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; + // ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count). + if msg.active.is_some() { + return Err(Error::ProtocolViolation); + } tracing::debug!(message = ?msg, "publish namespace ok"); } (_, ietf::RequestError::ID) => { @@ -1574,6 +1578,10 @@ 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). + if msg.active.is_some() { + return Err(Error::ProtocolViolation); + } tracing::debug!(message = ?msg, "publish_namespace update ok"); Ok(Refused::No) } @@ -1767,7 +1775,7 @@ where // 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 the active namespaces it replays (MoQ Active Count). + // 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 @@ -1795,8 +1803,8 @@ where let mut announced = origin.announced(); - // MoQ Active Count: take what is active now, so the OK can say how many - // NAMESPACE messages replay it. They go out first, ahead of any change. + // 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); @@ -4293,6 +4301,50 @@ 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 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:?}"); + } + /// 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 diff --git a/rs/moq-net/src/ietf/request.rs b/rs/moq-net/src/ietf/request.rs index 5fa7f4e78c..dc0cfb2889 100644 --- a/rs/moq-net/src/ietf/request.rs +++ b/rs/moq-net/src/ietf/request.rs @@ -86,8 +86,8 @@ impl Message for RequestsBlocked { #[derive(Clone, Debug)] pub struct RequestOk { pub request_id: Option, - /// MoQ Active Count: how many active namespaces the NAMESPACE messages right after this - /// replay, when it answers a SUBSCRIBE_NAMESPACE on a session that negotiated it. + /// 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, } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index aa36e491e9..876e25770b 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -789,7 +789,7 @@ where tracing::debug!(%prefix, ?count, "subscribe_namespace ok"); - // MoQ Active Count says how many active namespaces the NAMESPACE messages replay. + // 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. From 96c060280b93422de8f50eb72507df094d00ba1a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 13:38:47 -0700 Subject: [PATCH 5/6] fix(net): define Active Count's caught up from the subscriber's side The subscriber is caught up after reading Count NAMESPACE messages; the publisher only has to send exactly that many first. Also covers a counted OK to a REQUEST_UPDATE. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-active-count.md | 14 +++--- rs/moq-net/src/ietf/publisher.rs | 60 ++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 8 deletions(-) diff --git a/drafts/draft-lcurley-moq-active-count.md b/drafts/draft-lcurley-moq-active-count.md index 611d845887..3be4d75db7 100644 --- a/drafts/draft-lcurley-moq-active-count.md +++ b/drafts/draft-lcurley-moq-active-count.md @@ -23,8 +23,8 @@ informative: --- abstract -This document defines an extension for MoQ Transport {{moqt}} that tells a subscriber how many NAMESPACE messages a publisher sends in answer to its SUBSCRIBE_NAMESPACE before the subscription is caught up. -The subscriber then knows when that is, instead of guessing from a quiet stream. +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 @@ -37,7 +37,7 @@ Submit an [issue](https://github.com/moq-dev/moq/issues) or [PR](https://github. {::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 subscription is **caught up** once every suffix the publisher advertised to it when it sent REQUEST_OK is active. +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 @@ -45,7 +45,7 @@ A publisher answers SUBSCRIBE_NAMESPACE with a NAMESPACE message for each suffix 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 the number of NAMESPACE messages that precede the subscription being caught up on its REQUEST_OK. +This extension has the publisher put on its REQUEST_OK how many NAMESPACE messages come before the first change. # Setup Negotiation @@ -75,12 +75,10 @@ ACTIVE_COUNT Parameter { } ~~~ -Count is the number of suffixes the publisher makes active before the subscription is caught up, one NAMESPACE message each. -The publisher MUST send exactly Count NAMESPACE messages, each with a distinct suffix, immediately after REQUEST_OK and before any other message on the stream; later changes follow as usual. +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 is caught up once it has read Count NAMESPACE messages, immediately on a Count of 0. -It counts every one of them, including any it discards on receipt. +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. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index e3949c23eb..fd5b520ad8 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4723,6 +4723,66 @@ 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:?}"); + } + /// 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. From 6b08c192ec57d283572655d6be4bc51c52e55548 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 26 Sep 2026 19:56:39 -0700 Subject: [PATCH 6/6] fix(net): close the session on a misplaced ACTIVE_COUNT On the draft-15 solicited path, advertise_namespace runs inside a request task that only logs errors, so the violation ended one stream instead of the session the draft requires. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index fd5b520ad8..e7dd065aaa 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -1496,8 +1496,14 @@ where } (_, ietf::RequestOk::ID) => { let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; - // ACTIVE_COUNT only answers SUBSCRIBE_NAMESPACE (MoQ Active Count). + // 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"); @@ -1578,8 +1584,13 @@ 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). + // 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"); @@ -4323,6 +4334,7 @@ mod tests { .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 { @@ -4343,6 +4355,9 @@ mod tests { .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`, @@ -4781,6 +4796,9 @@ mod tests { .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