moq-net: RAII broadcast announcements via OriginPublish; remove broadcast abort/close

rs/hang/examples/video.rs

@@ -1,4 +1,5 @@
 // cargo run --example video
+use anyhow::Context;
 use bytes::Bytes;
 
 #[tokio::main]
@@ -94,7 +95,9 @@ async fn run_broadcast(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 
 	// NOTE: The path is empty because we're using the URL to scope the broadcast.
 	// OPTIONAL: We publish after inserting the tracks just to avoid a nearly impossible race condition.
-	origin.publish_broadcast("", broadcast.consume());
+	let _publish = origin
+		.publish_broadcast("", broadcast.consume())
+		.context("failed to publish broadcast")?;
 
 	// Wrap in a Producer for keyframe-based group management.
 	let mut producer = moq_mux::container::Producer::new(track, moq_mux::catalog::hang::Container::Legacy);

rs/libmoq/src/api.rs

@@ -214,7 +214,9 @@ pub extern "C" fn moq_origin_create() -> i32 {
 ///
 /// The broadcast will be announced to any origin consumers, such as over the network.
 ///
-/// Returns a zero on success, or a negative code on failure.
+/// Returns a positive publish handle on success, or a negative code on failure. The broadcast
+/// stays announced until the handle is passed to [moq_origin_unpublish]; closing the broadcast
+/// itself does not unannounce it.
 ///
 /// # Safety
 /// - The caller must ensure that path is a valid pointer to path_len bytes of data.
@@ -231,6 +233,18 @@ pub unsafe extern "C" fn moq_origin_publish(origin: u32, path: *const c_char, pa
 	})
 }
 
+/// Unannounce a broadcast previously published with [moq_origin_publish].
+///
+/// Takes the publish handle returned by [moq_origin_publish]. Returns zero on success, or a
+/// negative code on failure.
+#[unsafe(no_mangle)]
+pub extern "C" fn moq_origin_unpublish(publish: u32) -> i32 {
+	ffi::enter(move || {
+		let publish = ffi::parse_id(publish)?;
+		State::lock().origin.unpublish(publish)
+	})
+}
+
 /// Learn about all broadcasts published to an origin.
 ///
 /// `on_announce` is invoked with a positive announced ID for each broadcast,

rs/libmoq/src/origin.rs

@@ -25,6 +25,10 @@ pub struct Origin {
 	/// Active origin producers for publishing and consuming broadcasts.
 	active: NonZeroSlab<moq_net::OriginProducer>,
 
+	/// Announcement guards from `publish`. Removing an entry (via `unpublish`) drops the
+	/// guard, which unannounces the broadcast.
+	published: NonZeroSlab<moq_net::OriginPublish>,
+
 	/// Broadcast announcement information (path, active status).
 	announced: NonZeroSlab<(String, bool)>,
 
@@ -185,14 +189,25 @@ impl Origin {
 		Ok(())
 	}
 
+	/// Announce `broadcast` under `path`, returning a publish handle. The announcement stays
+	/// live until [`Self::unpublish`] is called with that handle (independent of the broadcast's
+	/// own lifetime). Errors with [`Error::Moq`] if the path is outside the origin's scope.
 	pub fn publish<P: moq_net::AsPath>(
 		&mut self,
 		origin: Id,
 		path: P,
 		broadcast: moq_net::BroadcastConsumer,
-	) -> Result<(), Error> {
-		let origin = self.active.get_mut(origin).ok_or(Error::OriginNotFound)?;
-		origin.publish_broadcast(path, broadcast);
+	) -> Result<Id, Error> {
+		let origin = self.active.get(origin).ok_or(Error::OriginNotFound)?;
+		let publish = origin.publish_broadcast(path, broadcast)?;
+		self.published.insert(publish)
+	}
+
+	/// Drop a publish handle from [`Self::publish`], unannouncing the broadcast.
+	pub fn unpublish(&mut self, publish: Id) -> Result<(), Error> {
+		// Dropping the removed guard is what unannounces the broadcast.
+		let publish = self.published.remove(publish).ok_or(Error::BroadcastNotFound)?;
+		drop(publish);
 		Ok(())
 	}
 

rs/moq-boy/src/main.rs

@@ -219,7 +219,9 @@ async fn run(config: &Config) -> Result<()> {
 	let viewer_prefix = config.prefix_viewer.as_deref().unwrap_or(&default_viewer_prefix);
 
 	let broadcast_path = format!("{game_prefix}/{name}");
-	publish_origin.publish_broadcast(&broadcast_path, broadcast.consume());
+	let _publish = publish_origin
+		.publish_broadcast(&broadcast_path, broadcast.consume())
+		.context("failed to publish broadcast")?;
 
 	// Consume origin: viewer broadcasts under the viewer prefix.
 	// JS publishes viewer feedback at "{viewer_prefix}/{name}/{viewerId}"

rs/moq-cli/src/client.rs

@@ -1,12 +1,15 @@
 use crate::Publish;
 
+use anyhow::Context;
 use hang::moq_net;
 use url::Url;
 
 pub async fn run_client(client: moq_native::Client, url: Url, name: String, publish: Publish) -> anyhow::Result<()> {
 	// Create an origin producer to publish to the broadcast.
 	let origin = moq_net::Origin::random().produce();
-	origin.publish_broadcast(&name, publish.consume());
+	let _publish = origin
+		.publish_broadcast(&name, publish.consume())
+		.context("failed to publish broadcast")?;
 
 	tracing::info!(%url, %name, "connecting");
 

rs/moq-cli/src/server.rs

@@ -1,3 +1,4 @@
+use anyhow::Context;
 use hang::moq_net;
 
 pub async fn run_server(
@@ -40,7 +41,9 @@ async fn run_serve_session(
 ) -> anyhow::Result<()> {
 	// Create an origin producer to publish to the broadcast.
 	let origin = moq_net::Origin::random().produce();
-	origin.publish_broadcast(&name, consumer);
+	let _publish = origin
+		.publish_broadcast(&name, consumer)
+		.context("failed to publish broadcast")?;
 
 	// Blindly accept the session (WebTransport or QUIC), regardless of the URL.
 	let session = session.with_publisher(origin.clone()).ok().await?;

rs/moq-ffi/src/origin.rs

@@ -114,9 +114,16 @@ impl MoqOriginProducer {
 	pub fn announce(&self, path: String, broadcast: &MoqBroadcastProducer) -> Result<(), MoqError> {
 		let _guard = crate::ffi::RUNTIME.enter();
 		let consumer = broadcast.consume_inner()?;
-		if !self.inner.publish_broadcast(path.as_str(), consumer) {
-			return Err(MoqError::Unauthorized);
-		}
+		// Surfaces Error::Unauthorized (out of scope) via the MoqError::Protocol conversion.
+		let publish = self.inner.publish_broadcast(path.as_str(), consumer.clone())?;
+
+		// Auto-unannounce when the broadcast closes (all producers dropped). The origin no longer
+		// watches closure itself, so the spawn lives here at the runtime-bound FFI boundary.
+		tokio::spawn(async move {
+			consumer.closed().await;
+			drop(publish);
+		});
+
 		Ok(())
 	}
 }

rs/moq-gst/src/sink/imp.rs

@@ -393,11 +393,10 @@ async fn run_session(
 
 	let catalog = moq_mux::catalog::hang::Producer::new(&mut broadcast)?;
 
-	anyhow::ensure!(
-		origin.publish_broadcast(&settings.broadcast, broadcast_consumer),
-		"failed to publish broadcast {}",
-		settings.broadcast
-	);
+	// Held for the lifetime of this task; dropping it (on return) unannounces the broadcast.
+	let _publish = origin
+		.publish_broadcast(&settings.broadcast, broadcast_consumer)
+		.context("failed to publish broadcast")?;
 
 	let client = client.with_publisher(origin.clone());
 	let session = client.connect(settings.url.clone()).await?;

rs/moq-native/examples/chat.rs

@@ -1,5 +1,7 @@
 // cargo run --example chat
 
+use anyhow::Context;
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
 	// Optional: Use moq_native to configure a logger.
@@ -46,7 +48,9 @@ async fn run_broadcast(origin: moq_net::OriginProducer) -> anyhow::Result<()> {
 	// NOTE: The path is empty because we're using the URL to scope the broadcast.
 	// If you put "alice" here, it would be published as "anon/chat-example/alice".
 	// OPTIONAL: We publish after inserting the track just to avoid a nearly impossible race condition.
-	origin.publish_broadcast("", broadcast.consume());
+	let _publish = origin
+		.publish_broadcast("", broadcast.consume())
+		.context("failed to publish broadcast")?;
 
 	// Create a group.
 	// Each group is independent and the newest group(s) will be prioritized.

rs/moq-native/examples/clock.rs

@@ -70,7 +70,9 @@ async fn main() -> anyhow::Result<()> {
 			let track = broadcast.create_track(track, None)?;
 			let clock = Publisher::new(track);
 
-			origin.publish_broadcast(&config.broadcast, broadcast.consume());
+			let _publish = origin
+				.publish_broadcast(&config.broadcast, broadcast.consume())
+				.context("failed to publish broadcast")?;
 
 			let reconnect = client.with_publisher(origin.clone()).reconnect(config.url);
 

rs/moq-net/src/ietf/subscriber.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use crate::{
 	BroadcastDynamic, BroadcastInfo, Error, Frame, FrameProducer, Group, GroupProducer, MAX_FRAME_SIZE, OriginProducer,
-	Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer, TrackRequest,
+	OriginPublish, Path, PathOwned, StatsHandle, SubscriberStats, SubscriberTrack, TrackProducer, TrackRequest,
 	coding::{Reader, Stream},
 	ietf::{self, Control, FilterType, GroupOrder, RequestId},
 	model::BroadcastProducer,
@@ -43,6 +43,11 @@ struct BroadcastState {
 	// active number of PUBLISH or PUBLISH_NAMESPACE messages.
 	count: usize,
 
+	/// Announcement guard into our origin. Dropped when the entry is removed,
+	/// which unannounces the broadcast.
+	#[allow(dead_code)]
+	publish: OriginPublish,
+
 	/// Subscriber-side announce guard (bumps `announced` / `announced_closed`),
 	/// held for as long as the broadcast is announced into our origin.
 	_stats: SubscriberStats,
@@ -431,10 +436,12 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 			}
 			Entry::Vacant(entry) => {
 				let broadcast = BroadcastInfo::new().produce();
-				self.origin.publish_broadcast(path.clone(), broadcast.consume());
+				// Propagates Error::Unauthorized if the path is out of scope.
+				let publish = self.origin.publish_broadcast(path.clone(), broadcast.consume())?;
 				entry.insert(BroadcastState {
 					producer: broadcast.clone(),
 					count: 1,
+					publish,
 					_stats: self.stats.broadcast(&abs).subscriber(),
 				});
 				broadcast

rs/moq-net/src/lite/subscriber.rs

@@ -1,5 +1,5 @@
 use std::{
-	collections::{HashMap, hash_map::Entry},
+	collections::HashMap,
 	pin::Pin,
 	sync::{Arc, atomic},
 	task::Poll,
@@ -10,11 +10,10 @@ use futures::{FutureExt, StreamExt, future::BoxFuture, stream::FuturesUnordered}
 
 use crate::{
 	AsPath, BandwidthProducer, BroadcastDynamic, BroadcastInfo, Compression, Error, Frame, FrameProducer, Group,
-	GroupProducer, GroupRequest, MAX_FRAME_SIZE, OriginProducer, Path, PathOwned, StatsHandle, SubscriberStats,
-	SubscriberTrack, Subscription, Timescale, Timestamp, TrackInfo, TrackProducer, TrackRequest,
+	GroupProducer, GroupRequest, MAX_FRAME_SIZE, OriginProducer, OriginPublish, Path, PathOwned, StatsHandle,
+	SubscriberStats, SubscriberTrack, Subscription, Timescale, Timestamp, TrackInfo, TrackProducer, TrackRequest,
 	coding::{Reader, Stream},
 	lite,
-	model::BroadcastProducer,
 };
 
 use super::{ConnectingProducer, Version};
@@ -273,8 +272,8 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 					// The matching Active may have been silently dropped by
 					// start_announce as a reflected loop, in which case
 					// `producers` has no entry; that's expected, not an error.
-					if let Some(mut producer) = producers.remove(&path) {
-						producer.abort(Error::Cancel).ok();
+					// Dropping the entry drops its OriginPublish guard, which unannounces.
+					if producers.remove(&path).is_some() {
 						let abs = self.origin.absolute(&path).to_owned();
 						stats_guards.remove(&abs);
 					}
@@ -346,7 +345,7 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		// the full `[src...sender]` chain Lite04 stored. None for older versions,
 		// where the sender already appended itself.
 		responder_origin: Option<crate::Origin>,
-		producers: &mut HashMap<PathOwned, BroadcastProducer>,
+		producers: &mut HashMap<PathOwned, OriginPublish>,
 	) -> Result<bool, Error> {
 		if let Some(responder) = responder_origin {
 			// If the chain is already full, drop the announce — the same decision
@@ -369,25 +368,30 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 			return Ok(false);
 		}
 
+		// Make sure the peer doesn't double announce.
+		if producers.contains_key(&path) {
+			return Err(Error::Duplicate);
+		}
+
 		tracing::debug!(broadcast = %self.log_path(&path), hops = hops.len(), "announce");
 
 		let broadcast = BroadcastInfo { hops }.produce();
 
-		// Make sure the peer doesn't double announce.
-		match producers.entry(path.to_owned()) {
-			Entry::Occupied(_) => return Err(Error::Duplicate),
-			Entry::Vacant(entry) => entry.insert(broadcast.clone()),
-		};
-
 		// Create the dynamic handler BEFORE publishing, so that consumers
 		// see dynamic >= 1 immediately when they receive the announcement.
 		// Otherwise there's a race on multi-threaded runtimes where a consumer
 		// can call consume_track() before dynamic is incremented, getting NotFound.
 		let dynamic = broadcast.dynamic();
 
-		// Run the broadcast in the background until all consumers are dropped.
-		self.origin.publish_broadcast(path.clone(), broadcast.consume());
+		// Publish into the origin. An error means the path is outside our scope, so don't announce
+		// or spawn a server for it. Reflections are already filtered above.
+		let Ok(publish) = self.origin.publish_broadcast(path.clone(), broadcast.consume()) else {
+			return Ok(false);
+		};
+
+		producers.insert(path.clone(), publish);
 
+		// Run the broadcast in the background until all consumers are dropped.
 		web_async::spawn(self.clone().run_broadcast(path, dynamic));
 
 		Ok(true)
@@ -407,7 +411,7 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		// rebuild the full chain since the sender no longer stamps itself. None for older
 		// versions. See `start_announce`.
 		responder_origin: Option<crate::Origin>,
-		producers: &mut HashMap<PathOwned, BroadcastProducer>,
+		producers: &mut HashMap<PathOwned, OriginPublish>,
 	) -> Result<bool, Error> {
 		// Reflected loop (or a full chain): the replacement can't be used here. Retire the broadcast.
 		let reflected = match responder_origin {
@@ -416,9 +420,8 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		};
 		if reflected {
 			tracing::debug!(broadcast = %self.log_path(&path), "dropping reflected restart");
-			if let Some(mut old) = producers.remove(&path) {
-				old.abort(Error::Cancel).ok();
-			}
+			// Dropping the entry drops its guard, unannouncing the broadcast.
+			producers.remove(&path);
 			return Ok(false);
 		}
 
@@ -428,15 +431,18 @@ impl<S: web_transport_trait::Session> Subscriber<S> {
 		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());
+		// demoted to a backup and removed silently when we drop its guard below.
+		let Ok(publish) = self.origin.publish_broadcast(path.clone(), broadcast.consume()) else {
+			// Origin rejected the replacement; retire the existing broadcast.
+			producers.remove(&path);
+			return Ok(false);
+		};
 
-		let old = producers.insert(path.clone(), broadcast.clone());
+		let old = producers.insert(path.clone(), publish);
 		web_async::spawn(self.clone().run_broadcast(path.clone(), dynamic));
 
-		if let Some(mut old) = old {
-			old.abort(Error::Cancel).ok();
-		}
+		// Drop the replaced broadcast's guard last, unannouncing it now that the replacement is live.
+		drop(old);
 
 		Ok(true)
 	}