diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 6c1fd65b96..e8ef1a555a 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -88,12 +88,17 @@ prefixes it is told about. A subscriber watching under a root sees advertisements named relative to that root. The pattern scope filters which prefixes are visible without changing a route's prefix. Announce events carry the covered path, captures, and what -happened to it: Rust -`announce::Update { path, captures: Option>, route, kind }` and -TypeScript `Announce.Update { path, captures, route, kind }`, where the kind is -announced, updated (a reprice in place), or retracted. Captures are present when -the announced prefix pins every wildcard in the most-specific matching scope -member. The Rust consumer is a `Stream` and the TypeScript one an async iterable. +happened to it: Rust `announce::Event::{Announced, Updated, Retracted}`, each +holding an `announce::Announce { prefix, captures: Option>, route }`, +and TypeScript `Announce.Update { path, captures, route, kind }`, where the kind +is announced, updated (a reprice in place), or retracted. Captures are present +when the announced prefix pins every wildcard in the most-specific matching +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. 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/lib/rs/index.md b/doc/lib/rs/index.md index 17cd466441..4ab44798c9 100644 --- a/doc/lib/rs/index.md +++ b/doc/lib/rs/index.md @@ -50,8 +50,9 @@ let session = client.with_origin(origin.clone()).connect(url); // Subscribe: wait for a route, resolve the broadcast at its path, read the catalog. let consumer = origin.consume(); let mut announced = consumer.announced(); -while let Some(update) = announced.next().await { - if !update.kind.is_active() { continue } +while let Some(event) = announced.next().await { + // Skip retractions, and `Live`, which marks the end of what was already live. + let moq_net::announce::Event::Announced(update) = event else { continue }; let broadcast = consumer.request_broadcast(&update.prefix).await?; let catalog = broadcast .track(hang::Catalog::DEFAULT_NAME)? diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 33f67336aa..0f30796e69 100644 --- a/doc/lib/rs/moq-net.md +++ b/doc/lib/rs/moq-net.md @@ -126,14 +126,16 @@ overlaps that scope; exact creates and requests must match it, so a broad route can advertise the wire-compatible prefix while excluded requests are refused locally. A disjoint route is `Unauthorized`. -`origin.consume().announced()` yields `announce::Update` values: `path` is the -covered prefix relative to the consumer's root, `kind` is `Announced`, -`Updated` (a reprice in place), or `Retracted`, `captures` reports what the -most specific matching scope member's wildcards stood for when the prefix -pins them, and `route` carries hops and cost (on a retraction, its last -values). The consumer is also a `futures::Stream`. A prefix is not a -broadcast name; sessions request each scope member's literal head and filter -locally. +`origin.consume().announced()` yields `announce::Event`s: `Announced`, +`Updated` (a reprice in place), or `Retracted`, each holding an +`announce::Announce` with `prefix`, the covered prefix relative to the +consumer's root; `captures`, what the most specific matching scope member's +wildcards stood for when the prefix pins them; and `route`, its hops and cost +(on a retraction, its last values). A single `Event::Live` follows +the routes live at subscribe time, including every route a connected peer +was still sending, so a caller listing what is live stops there. The +consumer is also a `futures::Stream`. A prefix is not a broadcast name; +sessions request each scope member's literal head and filter locally. ## Limiting reads diff --git a/quest/m1/cli-inspect/README.md b/quest/m1/cli-inspect/README.md index 0ff452df40..35315ac840 100644 --- a/quest/m1/cli-inspect/README.md +++ b/quest/m1/cli-inspect/README.md @@ -20,6 +20,5 @@ This README owns the guide, written once both verbs land: a new ## Quests -- [Caught up](/quest/m1/cli-inspect/caught-up.md) - moq-net's announce consumer yields a `Live` marker once the initial set has landed, and shell completion drops its settle timer - [ls](/quest/m1/cli-inspect/ls.md) - `moq ls` prints the live set and exits, or follows changes, as paths or JSON lines - [Fetch](/quest/m1/cli-inspect/fetch.md) - `moq fetch` writes one group of a track to stdout, raw or as JSON lines diff --git a/quest/m1/cli-inspect/caught-up.md b/quest/m1/cli-inspect/caught-up.md deleted file mode 100644 index dcec2f6787..0000000000 --- a/quest/m1/cli-inspect/caught-up.md +++ /dev/null @@ -1,41 +0,0 @@ -# [M] An announce consumer knows when it has caught up - -## Goal - -A Rust consumer of `origin::Consumer::announced()` is told, once and in -order, that the routes live at subscribe time have all been delivered, so -"list what is live" needs no guess. Today the stream replays the current -routes and continues live with no boundary, and moq-cli's `--broadcast` -completion (`rs/moq-cli/src/complete.rs`) stops on a 30ms settle / 500ms -budget timer instead. - -## Plan - -- `AnnounceConsumer::next()` yields an enum: each route update as today, - then a single `Live` marker once the initial set has been delivered, then - live updates. `Live` comes after every replayed update, so a caller that - stops at it has seen the whole set. This changes a published return type, - so the quest lands on `dev`. -- The fact is per source, but consumers read a merged origin, which today - has no source registry. Each remote announce subscription (a session's - subscriber, a cluster peer) holds a pending guard from the origin until - its initial set has landed; in-process routes are replayed synchronously - and are never pending. A cursor yields `Live` once every guard pending at - subscribe time has cleared. A source that closes before clearing drops - its guard, so a dead session cannot stall the marker. No cost while no - guard is pending. -- A source clears on the wire's boundary: `AnnounceOk.active` on lite-05+, - `AnnounceInit` on lite-01/02. Versions without one (lite-03/04, IETF) - clear on a settle timer owned by the session, the one place it lives. -- Move completion onto `Live`; the timer survives only in the session for - marker-less versions. Look at the other settle and deadline loops - (`moq-bench` startup, relay cluster discovery, test `settle()` helpers) - and move any that want the initial set. -- Test: a relay with N announced broadcasts yields `Live` after exactly N - updates on each lite version that carries a count, an empty relay yields - it immediately, and a session that closes before its count arrives does - not block it. - -Public API: breaking on moq-net (`next()` returns an enum). Wire: none. JS -parity is [a separate quest](/quest/m1/js-announce-caught-up.md), and an -IETF count is [another](/quest/m1/ietf-announce-count.md). diff --git a/quest/m1/cli-inspect/ls.md b/quest/m1/cli-inspect/ls.md index a5791a8d6a..67307711f7 100644 --- a/quest/m1/cli-inspect/ls.md +++ b/quest/m1/cli-inspect/ls.md @@ -14,6 +14,9 @@ it exits non-zero. ## Plan +- Snapshot mode reads the announce cursor up to its `announce::Event::Live` + marker, which follows the relay's whole initial set; `--follow` prints + through it. - A MoQ verb beside import/export/play, not a stageable one: it consumes the origin and never publishes. It is a subscriber-only session, so it works with a subscribe-only token. @@ -23,7 +26,3 @@ it exits non-zero. - Test: against an in-process relay, snapshot mode prints exactly the announced set and exits, `--follow` prints the initial replay then a `-` when a publisher leaves, and `--json` lines parse. - -## Required - -- [Caught up](/quest/m1/cli-inspect/caught-up.md) - snapshot mode exits on the consumer's `Live` marker, so `ls` follows it onto `dev` diff --git a/quest/m1/ietf-announce-count.md b/quest/m1/ietf-announce-count.md index 5c5b92695f..c077ad10bc 100644 --- a/quest/m1/ietf-announce-count.md +++ b/quest/m1/ietf-announce-count.md @@ -18,7 +18,3 @@ instead of the settle timer. Without the extension the timer stays. an un-negotiated one still clears on the timer. Public API: none. Wire: a new opt-in extension. - -## Required - -- [Caught up](/quest/m1/cli-inspect/caught-up.md) - the per-source clear this feeds diff --git a/quest/m1/js-announce-caught-up.md b/quest/m1/js-announce-caught-up.md index 46785d11e4..b537acba01 100644 --- a/quest/m1/js-announce-caught-up.md +++ b/quest/m1/js-announce-caught-up.md @@ -2,17 +2,15 @@ ## Goal -`@moq/net`'s announce consumer yields the same `Live` marker the Rust -consumer gets from [Caught up](/quest/m1/cli-inspect/caught-up.md), with -the same ordering and per-source semantics, so a browser app can render "no +`@moq/net`'s announce consumer yields the same `Live` marker as the Rust +`announce::Consumer`, with the same ordering and per-source semantics, so a browser app can render "no broadcasts" instead of a spinner that never resolves. ## Plan -Mirror the Rust shape, name, and marker-less fallback. Test the same cases +Mirror the Rust shape (one flat `announce::Event`: `Announced`, `Updated`, +`Retracted`, each holding an `Announce`, or `Live`), its +per-source guards (a session holds one per announce stream until the count, +ANNOUNCE_INIT, or a quiet stream lands it), and the marker-less fallback. Test the same cases in JS. Public API: the announce consumer's yield type changes, so it lands with the Rust break. Wire: none. - -## Required - -- [Caught up](/quest/m1/cli-inspect/caught-up.md) - settles the API shape this mirrors diff --git a/rs/hang/examples/subscribe.rs b/rs/hang/examples/subscribe.rs index eb1614df05..a832f96f96 100644 --- a/rs/hang/examples/subscribe.rs +++ b/rs/hang/examples/subscribe.rs @@ -43,9 +43,14 @@ async fn run_session(origin: moq_net::origin::Producer) -> anyhow::Result<()> { async fn run_subscribe(consumer: moq_net::origin::Consumer) -> anyhow::Result<()> { // Wait for a route to be announced, then resolve the broadcast at its path. // The convention is that a publisher announces each broadcast's exact path. - let update = consumer.announced().next().await.context("origin closed")?; - anyhow::ensure!(update.kind.is_active(), "route retracted: {}", update.prefix); - let path = update.prefix; + let mut announced = consumer.announced(); + let path = loop { + match announced.next().await.context("origin closed")? { + moq_net::announce::Event::Announced(announce) => break announce.prefix, + // Nothing else can come before the first announcement. + event => anyhow::ensure!(matches!(event, moq_net::announce::Event::Live), "unexpected {event:?}"), + } + }; tracing::info!(%path, "broadcast announced"); let broadcast = consumer.request_broadcast(&path).await?; diff --git a/rs/libmoq/src/origin.rs b/rs/libmoq/src/origin.rs index 23451023a3..04a77321bd 100644 --- a/rs/libmoq/src/origin.rs +++ b/rs/libmoq/src/origin.rs @@ -54,7 +54,7 @@ struct AnnouncedRecord { unsafe impl Send for AnnouncedRecord {} impl AnnouncedRecord { - fn new(update: moq_net::announce::Update) -> Self { + fn new(update: moq_net::announce::Announce, active: bool) -> Self { let captures = update.captures.map(|captures| { captures .into_iter() @@ -75,7 +75,7 @@ impl AnnouncedRecord { prefix: update.prefix.to_string(), captures, capture_views, - active: update.kind.is_active(), + active, } } } @@ -143,17 +143,25 @@ impl Origin { ) -> Result<(), Error> { loop { // `biased` so a pending close always wins over a ready announcement. - let update = tokio::select! { + let (update, active) = tokio::select! { biased; _ = &mut close => return Ok(()), next = consumer.next() => match next { - Some(announced) => announced, + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => { + (update, true) + } + Some(moq_net::announce::Event::Retracted(update)) => (update, false), + // The C API has no caught-up callback yet. + Some(moq_net::announce::Event::Live) => continue, None => return Ok(()), }, }; // Hold the lock only to buffer the announcement; release it before the callback. - let announced_id = State::lock().origin.announced.insert(AnnouncedRecord::new(update))?; + let announced_id = State::lock() + .origin + .announced + .insert(AnnouncedRecord::new(update, active))?; callback.call(announced_id); } } diff --git a/rs/moq-bench/src/connection.rs b/rs/moq-bench/src/connection.rs index 1502ab5c87..9dcc527ec8 100644 --- a/rs/moq-bench/src/connection.rs +++ b/rs/moq-bench/src/connection.rs @@ -335,10 +335,11 @@ async fn subscribe( biased; _ = &mut deadline => break, update = announced.next() => { - let Some(update) = update else { break }; - if !update.kind.is_active() { - continue; - } + let update = match update { + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => update, + Some(moq_net::announce::Event::Retracted(_) | moq_net::announce::Event::Live) => continue, + None => break, + }; let path = update.prefix.to_string(); if own.contains(&path) || !seen.insert(path.clone()) { continue; @@ -360,12 +361,11 @@ async fn subscribe( // Top up from late announcements, first-come: the pool was too small, so // there is nothing to spread over. while selected < want { - let Some(update) = announced.next().await else { - break; + let update = match announced.next().await { + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => update, + Some(moq_net::announce::Event::Retracted(_) | moq_net::announce::Event::Live) => continue, + None => break, }; - if !update.kind.is_active() { - continue; - } let path = update.prefix.to_string(); if own.contains(&path) || !seen.insert(path.clone()) { continue; diff --git a/rs/moq-boy/src/input.rs b/rs/moq-boy/src/input.rs index cf6e658ea7..d8e48b6432 100644 --- a/rs/moq-boy/src/input.rs +++ b/rs/moq-boy/src/input.rs @@ -66,13 +66,18 @@ pub async fn handle_viewers( ) -> anyhow::Result<()> { let mut announced = viewer_origin.announced(); loop { - let Some(update) = announced.next().await else { - break; + let (update, active) = match announced.next().await { + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => { + (update, true) + } + Some(moq_net::announce::Event::Retracted(update)) => (update, false), + Some(moq_net::announce::Event::Live) => continue, + None => break, }; let viewer_id = update.prefix.to_string(); - if update.kind.is_active() { + if active { let Ok(broadcast) = viewer_origin.request_broadcast(&update.prefix).await else { continue; }; diff --git a/rs/moq-cli/src/complete.rs b/rs/moq-cli/src/complete.rs index 104f64fac2..d495eedd08 100644 --- a/rs/moq-cli/src/complete.rs +++ b/rs/moq-cli/src/complete.rs @@ -47,13 +47,6 @@ const BUDGET: Duration = Duration::from_millis(500); /// (a blocking device enumeration) still cannot hold the answer back. const CEILING: Duration = Duration::from_millis(1_500); -/// How long the announce sweep waits for a sibling after an announcement lands. -/// -/// A relay sends its whole announced set back to back, so the gap between the -/// first and the last is a round trip, not a budget. Without this the sweep would -/// always cost [`BUDGET`], even when the answer arrived in a few milliseconds. -const SETTLE: Duration = Duration::from_millis(30); - /// The completers every build has, keyed by the *value* name they answer for. /// /// The value name, not the flag: that is what Usage matches an overlay on, and the @@ -481,21 +474,22 @@ fn broadcasts(_ctx: CompleteCtx<'_>) -> CompletionFuture<'static> { // sweep is running is one the user cannot name by the time they press enter. let mut announced = origin.consume().announced(); let mut live = BTreeSet::new(); - // The first announcement gets the whole remaining budget; each one after it - // only has to beat its siblings, which are already on the wire. - let mut until = deadline; - while let Ok(Some(update)) = timeout_at(until, announced.next()).await { - until = deadline.min(Instant::now() + SETTLE); - let path = update.prefix.to_string(); + // The marker says the relay's whole set has arrived; the budget only bounds + // a relay that is slow to send it. + while let Ok(Some(event)) = timeout_at(deadline, announced.next()).await { // The root broadcast is the connection path itself, which an unset // `--broadcast` already names; there is no word to insert for it. - if path.is_empty() { - continue; + match event { + moq_net::announce::Event::Announced(announce) | moq_net::announce::Event::Updated(announce) => { + if !announce.prefix.is_empty() { + live.insert(announce.prefix.to_string()); + } + } + moq_net::announce::Event::Retracted(announce) => { + live.remove(announce.prefix.as_str()); + } + moq_net::announce::Event::Live => break, } - match update.kind.is_active() { - true => live.insert(path), - false => live.remove(&path), - }; } drop(connection); diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 0169459ed9..250b2d5929 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -143,14 +143,25 @@ impl OriginDynamic { impl Announced { async fn next(&mut self) -> Result>, MoqError> { - match self.inner.next().await { - Some(update) => Ok(Some(Arc::new(MoqAnnounceUpdate { + // The bindings have no caught-up marker yet. + let update = loop { + match self.inner.next().await { + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => { + break Some((update, true)); + } + Some(moq_net::announce::Event::Retracted(update)) => break Some((update, false)), + Some(moq_net::announce::Event::Live) => continue, + None => break None, + } + }; + match update { + Some((update, active)) => Ok(Some(Arc::new(MoqAnnounceUpdate { prefix: update.prefix.to_string(), captures: update .captures .map(|captures| captures.into_iter().map(|capture| capture.to_string()).collect()), route: update.route.into(), - active: update.kind.is_active(), + active, }))), None => Ok(None), } diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 04e1be686e..58478c3021 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -267,7 +267,7 @@ fn bench_subscribe(c: &mut Criterion) { b.iter(|| { let mut cursor = fleet.consumer.announced(); let mut replayed = 0; - while cursor.next().now_or_never().flatten().is_some() { + while let Some(announce::Event::Announced(_)) = cursor.next().now_or_never().flatten() { replayed += 1; } assert_eq!(replayed, publishers); diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 4f567d715b..c96cf6659a 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -735,6 +735,37 @@ mod tests { .expect("connect failed"); } + /// A peer that never delivers its announce count cannot stall the live + /// marker past its session: dropping the session lands the source. + #[tokio::test(start_paused = true)] + async fn a_dead_session_does_not_hold_the_live_marker() { + let gate = kio::Producer::new(true); + let transport = crate::lite::test_transport::SinkSession::gated_bi(gate.consume()) + .with_protocol(crate::version::ALPN_LITE_05); + + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let client = Client::new() + .with_versions([Version::Lite(lite::Version::Lite05)].into()) + .with_subscriber(origin.clone()); + let (session, driver) = client + .connect(tokio::time::Instant::now().into_std(), transport) + .await + .expect("connect failed"); + + let mut announced = origin.consume().announced(); + let mut next = std::pin::pin!(announced.next()); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(5), next.as_mut()) + .await + .is_err(), + "live before the peer answered" + ); + + drop(driver); + drop(session); + assert!(matches!(next.await, Some(crate::announce::Event::Live))); + } + #[tokio::test(start_paused = true)] async fn alpn_lite_falls_back_to_draft14_and_switches_version_post_setup() { run_alpn_lite_fallback_case(Some(ALPN_LITE)).await; diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index ec5332f378..c40bdeebe6 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -214,8 +214,9 @@ impl Namespaces { enum NamespaceEvent { /// The session or stream ended, with the result to surface. Closed(Result<(), Error>), - /// An origin-level route (un)announce, `None` once the announce stream ends. - Update(Option), + /// An origin-level route (un)announce and whether it is active, `None` once + /// the announce stream ends. + Update(Option<(crate::announce::Announce, bool)>), /// The retry sleep fired: re-offer whatever the peer should be holding and isn't. Retry, } @@ -1777,8 +1778,20 @@ where if let Poll::Ready(res) = target.poll_closed(&mut cx) { return Poll::Ready(NamespaceEvent::Closed(res)); } - if let Poll::Ready(update) = announced.poll_next(waiter) { - return Poll::Ready(NamespaceEvent::Update(update)); + // The origin's live marker means nothing to the peer here. + while let Poll::Ready(next) = announced.poll_next(waiter) { + match next { + Some(crate::announce::Event::Live) => continue, + Some( + crate::announce::Event::Announced(update) | crate::announce::Event::Updated(update), + ) => { + return Poll::Ready(NamespaceEvent::Update(Some((update, true)))); + } + Some(crate::announce::Event::Retracted(update)) => { + return Poll::Ready(NamespaceEvent::Update(Some((update, false)))); + } + None => return Poll::Ready(NamespaceEvent::Update(None)), + } } if retry.poll(waiter).is_ready() { return Poll::Ready(NamespaceEvent::Retry); @@ -1819,14 +1832,14 @@ where stream.writer.finish()?; return stream.writer.closed().await; } - NamespaceEvent::Update(Some(update)) => { + NamespaceEvent::Update(Some((update, active))) => { let path = update.prefix; let suffix = path .strip_prefix(&prefix) .expect("origin returned invalid prefix") .to_owned(); - if update.kind.is_active() { + 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) { diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index 92093e65ef..f12351c168 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -8,8 +8,10 @@ use crate::{ }; use super::{ - Control, Message, Publisher, Subscriber, Version, adapter::ControlStreamAdapter, cluster, peer, solicit, - subscriber::is_protocol_violation, + Control, Message, Publisher, Subscriber, Version, + adapter::ControlStreamAdapter, + cluster, peer, solicit, + subscriber::{is_protocol_violation, subscribe_prefixes}, }; /// Everything one moq-transport session needs to start. @@ -90,6 +92,10 @@ where // server to open connections (draft-19 sect 10.4). let (goaway_handle, goaway) = crate::goaway::Handle::new(!client); + // One SUBSCRIBE_NAMESPACE per permitted prefix, like `lite::Subscriber`: the + // scope is what we may ask for, and it is not the origin's root. + let namespaces = subscribe.as_ref().map(subscribe_prefixes).unwrap_or_default(); + let driver = async move { // Our own Hop ID, taken from whichever origin the caller actually supplied so // every session out of this process stamps the same one and cross-session loop @@ -203,11 +209,9 @@ where // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); - // One SUBSCRIBE_NAMESPACE per permitted prefix, like `lite::Subscriber`: - // the scope is what we may ask for, and it is not the origin's root. let mut sub_ns_run = std::pin::pin!(err_only(async { let mut prefixes = futures::stream::FuturesUnordered::new(); - for prefix in sub_ns.subscribe_prefixes() { + for (prefix, replaying) in namespaces { let mut sub_ns = sub_ns.clone(); let sub_ns_adapter = sub_ns_adapter.clone(); prefixes.push(async move { @@ -222,7 +226,7 @@ where } _ => Stream::open(&mut sub_ns_adapter.clone(), version).await?, }; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix).await { + if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, replaying).await { // The peer breaking the protocol is fatal, and the driver // below turns this into the session close the draft wants. if is_protocol_violation(&err) { @@ -340,16 +344,15 @@ where // Unsolicited PUBLISH_NAMESPACE unless the peer requires solicitation; // see `Publisher::run_publish_namespaces`. let mut pub_ns_run = std::pin::pin!(err_only(publisher.clone().run_publish_namespaces())); - // One SUBSCRIBE_NAMESPACE per permitted prefix; see the draft-16 arm. let mut sub_ns_run = std::pin::pin!(err_only(async { let mut prefixes = futures::stream::FuturesUnordered::new(); - for prefix in sub_ns.subscribe_prefixes() { + for (prefix, replaying) in namespaces { let mut sub_ns = sub_ns.clone(); let sub_ns_session = sub_ns_session.clone(); prefixes.push(async move { let mut sub_ns_session = sub_ns_session; let stream = Stream::open(&mut sub_ns_session, version).await?; - if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix).await { + if let Err(err) = sub_ns.run_subscribe_namespace(stream, prefix, replaying).await { // The peer breaking the protocol is fatal, and the driver // below turns this into the session close the draft wants. if is_protocol_violation(&err) { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1b2fab1598..b973b0033f 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -415,6 +415,28 @@ pub(super) struct Subscriber { going_away: crate::goaway::GoingAway, } +/// The prefixes to issue SUBSCRIBE_NAMESPACE for: `origin`'s permitted scope, +/// relative to its root. +/// +/// The scope is what we may ASK the peer for; the root is where what comes back +/// MOUNTS locally. Those are independent, and only coincide when the peer shares +/// our namespace -- a peer outside it has never heard of our root, so a rooted +/// subscriber asks for its scope and mounts the replies under the root. +/// +/// Asked unconditionally, without waiting on the peer's SETUP: a peer with nothing to +/// advertise answers with an empty set, which costs one stream, while waiting to find +/// out costs a round trip on every session. +/// +/// Each comes with the guard its stream holds until the peer's initial set has +/// landed. Taken when the session starts, before it is handed out, so no announce +/// cursor misses the replay. +pub(super) fn subscribe_prefixes(origin: &origin::Producer) -> Vec<(PathOwned, crate::model::Replaying)> { + crate::model::interest_prefixes(&origin.allowed()) + .into_iter() + .map(|prefix| (prefix.clone(), origin.replaying(prefix))) + .collect() +} + /// Resolve the subscription a data stream belongs to. /// /// SUBSCRIBE_OK can be reordered behind the stream it describes, so an alias we have not @@ -626,21 +648,6 @@ where Some(track) } - /// The prefixes to issue SUBSCRIBE_NAMESPACE for: this handle's permitted scope, - /// relative to its root. - /// - /// The scope is what we may ASK the peer for; the root is where what comes back - /// MOUNTS locally. Those are independent, and only coincide when the peer shares - /// our namespace -- a peer outside it has never heard of our root, so a rooted - /// subscriber asks for its scope and mounts the replies under the root. - /// - /// Asked unconditionally, without waiting on the peer's SETUP: a peer with nothing to - /// advertise answers with an empty set, which costs one stream, while waiting to find - /// out costs a round trip on every session. - pub fn subscribe_prefixes(&self) -> Vec { - crate::model::interest_prefixes(&self.origin.allowed()) - } - /// Send SUBSCRIBE_NAMESPACE for one prefix on a bidi stream. /// The caller is responsible for opening the appropriate stream type /// (virtual for v14/v15, real bidi for v16+), one per prefix. @@ -652,6 +659,7 @@ where &mut self, mut stream: Stream, prefix: PathOwned, + replaying: crate::model::Replaying, ) -> Result<(), Error> { // A peer that sent GOAWAY told us to stop opening requests on this session, // announce-interest included (draft-19 sect 10.4). @@ -714,6 +722,10 @@ where tracing::debug!(%prefix, "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))); + // The extension changes the NAMESPACE encoding, so we can't parse one until // the peer's SETUP says whether it negotiated. let peer = self.peer().await; @@ -731,7 +743,9 @@ where // its channel without being withdrawn, so hold the front open for a reconnect. // This is what moq-lite already does, where the equivalent map is a local whose // guards drop. - let res = self.run_namespace_entries(&mut stream, &prefix, &peer, &mut live).await; + let res = self + .run_namespace_entries(&mut stream, &prefix, &peer, &mut live, &mut landing) + .await; for path in live { let _ = self.stop_announce(path, Detach::Abrupt); } @@ -749,12 +763,29 @@ 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)>, ) -> Result<(), Error> { loop { - let type_id: u64 = match stream.reader.decode_maybe().await? { + 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() + { + *landing = None; + } + waiter.poll_future(decode.as_mut()) + }) + .await + }; + let type_id: u64 = match next? { Some(id) => id, None => break, // Stream closed }; + if let Some((_, quiet)) = landing { + quiet.heard(); + } let size: u16 = stream.reader.decode().await?; let mut data = stream.reader.read_exact(size as usize).await?; @@ -2894,14 +2925,16 @@ mod tests { Default::default(), ); + let mut namespaces = subscribe_prefixes(&subscriber.origin); assert_eq!( - subscriber.subscribe_prefixes(), + namespaces.iter().map(|(prefix, _)| prefix.clone()).collect::>(), vec![crate::Path::new("cam").to_owned()], "one SUBSCRIBE_NAMESPACE per permitted prefix, relative to the root", ); + let (prefix, replaying) = namespaces.pop().unwrap(); let stream = Stream::open(&mut session.clone(), Version::Draft16).await.unwrap(); - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("cam").to_owned())); + let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, replaying)); // Parks awaiting the peer's response; the request is already on the wire. assert!(futures::poll!(run.as_mut()).is_pending()); @@ -2966,10 +2999,10 @@ mod tests { Default::default(), ); - let prefix = subscriber.subscribe_prefixes().pop().expect("one prefix"); + let (prefix, replaying) = subscribe_prefixes(&subscriber.origin).pop().expect("one prefix"); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); // Parks on the read after the scripted NAMESPACE is consumed. - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix)); + let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, prefix, replaying)); for _ in 0..100 { // The result is deliberately ignored: a regressed mount lands out of scope // and errors here, which the assertions below name far better than a poll @@ -3999,7 +4032,7 @@ mod tests { let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); subscriber - .run_subscribe_namespace(stream, crate::Path::new("").to_owned()) + .run_subscribe_namespace(stream, crate::Path::new("").to_owned(), subscriber.origin.replaying("")) .await .expect("a clean FIN is not an error"); settle().await; @@ -4069,7 +4102,9 @@ mod tests { ); let stream = Stream::open(&mut session.clone(), VERSION).await.unwrap(); - let mut run = std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned())); + let replaying = subscriber.origin.replaying(""); + let mut run = + std::pin::pin!(subscriber.run_subscribe_namespace(stream, crate::Path::new("").to_owned(), replaying)); for _ in 0..100 { assert!( futures::poll!(run.as_mut()).is_pending(), diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index e618843c2d..cb7274fd10 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -601,7 +601,7 @@ impl AnnounceRun { /// Where an update travels on this stream: its prefix relative to the requested /// prefix, which the origin's scope guarantees it sits under. - fn suffix(&self, update: &announce::Update) -> crate::PathOwned { + fn suffix(&self, update: &announce::Announce) -> crate::PathOwned { update .prefix .strip_prefix(&self.prefix) @@ -685,11 +685,18 @@ impl AnnounceRun { // Send ANNOUNCE_INIT as the first message with all currently active routes. // We use `try_next()` to synchronously get the initial updates. - while let Some(update) = announced.try_next() { + while let Some(event) = announced.try_next() { + let (update, active) = match event { + announce::Event::Announced(update) | announce::Event::Updated(update) => (update, true), + announce::Event::Retracted(update) => (update, false), + // The marker only says the origin caught up; the peer learns the + // initial set's end from the version's own framing. + announce::Event::Live => continue, + }; let absolute = origin.absolute(&update.prefix); let suffix = self.suffix(&update); - if update.kind.is_active() { + if active { if self.outgoing(&update.route, &absolute).is_none() { continue; } @@ -713,11 +720,18 @@ impl AnnounceRun { // them afterward. The receiver stamps our origin onto each hop chain, so we // forward the stored chain as-is (no self push here). let mut initial: Vec<(crate::PathOwned, Hops, crate::origin::Cost)> = Vec::new(); - while let Some(update) = announced.try_next() { + while let Some(event) = announced.try_next() { + let (update, active) = match event { + announce::Event::Announced(update) | announce::Event::Updated(update) => (update, true), + announce::Event::Retracted(update) => (update, false), + // The marker only says the origin caught up; the peer learns the + // initial set's end from the version's own framing. + announce::Event::Live => continue, + }; let absolute = origin.absolute(&update.prefix); let suffix = self.suffix(&update); - if update.kind.is_active() { + if active { let Some((hops, cost)) = self.outgoing(&update.route, &absolute) else { continue; }; @@ -785,18 +799,23 @@ impl AnnounceRun { return Poll::Pending; }; - let Some(update) = next else { - // The buffer is empty (flushed at the loop top), so FIN now and - // wait for the acknowledgement. - stream.writer.finish()?; - self.phase = AnnouncePhase::Closing; - continue; + let (update, active) = match next { + Some(announce::Event::Announced(update) | announce::Event::Updated(update)) => (update, true), + Some(announce::Event::Retracted(update)) => (update, false), + Some(announce::Event::Live) => continue, + None => { + // The buffer is empty (flushed at the loop top), so FIN now and + // wait for the acknowledgement. + stream.writer.finish()?; + self.phase = AnnouncePhase::Closing; + continue; + } }; let absolute = origin.absolute(&update.prefix); let suffix = self.suffix(&update); - if !update.kind.is_active() { + if !active { self.retract(stream, suffix, &absolute)?; continue; } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 729a23e4bc..93d22504da 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -1060,6 +1060,20 @@ struct AnnouncePrefix { subscriber: Subscriber, prefix: PathOwned, state: PrefixState, + /// Held from the request until the peer's initial set has landed in the + /// origin, so announce cursors know when they have caught up. Taken at + /// construction, before the session is handed out, so no cursor misses it. + replaying: Option, +} + +/// How the peer's initial set ends on this version. +enum Landing { + /// Lite05+: after this many more announces, the count in ANNOUNCE_OK. + Count(u64), + /// Lite03/04 carry no boundary: once the stream goes quiet. + Quiet(crate::model::Quiet), + /// It has landed. Lite01/02 land it in ANNOUNCE_INIT, before the run. + Landed, } enum PrefixState { @@ -1073,6 +1087,8 @@ enum PrefixState { Cost { stream: Stream, responder_origin: Option, + /// Lite05+: the initial set's size, from ANNOUNCE_OK. + active: Option, }, /// Lite01/02: reading the ANNOUNCE_INIT set. ReadInit { stream: Stream, run: PrefixRun }, @@ -1095,17 +1111,32 @@ struct PrefixRun { // but a peer may. next_announce_id: u64, announced_by_id: HashMap, + landing: Landing, } impl AnnouncePrefix { fn new(subscriber: Subscriber, prefix: PathOwned) -> Self { Self { + replaying: Some(subscriber.origin.replaying(&prefix)), subscriber, prefix, state: PrefixState::Open, } } + /// Drop the guard once the initial set has landed. + fn poll_landing(replaying: &mut Option, landing: &mut Landing, waiter: &kio::Waiter) { + let landed = match landing { + Landing::Count(remaining) => *remaining == 0, + Landing::Quiet(quiet) => quiet.poll(waiter).is_ready(), + Landing::Landed => return, + }; + if landed { + *landing = Landing::Landed; + *replaying = None; + } + } + fn poll(&mut self, waiter: &kio::Waiter) -> Poll> { let mut cx = std::task::Context::from_waker(waiter.waker()); loop { @@ -1140,16 +1171,14 @@ impl AnnouncePrefix { false => PrefixState::Cost { stream, responder_origin: None, + active: None, }, }; } PrefixState::ReadOk { stream } => { // Lite05+: the publisher reports its own origin id, which we stamp onto // every received Announce's hop chain since it no longer does so itself. - // Its `active` count marks where the initial set ends; nothing here needs - // that boundary, so it is read and dropped. Callers that must not race an - // announcement use `origin::Consumer::announced_broadcast`, which waits - // for the path itself. + // Its `active` count marks where the initial set ends. let ok = ready!(stream.reader.poll_decode::(&mut cx))?; // A peer may legally report id 0 (no identity). Keep it: the assigned // identity stays on `via` and is never forwarded as a hop. @@ -1160,6 +1189,7 @@ impl AnnouncePrefix { self.state = PrefixState::Cost { stream, responder_origin: Some(origin), + active: Some(ok.active), }; } PrefixState::Cost { .. } => { @@ -1167,17 +1197,24 @@ impl AnnouncePrefix { let PrefixState::Cost { stream, responder_origin, + active, } = std::mem::replace(&mut self.state, PrefixState::Open) else { unreachable!() }; + let landing = match (active, self.subscriber.version) { + (Some(active), _) => Landing::Count(active), + (None, Version::Lite01 | Version::Lite02) => Landing::Landed, + (None, _) => Landing::Quiet(crate::model::Quiet::new(&self.subscriber.runtime)), + }; let run = PrefixRun { responder_origin, link_cost, announced: Announced::default(), next_announce_id: 0, announced_by_id: HashMap::new(), + landing, }; // Lite01/02 send the initial set as one ANNOUNCE_INIT message, so they @@ -1208,6 +1245,7 @@ impl AnnouncePrefix { else { unreachable!() }; + self.replaying = None; self.state = PrefixState::Run { stream, run }; } PrefixState::Run { stream, run } => { @@ -1225,9 +1263,19 @@ impl AnnouncePrefix { // request that arrives in between: its wake finds no waiter, and // nothing else re-polls this machine. loop { + // Land before decoding past the boundary, so no live update + // enters the origin ahead of the marker. + Self::poll_landing(&mut self.replaying, &mut run.landing, waiter); match stream.reader.poll_decode_maybe::(&mut cx) { Poll::Ready(Ok(Some(announce))) => { + // The count is of ANNOUNCE_STARTs, not every message. + let start = matches!(announce, lite::AnnounceBroadcast::Active { .. }); self.subscriber.handle_announce(&self.prefix, announce, run)?; + match &mut run.landing { + Landing::Count(remaining) if start => *remaining = remaining.saturating_sub(1), + Landing::Quiet(quiet) => quiet.heard(), + Landing::Count(_) | Landing::Landed => {} + } } Poll::Ready(Ok(None)) => { // The publisher FINed: it has nothing (more) to announce for this @@ -2413,6 +2461,54 @@ mod tests { announced.retire(&path.clone()); cursor.assert_next_ended("room/host"); } + + /// ANNOUNCE_OK's count lands the initial set at exactly that many + /// ANNOUNCE_STARTs: an unknown message does not count toward it, and a live + /// update buffered right behind the set does not enter the origin before the + /// marker. + #[tokio::test(start_paused = true)] + async fn the_count_lands_at_the_last_initial_start() { + const VERSION: Version = Version::Lite06; + let start = |suffix| lite::AnnounceBroadcast::Active { + suffix: Path::new(suffix), + hops: crate::Hops::new(), + cost: crate::origin::Cost::default(), + }; + let mut script = Vec::new(); + lite::AnnounceOk { + origin: crate::Hop::new(9).unwrap(), + active: 1, + } + .encode(&mut script, VERSION) + .unwrap(); + // An unknown announce type with an empty body, which decodes as `Skipped`. + script.extend([0x3f, 0x00]); + start("a").encode(&mut script, VERSION).unwrap(); + start("b").encode(&mut script, VERSION).unwrap(); + + let origin = origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let consumer = origin.consume(); + let subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::Clock::tokio(), + session: crate::lite::test_transport::ScriptedSession::new(script), + origin, + recv_bandwidth: None, + version: VERSION, + peer_setup: Default::default(), + cost: Some(1), + peer_hop: None, + going_away: Default::default(), + }); + let mut prefix = AnnouncePrefix::new(subscriber, Path::new("").to_owned()); + let mut cursor = consumer.announced(); + + let mut run = std::pin::pin!(kio::wait(|waiter| prefix.poll(waiter))); + assert!(futures::poll!(run.as_mut()).is_pending()); + + cursor.assert_next_active("a"); + cursor.assert_next_live(); + cursor.assert_next_active("b"); + } } /// The four wire fields a subscription's half-open range encodes to. diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 3b259d1540..8f0a7c7f39 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -43,7 +43,7 @@ pub mod origin { /// Subscribing to route (un)announcements from an origin. pub mod announce { - pub use super::origin_impl::{AnnounceConsumer as Consumer, AnnounceKind as Kind, AnnounceUpdate as Update}; + pub use super::origin_impl::{Announce, AnnounceConsumer as Consumer, AnnounceEvent as Event}; } // Hop identity and the `Consume` conversion trait aren't part of a role @@ -53,6 +53,9 @@ pub use origin_impl::{Consume, Hop, Hops, InvalidHop}; // The announce-interest prefixes a scope needs on a prefix-shaped wire. pub(crate) use origin_impl::interest_prefixes; +// Held by a session until the peer's initial announce set has landed. +pub(crate) use origin_impl::{Quiet, Replaying}; + // The advertise-only route guard, for tests shaping the route table. #[cfg(test)] pub(crate) use origin_impl::AnnounceProducer; diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 2f7ab0f77b..89aaca5096 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -602,9 +602,59 @@ struct OriginConsumerState { /// Set by the origin's teardown: the cursor drains `pending`, then reports /// the end instead of parking forever on a table that can never fire again. ended: bool, + /// Progress toward the one [`AnnounceEvent::Live`] marker. + live: LiveState, +} + +/// Where a cursor stands on delivering its initial set. +enum LiveState { + /// This many remote sources pending at subscribe time are still replaying. + Waiting(usize), + /// Every source has landed: the marker follows once these prefixes, pending + /// when the last one landed, have been delivered (or cancelled). + Owed(BTreeSet), + /// The marker was delivered. + Done, +} + +impl Default for LiveState { + fn default() -> Self { + Self::Waiting(0) + } } impl OriginConsumerState { + /// One source the cursor was waiting on has landed its initial set. + fn landed(&mut self) { + if let LiveState::Waiting(count) = &mut self.live { + *count = count.saturating_sub(1); + if *count == 0 { + self.live = LiveState::Owed(self.pending.keys().cloned().collect()); + } + } + } + + /// `prefix` left the pending set (delivered or cancelled), so the marker no + /// longer waits on it. A prefix still pending (the announce trailing a + /// retraction) stays owed. + fn settled(&mut self, prefix: &PathOwned) { + if let LiveState::Owed(owed) = &mut self.live + && !self.pending.contains_key(prefix) + { + owed.remove(prefix); + } + } + + /// Whether the marker is due now. + fn live_due(&self) -> bool { + matches!(&self.live, LiveState::Owed(owed) if owed.is_empty()) + } + + /// Whether [`Self::take`] has something to hand out, or the cursor ended. + fn ready(&self) -> bool { + !self.pending.is_empty() || self.ended || self.live_due() + } + fn apply_announce(&mut self, prefix: PathOwned, meta: RouteMeta, captures: Option>) { let meta = (meta, captures); let new = match self.pending.remove(&prefix) { @@ -623,7 +673,7 @@ impl OriginConsumerState { match self.pending.remove(&prefix) { // The pending announce was never delivered and neither was any earlier // one, so the pair cancels entirely. - Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => {} + Some(PendingUpdate::Announce(_)) if !self.delivered.contains(&prefix) => self.settled(&prefix), // Either nothing is pending or the pending announce was a metadata // update on a delivered route; the consumer still owes a retraction. None | Some(PendingUpdate::Announce(_) | PendingUpdate::Unannounce(_)) => { @@ -637,31 +687,37 @@ impl OriginConsumerState { } } - /// Take one update to deliver to the consumer, if any. - fn take(&mut self) -> Option { + /// Take one event to deliver to the consumer, if any: the marker as soon as + /// it is due, otherwise the next update. + fn take(&mut self) -> Option { + if self.live_due() { + self.live = LiveState::Done; + return Some(AnnounceEvent::Live); + } let prefix = self.pending.keys().next()?.clone(); - let ((meta, captures), kind) = match self.pending.remove(&prefix).unwrap() { + let ((meta, captures), kind): (_, fn(Announce) -> AnnounceEvent) = match self.pending.remove(&prefix).unwrap() { PendingUpdate::Announce(meta) => { // The consumer has seen this prefix before, so it is a metadata update. let kind = match self.delivered.insert(prefix.clone()) { - true => AnnounceKind::Announced, - false => AnnounceKind::Updated, + true => AnnounceEvent::Announced, + false => AnnounceEvent::Updated, }; (meta, kind) } PendingUpdate::Unannounce(meta) => { self.delivered.remove(&prefix); - (meta, AnnounceKind::Retracted) + (meta, AnnounceEvent::Retracted) } PendingUpdate::UnannounceAnnounce { old, new } => { // Deliver the retraction now; leave the trailing announce pending so // the next take returns it for the same prefix. self.delivered.remove(&prefix); self.pending.insert(prefix.clone(), PendingUpdate::Announce(new)); - (old, AnnounceKind::Retracted) + (old, AnnounceEvent::Retracted) } }; - Some(AnnounceUpdate { + self.settled(&prefix); + Some(kind(Announce { prefix, captures, route: Route { @@ -669,8 +725,7 @@ impl OriginConsumerState { cost: meta.1, via: Hop::UNKNOWN, }, - kind, - }) + })) } } @@ -938,25 +993,22 @@ pub(crate) fn interest_prefixes(allowed: &Patterns) -> Vec { heads } -/// What an [`AnnounceUpdate`] reports about its path. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum AnnounceKind { - /// A route now covers the path; the cursor had none there. - Announced, - /// The route covering the path changed hops or cost; it is delivered in place. - Updated, - /// No route covers the path any more. - Retracted, -} - -impl AnnounceKind { - /// Whether a route covers the path after this update. - pub fn is_active(self) -> bool { - !matches!(self, Self::Retracted) - } +/// What an [`AnnounceConsumer`] yields. +#[derive(Clone, Debug)] +pub enum AnnounceEvent { + /// A route now covers the prefix; the cursor had none there. + Announced(Announce), + /// The route covering the prefix changed hops or cost; it is delivered in place. + Updated(Announce), + /// No route covers the prefix any more. Carries its last advertised route. + Retracted(Announce), + /// Every route live at subscribe time has been delivered; what follows is + /// live changes. Yielded once, and only after each remote source feeding the + /// origin at subscribe time has landed its initial set. + Live, } -/// A route announcement, update, or retraction, delivered by [`AnnounceConsumer`]. +/// A route over a prefix, delivered by [`AnnounceConsumer`] inside an [`AnnounceEvent`]. /// /// An announcement is always a prefix, never a broadcast: it advertises that /// [`prefix`](Self::prefix) and every path beneath it are servable. A broadcast @@ -964,7 +1016,7 @@ impl AnnounceKind { /// [`Consumer::request_broadcast`]; the application decides which paths name /// broadcasts, and filters with a [`Pattern`] locally when it wants a subset. #[derive(Clone, Debug)] -pub struct AnnounceUpdate { +pub struct Announce { /// The prefix the route covers, relative to the consuming cursor's root. pub prefix: PathOwned, /// What the scope's wildcards stood for when the announced prefix pins all of @@ -973,8 +1025,6 @@ pub struct AnnounceUpdate { /// The route serving the prefix. On a retraction this carries its last /// advertised metadata. pub route: Route, - /// Whether the prefix was announced, re-priced, or retracted. - pub kind: AnnounceKind, } /// Publishes broadcasts and announces routes into an origin. @@ -1236,6 +1286,35 @@ impl Producer { ) } + /// Register a remote announce source replaying its initial set under + /// `prefix`, relative to this producer's root. + /// + /// Every announce cursor overlapping that prefix and registered while the + /// guard lives withholds its [`AnnounceEvent::Live`] marker until the guard + /// drops, so a session drops it once the peer's initial routes are in the + /// table (or the session dies). + pub(crate) fn replaying(&self, prefix: impl AsPath) -> Replaying { + // A subtree too complex to intersect waits on the whole scope instead. + let scope = Pattern::subtree(self.absolute(prefix).as_str()) + .ok() + .and_then(|subtree| self.scope.allowed.intersect(&subtree.into()).ok()) + .unwrap_or_else(|| self.scope.allowed.clone()); + let mut state = self.shared.lock(); + let id = state.next_replaying; + state.next_replaying += 1; + state.replaying.insert( + id, + ReplayingSource { + scope, + waiting: Vec::new(), + }, + ); + Replaying { + shared: self.shared.clone(), + id, + } + } + /// Advertise a route over `prefix` and serve the requests beneath it. /// /// A route is always a prefix: it claims `prefix` and every path beneath it @@ -2533,11 +2612,24 @@ struct OriginState { // a front dies with its watcher and a later request re-creates it. fronts: WeakCache, + // Remote announce sources still replaying their initial set, keyed by + // [`Replaying`] id. Empty outside a session's first round trip. + replaying: HashMap, + next_replaying: u64, + // Set when the origin's driver dropped: new requests fail with `Closed` // immediately and handlers observe the end instead of parking forever. closed: bool, } +/// A remote announce source still replaying its initial set. +struct ReplayingSource { + /// The absolute patterns the source may announce under. + scope: Patterns, + /// The cursors registered while it replayed, each owed its landing. + waiting: Vec, +} + impl OriginState { /// Re-deliver the best route at every presented prefix `prefix` maps to, on /// every cursor it can present on. Called after an entry covering `prefix` @@ -2659,6 +2751,19 @@ impl OriginState { for p in &presented { Self::sync_cursor(&self.routes, &mut cursor, p); } + // The marker also waits on every source still replaying into this + // cursor's scope; with none, it follows the replay above. + let mut waiting = 0; + for source in self.replaying.values_mut() { + if source.scope.iter().any(|pattern| cursor.allowed.overlaps(pattern)) { + source.waiting.push(id); + waiting += 1; + } + } + if let Ok(mut state) = cursor.state.write() { + state.live = LiveState::Waiting(waiting + 1); + state.landed(); + } for head in &cursor.heads { self.routes.add_cursor(head, id); } @@ -2702,6 +2807,66 @@ impl OriginState { } } +/// A remote announce source still replaying its initial set, from +/// [`Producer::replaying`]. Drop it once the set has landed. +pub(crate) struct Replaying { + shared: kio::Shared, + id: u64, +} + +impl Drop for Replaying { + fn drop(&mut self) { + let mut state = self.shared.lock(); + let Some(source) = state.replaying.remove(&self.id) else { + return; + }; + for id in source.waiting { + // A cursor dropped since registering is simply gone. + if let Some(cursor) = state.cursors.get(&id) + && let Ok(mut cursor) = cursor.state.write() + { + cursor.landed(); + } + } + } +} + +/// 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. +/// +/// 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. +pub(crate) struct Quiet { + deadline: crate::runtime::Deadline, + clock: Clock, +} + +impl Quiet { + /// How long the stream may stay silent before its first announcement. + const FIRST: Duration = Duration::from_millis(500); + /// How long the stream may stay silent between announcements. + const GAP: Duration = Duration::from_millis(30); + + /// Start counting from now. + pub(crate) fn new(clock: &Clock) -> Self { + Self { + deadline: crate::runtime::Deadline::after(clock, Self::FIRST), + clock: clock.clone(), + } + } + + /// An announcement arrived: the set is still landing. + pub(crate) fn heard(&mut self) { + let now = crate::runtime::Timers::now(&self.clock); + self.deadline.set(now.checked_add(Self::GAP)); + } + + /// Ready once the stream has gone quiet. + pub(crate) fn poll(&mut self, waiter: &kio::Waiter) -> Poll<()> { + self.deadline.poll(waiter) + } +} + /// One-shot result of a dynamic broadcast request. /// /// Stays `None` until a handler [`accept`](Request::accept)s (yielding the served @@ -3217,9 +3382,10 @@ impl Consumer { // forwarding, so it must not drive the announce guards. let mut announced = consumer.untagged().announced(); loop { - let update = announced.next().await?; - if update.kind.is_active() && path.has_prefix(&update.prefix) { - return Some(update.route); + if let AnnounceEvent::Announced(announce) | AnnounceEvent::Updated(announce) = announced.next().await? + && path.has_prefix(&announce.prefix) + { + return Some(announce.route); } } } @@ -3481,50 +3647,52 @@ impl AnnounceConsumer { } /// Drive the egress announce guards for one update. - fn hand_out(&mut self, update: AnnounceUpdate) -> AnnounceUpdate { - let absolute = self.root.join(&update.prefix).to_owned(); - if update.kind.is_active() { - let scope = self.stats.egress(&absolute); - self.guards - .entry(update.prefix.clone()) - .or_insert_with(|| scope.announce()); - } else { - self.guards.remove(&update.prefix); + fn hand_out(&mut self, event: AnnounceEvent) -> AnnounceEvent { + match &event { + AnnounceEvent::Announced(announce) | AnnounceEvent::Updated(announce) => { + let scope = self.stats.egress(self.root.join(&announce.prefix)); + self.guards + .entry(announce.prefix.clone()) + .or_insert_with(|| scope.announce()); + } + AnnounceEvent::Retracted(announce) => { + self.guards.remove(&announce.prefix); + } + AnnounceEvent::Live => {} } - update + event } /// Returns the next route announcement, update, or retraction, its prefix - /// relative to this cursor's root. + /// relative to this cursor's root, or the one [`AnnounceEvent::Live`] marker. /// - /// A retraction is only delivered for a previously announced prefix, and a + /// The routes live at subscribe time come first, then the marker, then live + /// changes, so a caller listing what is live stops at the marker. A + /// retraction is only delivered for a previously announced prefix, and a /// repeated announcement for the same prefix is a metadata update. Returns /// None if the cursor is closed. The consumer is also a [`futures::Stream`] - /// of the same updates. - pub async fn next(&mut self) -> Option { + /// of the same events. + pub async fn next(&mut self) -> Option { kio::wait(|waiter| self.poll_next(waiter)).await } - /// Poll for the next update, without blocking. + /// Poll for the next event, without blocking. /// - /// Returns `Poll::Ready(Some(_))` for an update, `Poll::Ready(None)` if the + /// Returns `Poll::Ready(Some(_))` for an event, `Poll::Ready(None)` if the /// cursor is closed, or `Poll::Pending` after registering `waiter` to be - /// notified when the next update arrives. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { - let update = { - let mut state = match ready!(self.state.poll(waiter, |state| { - if state.pending.is_empty() && !state.ended { - Poll::Pending - } else { - Poll::Ready(()) - } + /// notified when the next event arrives. + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll> { + let event = { + let mut state = match ready!(self.state.poll(waiter, |state| match state.ready() { + true => Poll::Ready(()), + false => Poll::Pending, })) { Ok(state) => state, // Closed: discard the Ref so its MutexGuard doesn't escape this call. Err(_) => return Poll::Ready(None), }; match state.take() { - Some(update) => update, + Some(event) => event, None => { // Ended by the origin's teardown, pending updates already // drained; close the channel so every closure signal agrees. @@ -3533,16 +3701,16 @@ impl AnnounceConsumer { } } }; - Poll::Ready(Some(self.hand_out(update))) + Poll::Ready(Some(self.hand_out(event))) } - /// Returns the next update without blocking. + /// Returns the next event without blocking. /// - /// Returns None if there is no update available; NOT because the cursor is closed. + /// Returns None if there is no event available; NOT because the cursor is closed. /// Use [`Self::is_closed`] to check if the cursor is closed. - pub fn try_next(&mut self) -> Option { - let update = self.state.write().ok()?.take()?; - Some(self.hand_out(update)) + pub fn try_next(&mut self) -> Option { + let event = self.state.write().ok()?.take()?; + Some(self.hand_out(event)) } /// Returns true if the cursor is closed (no more updates will arrive). @@ -3563,7 +3731,7 @@ impl AnnounceConsumer { } impl futures::Stream for AnnounceConsumer { - type Item = AnnounceUpdate; + type Item = AnnounceEvent; fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll> { let this = self.get_mut(); @@ -3589,35 +3757,70 @@ use futures::FutureExt; #[cfg(test)] #[allow(missing_docs)] // test-only assertion helpers impl AnnounceConsumer { + /// The next route event without blocking, skipping the + /// [`AnnounceEvent::Live`] marker: these helpers pin routes, and the marker + /// has its own tests. + fn next_route_now(&mut self) -> Option> { + loop { + match self.next().now_or_never()? { + Some(AnnounceEvent::Live) => continue, + event => return Some(event), + } + } + } + + /// The `try_next` counterpart of [`Self::next_route_now`]. + fn try_next_route(&mut self) -> Option { + loop { + match self.try_next()? { + AnnounceEvent::Live => continue, + event => return Some(event), + } + } + } + + /// The route of an active event at `expected`. + fn active(event: AnnounceEvent, expected: &Path) -> Route { + match event { + AnnounceEvent::Announced(announce) | AnnounceEvent::Updated(announce) => { + assert_eq!(announce.prefix, *expected, "wrong prefix"); + announce.route + } + other => panic!("should be an active route: got {other:?}"), + } + } + /// The next update must be an active route at `expected`; returns it. pub fn assert_next_active(&mut self, expected: impl AsPath) -> Route { - let expected = expected.as_path(); - let update = self.next().now_or_never().expect("next blocked").expect("no next"); - assert_eq!(update.prefix, expected, "wrong prefix"); - assert!(update.kind.is_active(), "should be an active route"); - update.route + let event = self.next_route_now().expect("next blocked").expect("no next"); + Self::active(event, &expected.as_path()) } /// The `try_next` counterpart of [`Self::assert_next_active`]. pub fn assert_try_next_active(&mut self, expected: impl AsPath) -> Route { - let expected = expected.as_path(); - let update = self.try_next().expect("no next"); - assert_eq!(update.prefix, expected, "wrong prefix"); - assert!(update.kind.is_active(), "should be an active route"); - update.route + let event = self.try_next_route().expect("no next"); + Self::active(event, &expected.as_path()) } /// The next update must be a retraction at `expected`. pub fn assert_next_ended(&mut self, expected: impl AsPath) { - let expected = expected.as_path(); - let update = self.next().now_or_never().expect("next blocked").expect("no next"); - assert_eq!(update.prefix, expected, "wrong prefix"); - assert_eq!(update.kind, AnnounceKind::Retracted, "should be a retraction"); + match self.next_route_now().expect("next blocked").expect("no next") { + AnnounceEvent::Retracted(announce) => assert_eq!(announce.prefix, expected.as_path(), "wrong prefix"), + other => panic!("should be a retraction: got {other:?}"), + } } pub fn assert_next_wait(&mut self) { - if let Some(res) = self.next().now_or_never() { - panic!("next should block: got {:?}", res.map(|u| u.prefix)); + if let Some(event) = self.next_route_now() { + panic!("next should block: got {event:?}"); + } + } + + /// The next event must be the [`AnnounceEvent::Live`] marker. + pub fn assert_next_live(&mut self) { + match self.next().now_or_never() { + Some(Some(AnnounceEvent::Live)) => {} + other => panic!("expected the live marker: got {other:?}"), } } } @@ -3751,6 +3954,88 @@ mod tests { announced.assert_next_wait(); } + #[tokio::test] + async fn an_empty_origin_is_live_at_once() { + let producer = origin(1).produce(); + let mut announced = producer.consume().announced(); + announced.assert_next_live(); + announced.assert_next_wait(); + + // Later routes are live changes; the marker never repeats. + let _alice = producer.announce("alice", Route::default()).unwrap(); + announced.assert_next_active("alice"); + assert!(announced.next().now_or_never().is_none()); + } + + #[tokio::test] + async fn live_follows_the_replayed_routes() { + let producer = origin(1).produce(); + let _b = producer.announce("b", Route::default()).unwrap(); + let _c = producer.announce("c", Route::default()).unwrap(); + let mut announced = producer.consume().announced(); + + // A route arriving before the replay drains may come either side of the + // marker, but never pushes it past the replay. + let _a = producer.announce("a", Route::default()).unwrap(); + let mut seen = Vec::new(); + loop { + match announced.next().now_or_never().expect("blocked").expect("closed") { + AnnounceEvent::Live => break, + AnnounceEvent::Announced(announce) => seen.push(announce.prefix.to_string()), + other => panic!("only announcements before the marker: got {other:?}"), + } + } + assert!( + seen.contains(&"b".to_string()) && seen.contains(&"c".to_string()), + "{seen:?}" + ); + } + + #[tokio::test] + async fn a_replayed_route_retracted_before_delivery_does_not_hold_live() { + let producer = origin(1).produce(); + let alice = producer.announce("alice", Route::default()).unwrap(); + let mut announced = producer.consume().announced(); + drop(alice); + announced.assert_next_live(); + announced.assert_next_wait(); + } + + #[tokio::test] + async fn a_replaying_source_withholds_live_until_it_lands() { + let producer = origin(1).produce(); + let room = producer.scope("", &scopes(&["room"])).unwrap(); + let replaying = room.replaying("room/alice"); + + let mut announced = producer.consume().announced(); + // Out of the source's scope: nothing to wait for. + let mut elsewhere = producer.consume().scope("", &scopes(&["other"])).unwrap().announced(); + elsewhere.assert_next_live(); + // In its scope but beside the prefix it replays: nothing to wait for either. + let mut bob = producer + .consume() + .scope("", &scopes(&["room/bob"])) + .unwrap() + .announced(); + bob.assert_next_live(); + + // Routes the source lands arrive before the marker. + let _alice = room.announce("room/alice", Route::default()).unwrap(); + announced.assert_next_active("room/alice"); + assert!( + announced.next().now_or_never().is_none(), + "live before the source landed" + ); + + drop(replaying); + announced.assert_next_live(); + + // A cursor registered after the source landed does not wait on it. + let mut later = producer.consume().announced(); + later.assert_next_active("room/alice"); + later.assert_next_live(); + } + #[tokio::test] async fn broadcast_announces_its_own_path() { let producer = origin(1).produce(); @@ -4113,19 +4398,22 @@ mod tests { .unwrap(); let mut announced = consumer.announced(); - let first = announced.next().now_or_never().expect("next").expect("announce"); + let Some(Some(AnnounceEvent::Announced(first))) = announced.next_route_now() else { + panic!("expected the announcement"); + }; assert_eq!(first.prefix.as_str(), ""); - assert_eq!(first.kind, AnnounceKind::Announced); assert_eq!(first.captures, Some(Vec::new())); drop(exact); - let retracted = announced.next().now_or_never().expect("next").expect("retract"); + let Some(Some(AnnounceEvent::Retracted(retracted))) = announced.next_route_now() else { + panic!("expected the retraction"); + }; assert_eq!(retracted.prefix.as_str(), ""); - assert_eq!(retracted.kind, AnnounceKind::Retracted); assert_eq!(retracted.captures, Some(Vec::new())); - let replacement = announced.next().now_or_never().expect("next").expect("announce"); + let Some(Some(AnnounceEvent::Announced(replacement))) = announced.next_route_now() else { + panic!("expected the replacement"); + }; assert_eq!(replacement.prefix.as_str(), ""); - assert_eq!(replacement.kind, AnnounceKind::Announced); assert_eq!(replacement.captures, None); } @@ -4183,16 +4471,18 @@ mod tests { let second = producer.dynamic("live", Route::default().with_cost(1)).unwrap(); let mut announced = producer.consume().announced(); - let update = announced.next().now_or_never().expect("next").expect("no next"); + let Some(Some(AnnounceEvent::Announced(update))) = announced.next_route_now() else { + panic!("expected the announcement"); + }; assert_eq!(update.prefix.as_str(), "live"); - assert_eq!(update.kind, AnnounceKind::Announced); assert_eq!(update.route.cost, Cost::new(1)); announced.assert_next_wait(); drop(second); - let update = announced.next().now_or_never().expect("next").expect("no next"); + let Some(Some(AnnounceEvent::Updated(update))) = announced.next_route_now() else { + panic!("expected the reprice"); + }; assert_eq!(update.prefix.as_str(), "live"); - assert_eq!(update.kind, AnnounceKind::Updated); assert_eq!(update.route.cost, Cost::new(3)); drop(first); @@ -4288,19 +4578,15 @@ mod tests { let producer = origin(1).produce(); let server = producer.dynamic("live", Route::default()).unwrap(); let mut announced = producer.consume().announced(); - let update = StreamExt::next(&mut announced) - .now_or_never() - .expect("next") - .expect("no next"); + let mut next = || StreamExt::next(&mut announced).now_or_never(); + let Some(Some(AnnounceEvent::Announced(update))) = next() else { + panic!("expected the replayed route"); + }; assert_eq!(update.prefix.as_str(), "live"); - assert_eq!(update.kind, AnnounceKind::Announced); - assert!(StreamExt::next(&mut announced).now_or_never().is_none()); + assert!(matches!(next(), Some(Some(AnnounceEvent::Live)))); + assert!(next().is_none()); drop(server); - let update = StreamExt::next(&mut announced) - .now_or_never() - .expect("next") - .expect("no next"); - assert_eq!(update.kind, AnnounceKind::Retracted); + assert!(matches!(next(), Some(Some(AnnounceEvent::Retracted(_))))); } #[tokio::test] @@ -5703,7 +5989,9 @@ mod tests { let alice = producer.create_broadcast("room/alice/chat").unwrap(); alice.announce(Route::default()).unwrap(); - let update = announced.try_next().expect("alice's chat"); + let Some(AnnounceEvent::Announced(update)) = announced.try_next_route() else { + panic!("expected alice's chat"); + }; assert_eq!(update.prefix.as_str(), "room/alice/chat"); assert_eq!(update.captures, Some(vec!["alice".parse::().unwrap()])); @@ -5712,7 +6000,9 @@ mod tests { announced.assert_next_wait(); let broad = producer.announce("room", Route::default()).unwrap(); - let update = announced.try_next().expect("overlapping broad route"); + let Some(AnnounceEvent::Announced(update)) = announced.try_next_route() else { + panic!("expected the overlapping broad route"); + }; assert_eq!(update.prefix.as_str(), "room"); assert_eq!(update.captures, None, "an overlap does not pin the wildcard"); @@ -5729,7 +6019,9 @@ mod tests { local.announce(Route::default()).unwrap(); let mut announced = producer.consume().announced(); - let update = announced.try_next().expect("one winning route"); + let Some(AnnounceEvent::Announced(update)) = announced.try_next_route() else { + panic!("expected one winning route"); + }; assert_eq!(update.prefix.as_str(), "room/alice"); assert_eq!(update.route.cost, Cost::default()); announced.assert_next_wait(); diff --git a/rs/moq-net/tests/caught_up.rs b/rs/moq-net/tests/caught_up.rs new file mode 100644 index 0000000000..bbf55344c0 --- /dev/null +++ b/rs/moq-net/tests/caught_up.rs @@ -0,0 +1,76 @@ +//! An announce cursor on a session's origin yields `Live` once the peer's +//! initial set has landed, on every version. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Version, announce, origin}; +use support::harness::{MockConnectOptions, connect_mock}; + +fn produce_origin(hop: Hop) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(hop)); + tokio::spawn(support::harness::run(driver)); + 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 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"]; + +/// 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 { + let version: Version = version.parse().unwrap(); + let published = produce_origin(Hop::new(1).unwrap()); + let broadcasts: Vec<_> = paths + .iter() + .map(|path| { + let broadcast = published.create_broadcast(*path).unwrap(); + broadcast.announce(Default::default()).unwrap(); + broadcast + }) + .collect(); + + let subscribed = produce_origin(Hop::new(2).unwrap()); + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(published.clone()); + options.client_subscribe = Some(subscribed.clone()); + let _pair = connect_mock(options).await; + + let mut announced = subscribed.consume().announced(); + let mut live = Vec::new(); + let read = async { + loop { + match announced.next().await.expect("cursor closed") { + announce::Event::Live => break, + announce::Event::Announced(update) => live.push(update.prefix.to_string()), + other => panic!("{version}: only announcements before the marker: got {other:?}"), + } + } + }; + tokio::time::timeout(Duration::from_secs(10), read) + .await + .unwrap_or_else(|_| panic!("{version}: never caught up")); + + drop(broadcasts); + live.sort(); + live +} + +#[tokio::test] +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}"); + } +} + +#[tokio::test] +async fn an_empty_peer_is_caught_up() { + for version in COUNTED.iter().chain(QUIET) { + assert!(caught_up(version, &[]).await.is_empty(), "{version}"); + } +} diff --git a/rs/moq-net/tests/goaway.rs b/rs/moq-net/tests/goaway.rs index 2eefac0f0a..69e8c3198e 100644 --- a/rs/moq-net/tests/goaway.rs +++ b/rs/moq-net/tests/goaway.rs @@ -443,11 +443,12 @@ async fn goaway_drains_routes(version: Version) { // re-prices the route in place, which arrives as another active update. let mut announced = sub.announced(); loop { - let update = announced.next().await.expect("update"); - if update.kind.is_active() - && update.prefix.as_str() == "test" - && update.route.cost == moq_net::origin::Cost::DRAIN - { + let (moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) = + announced.next().await.expect("update") + else { + continue; + }; + if update.prefix.as_str() == "test" && update.route.cost == moq_net::origin::Cost::DRAIN { break; } } diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index e274431079..4bab2fd09c 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -1585,7 +1585,12 @@ impl Cluster { loop { tokio::select! { ann = announced.next() => { - let Some(update) = ann else { return; }; + let (update, active) = match ann { + Some(moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) => (update, true), + Some(moq_net::announce::Event::Retracted(update)) => (update, false), + Some(moq_net::announce::Event::Live) => continue, + None => return, + }; let relative = update.prefix; // The address to dial, which keeps its query: `run_remote` reads // `?cost=` and `?jwt=` off it. The key is only its identity. @@ -1603,7 +1608,7 @@ impl Cluster { continue; } let advertisement = relative.as_str().to_owned(); - match update.kind.is_active() { + match active { true => { let target = live.announce(advertisement, target); let mut spawn = |target: DialTarget| { @@ -2217,6 +2222,19 @@ mod tests { use super::*; use crate::Config as RelayConfig; + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + fn new_cluster(config: Config) -> anyhow::Result { Cluster::new(Options::new(config)) } @@ -2951,11 +2969,11 @@ mod tests { .announced(); let registration = origin.create_broadcast(&path).expect("node advertise"); registration.announce(Default::default()).expect("announce node"); - let update = tokio::time::timeout(Duration::from_secs(2), announced.next()) + let (_, active) = tokio::time::timeout(Duration::from_secs(2), next_update(&mut announced)) .await .expect("node advertised") .expect("announce"); - assert!(update.kind.is_active()); + assert!(active); let snapshot = cluster.nodes.snapshot(); assert!( snapshot.nodes.iter().any(|node| node.node.contains("peer.example")), @@ -3008,9 +3026,12 @@ mod tests { tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; // The self-registration route must be visible on the origin. - let update = watcher.try_next().expect("self-registration must be published"); + // The watcher subscribed to an empty origin, so its marker comes first. + assert!(matches!(watcher.try_next(), Some(moq_net::announce::Event::Live))); + let Some(moq_net::announce::Event::Announced(update)) = watcher.try_next() else { + panic!("self-registration must be published"); + }; assert_eq!(update.prefix.as_str(), ".internal/origins/rendezvous.example.com:4443"); - assert!(update.kind.is_active()); // run() must NOT have returned: dropping the broadcast (via run returning) // would unannounce the registration immediately. Use a short timeout to @@ -3589,7 +3610,7 @@ mod tests { let _dial = fingerprint.dial_lan_target(&target).expect("dial"); let mut announced = fingerprint.origin.consume().announced(); - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, _) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("timed out waiting for from-node") .expect("origin closed"); @@ -3599,7 +3620,7 @@ mod tests { _from_fp.announce(Default::default()).expect("announce"); let mut announced = node.origin.consume().announced(); loop { - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, _) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("timed out waiting for from-fingerprint") .expect("origin closed"); diff --git a/rs/moq-relay/src/internal.rs b/rs/moq-relay/src/internal.rs index c3e576a8a8..514dbae25c 100644 --- a/rs/moq-relay/src/internal.rs +++ b/rs/moq-relay/src/internal.rs @@ -626,6 +626,19 @@ fn render_uring(_out: &mut String, _workers: &[UringWorker]) {} mod tests { use super::*; + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + /// An `Config` whose listener is enabled, so `Internal` registers its own. fn listening() -> Config { Config { @@ -919,8 +932,8 @@ mod tests { // Leave 46 bytes across two frames behind the live edge, then read 1234 // egress bytes out of the default-tier broadcast. - let update = announced.next().await.unwrap(); - assert!(update.kind.is_active()); + let (update, active) = next_update(&mut announced).await.unwrap(); + assert!(active); let bc = egress .request_broadcast(moq_net::Path::new(update.prefix.as_str())) .await diff --git a/rs/moq-relay/src/nodes.rs b/rs/moq-relay/src/nodes.rs index 28bf31ab77..35be6a55a9 100644 --- a/rs/moq-relay/src/nodes.rs +++ b/rs/moq-relay/src/nodes.rs @@ -167,13 +167,14 @@ impl Nodes { fn scan_announced(&self, announced: &mut moq_net::announce::Consumer) -> Announced { let mut scanned = Announced::default(); - while let Some(update) = announced.try_next() { + while let Some(event) = announced.try_next() { // A retraction can land mid-drain (a re-announce is a metadata update, // not a retract-and-announce). Skip it rather than end the scan, which // would drop every node still queued behind it. - if !update.kind.is_active() { + let (moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update)) = event + else { continue; - } + }; let key = canonical_announced_node(update.prefix.as_str()); let route = update.route; @@ -384,7 +385,9 @@ mod tests { // Take relay-a's replayed announce, then retire it so the cursor queues a // bare unannounce ahead of relay-b's still-pending announce. - let first_update = announced.try_next().expect("replayed announce"); + let Some(moq_net::announce::Event::Announced(first_update)) = announced.try_next() else { + panic!("replayed announce"); + }; assert_eq!( canonical_announced_node(first_update.prefix.as_str()), "https://relay-a.example/" diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 8b89b5ff87..15e6bba4ff 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -865,8 +865,8 @@ async fn serve_announced( let mut announced = origin.consume().announced(); let mut broadcasts = Vec::new(); - while let Some(update) = announced.try_next() { - if update.kind.is_active() { + while let Some(event) = announced.try_next() { + if let moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update) = event { broadcasts.push(update.prefix); } } diff --git a/rs/moq-relay/tests/auth_lifetime.rs b/rs/moq-relay/tests/auth_lifetime.rs index 7c4afbfe1e..d74b42b0bd 100644 --- a/rs/moq-relay/tests/auth_lifetime.rs +++ b/rs/moq-relay/tests/auth_lifetime.rs @@ -253,12 +253,12 @@ async fn connect_and_round_trip(url: &url::Url) -> (moq_tokio::Connection, moq_t .expect("subscriber connect timeout") .expect("subscriber connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = sub_consumer .request_broadcast("test") .await @@ -511,12 +511,12 @@ async fn http_routes_hold_a_lease() { .await .expect("subscriber connect timeout") .expect("subscriber connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let http = reqwest::Client::new(); let announced = http @@ -940,3 +940,16 @@ async fn a_relay_without_an_auth_source_is_decided_by_the_embedder() { .expect("relay task panicked") .expect("relay exited with an error"); } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/cluster_unknown.rs b/rs/moq-relay/tests/cluster_unknown.rs index 9ad71cdde3..69a230d2a3 100644 --- a/rs/moq-relay/tests/cluster_unknown.rs +++ b/rs/moq-relay/tests/cluster_unknown.rs @@ -190,8 +190,8 @@ async fn watch_announces(port: u16, window: Duration) -> Vec<(String, bool)> { let mut updates = Vec::new(); let deadline = tokio::time::Instant::now() + window; - while let Ok(Some(update)) = tokio::time::timeout_at(deadline, announced.next()).await { - updates.push((update.prefix.as_str().to_string(), update.kind.is_active())); + while let Ok(Some((update, active))) = tokio::time::timeout_at(deadline, next_update(&mut announced)).await { + updates.push((update.prefix.as_str().to_string(), active)); } updates } @@ -380,3 +380,16 @@ async fn unknown_publisher_does_not_flap_across_a_lite04_cluster_triangle() { let version = "moq-lite-04".parse().expect("parse version"); assert_unknown_publisher_stays_announced(Some(version), true).await; } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/drills.rs b/rs/moq-relay/tests/drills.rs index 3aea1c105b..cede1d8754 100644 --- a/rs/moq-relay/tests/drills.rs +++ b/rs/moq-relay/tests/drills.rs @@ -575,11 +575,11 @@ async fn interrupted_publisher_republishes_new_content() { /// Wait for `path` to be announced (`want`) or unannounced (`!want`). async fn expect_announce(announced: &mut moq_net::announce::Consumer, path: &str, want: bool, who: &str) { loop { - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(announced)) .await .unwrap_or_else(|_| panic!("{who}: no announcement change within {TIMEOUT:?}")) .unwrap_or_else(|| panic!("{who}: the announcement stream closed")); - if update.prefix.as_str() == path && update.kind.is_active() == want { + if update.prefix.as_str() == path && active == want { return; } } @@ -615,8 +615,8 @@ async fn no_publisher_never_delivers() { let quiet = Duration::from_secs(2); let mut announced = subscribed.announced(); - if let Ok(update) = tokio::time::timeout(quiet, announced.next()).await { - let path = update.map(|update| update.prefix.to_string()); + if let Ok(update) = tokio::time::timeout(quiet, next_update(&mut announced)).await { + let path = update.map(|(update, _)| update.prefix.to_string()); panic!("the announcement stream reported {path:?} with no publisher"); } @@ -630,3 +630,16 @@ async fn no_publisher_never_delivers() { drop(session); drop(relay); } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/embed.rs b/rs/moq-relay/tests/embed.rs index e65b5577e9..bc2738f632 100644 --- a/rs/moq-relay/tests/embed.rs +++ b/rs/moq-relay/tests/embed.rs @@ -247,12 +247,12 @@ async fn embed_and_stop(mut config: Config) { .expect("connect timeout") .expect("connect failed"); - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("announcement timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let announced = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("test")) .await .expect("request timeout") @@ -465,3 +465,16 @@ fn embedded_cli_merges_only_relay_settings() { assert_eq!(parsed.worker_name.as_deref(), Some("recorder")); assert_eq!(parsed.relay.cluster.id, Some(9)); } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/goaway_cluster.rs b/rs/moq-relay/tests/goaway_cluster.rs index c2b6a061c5..c0afc746ab 100644 --- a/rs/moq-relay/tests/goaway_cluster.rs +++ b/rs/moq-relay/tests/goaway_cluster.rs @@ -245,7 +245,7 @@ async fn cluster_migrates_on_upstream_goaway_inner() { // unannounce the path (metadata updates are expected: the drain // re-prices the old route and the sibling announces its own). let mut announcements = cluster.origin.consume().announced(); - let first = announcements.next().await.expect("initial announce"); + let (first, _) = next_update(&mut announcements).await.expect("initial announce"); assert_eq!(first.prefix.as_str(), "cam"); // ── sibling A drains with a redirect to sibling B ──────────────── @@ -283,9 +283,9 @@ async fn cluster_migrates_on_upstream_goaway_inner() { // updates (the drain re-pricing, the sibling's route) are expected and // harmless; an inactive event means the path flapped. loop { - match tokio::time::timeout(Duration::from_millis(500), announcements.next()).await { + match tokio::time::timeout(Duration::from_millis(500), next_update(&mut announcements)).await { Err(_) => break, - Ok(Some(update)) if update.kind.is_active() => continue, + Ok(Some((_, true))) => continue, Ok(event) => panic!("migration must not retract the path on the cluster origin: {event:?}"), } } @@ -452,9 +452,12 @@ async fn cluster_diamond_goaway_seamless_failover_inner() { // Watch announcements for the whole test: the failover must never // unannounce the broadcast under the subscriber. let mut announcements = sub_origin.consume().announced(); - let first = within("broadcast announced through the MID-A leg", announcements.next()) - .await - .expect("origin closed before the announce"); + let (first, _) = within( + "broadcast announced through the MID-A leg", + next_update(&mut announcements), + ) + .await + .expect("origin closed before the announce"); assert_eq!(first.prefix.as_str(), "diamond"); let bc = within("broadcast resolves on the subscriber origin", async { @@ -603,9 +606,9 @@ async fn cluster_diamond_goaway_seamless_failover_inner() { // ── announcement stability: the path never retracted under the swap ── // Metadata updates (route re-pricing, the new leg's hops) are expected. loop { - match tokio::time::timeout(Duration::from_millis(500), announcements.next()).await { + match tokio::time::timeout(Duration::from_millis(500), next_update(&mut announcements)).await { Err(_) => break, - Ok(Some(update)) if update.kind.is_active() => continue, + Ok(Some((_, true))) => continue, Ok(event) => panic!("failover must not retract the path under the subscriber: {event:?}"), } } @@ -836,3 +839,16 @@ async fn goaway_handover_is_enforced_while_the_replacement_dial_hangs_inner() { drop(connection); } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/runtime_uring.rs b/rs/moq-relay/tests/runtime_uring.rs index 8a5dab09a8..24f9b08bc6 100644 --- a/rs/moq-relay/tests/runtime_uring.rs +++ b/rs/moq-relay/tests/runtime_uring.rs @@ -156,12 +156,12 @@ async fn uring_workers_serve_webtransport_and_raw_quic() { } for (index, (_connection, consumer, announced)) in subscribers.iter_mut().enumerate() { - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(announced)) .await .unwrap_or_else(|_| panic!("subscriber {index} announcement timeout")) .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let broadcast = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("test")) .await .unwrap_or_else(|_| panic!("subscriber {index} request timeout")) @@ -312,12 +312,12 @@ async fn an_mtls_client_authenticates_without_a_token() { let mut announced = consumer.announced(); let subscriber = connect(client().with_subscriber(subscriber_origin), url).await; - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("announcement timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let announced = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("test")) .await .expect("request timeout") @@ -391,11 +391,11 @@ async fn uring_workers_write_qlog_traces() { let consumer = subscriber_origin.consume(); let mut announced = consumer.announced(); let subscriber = connect(client().with_subscriber(subscriber_origin), url).await; - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (_, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("announcement timeout") .expect("origin closed"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); assert!(!running.is_finished(), "the relay stopped while serving"); drop(track); @@ -462,3 +462,16 @@ async fn spawn_auth_server(policy: moq_auth::serve::Policy) -> url::Url { tokio::spawn(async move { server.serve(listener).await }); url } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/runtime_workers.rs b/rs/moq-relay/tests/runtime_workers.rs index 7f6040ec1d..c1961c5e97 100644 --- a/rs/moq-relay/tests/runtime_workers.rs +++ b/rs/moq-relay/tests/runtime_workers.rs @@ -119,12 +119,12 @@ async fn workers_serve_quic_and_share_one_origin() { } for (index, (_connection, consumer, announced)) in subscribers.iter_mut().enumerate() { - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(announced)) .await .unwrap_or_else(|_| panic!("subscriber {index} announcement timeout")) .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let broadcast = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("test")) .await .unwrap_or_else(|_| panic!("subscriber {index} request timeout")) @@ -158,3 +158,16 @@ async fn workers_serve_quic_and_share_one_origin() { running.abort(); let _ = running.await; } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-relay/tests/smoke.rs b/rs/moq-relay/tests/smoke.rs index 77a5ef78ec..5b99363890 100644 --- a/rs/moq-relay/tests/smoke.rs +++ b/rs/moq-relay/tests/smoke.rs @@ -206,12 +206,12 @@ async fn relay_websocket_round_trip_uses_newest_version() { ); // ── data path ─────────────────────────────────────────────────── - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); let path = moq_net::Path::new(update.prefix.as_str()).to_owned(); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); // Auth root for `/smoke` is "smoke"; the broadcast "test" announces underneath. assert_eq!(path.as_str(), "test"); let bc = sub_consumer @@ -397,12 +397,12 @@ async fn relay_websocket_root_path_upgrades() { // ── data path ─────────────────────────────────────────────────── // The root auth scope is the empty path, so the broadcast announces at its // own name with no prefix. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); let path = moq_net::Path::new(update.prefix.as_str()).to_owned(); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); assert_eq!(path.as_str(), "test"); let bc = sub_consumer .request_broadcast(&path) @@ -485,11 +485,11 @@ async fn two_publish_only_clients_coexist() { let mut seen = std::collections::HashSet::new(); while seen.len() < 2 { - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); - if update.kind.is_active() { + if active { seen.insert(update.prefix.as_str().to_owned()); } } @@ -628,12 +628,12 @@ async fn internal_tcp_round_trip() { // ── data path ─────────────────────────────────────────────────── // The internal listener grants the empty root, so the broadcast announces // at its own name with no path prefix. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); let path = moq_net::Path::new(update.prefix.as_str()).to_owned(); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); assert_eq!(path.as_str(), "test"); let bc = sub_consumer .request_broadcast(&path) @@ -742,12 +742,12 @@ async fn internal_unix_round_trip() { .expect("subscriber connect failed"); // ── data path ─────────────────────────────────────────────────── - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timeout") @@ -825,7 +825,7 @@ async fn path_round_trip(version: moq_net::Version, pub_url: url::Url, sub_url: .expect("subscriber connect timeout") .expect("subscriber connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, _) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announcement timeout") .expect("origin closed"); @@ -1100,3 +1100,16 @@ async fn connect_once( let connection = client.clone().with_reconnect(false).connect(url).established().await?; Ok((client, connection)) } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-room/src/room.rs b/rs/moq-room/src/room.rs index 22666ca7a4..19e9d28c05 100644 --- a/rs/moq-room/src/room.rs +++ b/rs/moq-room/src/room.rs @@ -78,8 +78,11 @@ impl Room { return Poll::Ready(Some(event)); } - let Some(update) = ready!(self.announced.poll_next(waiter)) else { - return Poll::Ready(None); + let (update, active) = match ready!(self.announced.poll_next(waiter)) { + Some(announce::Event::Announced(update) | announce::Event::Updated(update)) => (update, true), + Some(announce::Event::Retracted(update)) => (update, false), + Some(announce::Event::Live) => continue, + None => return Poll::Ready(None), }; let path = update.prefix; let Some(parsed) = parse(&path) else { @@ -88,7 +91,7 @@ impl Room { if self.local.as_ref().is_some_and(|id| *id == parsed.identity) { continue; } - if !update.kind.is_active() { + if !active { return Poll::Ready(Some(Event { identity: parsed.identity, kind: parsed.kind, diff --git a/rs/moq-rtc/src/lib.rs b/rs/moq-rtc/src/lib.rs index 7217a4c8f6..bf7c9360e4 100644 --- a/rs/moq-rtc/src/lib.rs +++ b/rs/moq-rtc/src/lib.rs @@ -72,6 +72,19 @@ pub use server::{Response, Server, whep, whip}; mod tests { use std::time::Duration; + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + use axum::Router; use bytes::Bytes; @@ -103,12 +116,12 @@ mod tests { }, ) .expect("publish source packet"); - let announcement = tokio::time::timeout(TIMEOUT, announcements.next()) + let (announcement, announcement_active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("source announcement timed out") .expect("source origin closed"); assert_eq!(announcement.prefix.as_str(), "source"); - assert!(announcement.kind.is_active(), "source was unannounced"); + assert!(announcement_active, "source was unannounced"); drop(announcements); let server_origin = moq_tokio::origin::spawn(); diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index c494def001..b15c5da7c6 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -295,7 +295,13 @@ impl Merged { // unannounce. A closed stream ends the merged view. loop { match self.announce.poll_next(waiter) { - Poll::Ready(Some(update)) => changed |= self.apply_announce(update), + Poll::Ready(Some( + moq_net::announce::Event::Announced(update) | moq_net::announce::Event::Updated(update), + )) => changed |= self.apply_announce(update, true), + Poll::Ready(Some(moq_net::announce::Event::Retracted(update))) => { + changed |= self.apply_announce(update, false) + } + Poll::Ready(Some(moq_net::announce::Event::Live)) => {} Poll::Ready(None) => return Poll::Ready(Ok(None)), Poll::Pending => break, } @@ -319,7 +325,7 @@ impl Merged { /// Apply one announce update to the node set. Returns whether the merged view /// changed (only a non-sticky contribution leaving does; a sticky one is /// kept). - fn apply_announce(&mut self, update: moq_net::announce::Update) -> bool { + fn apply_announce(&mut self, update: moq_net::announce::Announce, active: bool) -> bool { let path = update.prefix; let absolute = self.announce.absolute(&path).to_owned(); @@ -330,7 +336,7 @@ impl Merged { return false; } - if update.kind.is_active() { + if active { // A route update on a node already tracked (a reprice, or a takeover // with different metadata) keeps the live reader: existing // subscriptions survive a takeover, and the reader re-resolves through @@ -489,6 +495,19 @@ mod tests { producer } + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + use std::time::Duration; use moq_net::{PathOwned, Timestamp, announce, broadcast, origin, track}; @@ -533,8 +552,8 @@ mod tests { source.announce(origin::Route::default()).expect("announce"); let track = source.create_track("video", None).expect("create_track"); - let update = announced.next().await.expect("announce"); - assert!(update.kind.is_active()); + let (_, active) = next_update(&mut announced).await.expect("announce"); + assert!(active); let consumer = egress.request_broadcast(path).await.expect("resolve"); let mut sub = consumer.track("video").unwrap().subscribe(None).await.unwrap(); diff --git a/rs/moq-stats/src/consume.rs b/rs/moq-stats/src/consume.rs index bc02b53f89..9ce49003d2 100644 --- a/rs/moq-stats/src/consume.rs +++ b/rs/moq-stats/src/consume.rs @@ -117,6 +117,19 @@ mod tests { producer } + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + use std::time::Duration; use moq_net::{Consume, PathOwned, Timestamp, announce, broadcast, origin, track}; @@ -167,8 +180,8 @@ mod tests { source.announce(origin::Route::default()).unwrap(); let track = source.clone().create_track("video", None).unwrap(); - let update = announced.next().await.expect("announce"); - assert!(update.kind.is_active()); + let (_, active) = next_update(&mut announced).await.expect("announce"); + assert!(active); let consumer = egress.request_broadcast(path).await.expect("resolve"); let sub = consumer.track("video").unwrap().subscribe(None).await.unwrap(); @@ -184,8 +197,8 @@ mod tests { async fn announced(origin: &origin::Producer) -> moq_net::broadcast::Consumer { let mut consumer = origin.consume().announced(); tokio::time::advance(Duration::from_millis(1)).await; - let update = consumer.next().await.expect("expected announce"); - assert!(update.kind.is_active()); + let (update, active) = next_update(&mut consumer).await.expect("expected announce"); + assert!(active); origin .consume() .request_broadcast(moq_net::Path::new(update.prefix.as_str())) diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index ce706424d4..0a8c40d1e4 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -1017,6 +1017,19 @@ mod tests { producer } + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + use std::collections::BTreeMap; use moq_net::stats::{Registry, Tier}; @@ -1068,8 +1081,8 @@ mod tests { source.announce(origin::Route::default()).expect("announce"); let producer = source.create_track("video", None).expect("create_track"); - let update = announced.next().await.expect("announce"); - assert!(update.kind.is_active()); + let (_, active) = next_update(&mut announced).await.expect("announce"); + assert!(active); let consumer = egress.request_broadcast(path).await.expect("resolve"); let sub = if subscribe { @@ -1109,8 +1122,8 @@ mod tests { async fn announced(origin: &origin::Producer) -> (String, moq_net::broadcast::Consumer) { let mut consumer = origin.consume().announced(); tokio::time::advance(Duration::from_millis(1)).await; - let update = consumer.next().await.expect("expected announce"); - assert!(update.kind.is_active()); + let (update, active) = next_update(&mut consumer).await.expect("expected announce"); + assert!(active); let broadcast = origin .consume() .request_broadcast(moq_net::Path::new(update.prefix.as_str())) diff --git a/rs/moq-tokio/examples/clock.rs b/rs/moq-tokio/examples/clock.rs index 42912d89c1..195ccead8a 100644 --- a/rs/moq-tokio/examples/clock.rs +++ b/rs/moq-tokio/examples/clock.rs @@ -110,17 +110,18 @@ async fn main() -> anyhow::Result<()> { loop { tokio::select! { - Some(update) = announced.next() => match update.kind.is_active() { - true => { + Some(event) = announced.next() => match event { + announce::Event::Announced(update) | announce::Event::Updated(update) => { tracing::info!(broadcast = %update.prefix, "broadcast is online, subscribing to track"); let broadcast = consumer.request_broadcast(&update.prefix).await?; let track = broadcast .track(&track)?.subscribe(None).await?; clock = Some(Subscriber::new(track)); } - false => { + announce::Event::Retracted(update) => { tracing::warn!(broadcast = %update.prefix, "broadcast is offline, waiting..."); } + announce::Event::Live => {} }, res = reconnect.closed() => return Ok(res?), // Drops the previous subscriber on each new announce. diff --git a/rs/moq-tokio/src/origin.rs b/rs/moq-tokio/src/origin.rs index b2911dfba5..51de8f3801 100644 --- a/rs/moq-tokio/src/origin.rs +++ b/rs/moq-tokio/src/origin.rs @@ -25,6 +25,19 @@ pub fn spawn_config(config: moq_net::origin::Config) -> moq_net::origin::Produce mod tests { use super::*; + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + /// The spawned driver runs the origin's lifecycle: an announced route /// reaches a consumer, and dropping the announcement retracts it. #[tokio::test] @@ -35,13 +48,13 @@ mod tests { let broadcast = origin.create_broadcast("cam").expect("create broadcast"); broadcast.announce(Default::default()).expect("create broadcast"); - let update = announced.next().await.expect("announce"); + let (update, active) = next_update(&mut announced).await.expect("announce"); assert_eq!(update.prefix.as_str(), "cam"); - assert!(update.kind.is_active()); + assert!(active); broadcast.finish(); - let update = announced.next().await.expect("retraction"); - assert!(!update.kind.is_active()); + let (_, active) = next_update(&mut announced).await.expect("retraction"); + assert!(!active); } /// Outside a runtime the spawn has nowhere to run the driver: it panics diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index ba6ccd00a1..9f39d8911c 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -1419,6 +1419,19 @@ impl Request { mod tests { use super::*; + /// The next route and whether it is active, skipping the caught-up marker. + async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } + } + #[test] fn version_help_lists_every_parseable_name() { #[derive(usage::Cli)] @@ -1582,12 +1595,12 @@ mod tests { // Without the server's publisher the session announces nothing, so this is // where the regression shows up. - let update = tokio::time::timeout(TIMEOUT, announced.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announced)) .await .expect("announce timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active()); + assert!(active); let broadcast = consumer.request_broadcast("test").await.expect("resolve"); let mut track = broadcast diff --git a/rs/moq-tokio/tests/backend.rs b/rs/moq-tokio/tests/backend.rs index 15d850c24b..06871ac776 100644 --- a/rs/moq-tokio/tests/backend.rs +++ b/rs/moq-tokio/tests/backend.rs @@ -176,12 +176,12 @@ async fn connect_test(config: ConnectTest<'_>) { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -629,12 +629,12 @@ async fn iroh_connect() { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -859,3 +859,16 @@ async fn window_test(scheme: &str) { async fn noq_windows() { window_test("moqt").await; } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +} diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index ddc1db8b78..65378a8040 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -83,13 +83,13 @@ async fn broadcast_test(scheme: &str, client_version: Option<&str>, server_versi .expect("client connect failed"); // Wait for the broadcast announcement. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -197,12 +197,12 @@ async fn lite05_timestamp_roundtrip(scheme: &str) { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -320,12 +320,12 @@ async fn lite05_fetch_roundtrip(scheme: &str) { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -450,12 +450,12 @@ async fn lite05_fetch_during_subscribe(scheme: &str) { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -561,11 +561,11 @@ async fn broadcast_moq_lite_05_default_timescale() { .expect("connect timeout") .expect("connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timeout") .expect("origin closed"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); let bc = tokio::time::timeout( TIMEOUT, sub_consumer.request_broadcast(moq_net::Path::new(update.prefix.as_str())), @@ -661,9 +661,9 @@ async fn broadcast_moq_transport_20_current_group_join() { .expect("connect timed out") .expect("connect failed"); - let announced = next_announce(&mut announcements).await; + let (announced, announced_active) = next_announce(&mut announcements).await; assert_eq!(announced.prefix.as_str(), "test"); - assert!(announced.kind.is_active(), "expected an announce"); + assert!(announced_active, "expected an announce"); let remote = tokio::time::timeout( TIMEOUT, sub_consumer.request_broadcast(moq_net::Path::new(announced.prefix.as_str())), @@ -710,8 +710,8 @@ async fn broadcast_moq_transport_20_current_group_join() { } /// Wait for the next announce event, failing the test on a timeout or a closed origin. -async fn next_announce(announcements: &mut moq_net::announce::Consumer) -> moq_net::announce::Update { - tokio::time::timeout(TIMEOUT, announcements.next()) +async fn next_announce(announcements: &mut moq_net::announce::Consumer) -> (moq_net::announce::Announce, bool) { + tokio::time::timeout(TIMEOUT, next_update(announcements)) .await .expect("announce timeout") .expect("origin closed") @@ -763,50 +763,50 @@ async fn broadcast_moq_lite_06_announce_lifecycle() { .expect("connect failed"); // The initial set: "first" was announced before the session existed. - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "first"); - assert!(update.kind.is_active(), "expected initial announce"); + assert!(active, "expected initial announce"); // A live announce after the initial set. let second = pub_origin.create_broadcast("second").expect("create broadcast"); second.announce(Default::default()).expect("create broadcast"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "second"); - assert!(update.kind.is_active(), "expected live announce"); + assert!(active, "expected live announce"); // Unannounce: retracted by announce id on the wire. Dropping the announcement // retracts the route; the broadcast's own end is independent. second.finish(); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "second"); - assert!(!update.kind.is_active(), "expected retraction"); + assert!(!active, "expected retraction"); // Re-announce the same path: a fresh announce assigning a fresh id. let _second = pub_origin.create_broadcast("second").expect("create broadcast"); _second.announce(Default::default()).expect("create broadcast"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "second"); - assert!(update.kind.is_active(), "expected re-announce"); + assert!(active, "expected re-announce"); // Replace the route at "first": retract the original (retiring its announce // id on the wire), then announce the same path again (assigning a fresh id). // Await the retraction first so the events cannot coalesce away. first.finish(); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "first"); - assert!(!update.kind.is_active(), "expected the replaced retraction"); + assert!(!active, "expected the replaced retraction"); let _replacement = pub_origin.create_broadcast("first").expect("create replacement"); _replacement.announce(Default::default()).expect("create replacement"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "first"); - assert!(update.kind.is_active(), "expected the replacement announce"); + assert!(active, "expected the replacement announce"); // A sentinel proves no stray event for "first" snuck in behind the replacement. let _sentinel = pub_origin.create_broadcast("sentinel").expect("create broadcast"); _sentinel.announce(Default::default()).expect("create broadcast"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "sentinel"); - assert!(update.kind.is_active(), "expected sentinel announce"); + assert!(active, "expected sentinel announce"); drop(connection); server_handle @@ -947,9 +947,9 @@ async fn broadcast_route_migration() { // One path, announced once even though two sessions route it (the cheaper // route wins the advertisement). - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); // Resolve and subscribe: the cheaper route (A) serves the track. let subscription = moq_net::track::Subscription::default() @@ -985,8 +985,11 @@ async fn broadcast_route_migration() { // The path was never retracted across the swap: any events delivered are // active metadata updates (the standby's chain taking over). - while let Some(update) = announcements.try_next() { - assert!(update.kind.is_active(), "failover must not retract the route"); + while let Some(event) = announcements.try_next() { + assert!( + !matches!(event, moq_net::announce::Event::Retracted(_)), + "failover must not retract the route" + ); } handle_a.await.expect("server a panicked").expect("server a failed"); @@ -1062,9 +1065,9 @@ async fn route_reannounce_test(version: Option<&str>) { .expect("connect timeout") .expect("connect failed"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); let initial = update.route; let broadcast = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) @@ -1089,9 +1092,9 @@ async fn route_reannounce_test(version: Option<&str>) { .expect("update route"); // The subscriber sees the new chain as another active update for the prefix... - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "a restart must not retract the route"); + assert!(active, "a restart must not retract the route"); assert_ne!(initial.hops, update.route.hops, "route must change"); assert!( update.route.hops.iter().any(|h| h.id() == 0x5555), @@ -1387,9 +1390,9 @@ async fn max_age_test(version: &str) -> Duration { .expect("connect timeout") .expect("connect failed"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); let broadcast = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -1702,13 +1705,13 @@ async fn broadcast_websocket() { .expect("client connect failed"); // Wait for the broadcast announcement. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -1824,13 +1827,13 @@ async fn broadcast_websocket_fallback() { .expect("client connect failed"); // Wait for the broadcast announcement. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -2091,11 +2094,11 @@ async fn quic_driver_task_inherits_connection_span() { .expect("client connect timed out") .expect("client connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout( TIMEOUT, sub_consumer.request_broadcast(moq_net::Path::new(update.prefix.as_str())), @@ -2209,12 +2212,12 @@ async fn resubscribe_keeps_flowing_moq_lite_03() { .expect("connect timeout") .expect("connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timeout") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -2348,11 +2351,11 @@ async fn idle_subscription_releases_the_viewer_count() { .expect("connect timeout") .expect("connect failed"); - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timeout") .expect("origin closed"); - assert!(update.kind.is_active(), "expected announce"); + assert!(active, "expected announce"); let bc = tokio::time::timeout( TIMEOUT, sub_consumer.request_broadcast(moq_net::Path::new(update.prefix.as_str())), @@ -2673,9 +2676,9 @@ async fn a_dead_session_unannounces_while_the_reconnect_retries() { .expect("connect timed out") .expect("connect failed"); - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "live"); - assert!(update.kind.is_active(), "expected the initial announce"); + assert!(active, "expected the initial announce"); // The server kills the session; the loop starts redialing into the void. let session = tokio::time::timeout(TIMEOUT, session_rx) @@ -2685,9 +2688,9 @@ async fn a_dead_session_unannounces_while_the_reconnect_retries() { session.abort(moq_net::Error::Cancel); // The retraction lands promptly, while the connection is still retrying. - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "live"); - assert!(!update.kind.is_active(), "a dead session must retract, not linger"); + assert!(!active, "a dead session must retract, not linger"); drop(connection); server_handle.abort(); @@ -2758,12 +2761,12 @@ async fn announce_interest_unauthorized_keeps_session_alive() { .expect("client connect failed"); // The "allowed" announce stream still delivers even though "denied" was FINed. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "allowed/test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); // The unauthorized "denied" interest must not have torn down the session. assert!( @@ -2850,15 +2853,15 @@ async fn wildcard_scope_test(version: &str, server_scope: &str) { // Exactly alice's chat: bob's is outside the server's grant, alice's audio and the // lobby are outside the client's. The match pins the room. - let update = next_announce(&mut announcements).await; + let (update, active) = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "room/alice/chat"); assert_eq!( update.captures, Some(vec!["alice".parse::().unwrap()]) ); - assert!(update.kind.is_active()); + assert!(active); assert!( - tokio::time::timeout(Duration::from_millis(200), announcements.next()) + tokio::time::timeout(Duration::from_millis(200), next_update(&mut announcements)) .await .is_err(), "a path outside one of the grants was announced" @@ -2961,12 +2964,12 @@ async fn publish_only_client_to_subscribe_only_server() { // The client serves "allowed/test"; the "denied" interest is FINed but must not // tear down the session. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "allowed/test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("allowed/test")) .await .expect("request timed out") @@ -3142,12 +3145,12 @@ async fn goaway_test(scheme: &str, version: &str, expect_wire_timeout: bool) { .expect("client connect failed"); // Subscribe and read the pre-GOAWAY group. - let update = tokio::time::timeout(TIMEOUT, announcements.next()) + let (update, active) = tokio::time::timeout(TIMEOUT, next_update(&mut announcements)) .await .expect("announce timed out") .expect("origin closed"); assert_eq!(update.prefix.as_str(), "test"); - assert!(update.kind.is_active(), "expected announce, got retraction"); + assert!(active, "expected announce, got retraction"); let bc = tokio::time::timeout(TIMEOUT, sub_consumer.request_broadcast("test")) .await .expect("request timed out") @@ -3471,3 +3474,16 @@ async fn abort_carries_its_code_to_the_peer() { server_handle.abort(); } + +/// The next route and whether it is active, skipping the caught-up marker. +async fn next_update(announced: &mut moq_net::announce::Consumer) -> Option<(moq_net::announce::Announce, bool)> { + loop { + return match announced.next().await? { + moq_net::announce::Event::Announced(route) | moq_net::announce::Event::Updated(route) => { + Some((route, true)) + } + moq_net::announce::Event::Retracted(route) => Some((route, false)), + moq_net::announce::Event::Live => continue, + }; + } +}