moq-net: RAII broadcast announcements via OriginPublish; remove broadcast abort/close (#1640)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: kixelated

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