From 37741d2563fb32a14ab94799ea420eab8f3c1568 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sat, 30 May 2026 17:14:06 +0000
Subject: [PATCH] moq-net: REANNOUNCE via duplicate ANNOUNCE; announced()
 returns (path, Announced)

REANNOUNCE is an atomic broadcast replacement (relay failover, upstream restart,
shorter hop path). On the wire it's a duplicate ANNOUNCE: an active announcement
for a path that is already announced, with no intervening unannounce. No new
status byte and no version gating.

AnnounceConsumer::{next,try_next,poll_next} now yield (PathOwned, Announced)
instead of (PathOwned, Option<BroadcastConsumer>). The `Announced` enum is
Active/Restart/Ended; `.broadcast()` gives the legacy Option<BroadcastConsumer>
view. The origin coalesces atomic replacements (backup promotion, shorter hop
path) into a single Announced::Restart delivery.

Lite publisher sends a restart as a duplicate Announce::Active. Lite subscriber
treats an Active for an already-announced path as an atomic replacement instead
of a Duplicate error. The lite decoder also tolerates the draft's explicit
`restart` status (2), surfacing it as an Active. IETF moq-transport has no
restart, so it splits into namespace_done + namespace.

Examples use `.context`/`event.broadcast()` for ergonomics.
---
 rs/hang/examples/subscribe.rs     |  12 +-
 rs/libmoq/src/origin.rs           |   5 +-
 rs/moq-boy/src/input.rs           |   2 +-
 rs/moq-ffi/src/origin.rs          |  18 +--
 rs/moq-native/examples/clock.rs   |   6 +-
 rs/moq-native/tests/backend.rs    |   4 +-
 rs/moq-native/tests/broadcast.rs  |   8 +-
 rs/moq-net/src/ietf/publisher.rs  | 206 +++++++++++++++++-------------
 rs/moq-net/src/lite/announce.rs   |  75 +++++++++++
 rs/moq-net/src/lite/publisher.rs  | 158 +++++++++++++----------
 rs/moq-net/src/lite/subscriber.rs |  54 +++++++-
 rs/moq-net/src/model/origin.rs    | 181 ++++++++++++++++++++------
 rs/moq-net/src/stats.rs           |  24 ++--
 rs/moq-relay/src/cluster.rs       |  18 +--
 rs/moq-relay/src/web.rs           |   4 +-
 rs/moq-relay/tests/smoke.rs       |   2 +-
 16 files changed, 531 insertions(+), 246 deletions(-)

diff --git a/rs/hang/examples/subscribe.rs b/rs/hang/examples/subscribe.rs
index 628aaa90a..977d1ca2a 100644
--- a/rs/hang/examples/subscribe.rs
+++ b/rs/hang/examples/subscribe.rs
@@ -2,6 +2,8 @@
 
 use std::time::Duration;
 
+use anyhow::Context;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
 	// Optional: Use moq_native to configure a logger.
@@ -40,13 +42,11 @@ async fn run_session(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 // Subscribe to a broadcast and read media frames.
 async fn run_subscribe(consumer: moq_net::OriginConsumer) -> anyhow::Result<()> {
 	// Wait for a broadcast to be announced.
-	let (path, broadcast) = consumer
-		.announced()
-		.next()
-		.await
-		.ok_or_else(|| anyhow::anyhow!("origin closed"))?;
+	let (path, broadcast) = consumer.announced().next().await.context("origin closed")?;
 
-	let broadcast = broadcast.ok_or_else(|| anyhow::anyhow!("broadcast unannounced: {path}"))?;
+	let broadcast = broadcast
+		.broadcast()
+		.with_context(|| format!("broadcast unannounced: {path}"))?;
 
 	tracing::info!(%path, "broadcast announced");
 
diff --git a/rs/libmoq/src/origin.rs b/rs/libmoq/src/origin.rs
index 68f076d08..350f5fb25 100644
--- a/rs/libmoq/src/origin.rs
+++ b/rs/libmoq/src/origin.rs
@@ -76,7 +76,7 @@ impl Origin {
 	) -> Result<(), Error> {
 		loop {
 			// `biased` so a pending close always wins over a ready announcement.
-			let (path, broadcast) = tokio::select! {
+			let (path, event) = tokio::select! {
 				biased;
 				_ = &mut close => return Ok(()),
 				next = consumer.next() => match next {
@@ -86,10 +86,11 @@ impl Origin {
 			};
 
 			// Hold the lock only to buffer the announcement; release it before the callback.
+			// Active and Restart both carry a broadcast; Ended does not.
 			let announced_id = State::lock()
 				.origin
 				.announced
-				.insert((path.to_string(), broadcast.is_some()))?;
+				.insert((path.to_string(), event.broadcast().is_some()))?;
 			callback.call(announced_id);
 		}
 	}
diff --git a/rs/moq-boy/src/input.rs b/rs/moq-boy/src/input.rs
index 08b6816c5..d4635fdda 100644
--- a/rs/moq-boy/src/input.rs
+++ b/rs/moq-boy/src/input.rs
@@ -71,7 +71,7 @@ pub async fn handle_viewers(
 
 		let viewer_id = path.to_string();
 
-		if let Some(broadcast) = broadcast {
+		if let Some(broadcast) = broadcast.broadcast() {
 			tracing::info!(%viewer_id, "viewer connected");
 			let cmd_tx = cmd_tx.clone();
 			let vid = viewer_id.clone();
diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs
index 32785f694..8a51309d1 100644
--- a/rs/moq-ffi/src/origin.rs
+++ b/rs/moq-ffi/src/origin.rs
@@ -28,14 +28,16 @@ impl Announced {
 	async fn next(&mut self) -> Result<Option<Arc<MoqAnnouncement>>, MoqError> {
 		loop {
 			match self.inner.next().await {
-				Some((path, Some(broadcast))) => {
+				// Active and Restart both carry a broadcast; skip unannounce events.
+				Some((path, event)) => {
+					let Some(broadcast) = event.broadcast() else {
+						continue;
+					};
 					return Ok(Some(Arc::new(MoqAnnouncement {
 						path: path.to_string(),
 						broadcast: Arc::new(MoqBroadcastConsumer::new(broadcast)),
 					})));
 				}
-				// TODO moq-lite will change to not emit None (unannounce) events here.
-				Some((_path, None)) => continue,
 				None => return Ok(None),
 			}
 		}
@@ -44,11 +46,11 @@ impl Announced {
 	async fn available(&mut self) -> Result<Arc<MoqBroadcastConsumer>, MoqError> {
 		loop {
 			match self.inner.next().await {
-				Some((_path, Some(broadcast))) => {
-					return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast)));
-				}
-				// TODO moq-lite will change to not emit None (unannounce) events here.
-				Some((_path, None)) => continue,
+				// Active and Restart both carry a broadcast; skip unannounce events.
+				Some((_path, event)) => match event.broadcast() {
+					Some(broadcast) => return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast))),
+					None => continue,
+				},
 				None => return Err(MoqError::Closed),
 			}
 		}
diff --git a/rs/moq-native/examples/clock.rs b/rs/moq-native/examples/clock.rs
index d8ff48736..4ea556f1b 100644
--- a/rs/moq-native/examples/clock.rs
+++ b/rs/moq-native/examples/clock.rs
@@ -101,13 +101,13 @@ async fn main() -> anyhow::Result<()> {
 
 			loop {
 				tokio::select! {
-					Some(announce) = origin.next() => match announce {
-						(path, Some(broadcast)) => {
+					Some((path, event)) = origin.next() => match event.broadcast() {
+						Some(broadcast) => {
 							tracing::info!(broadcast = %path, "broadcast is online, subscribing to track");
 							let track = broadcast.subscribe_track(&track)?;
 							clock = Some(Subscriber::new(track));
 						}
-						(path, None) => {
+						None => {
 							tracing::warn!(broadcast = %path, "broadcast is offline, waiting...");
 						}
 					},
diff --git a/rs/moq-native/tests/backend.rs b/rs/moq-native/tests/backend.rs
index 2b2e3796a..7f2ffd1ab 100644
--- a/rs/moq-native/tests/backend.rs
+++ b/rs/moq-native/tests/backend.rs
@@ -67,7 +67,7 @@ async fn backend_test(scheme: &str, backend: moq_native::QuicBackend) {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	let mut track_sub = bc
 		.subscribe_track(&Track::new("video"))
@@ -219,7 +219,7 @@ async fn iroh_connect() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	let mut track_sub = bc
 		.subscribe_track(&Track::new("video"))
diff --git a/rs/moq-native/tests/broadcast.rs b/rs/moq-native/tests/broadcast.rs
index 09bf5da14..f3a800023 100644
--- a/rs/moq-native/tests/broadcast.rs
+++ b/rs/moq-native/tests/broadcast.rs
@@ -84,7 +84,7 @@ async fn broadcast_test(scheme: &str, client_version: Option<&str>, server_versi
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -496,7 +496,7 @@ async fn broadcast_websocket() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -603,7 +603,7 @@ async fn broadcast_websocket_fallback() {
 		.expect("origin closed");
 
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	// Subscribe to the track.
 	let mut track_sub = bc
@@ -844,7 +844,7 @@ async fn linger_resubscribe_keeps_flowing_moq_lite_03() {
 		.expect("announce timeout")
 		.expect("origin closed");
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce");
+	let bc = bc.broadcast().expect("expected announce");
 
 	// First subscription: receive group 0.
 	let mut sub1 = bc.subscribe_track(&Track::new("video")).expect("subscribe1");
diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs
index 2f390bf7e..e59fc1e21 100644
--- a/rs/moq-net/src/ietf/publisher.rs
+++ b/rs/moq-net/src/ietf/publisher.rs
@@ -448,79 +448,94 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 				next = announced.next() => next,
 			};
 
-			let Some((path, active)) = next else {
+			let Some((path, event)) = next else {
 				break;
 			};
 
 			let suffix = path.to_owned();
 
-			if active.is_some() {
-				tracing::debug!(broadcast = %self.origin.absolute(&path), "announce");
-
-				let request_id = self.control.next_request_id().await?;
-				let mut stream = Stream::open(&self.session, self.version).await?;
-
-				// Write the PublishNamespace message
-				stream.writer.encode(&ietf::PublishNamespace::ID).await?;
-				stream
-					.writer
-					.encode(&ietf::PublishNamespace {
-						request_id,
-						track_namespace: suffix.as_path(),
-					})
-					.await?;
-
-				// Read response from stream.reader
-				let type_id: u64 = stream.reader.decode().await?;
-				let size: u16 = stream.reader.decode().await?;
-				let mut data = stream.reader.read_exact(size as usize).await?;
-
-				match (self.version, type_id) {
-					// Draft14 uses PublishNamespaceOk (0x07) / PublishNamespaceError (0x08)
-					(Version::Draft14, ietf::PublishNamespaceOk::ID) => {
-						let msg = ietf::PublishNamespaceOk::decode_msg(&mut data, self.version)?;
-						tracing::debug!(message = ?msg, "publish namespace ok");
-						namespace_streams.insert(suffix, (request_id, stream));
-					}
-					(Version::Draft14, ietf::PublishNamespaceError::ID) => {
-						let msg = ietf::PublishNamespaceError::decode_msg(&mut data, self.version)?;
-						tracing::warn!(message = ?msg, "publish namespace error");
-					}
-					// Draft15+ uses RequestOk (0x07) / RequestError (0x05)
-					(_, ietf::RequestOk::ID) => {
-						let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?;
-						tracing::debug!(message = ?msg, "publish namespace ok");
-						namespace_streams.insert(suffix, (request_id, stream));
-					}
-					(_, ietf::RequestError::ID) => {
-						let msg = ietf::RequestError::decode_msg(&mut data, self.version)?;
-						tracing::warn!(message = ?msg, "publish namespace error");
-					}
-					_ => return Err(Error::UnexpectedMessage),
+			match event {
+				crate::Announced::Active(_) => {
+					self.announce_namespace(suffix, &mut namespace_streams).await?;
 				}
-			} else {
-				tracing::debug!(broadcast = %self.origin.absolute(&path), "unannounce");
-				if let Some((request_id, mut stream)) = namespace_streams.remove(&suffix) {
-					// v14-16 sends PublishNamespaceDone; v17+ just closes the stream.
-					match self.version {
-						Version::Draft14 | Version::Draft15 | Version::Draft16 => {
-							let _ = stream
-								.writer
-								.encode_message(&ietf::PublishNamespaceDone {
-									track_namespace: suffix.as_path(),
-									request_id,
-								})
-								.await;
-						}
-						_ => {}
-					}
-					stream.writer.finish().ok();
+				crate::Announced::Restart(_) => {
+					// moq-transport has no RESTART; fall back to done + re-announce.
+					self.unannounce_namespace(&suffix, &mut namespace_streams).await;
+					self.announce_namespace(suffix, &mut namespace_streams).await?;
+				}
+				crate::Announced::Ended => {
+					self.unannounce_namespace(&suffix, &mut namespace_streams).await;
 				}
 			}
 		}
 
 		// Clean up remaining streams
-		for (suffix, (request_id, mut stream)) in namespace_streams {
+		let suffixes: Vec<crate::PathOwned> = namespace_streams.keys().cloned().collect();
+		for suffix in suffixes {
+			self.unannounce_namespace(&suffix, &mut namespace_streams).await;
+		}
+
+		Ok(())
+	}
+
+	/// Open a bidi stream and send a PublishNamespace, recording the stream for later teardown.
+	async fn announce_namespace(
+		&self,
+		suffix: crate::PathOwned,
+		namespace_streams: &mut HashMap<crate::PathOwned, (RequestId, Stream<S, Version>)>,
+	) -> Result<(), Error> {
+		tracing::debug!(broadcast = %self.origin.absolute(&suffix), "announce");
+
+		let request_id = self.control.next_request_id().await?;
+		let mut stream = Stream::open(&self.session, self.version).await?;
+
+		stream.writer.encode(&ietf::PublishNamespace::ID).await?;
+		stream
+			.writer
+			.encode(&ietf::PublishNamespace {
+				request_id,
+				track_namespace: suffix.as_path(),
+			})
+			.await?;
+
+		let type_id: u64 = stream.reader.decode().await?;
+		let size: u16 = stream.reader.decode().await?;
+		let mut data = stream.reader.read_exact(size as usize).await?;
+
+		match (self.version, type_id) {
+			(Version::Draft14, ietf::PublishNamespaceOk::ID) => {
+				let msg = ietf::PublishNamespaceOk::decode_msg(&mut data, self.version)?;
+				tracing::debug!(message = ?msg, "publish namespace ok");
+				namespace_streams.insert(suffix, (request_id, stream));
+			}
+			(Version::Draft14, ietf::PublishNamespaceError::ID) => {
+				let msg = ietf::PublishNamespaceError::decode_msg(&mut data, self.version)?;
+				tracing::warn!(message = ?msg, "publish namespace error");
+			}
+			(_, ietf::RequestOk::ID) => {
+				let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?;
+				tracing::debug!(message = ?msg, "publish namespace ok");
+				namespace_streams.insert(suffix, (request_id, stream));
+			}
+			(_, ietf::RequestError::ID) => {
+				let msg = ietf::RequestError::decode_msg(&mut data, self.version)?;
+				tracing::warn!(message = ?msg, "publish namespace error");
+			}
+			_ => return Err(Error::UnexpectedMessage),
+		}
+
+		Ok(())
+	}
+
+	/// Tear down the namespace stream for a suffix, sending PublishNamespaceDone where required.
+	async fn unannounce_namespace(
+		&self,
+		suffix: &crate::PathOwned,
+		namespace_streams: &mut HashMap<crate::PathOwned, (RequestId, Stream<S, Version>)>,
+	) {
+		tracing::debug!(broadcast = %self.origin.absolute(suffix), "unannounce");
+		if let Some((request_id, mut stream)) = namespace_streams.remove(suffix) {
+			// v14-16 sends PublishNamespaceDone; v17+ just closes the stream.
 			match self.version {
 				Version::Draft14 | Version::Draft15 | Version::Draft16 => {
 					let _ = stream
@@ -535,8 +550,6 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 			}
 			stream.writer.finish().ok();
 		}
-
-		Ok(())
 	}
 
 	/// Handle a SUBSCRIBE_NAMESPACE on its bidi stream.
@@ -589,17 +602,17 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 				let mut announced = origin.announced();
 
 				// Send initial NAMESPACE messages for currently active namespaces
-				while let Some((path, active)) = announced.try_next() {
-					let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path");
-					if active.is_some() {
+				// Send initial NAMESPACE messages for currently active namespaces.
+				// A restart is indistinguishable from an active here (the namespace is present).
+				while let Some((path, event)) = announced.try_next() {
+					if event.broadcast().is_some() {
+						let suffix = path
+							.strip_prefix(&prefix)
+							.expect("origin returned invalid path")
+							.to_owned();
 						tracing::debug!(broadcast = %origin.absolute(&path), "namespace");
 						stream.writer.encode(&ietf::Namespace::ID).await?;
-						stream
-							.writer
-							.encode(&ietf::Namespace {
-								suffix: suffix.to_owned(),
-							})
-							.await?;
+						stream.writer.encode(&ietf::Namespace { suffix }).await?;
 					}
 				}
 
@@ -608,27 +621,38 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 					tokio::select! {
 						biased;
 						res = stream.reader.closed() => return res,
-						next = announced.next() => {
-							match next {
-								Some((path, active)) => {
-									let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned();
-									if active.is_some() {
-										tracing::debug!(broadcast = %origin.absolute(&path), "namespace");
-										stream.writer.encode(&ietf::Namespace::ID).await?;
-										stream.writer.encode(&ietf::Namespace { suffix }).await?;
-									} else {
-										tracing::debug!(broadcast = %origin.absolute(&path), "namespace_done");
-										stream.writer.encode(&ietf::NamespaceDone::ID).await?;
-										stream.writer.encode(&ietf::NamespaceDone { suffix }).await?;
-									}
-								}
-								None => {
-									stream.writer.finish()?;
-									return stream.writer.closed().await;
-								}
+					next = announced.next() => {
+						let Some((path, event)) = next else {
+							stream.writer.finish()?;
+							return stream.writer.closed().await;
+						};
+
+						let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned();
+						let absolute = origin.absolute(&path).to_owned();
+
+						match event {
+							crate::Announced::Active(_) => {
+								tracing::debug!(broadcast = %absolute, "namespace");
+								stream.writer.encode(&ietf::Namespace::ID).await?;
+								stream.writer.encode(&ietf::Namespace { suffix }).await?;
+							}
+							crate::Announced::Restart(_) => {
+								// moq-transport has no RESTART; fall back to namespace_done + namespace.
+								tracing::debug!(broadcast = %absolute, "namespace_done");
+								stream.writer.encode(&ietf::NamespaceDone::ID).await?;
+								stream.writer.encode(&ietf::NamespaceDone { suffix: suffix.clone() }).await?;
+								tracing::debug!(broadcast = %absolute, "namespace");
+								stream.writer.encode(&ietf::Namespace::ID).await?;
+								stream.writer.encode(&ietf::Namespace { suffix }).await?;
+							}
+							crate::Announced::Ended => {
+								tracing::debug!(broadcast = %absolute, "namespace_done");
+								stream.writer.encode(&ietf::NamespaceDone::ID).await?;
+								stream.writer.encode(&ietf::NamespaceDone { suffix }).await?;
 							}
 						}
 					}
+					}
 				}
 			}
 		}
diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs
index f0738d269..e3e2fb568 100644
--- a/rs/moq-net/src/lite/announce.rs
+++ b/rs/moq-net/src/lite/announce.rs
@@ -4,6 +4,18 @@ use crate::{Origin, OriginList, Path, coding::*};
 
 use super::{Message, Version};
 
+/// Whether the negotiated version carries restart (REANNOUNCE) semantics: a duplicate ANNOUNCE
+/// (and the draft's explicit `restart` status) for an already-announced path. Older versions never
+/// defined this, so we neither send nor interpret it there; a restart is sent as an unannounce
+/// followed by a fresh announce instead.
+pub fn restart_supported(version: Version) -> bool {
+	// Explicitly list older versions so future versions default to supported.
+	!matches!(
+		version,
+		Version::Lite01 | Version::Lite02 | Version::Lite03 | Version::Lite04
+	)
+}
+
 /// Sent by the publisher to announce the availability of a track.
 /// The payload contains the contents of the wildcard.
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -43,6 +55,13 @@ impl Message for Announce<'_> {
 		Ok(match status {
 			AnnounceStatus::Active => Self::Active { suffix, hops },
 			AnnounceStatus::Ended => Self::Ended { suffix, hops },
+			// We encode a restart as a duplicate ANNOUNCE (a second `Active`), but on versions that
+			// support restart we also accept the draft's explicit `restart` status and treat it the
+			// same. For an already-announced path the subscriber turns it into a restart; for an
+			// unknown path it's a fresh announce. Older versions never defined this status, so it's
+			// an invalid value there.
+			AnnounceStatus::Restart if restart_supported(version) => Self::Active { suffix, hops },
+			AnnounceStatus::Restart => return Err(DecodeError::InvalidValue),
 		})
 	}
 
@@ -110,6 +129,9 @@ impl Message for AnnounceInterest<'_> {
 enum AnnounceStatus {
 	Ended = 0,
 	Active = 1,
+	/// The draft's explicit restart status. We never encode it (a restart goes out as a duplicate
+	/// `Active`), but we accept it on decode for forward/cross-compatibility.
+	Restart = 2,
 }
 
 impl Decode<Version> for AnnounceStatus {
@@ -173,3 +195,56 @@ impl Message for AnnounceInit<'_> {
 		Ok(())
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use bytes::Buf;
+
+	// Forge an ANNOUNCE with the draft's explicit `restart` status (2) for the given version.
+	fn encode_forged_restart(version: Version) -> bytes::Bytes {
+		// Encode a normal Active, then flip its status byte (1 -> 2).
+		let mut buf = bytes::BytesMut::new();
+		Announce::Active {
+			suffix: Path::new("foo/bar"),
+			hops: OriginList::new(),
+		}
+		.encode(&mut buf, version)
+		.expect("encode");
+
+		// Layout: <size varint><status u8><...>. The message is small, so the size is one byte and
+		// the status byte sits at index 1.
+		assert_eq!(
+			buf[1],
+			u8::from(AnnounceStatus::Active),
+			"expected an Active status byte"
+		);
+		buf[1] = u8::from(AnnounceStatus::Restart);
+		buf.freeze()
+	}
+
+	// On lite-05+ the explicit `restart` status is accepted and surfaced as an `Active` (the
+	// subscriber turns it into a restart for an already-announced path).
+	#[test]
+	fn decodes_explicit_restart_status_as_active_on_lite05() {
+		let version = Version::Lite05Wip;
+		let mut slice = encode_forged_restart(version);
+		let decoded = Announce::decode(&mut slice, version).expect("explicit restart must decode");
+		assert!(!slice.has_remaining(), "trailing bytes after decode");
+		assert!(
+			matches!(decoded, Announce::Active { .. }),
+			"restart should decode as Active"
+		);
+	}
+
+	// Older versions never defined the restart status, so it's an invalid value there.
+	#[test]
+	fn rejects_explicit_restart_status_before_lite05() {
+		let version = Version::Lite04;
+		let mut slice = encode_forged_restart(version);
+		assert!(
+			matches!(Announce::decode(&mut slice, version), Err(DecodeError::InvalidValue)),
+			"restart status must be rejected before lite-05"
+		);
+	}
+}
diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs
index b01126b54..2c8648737 100644
--- a/rs/moq-net/src/lite/publisher.rs
+++ b/rs/moq-net/src/lite/publisher.rs
@@ -202,21 +202,27 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 
 				// Send ANNOUNCE_INIT as the first message with all currently active paths
 				// We use `try_next()` to synchronously get the initial updates.
-				while let Some((path, active)) = announced.try_next() {
-					let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path");
-
-					if active.is_some() {
-						tracing::debug!(broadcast = %origin.absolute(&path), "announce");
-						let absolute = origin.absolute(&path).to_owned();
+				while let Some((path, event)) = announced.try_next() {
+					let suffix = path
+						.strip_prefix(&prefix)
+						.expect("origin returned invalid path")
+						.to_owned();
+					let absolute = origin.absolute(&path).to_owned();
+
+					// Lite01/02 only carries the set of active paths, so a restart is
+					// indistinguishable from an active here.
+					if event.broadcast().is_some() {
+						tracing::debug!(broadcast = %absolute, "announce");
 						let guard = stats.broadcast(&absolute).publisher();
-						let prev = stats_guards.insert(absolute, guard);
-						debug_assert!(prev.is_none(), "origin announced a path that was already active");
-						init.push(suffix.to_owned());
+						stats_guards.entry(absolute).or_insert(guard);
+						if !init.contains(&suffix) {
+							init.push(suffix);
+						}
 					} else {
 						// A potential race.
-						tracing::debug!(broadcast = %origin.absolute(&path), "unannounce");
-						stats_guards.remove(&origin.absolute(&path).to_owned());
-						init.retain(|path| path != &suffix);
+						tracing::debug!(broadcast = %absolute, "unannounce");
+						stats_guards.remove(&absolute);
+						init.retain(|p| p != &suffix);
 					}
 				}
 
@@ -234,70 +240,92 @@ impl<S: web_transport_trait::Session> Publisher<S> {
 				biased;
 				res = stream.reader.closed() => return res,
 				next = announced.next() => {
-					match next {
-						Some((path, active)) => {
-							let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned();
-
-							if let Some(active) = active {
-								// Skip if the peer asked us to exclude announces whose hop chain
-								// contains their id — they already saw this broadcast upstream.
-								if exclude_hop != 0 && active.hops.iter().any(|h| h.id == exclude_hop) {
-									tracing::debug!(
-										broadcast = %origin.absolute(&path),
-										%exclude_hop,
-										"skipping announce per peer's exclude_hop",
-									);
-									continue;
-								}
-								// Defense in depth: never echo an announce that already passed
-								// through us. The subscriber should drop these before they reach
-								// our origin, but if one slips through, don't propagate the loop.
-								if active.hops.contains(&self_origin) {
-									tracing::debug!(
-										broadcast = %origin.absolute(&path),
-										"skipping reflected announce",
-									);
-									continue;
-								}
-								tracing::debug!(broadcast = %origin.absolute(&path), "announce");
-								// Append our origin id to the hops so the next relay can detect loops.
-								// If the chain is already at MAX_HOPS, skip the announce — this link is
-								// effectively unreachable and the peer will eventually prune the loop.
-								let mut hops = active.hops.clone();
-								if hops.push(self_origin).is_err() {
-									tracing::warn!(
-										broadcast = %origin.absolute(&path),
-										"dropping announce; hop chain at MAX_HOPS (possible loop)",
-									);
+						let Some((path, event)) = next else {
+							stream.writer.finish()?;
+							return stream.writer.closed().await;
+						};
+
+						let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned();
+						let absolute = origin.absolute(&path).to_owned();
+
+						match event {
+							crate::Announced::Active(active) => {
+								let Some(hops) = Self::prepare_active_hops(&active.hops, self_origin, exclude_hop, &absolute) else {
 									continue;
-								}
-								let absolute = origin.absolute(&path).to_owned();
+								};
+								tracing::debug!(broadcast = %absolute, "announce");
 								let guard = stats.broadcast(&absolute).publisher();
 								let prev = stats_guards.insert(absolute, guard);
 								debug_assert!(prev.is_none(), "origin announced a path that was already active");
-								let msg = lite::Announce::Active { suffix, hops };
-								stream.writer.encode(&msg).await?;
-							} else {
-								tracing::debug!(broadcast = %origin.absolute(&path), "unannounce");
-								stats_guards.remove(&origin.absolute(&path).to_owned());
-								// An ended announce doesn't need hops — the receiver matches on path only.
-								let msg = lite::Announce::Ended {
-									suffix,
-									hops: OriginList::new(),
-								};
-								stream.writer.encode(&msg).await?;
+								stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?;
+							}
+							crate::Announced::Restart(active) => {
+								// On lite-05+ a restart travels as a duplicate ANNOUNCE (a second
+								// `Active` for an already-announced path). Older versions never defined
+								// that, so split it into an unannounce followed by a fresh announce.
+								match Self::prepare_active_hops(&active.hops, self_origin, exclude_hop, &absolute) {
+									Some(hops) => {
+										tracing::debug!(broadcast = %absolute, "restart");
+										// Continuity: keep the existing stats guard (no close + reopen).
+										if lite::restart_supported(version) {
+											stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?;
+										} else {
+											stream
+												.writer
+												.encode(&lite::Announce::Ended {
+													suffix: suffix.clone(),
+													hops: OriginList::new(),
+												})
+												.await?;
+											stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?;
+										}
+									}
+									None => {
+										// The replacement loops back to us; from this peer's view the broadcast is gone.
+										tracing::debug!(broadcast = %absolute, "restart replacement looped; unannouncing");
+										stats_guards.remove(&absolute);
+										stream.writer.encode(&lite::Announce::Ended { suffix, hops: OriginList::new() }).await?;
+									}
+								}
+							}
+							crate::Announced::Ended => {
+								tracing::debug!(broadcast = %absolute, "unannounce");
+								stats_guards.remove(&absolute);
+								// An ended announce doesn't need hops; the receiver matches on path only.
+								stream.writer.encode(&lite::Announce::Ended { suffix, hops: OriginList::new() }).await?;
 							}
-						},
-						None => {
-							stream.writer.finish()?;
-							return stream.writer.closed().await;
 						}
 					}
-				}
 			}
 		}
 	}
 
+	/// Decide whether to forward an active announcement and compute the outgoing hop chain.
+	///
+	/// Returns `None` when the announce should be skipped: the peer asked us to exclude it
+	/// (`exclude_hop`), it already passed through us (reflected loop), or the hop chain is full.
+	fn prepare_active_hops(
+		hops: &OriginList,
+		self_origin: Origin,
+		exclude_hop: u64,
+		absolute: &crate::Path,
+	) -> Option<OriginList> {
+		if exclude_hop != 0 && hops.iter().any(|h| h.id == exclude_hop) {
+			tracing::debug!(broadcast = %absolute, %exclude_hop, "skipping announce per peer's exclude_hop");
+			return None;
+		}
+		if hops.contains(&self_origin) {
+			tracing::debug!(broadcast = %absolute, "skipping reflected announce");
+			return None;
+		}
+		let mut hops = hops.clone();
+		if hops.push(self_origin).is_err() {
+			tracing::warn!(broadcast = %absolute, "dropping announce; hop chain at MAX_HOPS (possible loop)");
+			return None;
+		}
+		Some(hops)
+	}
+
 	pub async fn recv_subscribe(&mut self, mut stream: Stream<S, Version>) -> Result<(), Error> {
 		let subscribe = stream.reader.decode::<lite::Subscribe>().await?;
 
diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs
index 94aa9fd9e..d92f766fa 100644
--- a/rs/moq-net/src/lite/subscriber.rs
+++ b/rs/moq-net/src/lite/subscriber.rs
@@ -194,7 +194,19 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 				lite::Announce::Active { suffix, hops } => {
 					let path = prefix.join(&suffix);
 					let abs = self.origin.absolute(&path).to_owned();
-					if self.start_announce(path.clone(), hops, &mut producers)? {
+					if lite::restart_supported(self.version) && producers.contains_key(&path) {
+						// lite-05+ only: a duplicate ANNOUNCE for an already-announced path is a RESTART;
+						// atomically replace the broadcast. Older versions fall through to start_announce,
+						// which rejects the duplicate (Error::Duplicate).
+						if self.restart_announce(path.clone(), hops, &mut producers)? {
+							// Continuity: keep the existing stats guard if present.
+							stats_guards
+								.entry(abs.clone())
+								.or_insert_with(|| self.stats.broadcast(&abs).subscriber());
+						} else {
+							stats_guards.remove(&abs);
+						}
+					} else if self.start_announce(path.clone(), hops, &mut producers)? {
 						stats_guards.insert(abs.clone(), self.stats.broadcast(&abs).subscriber());
 					}
 				}
@@ -308,6 +320,46 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		Ok(true)
 	}
 
+	/// Handle a RESTART (a duplicate ANNOUNCE): atomically replace the broadcast at `path`.
+	///
+	/// Publishing the replacement before retiring the old producer lets the origin demote the old
+	/// broadcast to a backup and emit a single restart downstream, rather than an
+	/// unannounce/announce pair with a visible gap. Returns `Ok(false)` if the replacement was a
+	/// reflected loop (the broadcast is now gone), `Ok(true)` otherwise.
+	fn restart_announce(
+		&mut self,
+		path: PathOwned,
+		hops: crate::OriginList,
+		producers: &mut HashMap<PathOwned, BroadcastProducer>,
+	) -> Result<bool, Error> {
+		// Reflected loop: the replacement passed through us. Retire the broadcast.
+		if hops.contains(&self.self_origin) {
+			tracing::debug!(broadcast = %self.log_path(&path), "dropping reflected restart");
+			if let Some(mut old) = producers.remove(&path) {
+				old.abort(Error::Cancel).ok();
+			}
+			return Ok(false);
+		}
+
+		tracing::debug!(broadcast = %self.log_path(&path), hops = hops.len(), "restart");
+
+		let broadcast = Broadcast { hops }.produce();
+		let dynamic = broadcast.dynamic();
+
+		// Publish the replacement first so the origin restarts atomically; the old broadcast is
+		// demoted to a backup and dropped silently when we abort it below.
+		self.origin.publish_broadcast(path.clone(), broadcast.consume());
+
+		let old = producers.insert(path.clone(), broadcast.clone());
+		web_async::spawn(self.clone().run_broadcast(path.clone(), dynamic));
+
+		if let Some(mut old) = old {
+			old.abort(Error::Cancel).ok();
+		}
+
+		Ok(true)
+	}
+
 	async fn run_broadcast(self, path: PathOwned, mut broadcast: BroadcastDynamic) {
 		// Actually start serving subscriptions.
 		loop {
diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs
index 02db50141..060439f96 100644
--- a/rs/moq-net/src/model/origin.rs
+++ b/rs/moq-net/src/model/origin.rs
@@ -231,18 +231,21 @@ struct OriginBroadcast {
 	backup: VecDeque<BroadcastConsumer>,
 }
 
-/// One coalesced update queued for an `OriginConsumer`.
+/// One coalesced update queued for an `AnnounceConsumer`.
 ///
 /// At most one entry exists per path, so a slow consumer's pending set is bounded
-/// by the number of distinct paths. `UnannounceAnnounce` preserves the
-/// signal that the broadcast at a path was replaced (the consumer must see
-/// `(path, None)` before `(path, Some(new))`), while a stale
-/// `Announce` cancels with a subsequent `unannounce` because the consumer
-/// has not yet observed it.
+/// by the number of distinct paths. `UnannounceAnnounce` preserves the signal
+/// that a broadcast genuinely went away and a different one took its place (the
+/// consumer must see [`Announced::Ended`] before [`Announced::Active`]), while a
+/// stale `Announce` cancels with a subsequent `unannounce` because the consumer
+/// has not yet observed it. `Restart` is the atomic replacement: the broadcast
+/// at the path never became unavailable, so it collapses to a single
+/// [`Announced::Restart`] delivery.
 enum PendingUpdate {
 	Announce(BroadcastConsumer),
 	Unannounce,
 	UnannounceAnnounce(BroadcastConsumer),
+	Restart(BroadcastConsumer),
 }
 
 /// Pending updates keyed by path. `BTreeMap` keeps memory strictly bounded by
@@ -263,6 +266,22 @@ impl OriginConsumerState {
 			Some(PendingUpdate::Unannounce | PendingUpdate::UnannounceAnnounce(_)) => {
 				PendingUpdate::UnannounceAnnounce(broadcast)
 			}
+			// A restart the consumer hasn't drained yet; just swap in the newer broadcast.
+			Some(PendingUpdate::Restart(_)) => PendingUpdate::Restart(broadcast),
+		};
+		self.pending.insert(path, new);
+	}
+
+	fn apply_restart(&mut self, path: PathOwned, broadcast: BroadcastConsumer) {
+		let new = match self.pending.remove(&path) {
+			// Consumer has already drained the prior active; replace it atomically.
+			None => PendingUpdate::Restart(broadcast),
+			// Consumer hasn't seen the original announce yet; keep it a fresh announce.
+			Some(PendingUpdate::Announce(_)) => PendingUpdate::Announce(broadcast),
+			// Consumer saw the original active; collapse everything into one restart.
+			Some(PendingUpdate::Unannounce | PendingUpdate::UnannounceAnnounce(_) | PendingUpdate::Restart(_)) => {
+				PendingUpdate::Restart(broadcast)
+			}
 		};
 		self.pending.insert(path, new);
 	}
@@ -274,9 +293,9 @@ impl OriginConsumerState {
 			None | Some(PendingUpdate::Unannounce) => {
 				self.pending.insert(path, PendingUpdate::Unannounce);
 			}
-			// The embedded announce cancels with this unannounce; the consumer
-			// still needs the leading unannounce.
-			Some(PendingUpdate::UnannounceAnnounce(_)) => {
+			// The embedded/replacement announce cancels with this unannounce; the
+			// consumer still needs the leading unannounce.
+			Some(PendingUpdate::UnannounceAnnounce(_) | PendingUpdate::Restart(_)) => {
 				self.pending.insert(path, PendingUpdate::Unannounce);
 			}
 		}
@@ -285,16 +304,17 @@ impl OriginConsumerState {
 	/// Take one update to deliver to the consumer, if any.
 	fn take(&mut self) -> Option<OriginAnnounce> {
 		let path = self.pending.keys().next()?.clone();
-		match self.pending.remove(&path).unwrap() {
-			PendingUpdate::Announce(broadcast) => Some((path, Some(broadcast))),
-			PendingUpdate::Unannounce => Some((path, None)),
+		Some(match self.pending.remove(&path).unwrap() {
+			PendingUpdate::Announce(broadcast) => (path, Announced::Active(broadcast)),
+			PendingUpdate::Unannounce => (path, Announced::Ended),
 			PendingUpdate::UnannounceAnnounce(broadcast) => {
 				// Deliver the unannounce now; leave the trailing announce pending so
 				// the next take returns it for the same path.
 				self.pending.insert(path.clone(), PendingUpdate::Announce(broadcast));
-				Some((path, None))
+				(path, Announced::Ended)
 			}
-		}
+			PendingUpdate::Restart(broadcast) => (path, Announced::Restart(broadcast)),
+		})
 	}
 }
 
@@ -314,11 +334,13 @@ impl AnnounceConsumerNotify {
 			.apply_announce(path, broadcast);
 	}
 
-	fn reannounce(&self, path: impl AsPath, broadcast: BroadcastConsumer) {
+	fn restart(&self, path: impl AsPath, broadcast: BroadcastConsumer) {
 		let path = path.as_path().strip_prefix(&self.root).unwrap().to_owned();
-		let mut state = self.state.write().ok().expect("consumer closed");
-		state.apply_unannounce(path.clone());
-		state.apply_announce(path, broadcast);
+		self.state
+			.write()
+			.ok()
+			.expect("consumer closed")
+			.apply_restart(path, broadcast);
 	}
 
 	fn unannounce(&self, path: impl AsPath) {
@@ -353,13 +375,13 @@ impl NotifyNode {
 		}
 	}
 
-	fn reannounce(&mut self, path: impl AsPath, broadcast: &BroadcastConsumer) {
+	fn restart(&mut self, path: impl AsPath, broadcast: &BroadcastConsumer) {
 		for consumer in self.consumers.values() {
-			consumer.reannounce(path.as_path(), broadcast.clone());
+			consumer.restart(path.as_path(), broadcast.clone());
 		}
 
 		if let Some(parent) = &self.parent {
-			parent.lock().reannounce(path, broadcast);
+			parent.lock().restart(path, broadcast);
 		}
 	}
 
@@ -426,7 +448,7 @@ impl OriginNode {
 			//
 			// Drop duplicates (same underlying broadcast delivered via multiple links) so the
 			// backup queue can't accumulate clones of the active entry and trigger redundant
-			// reannouncements when a peer churns.
+			// restartments when a peer churns.
 			if existing.active.is_clone(broadcast) || existing.backup.iter().any(|b| b.is_clone(broadcast)) {
 				return;
 			}
@@ -436,7 +458,7 @@ impl OriginNode {
 				existing.active = broadcast.clone();
 				existing.backup.push_back(old);
 
-				self.notify.lock().reannounce(full, broadcast);
+				self.notify.lock().restart(full, broadcast);
 			} else {
 				// Longer path: keep as a backup in case the active one drops.
 				existing.backup.push_back(broadcast.clone());
@@ -529,7 +551,7 @@ impl OriginNode {
 			if let Some(idx) = best {
 				let active = entry.backup.remove(idx).expect("index in range");
 				entry.active = active;
-				self.notify.lock().reannounce(full, &entry.active);
+				self.notify.lock().restart(full, &entry.active);
 			} else {
 				// No more backups, so remove the entry.
 				self.broadcast = None;
@@ -626,8 +648,39 @@ impl Default for OriginNodes {
 	}
 }
 
-/// A broadcast path and its associated consumer, or None if closed.
-pub type OriginAnnounce = (PathOwned, Option<BroadcastConsumer>);
+/// A path and what happened to the broadcast there, delivered by [`AnnounceConsumer`].
+pub type OriginAnnounce = (PathOwned, Announced);
+
+/// What happened to a broadcast at a path.
+#[derive(Clone)]
+pub enum Announced {
+	/// A broadcast became available.
+	Active(BroadcastConsumer),
+	/// The broadcast was replaced without an interruption in availability (e.g. a
+	/// relay failover or a shorter hop path arriving).
+	///
+	/// Carries the replacement broadcast. On the wire this is a duplicate ANNOUNCE
+	/// (an active announcement for a path that is already announced, with no
+	/// intervening unannounce); there is no distinct status byte.
+	Restart(BroadcastConsumer),
+	/// The broadcast is no longer available.
+	Ended,
+}
+
+impl Announced {
+	/// The broadcast consumer, or `None` if the broadcast ended.
+	///
+	/// Both [`Active`](Self::Active) and [`Restart`](Self::Restart) carry a
+	/// broadcast; [`Ended`](Self::Ended) does not. This is the legacy
+	/// `Option<BroadcastConsumer>` view for callers that don't distinguish a fresh
+	/// announce from a restart.
+	pub fn broadcast(self) -> Option<BroadcastConsumer> {
+		match self {
+			Self::Active(broadcast) | Self::Restart(broadcast) => Some(broadcast),
+			Self::Ended => None,
+		}
+	}
+}
 
 /// Announces broadcasts to consumers over the network.
 #[derive(Clone)]
@@ -679,7 +732,7 @@ impl OriginProducer {
 	/// If there is already a broadcast with the same path, the new one replaces the active only
 	/// if its hop path is shorter or equal; otherwise it is queued as a backup.
 	/// When the active broadcast closes, the backup with the shortest hop path is promoted and
-	/// reannounced. Backups that close before being promoted are silently dropped.
+	/// restartd. Backups that close before being promoted are silently dropped.
 	///
 	/// Returns false if the broadcast is not allowed to be published.
 	pub fn publish_broadcast(&self, path: impl AsPath, broadcast: BroadcastConsumer) -> bool {
@@ -850,10 +903,10 @@ impl OriginConsumer {
 
 		let mut announced = consumer.announced();
 		loop {
-			let (announced_path, broadcast) = announced.next().await?;
+			let (announced_path, event) = announced.next().await?;
 			// `scope` narrows by prefix, but we only want an exact-path match.
 			if announced_path.as_path() == path {
-				if let Some(broadcast) = broadcast {
+				if let Some(broadcast) = event.broadcast() {
 					return Some(broadcast);
 				}
 			}
@@ -1031,23 +1084,41 @@ use futures::FutureExt;
 impl AnnounceConsumer {
 	pub fn assert_next(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) {
 		let expected = expected.as_path();
-		let (path, active) = self.next().now_or_never().expect("next blocked").expect("no next");
+		let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next");
+		assert!(matches!(event, Announced::Active(_)), "should be an active announce");
+		assert_eq!(path, expected, "wrong path");
+		assert!(
+			event.broadcast().unwrap().is_clone(broadcast),
+			"should be the same broadcast"
+		);
+	}
+
+	pub fn assert_next_restart(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) {
+		let expected = expected.as_path();
+		let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next");
+		assert!(matches!(event, Announced::Restart(_)), "should be a restart");
 		assert_eq!(path, expected, "wrong path");
-		assert!(active.unwrap().is_clone(broadcast), "should be the same broadcast");
+		assert!(
+			event.broadcast().unwrap().is_clone(broadcast),
+			"should be the same broadcast"
+		);
 	}
 
 	pub fn assert_try_next(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) {
 		let expected = expected.as_path();
-		let (path, active) = self.try_next().expect("no next");
+		let (path, event) = self.try_next().expect("no next");
 		assert_eq!(path, expected, "wrong path");
-		assert!(active.unwrap().is_clone(broadcast), "should be the same broadcast");
+		assert!(
+			event.broadcast().unwrap().is_clone(broadcast),
+			"should be the same broadcast"
+		);
 	}
 
 	pub fn assert_next_none(&mut self, expected: impl AsPath) {
 		let expected = expected.as_path();
-		let (path, active) = self.next().now_or_never().expect("next blocked").expect("no next");
+		let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next");
 		assert_eq!(path, expected, "wrong path");
-		assert!(active.is_none(), "should be unannounced");
+		assert!(event.broadcast().is_none(), "should be unannounced");
 	}
 
 	pub fn assert_next_wait(&mut self) {
@@ -1179,7 +1250,7 @@ mod tests {
 		origin.publish_broadcast("test", consumer3.clone());
 		assert!(consumer.get_broadcast("test").is_some());
 
-		// On equal hop lengths, each new publish replaces the active and reannounces.
+		// On equal hop lengths, each new publish replaces the active and restarts.
 		// Because the cursor hasn't drained between publishes, the stale announces
 		// collapse with their following unannounces and only the final active broadcast
 		// is delivered.
@@ -1195,15 +1266,14 @@ mod tests {
 		assert!(consumer.get_broadcast("test").is_some());
 		announced.assert_next_wait();
 
-		// Drop the active, we should reannounce with the remaining backup.
+		// Drop the active, we should restart with the remaining backup.
 		drop(broadcast3);
 
 		// Wait for the async task to run.
 		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
 
 		assert!(consumer.get_broadcast("test").is_some());
-		announced.assert_next_none("test");
-		announced.assert_next("test", &consumer1);
+		announced.assert_next_restart("test", &consumer1);
 
 		// Drop the final broadcast, we should unannounce.
 		drop(broadcast1);
@@ -1216,6 +1286,39 @@ mod tests {
 		announced.assert_next_wait();
 	}
 
+	#[tokio::test]
+	async fn test_restart_after_drain() {
+		// Replacing the active broadcast (equal hops) after the consumer has drained
+		// the original announce is delivered as a single atomic restart.
+		let origin = Origin::random().produce();
+		let a = Broadcast::new().produce();
+		let b = Broadcast::new().produce();
+
+		let mut announced = origin.consume().announced();
+		origin.publish_broadcast("test", a.consume());
+		announced.assert_next("test", &a.consume());
+
+		origin.publish_broadcast("test", b.consume());
+		announced.assert_next_restart("test", &b.consume());
+		announced.assert_next_wait();
+	}
+
+	#[tokio::test]
+	async fn test_restart_undrained_stays_active() {
+		// If the consumer hasn't observed the original announce yet, a restart just
+		// swaps in the newer broadcast and is still delivered as a fresh Active.
+		let origin = Origin::random().produce();
+		let a = Broadcast::new().produce();
+		let b = Broadcast::new().produce();
+
+		let mut announced = origin.consume().announced();
+		origin.publish_broadcast("test", a.consume());
+		origin.publish_broadcast("test", b.consume());
+
+		announced.assert_next("test", &b.consume());
+		announced.assert_next_wait();
+	}
+
 	#[tokio::test]
 	async fn test_duplicate_reverse() {
 		tokio::time::pause();
diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs
index 60d2ba948..fee117c2e 100644
--- a/rs/moq-net/src/stats.rs
+++ b/rs/moq-net/src/stats.rs
@@ -1063,8 +1063,8 @@ mod tests {
 		let _t2 = bs2.publisher().track("video");
 
 		tokio::time::advance(Duration::from_millis(1)).await;
-		let (path, broadcast) = consumer.next().await.expect("expected announce");
-		assert!(broadcast.is_some());
+		let (path, event) = consumer.next().await.expect("expected announce");
+		assert!(event.broadcast().is_some());
 		assert_eq!(path.as_str(), ".stats/node/sjc/1");
 	}
 
@@ -1078,8 +1078,8 @@ mod tests {
 		let _t = bs.publisher().track("video");
 
 		tokio::time::advance(Duration::from_millis(1)).await;
-		let (path, broadcast) = consumer.next().await.expect("expected announce");
-		assert!(broadcast.is_some());
+		let (path, event) = consumer.next().await.expect("expected announce");
+		assert!(event.broadcast().is_some());
 		assert_eq!(path.as_str(), ".stats/node");
 	}
 
@@ -1161,8 +1161,8 @@ mod tests {
 
 		tokio::time::advance(Duration::from_millis(1100)).await;
 
-		let (_path, broadcast) = consumer.next().await.expect("expected announce");
-		let broadcast = broadcast.expect("active");
+		let (_path, event) = consumer.next().await.expect("expected announce");
+		let broadcast = event.broadcast().expect("active");
 		let track = broadcast
 			.subscribe_track(&Track {
 				name: "publisher.json".into(),
@@ -1190,8 +1190,8 @@ mod tests {
 
 		tokio::time::advance(Duration::from_millis(1100)).await;
 
-		let (_path, broadcast) = consumer.next().await.expect("announce");
-		let broadcast = broadcast.expect("active");
+		let (_path, event) = consumer.next().await.expect("announce");
+		let broadcast = event.broadcast().expect("active");
 		let track = broadcast
 			.subscribe_track(&Track {
 				name: "publisher.json".into(),
@@ -1224,8 +1224,8 @@ mod tests {
 
 		tokio::time::advance(Duration::from_millis(1100)).await;
 
-		let (_path, broadcast) = consumer.next().await.expect("announce");
-		let broadcast = broadcast.expect("active");
+		let (_path, event) = consumer.next().await.expect("announce");
+		let broadcast = event.broadcast().expect("active");
 		let track = broadcast
 			.subscribe_track(&Track {
 				name: "publisher.json".into(),
@@ -1302,8 +1302,8 @@ mod tests {
 
 		drive_ticks(2).await;
 
-		let (_path, broadcast) = consumer.next().await.expect("announce");
-		let broadcast = broadcast.expect("active");
+		let (_path, event) = consumer.next().await.expect("announce");
+		let broadcast = event.broadcast().expect("active");
 
 		// External publisher slot SHOULD include foo/bar.
 		let pub_track = broadcast
diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs
index 2fae08108..3f2072917 100644
--- a/rs/moq-relay/src/cluster.rs
+++ b/rs/moq-relay/src/cluster.rs
@@ -20,7 +20,7 @@ const MESH_PREFIX: &str = ".internal/origins";
 const SWEEP_INTERVAL: Duration = Duration::from_secs(30);
 
 /// How long a peer must stay unannounced before we abort the dial. Must clear the
-/// "prefer shorter hop" reannounce flap (which arrives as unannounce-then-announce
+/// "prefer shorter hop" restart flap (which arrives as unannounce-then-announce
 /// within sub-milliseconds) plus reasonable churn from a peer restart.
 const STALE_AFTER: Duration = Duration::from_secs(60);
 
@@ -76,7 +76,7 @@ impl DialMap {
 	}
 
 	/// Clear any pending-unannounce on `peer`. Returns `true` if a timestamp was
-	/// actually cleared (useful for callers that want to log the reannounce).
+	/// actually cleared (useful for callers that want to log the restart).
 	fn mark_announced(&self, peer: &str) -> bool {
 		let mut map = self.inner.lock().expect("dial map poisoned");
 		map.get_mut(peer)
@@ -343,7 +343,7 @@ impl Cluster {
 	/// announced URL. Unannounces don't abort immediately — they just mark the
 	/// entry as "pending cleanup" with a timestamp. A periodic sweep evicts
 	/// entries whose unannounce has stuck for [`STALE_AFTER`]. The "prefer
-	/// shorter hop" path in OriginProducer delivers reannouncements as
+	/// shorter hop" path in OriginProducer delivers restartments as
 	/// unannounce-then-announce within sub-milliseconds, which clears the
 	/// pending-cleanup timestamp long before the sweep fires.
 	async fn run_discovery(self, self_url: String, token: String, dialed: DialMap) {
@@ -361,17 +361,17 @@ impl Cluster {
 		loop {
 			tokio::select! {
 				ann = announced.next() => {
-					let Some((relative, announced)) = ann else { return; };
+					let Some((relative, event)) = ann else { return; };
 					let peer = relative.as_str();
 					if peer == self_url {
 						continue;
 					}
 					let peer = peer.to_owned();
-					match announced {
+					match event.broadcast() {
 						Some(_) => {
 							if dialed.contains(&peer) {
 								if dialed.mark_announced(&peer) {
-									tracing::debug!(%peer, "reannounce within sweep window; keeping dial");
+									tracing::debug!(%peer, "restart within sweep window; keeping dial");
 								}
 								continue;
 							}
@@ -525,7 +525,7 @@ mod tests {
 		assert!(dialed.contains("healthy:4443"));
 	}
 
-	/// A reannounce after an unannounce clears the pending-sweep timestamp, so
+	/// A restart after an unannounce clears the pending-sweep timestamp, so
 	/// the entry survives even if the original unannounce was old enough to
 	/// otherwise trigger eviction.
 	#[tokio::test]
@@ -637,9 +637,9 @@ mod tests {
 		tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
 
 		// The self-registration broadcast must be visible on the origin.
-		let (path, broadcast) = watcher.try_next().expect("self-registration must be published");
+		let (path, event) = watcher.try_next().expect("self-registration must be published");
 		assert_eq!(path.as_str(), ".internal/origins/rendezvous.example.com:4443");
-		assert!(broadcast.is_some());
+		assert!(event.broadcast().is_some());
 
 		// run() must NOT have returned: dropping the broadcast (via run returning)
 		// would unannounce the registration immediately. Use a short timeout to
diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs
index 60c5e3f11..af4258e51 100644
--- a/rs/moq-relay/src/web.rs
+++ b/rs/moq-relay/src/web.rs
@@ -466,8 +466,8 @@ async fn serve_announced(
 	let mut announced = origin.consume().announced();
 	let mut broadcasts = Vec::new();
 
-	while let Some((suffix, active)) = announced.try_next() {
-		if active.is_some() {
+	while let Some((suffix, event)) = announced.try_next() {
+		if event.broadcast().is_some() {
 			broadcasts.push(suffix);
 		}
 	}
diff --git a/rs/moq-relay/tests/smoke.rs b/rs/moq-relay/tests/smoke.rs
index 8788736b1..2c45a9f80 100644
--- a/rs/moq-relay/tests/smoke.rs
+++ b/rs/moq-relay/tests/smoke.rs
@@ -158,7 +158,7 @@ async fn relay_websocket_round_trip_uses_newest_version() {
 		.expect("origin closed");
 	// Auth root for `/smoke` is "smoke"; the broadcast "test" announces underneath.
 	assert_eq!(path.as_str(), "test");
-	let bc = bc.expect("expected announce, got unannounce");
+	let bc = bc.broadcast().expect("expected announce, got unannounce");
 
 	let mut track_sub = bc.subscribe_track(&Track::new("video")).expect("subscribe_track");
 	let mut group_sub = tokio::time::timeout(TIMEOUT, track_sub.recv_group())
