feat(net): end a broadcast with close()

doc/lib/rs/moq-net.md

@@ -112,7 +112,11 @@ Three operations, on an origin:
   `broadcast.announce(route)`.
 - `broadcast.announce(route)` / `broadcast.unannounce()` own that
   advertisement. Announcing again re-prices the standing route. The route
-  retracts on `unannounce()`, `finish()`, or the last producer dropping.
+  retracts on `unannounce()`, `close()`, or the last producer dropping.
+- `broadcast.close()` ends the broadcast for good: it retracts, leaves local
+  discovery, and answers every later track lookup with `Unroutable`. Tracks
+  already subscribed carry on to their own end. It can never be announced
+  again. Dropping the last producer does the same.
 - `origin.dynamic(prefix, route)` claims `prefix` and every path beneath it
   (`""` claims everything). Hold the returned `origin::Dynamic` while the
   claim should stay advertised; drop it to retract. A request beneath it with

js/net/src/broadcast.ts

@@ -262,7 +262,10 @@ export class Producer {
 		this.#announcer?.unannounce();
 	}
 
-	/** Close the broadcast, optionally with an error to abort waiters. Idempotent. */
+	/** End the broadcast for good: retract it, serve no new tracks, and refuse a later {@link announce}. Idempotent. */
+	close(): void;
+	/** @deprecated A broadcast end carries no cause; call `close()` without one. */
+	close(abort?: Error): void;
 	close(abort?: Error) {
 		this.#announcer?.unannounce();
 		this.#announcer = undefined;
@@ -352,9 +355,12 @@ export class Consumer {
 	}
 
 	/**
-	 * Release this handle. The broadcast is closed (optionally with an error to abort waiters)
-	 * once this was the last live handle; while other {@link clone}s remain open it stays live.
+	 * Release this handle. The broadcast is closed once this was the last live handle;
+	 * while other {@link clone}s remain open it stays live.
 	 */
+	close(): void;
+	/** @deprecated A broadcast end carries no cause; call `close()` without one. */
+	close(abort?: Error): void;
 	close(abort?: Error) {
 		if (this.#closed) return;
 		this.#closed = true;

js/net/src/origin.test.ts

@@ -260,8 +260,9 @@ test("closing the origin closes every routed broadcast", async () => {
 
 	expect(origin.closed.peek()).toBe(abort);
 	expect(consumer.closed.peek()).toBe(abort);
-	expect(a.closed.peek()).toBe(abort);
-	expect(b.closed.peek()).toBe(abort);
+	// A broadcast end carries no cause, so the routed broadcasts close cleanly.
+	expect(a.closed.peek()).toBeNull();
+	expect(b.closed.peek()).toBeNull();
 
 	expect(wireOf(consumer).routes(Path.from("a"))).toBe(false);
 	expect(() => publish(origin, Path.from("late"))).toThrow();

js/net/src/origin.ts

@@ -143,7 +143,7 @@ class ServeState {
 			finishRequest(request, err);
 		}
 		for (const [path, front] of this.served) {
-			front.close(abort);
+			front.close();
 			this.onChange(path);
 		}
 		this.served.clear();
@@ -718,7 +718,7 @@ export class Producer implements Table {
 		this.#state.closed.set(abort ?? null);
 		this.#state.local.update((broadcasts) => {
 			for (const front of broadcasts?.values() ?? []) {
-				front.close(abort);
+				front.close();
 			}
 			return undefined;
 		});

quest/m1/broadcast-close/README.md

@@ -34,12 +34,10 @@ object ended", not "the path went offline": announcements say whether a path
 is live, and a path can be announced again by a new object. Caches and moq-hls
 use it to tell a live object from a replaced one at the same path.
 
-Stage it as the children below: Rust first, then the bindings, then the `dev`
-removal.
+Stage it as the children below: the bindings, then the `dev` removal.
 
 ## Quests
 
-- [Rust close](/quest/m1/broadcast-close/rust.md) - moq-net gains `close()`, deprecates `finish`/`abort`/`is_finished`, and every Rust and JS caller moves over
 - [Binding close](/quest/m1/broadcast-close/bindings.md) - moq-ffi, libmoq, and every wrapper expose `close()` and deprecate `finish`
 - [Remove finish](/quest/m1/broadcast-close/remove.md) - on dev, the deprecated broadcast end APIs are gone and `closed()` carries no cause
 

quest/m1/broadcast-close/bindings.md

@@ -18,12 +18,8 @@ Every binding ends a broadcast with `close()`, mirroring Rust, and its
   `Publish.Close()`. Kotlin and Dart only alias the generated type, so they
   pick `close` up from moq-ffi.
 - Move the binding tests over, keeping one test per binding that a second
-  `close` errors or no-ops, whichever Rust settles on.
+  `close` is a no-op, as in Rust.
 - Update `doc/lib/{py,swift,kt,go,dart,c}`, including `doc/lib/go/index.md`'s
   `broadcast.Finish()` sample.
 - Fix the moq-ffi `origin.rs` doc comment that points users at a
   `broadcast.closed()` the bindings don't have.
-
-## Required
-
-- [Rust close](/quest/m1/broadcast-close/rust.md) - the binding forwards to `broadcast::Producer::close`

rs/libmoq/src/publish.rs

@@ -170,9 +170,8 @@ impl Publish {
 			}
 			guard.commit()?;
 		}
-		// Finish the broadcast first so the clean end reaches subscribers even if
-		// finalizing the catalog fails.
-		producer.finish();
+		// Close the broadcast first so it ends even if finalizing the catalog fails.
+		producer.close();
 		catalog.finish()?;
 		Ok(())
 	}

rs/moq-bench/src/connection.rs

@@ -799,7 +799,7 @@ mod tests {
 
 		task.await.unwrap().unwrap();
 		assert_eq!(stats.groups_recv.load(Ordering::Relaxed), 1);
-		broadcast.finish();
+		broadcast.close();
 	}
 
 	/// The relay fails a group it gave up on (`Error::Lagged` once a subscriber
@@ -850,7 +850,7 @@ mod tests {
 		write_group(&mut track);
 		wait_for(&stats.groups_recv, 3).await;
 		track.finish().unwrap();
-		broadcast.finish();
+		broadcast.close();
 
 		task.await
 			.unwrap()

rs/moq-boy/src/main.rs

@@ -305,9 +305,8 @@ async fn run(config: &Config) -> Result<()> {
 		res = input::handle_viewers(&viewer_consumer, &cmd_tx) => res,
 	};
 
-	// Cleanly close the broadcast so subscribers see a normal end rather than
-	// Error::Dropped.
-	broadcast.finish();
+	// Close the broadcast now, even if the emulator thread still holds a clone.
+	broadcast.close();
 	result
 }
 

rs/moq-ffi/src/producer.rs

@@ -477,9 +477,8 @@ impl MoqBroadcastProducer {
 		let _guard = crate::ffi::enter();
 		let mut guard = self.state.lock().unwrap();
 		let mut state = guard.take().ok_or(MoqError::Closed)?;
-		// Finish the broadcast first so the clean end reaches subscribers even if
-		// finalizing the catalog fails.
-		state.broadcast.finish();
+		// Close the broadcast first so it ends even if finalizing the catalog fails.
+		state.broadcast.close();
 		state.catalog.finish()?;
 		Ok(())
 	}

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

@@ -641,9 +641,8 @@ impl MoqSink {
 		if let Some(err) = failure {
 			gst::warning!(CAT, "finalize on stop: {err:?}");
 		}
-		// Finish the broadcast (a deliberate end, so no dropped-without-finish
-		// warning) before reaping the session task.
-		state.broadcast.finish();
+		// Close the broadcast before reaping the session task.
+		state.broadcast.close();
 		state.session.stop();
 		self.notify_updates(updates);
 	}

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

@@ -1271,7 +1271,7 @@ mod session_tests {
 		// End the served rendition and the broadcast, which answers for the reserved one.
 		catalog.finish().unwrap();
 		video.finish().unwrap();
-		broadcast.finish();
+		broadcast.close();
 
 		// No shutdown is sent: the session has to end on the media draining alone.
 		super::RUNTIME

rs/moq-hls/src/export/mod.rs

@@ -2053,11 +2053,11 @@ mod tests {
 
 		// Tear down the publisher mid-group. The track ends abruptly (the cursor drains the
 		// segments it already saw and ends; the still-open live-edge group is NOT finalized,
-		// since a reset can't vouch that its media is complete), while finishing the broadcast
-		// ends it cleanly rather than as a failure. (Clean-end finalization of the
+		// since a reset can't vouch that its media is complete), and the broadcast closes.
+		// (Clean-end finalization of the
 		// live edge is covered by segments::tests::next_after_walks_finalized_segments.)
 		drop((catalog, media, registration));
-		broadcast.finish();
+		broadcast.close();
 
 		let end = tokio::time::timeout(Duration::from_secs(5), segments.next())
 			.await

rs/moq-hls/src/export/rendition.rs

@@ -110,9 +110,8 @@ impl Media {
 		let mut handle = self.handle.lock().expect("media lock poisoned");
 		match handle.binding.poll_broadcast(&kio::Waiter::noop()) {
 			Poll::Ready(Ok(broadcast)) if broadcast.is_closed() => {
-				// The bound sibling ended (a different first hop replaces the publisher with
-				// Dropped). Origin finish()es that spliced front, so is_finished() cannot
-				// tell a rival publisher from a clean VOD end. Rows listed for it must not
+				// The bound sibling ended, and a broadcast end carries no cause, so a rival
+				// publisher looks the same as a clean VOD end. Rows listed for it must not
 				// be served from the replacement.
 				window.clear();
 				if let Ok(next) = source.bind(Some(rel)) {

rs/moq-hls/src/server/mod.rs

@@ -248,8 +248,8 @@ mod tests {
 		let broadcaster = Broadcaster::new(source, Config::default())
 			.await
 			.expect("catalog broadcast resolves while announced");
-		// Finish the publisher so the resolved broadcast (and the broadcaster) reports closed.
-		producer.finish();
+		// Close the publisher so the resolved broadcast (and the broadcaster) reports closed.
+		producer.close();
 		settle().await;
 		broadcaster
 	}
@@ -299,6 +299,6 @@ mod tests {
 
 		let cached = server.inner.broadcasters.lock().unwrap().get("live").cloned();
 		assert!(cached.is_some_and(|cached| Arc::ptr_eq(&cached, &new)));
-		new_producer.finish();
+		new_producer.close();
 	}
 }

rs/moq-net/benches/origin.rs

@@ -360,8 +360,8 @@ fn bench_handoff(c: &mut Criterion) {
 						total += started.elapsed();
 
 						drop(subscription);
-						incumbent.finish();
-						standby.finish();
+						incumbent.close();
+						standby.close();
 						// Wait for the front to close so the next iteration starts a fresh one.
 						resolved.closed().await;
 					}