feat(moq-net)!: announce consumers yield a Live marker once caught up

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<Vec<Pattern>>, 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<Vec<Pattern>>, 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

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)?

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
 

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

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`

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

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

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?;

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);
 		}
 	}

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;

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;
 			};

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);
 

rs/moq-ffi/src/origin.rs

@@ -143,14 +143,25 @@ impl OriginDynamic {
 
 impl Announced {
 	async fn next(&mut self) -> Result<Option<Arc<MoqAnnounceUpdate>>, 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),
 		}

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);