From baf8de5b48f5c3292d2e158370cc83175d677127 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 10:29:20 -0700 Subject: [PATCH 1/4] quest: claim broadcast-close/rust Co-Authored-By: Claude Opus 5.5 From f1b931e68559ce098b9e0609f4535811af59cbee Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 10:52:03 -0700 Subject: [PATCH 2/4] feat(net): end a broadcast with close() Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-net.md | 6 +- js/net/src/broadcast.ts | 12 +- js/net/src/origin.ts | 4 +- quest/m1/broadcast-close/README.md | 4 +- quest/m1/broadcast-close/bindings.md | 6 +- quest/m1/broadcast-close/rust.md | 49 --- rs/libmoq/src/publish.rs | 5 +- rs/moq-bench/src/connection.rs | 4 +- rs/moq-boy/src/main.rs | 5 +- rs/moq-ffi/src/producer.rs | 5 +- rs/moq-gst/src/sink/imp.rs | 5 +- rs/moq-gst/src/source/imp.rs | 2 +- rs/moq-hls/src/export/mod.rs | 6 +- rs/moq-hls/src/export/rendition.rs | 5 +- rs/moq-hls/src/server/mod.rs | 6 +- rs/moq-net/benches/origin.rs | 4 +- rs/moq-net/src/ietf/subscriber.rs | 88 ++--- rs/moq-net/src/lite/subscriber.rs | 24 +- rs/moq-net/src/model/broadcast.rs | 416 ++++++++++---------- rs/moq-net/src/model/origin.rs | 25 +- rs/moq-net/tests/broadcast_close.rs | 72 ++++ rs/moq-net/tests/finished_broadcast_mock.rs | 2 +- rs/moq-relay/src/cluster.rs | 4 +- rs/moq-rtc/src/server/mod.rs | 4 +- rs/moq-rtmp/src/dial.rs | 11 +- rs/moq-rtmp/src/server.rs | 11 +- rs/moq-srt/src/ts.rs | 11 +- rs/moq-stats/src/produce.rs | 5 +- rs/moq-tokio/examples/chat.rs | 6 +- rs/moq-tokio/examples/clock.rs | 5 +- rs/moq-tokio/src/origin.rs | 2 +- rs/moq-tokio/tests/broadcast.rs | 4 +- rs/moq-transcode/src/lib.rs | 2 +- 33 files changed, 402 insertions(+), 418 deletions(-) delete mode 100644 quest/m1/broadcast-close/rust.md create mode 100644 rs/moq-net/tests/broadcast_close.rs diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 33f67336aa..5b7f0fafa7 100644 --- a/doc/lib/rs/moq-net.md +++ b/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 diff --git a/js/net/src/broadcast.ts b/js/net/src/broadcast.ts index 1691dcbfff..c8c8dad427 100644 --- a/js/net/src/broadcast.ts +++ b/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; diff --git a/js/net/src/origin.ts b/js/net/src/origin.ts index 947e14515c..f4ef8abf12 100644 --- a/js/net/src/origin.ts +++ b/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; }); diff --git a/quest/m1/broadcast-close/README.md b/quest/m1/broadcast-close/README.md index 3abce70484..210ddd8e1f 100644 --- a/quest/m1/broadcast-close/README.md +++ b/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 diff --git a/quest/m1/broadcast-close/bindings.md b/quest/m1/broadcast-close/bindings.md index 86437ec9e4..5f1444c897 100644 --- a/quest/m1/broadcast-close/bindings.md +++ b/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` diff --git a/quest/m1/broadcast-close/rust.md b/quest/m1/broadcast-close/rust.md deleted file mode 100644 index b2f8ea9acc..0000000000 --- a/quest/m1/broadcast-close/rust.md +++ /dev/null @@ -1,49 +0,0 @@ -# [M] Rust close - -## Goal - -`moq_net::broadcast::Producer::close()` is the one way to end a broadcast, and -no Rust or JS code in the repository calls a deprecated end API. - -## Plan - -In `rs/moq-net/src/model/broadcast.rs`: - -- Add `close(&self)`: set `closing`, close the liveness token, and retire the - announcer, as `finish` does today without setting `finished`. Borrow, like - `finish`: any clone ends it. -- Once a broadcast has ended, by `close()` or its last producer dropping, every - new lookup on any consumer answers `Unroutable`: `Consumer::track`, and the - requests still pending for names nothing served. That is what a fresh - `request_broadcast` for the path answers, so a consumer can't tell a raw - handle from one reached through an origin, or a local source from a remote - one. Today the same lookup answers `NotFound` while clones live and `Dropped` - after. Tracks already read are untouched and end on their own FIN or reset. - This changes the error a published API returns; say so in the PR. -- Document `broadcast::Producer::consume()` as a view of this one publisher: a - new publisher at the path is never spliced into it, so a consumer that should - not care about its source goes through an origin. -- Mark `finish`, `abort`, and `Consumer::is_finished` `#[deprecated]` and - `#[doc(hidden)]` per `rs/CLAUDE.md`. `finish` forwards to `close`. Leave - `abort`'s behavior alone until the `dev` removal. -- Drop the "dropped without finish()" warning in `Drop for Alive`. -- `SourceGuard` (the lite and IETF subscribers' source handle) closes on both - a graceful end and a drop, since the abort it records on drop reaches no one - past this hop. Check the IETF `Detach::Abrupt` path still ends the source. - -Callers to move (production): `model/origin.rs` (front end), -`lite/subscriber.rs` (`Route::finish`), `ietf/subscriber.rs`, moq-srt -`ts.rs`, moq-rtmp `server.rs` and `dial.rs`, moq-stats `produce.rs`, moq-gst -`sink/imp.rs`, moq-rtc `server/mod.rs`, moq-transcode `lib.rs`, moq-relay -`cluster.rs` (gossip registration), moq-boy `main.rs`, and moq-tokio's -`clock` and `chat` examples. Tests and benches across moq-net, moq-tokio, -moq-hls, moq-bench, and moq-gst move too. - -moq-srt and moq-rtmp `Publisher::abort` drop the broadcast instead of ending -it; closing it explicitly makes that path deterministic. - -In `js/net/src/broadcast.ts`, `Producer.close()` and `Consumer.close()` stop -taking `abort`: deprecate the parameter in the JSDoc and move the two origin -teardown callers in `origin.ts`. JS `close()` already blocks re-announcing. - -Update `doc/lib/rs/moq-net.md`, which says the route retracts on `finish()`. diff --git a/rs/libmoq/src/publish.rs b/rs/libmoq/src/publish.rs index 51d6108e5c..a98742effd 100644 --- a/rs/libmoq/src/publish.rs +++ b/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(()) } diff --git a/rs/moq-bench/src/connection.rs b/rs/moq-bench/src/connection.rs index 1502ab5c87..383f7be991 100644 --- a/rs/moq-bench/src/connection.rs +++ b/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() diff --git a/rs/moq-boy/src/main.rs b/rs/moq-boy/src/main.rs index a167fbec44..df54bf2baa 100644 --- a/rs/moq-boy/src/main.rs +++ b/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 } diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 179768b1c5..47ff363f91 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/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(()) } diff --git a/rs/moq-gst/src/sink/imp.rs b/rs/moq-gst/src/sink/imp.rs index 9135696fae..1d902b5925 100644 --- a/rs/moq-gst/src/sink/imp.rs +++ b/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); } diff --git a/rs/moq-gst/src/source/imp.rs b/rs/moq-gst/src/source/imp.rs index c7cd09b910..cadaadecdb 100644 --- a/rs/moq-gst/src/source/imp.rs +++ b/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 diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index bb73229465..e696e65c7f 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/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 diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 862429c8c4..17d77bf55b 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/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)) { diff --git a/rs/moq-hls/src/server/mod.rs b/rs/moq-hls/src/server/mod.rs index c4a13be663..bd6a09d764 100644 --- a/rs/moq-hls/src/server/mod.rs +++ b/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(); } } diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 04e1be686e..b3dd142dd0 100644 --- a/rs/moq-net/benches/origin.rs +++ b/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; } diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1b2fab1598..7e49793f18 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -339,24 +339,6 @@ impl TrackState { } } -/// How the last source for a path detaches, which decides whether the origin closes -/// the broadcast now or holds it open for a replacement. -/// -/// Only the detach that drops the refcount to zero decides, matching the model's rule -/// for several sources at one path (the front's source selection): an earlier owner that vanished -/// does not outvote the last one still on the path. That keeps two advertisements on -/// one session behaving like the same two on separate sessions, where the model sees -/// two independent sources and the last one out decides. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum Detach { - /// The peer retracted the path, or we are rolling back an announce we just made. - /// Nothing is coming back, so close it now. - Graceful, - /// The stream carrying the path went away without retracting it. Abort the - /// source so viewers observe the loss as an error rather than a clean end. - Abrupt, -} - struct BroadcastState { // The route announced into our origin for this namespace, post-charge. route: crate::origin::Route, @@ -368,9 +350,8 @@ struct BroadcastState { // active number of PUBLISH_NAMESPACE messages. count: usize, - // One minted source per requested path under the namespace: finish() on a - // deliberate unannounce, dropping (a dying session) aborts them so viewers - // observe the loss as an error. + // One minted source per requested path under the namespace, each closed + // when its guard drops. sources: HashMap, } @@ -726,14 +707,11 @@ where // ends: a clean close, a decode error, or the peer resetting it. Without this // each namespace keeps its refcount and the source never detaches. // - // Abruptly, including on a clean FIN: closing the stream retracts nothing, since - // the protocol has NAMESPACE_DONE for that. Whatever is still live here outlived - // its channel without being withdrawn, so hold the front open for a reconnect. // This is what moq-lite already does, where the equivalent map is a local whose // guards drop. let res = self.run_namespace_entries(&mut stream, &prefix, &peer, &mut live).await; for path in live { - let _ = self.stop_announce(path, Detach::Abrupt); + let _ = self.stop_announce(path); } res } @@ -777,7 +755,7 @@ where // leave subscriptions on a path the peer no longer offers. tracing::debug!(%path, "dropping reflected namespace"); if live.remove(&path) { - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); } continue; }; @@ -807,7 +785,7 @@ where let path = prefix.join(&msg.suffix); tracing::debug!(%path, "namespace_done"); if live.remove(&path) { - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); } } _ => { @@ -942,7 +920,7 @@ where Ok(_) => { if let Err(err) = self.write_ok(&mut stream, request_id).await { // Local rollback, not a peer unannounce: don't count announce bytes. - let _ = self.stop_announce(path, Detach::Graceful); + let _ = self.stop_announce(path); return Err(err); } } @@ -966,15 +944,7 @@ where .await; if attached { - // Ending cleanly IS the retraction here, unlike a NAMESPACE stream: this stream - // carries exactly one advertisement, and withdrawing it is what ends the stream. - // Any other ending left the advertisement standing, so the peer never withdrew - // it and the loss reads as abrupt (an error, not a clean end). - let detach = match res.is_ok() { - true => Detach::Graceful, - false => Detach::Abrupt, - }; - self.stop_announce(path, detach)?; + self.stop_announce(path)?; } res @@ -1086,7 +1056,7 @@ where let Some(advert) = self.route(held.as_ref(), &peer) else { if std::mem::take(attached) { tracing::debug!(%path, "publish_namespace now loops back; detaching"); - let _ = self.stop_announce(path.clone(), Detach::Graceful); + let _ = self.stop_announce(path.clone()); } self.write_ok(stream, msg.request_id).await?; continue; @@ -1355,24 +1325,18 @@ where } } - fn stop_announce(&mut self, path: PathOwned, detach: Detach) -> Result<(), Error> { + /// Release one advertisement of `path`, closing its sources when it was the last. + fn stop_announce(&mut self, path: PathOwned) -> Result<(), Error> { let mut state = self.state.lock(); match state.broadcasts.entry(path.clone()) { Entry::Occupied(mut entry) => { entry.get_mut().count -= 1; if entry.get().count == 0 { - tracing::debug!(route = %self.origin.absolute(&path), ?detach, "unannounced"); - // Dropping the entry retracts the route (its announcement drops). - let removed = entry.remove(); - for (_, source) in removed.sources { - match detach { - Detach::Graceful => source.finish(), - // Dropping the guard aborts the source, so the loss reads - // as an error rather than a clean end. - Detach::Abrupt => {} - } - } + tracing::debug!(route = %self.origin.absolute(&path), "unannounced"); + // Dropping the entry retracts the route (its announcement drops) and + // closes its sources (their guards drop). + entry.remove(); } } Entry::Vacant(_) => return Err(Error::NotFound), @@ -4097,7 +4061,7 @@ mod tests { subscriber.start_announce(path.clone(), advert).unwrap(); settle().await; - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!( routed_now(&consumer, "room/host").is_none(), "an explicit NAMESPACE_DONE must retract the route", @@ -4218,7 +4182,7 @@ mod tests { /// Several advertisements share one refcounted source, so the detach that empties it /// is the one that counts: the broadcast survives the first stop and closes on the - /// last, whatever kind each detach is. + /// last. /// /// That is the model's own rule for several sources at one path (the front's source /// selection), @@ -4239,7 +4203,7 @@ mod tests { settle().await; // One advertisement's stream dies: the other still holds the source. - subscriber.stop_announce(path.clone(), Detach::Abrupt).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); settle().await; assert!( routed_now(&consumer, "room/host").is_some(), @@ -4247,7 +4211,7 @@ mod tests { ); // The last owner retracts: the broadcast closes with it. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); settle().await; assert!( routed_now(&consumer, "room/host").is_none(), @@ -4314,7 +4278,7 @@ mod tests { // One advertisement, so one unannounce detaches it. If the update had bumped the // refcount, this would leave the route stranded. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4367,7 +4331,7 @@ mod tests { ); // One advertisement, so one unannounce detaches it. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4396,9 +4360,9 @@ mod tests { assert!(routed_now(&consumer, "room/host").is_some()); // Two advertisements, so it takes two unannounces to retract. - subscriber.stop_announce(path.clone(), Detach::Graceful).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); assert!(routed_now(&consumer, "room/host").is_some()); - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4425,9 +4389,9 @@ mod tests { assert!(routed_now(&consumer, "room/host").is_some()); // Two advertisements, so it takes two unannounces to retract. - subscriber.stop_announce(path.clone(), Detach::Graceful).unwrap(); + subscriber.stop_announce(path.clone()).unwrap(); assert!(routed_now(&consumer, "room/host").is_some()); - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!(routed_now(&consumer, "room/host").is_none()); } @@ -4464,7 +4428,7 @@ mod tests { ); // That supersedes the advertisement it repeats, so the old route is retired. - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); assert!( routed_now(&consumer, "room/host").is_none(), "the superseded route must not stay attached" @@ -4924,7 +4888,7 @@ mod tests { // What the stream's exit path does with whatever it still holds. for path in live { - subscriber.stop_announce(path, Detach::Graceful).unwrap(); + subscriber.stop_announce(path).unwrap(); } assert!(routed_now(&consumer, "room/a").is_none(), "room/a leaked a refcount"); diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 729a23e4bc..c8d8b5ad15 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2496,9 +2496,8 @@ impl Announced { } fn declined(&mut self, path: PathOwned) { - if let Some(Some(route)) = self.0.insert(path, None) { - route.finish(); - } + // Dropping a replaced route closes its sources. + self.0.insert(path, None); } /// Record an advertisement before deciding what to do with it. @@ -2509,8 +2508,7 @@ impl Announced { /// at each rejection is what stops the next early return from silently freeing a path /// the peer still holds. /// Only valid on a prefix the peer does not already hold, which the caller establishes - /// with [`Self::contains`]. Overwriting an attached route here would drop its source - /// without finishing it, which is [`Self::declined`]'s job. + /// with [`Self::contains`]. Overwriting an attached route is [`Self::declined`]'s job. fn reserve(&mut self, path: PathOwned) { debug_assert!(!self.0.contains_key(&path), "reserved a prefix already advertised"); self.0.insert(path, None); @@ -2521,9 +2519,8 @@ impl Announced { } fn retire(&mut self, path: &PathOwned) { - if let Some(Some(route)) = self.0.remove(path) { - route.finish(); - } + // Dropping the route closes its sources. + self.0.remove(path); } /// Serve queued requests on every attached route: mint a source per requested @@ -2567,8 +2564,7 @@ struct AnnouncedRoute { route: crate::origin::Route, /// Dropping it retracts the route and rejects its queued requests. dynamic: crate::origin::Dynamic, - /// One minted source per requested path, finished on a clean retraction and - /// aborted (via drop) when the session dies. + /// One minted source per requested path, each closed when its guard drops. sources: HashMap, /// Whether the GOAWAY drain already re-priced this route. drained: bool, @@ -2584,14 +2580,6 @@ impl AnnouncedRoute { } } - /// The peer deliberately retracted the route: finish the minted sources so - /// their consumers observe a clean end, and retract the announcement. - fn finish(self) { - for (_, source) in self.sources { - source.finish(); - } - } - /// Update the announced route in place (a restart). fn update(&mut self, route: crate::origin::Route) { self.route = route.clone(); diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index 3f9aaf9660..f07809f3e6 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -2,8 +2,8 @@ //! //! A [Producer] creates tracks on demand: a [Consumer] subscribes by name, and the //! producer either serves a track it already has or is handed a [`track::Request`] to -//! fill. Both handles are refcounted clones of one broadcast, which closes on -//! [`Producer::finish`] or when the last producer drops. +//! fill. Both handles are refcounted clones of one broadcast, which ends on +//! [`Producer::close`] or when the last producer drops. //! //! [Info] is the broadcast's static metadata, fixed for its lifetime. use crate::{cache, stats, track}; @@ -67,7 +67,7 @@ impl Info { /// Consume this [Info] to create a producer that carries its metadata. /// /// Keep the returned [`Producer`] alive for as long as the broadcast should stay - /// available, and end it with [`Producer::finish`]. See the note on [`Producer`]. + /// available, and end it with [`Producer::close`]. See the note on [`Producer`]. pub fn produce(self) -> Producer { Producer::new(self) } @@ -94,12 +94,11 @@ struct BroadcastState { // joined across per-session tracks. `None` for an ordinary broadcast. spliced: Option, - // Set by an explicit `Producer::finish()` or `Producer::abort()` so `Drop` can - // tell a deliberate shutdown apart from a producer dropped by accident. + // Set once the broadcast ends: `Producer::close()`, an abort, or the last + // producer-side handle dropping. Every lookup after it answers `Unroutable`. closing: bool, - // Set only by `Producer::finish()`: the broadcast ended deliberately, as - // opposed to aborting or losing its producer. + // Set only by the deprecated `Producer::finish()`, for `Consumer::is_finished`. finished: bool, // The error passed to `Producer::abort()`, reported by `Consumer::closed`. @@ -185,17 +184,10 @@ impl BroadcastState { /// /// # Lifetime /// -/// **You must keep this producer alive for as long as the broadcast should stay -/// available.** A broadcast lives as long as at least one [`Producer`] exists; -/// children do *not* keep it alive (cloning a [`Consumer`] or holding a -/// [`track::Producer`] does nothing for the broadcast's lifetime). When the last -/// producer goes away every consumer observes [`Error::Dropped`]. -/// -/// End the broadcast with [`Self::finish`] rather than dropping it. Dropping is an -/// easy footgun in garbage-collected bindings (Go, Python, ...), where the handle -/// can be collected the moment it falls out of scope even while you are still -/// publishing, tearing the stream down mid-broadcast. Dropping the last producer -/// without [`Self::finish`] logs a warning. +/// A broadcast lives until [`Self::close`] or until the last [`Producer`] (or +/// [`Dynamic`]) drops, whichever comes first; both end it the same way. Children +/// do *not* keep it alive: cloning a [`Consumer`] or holding a [`track::Producer`] +/// does nothing for the broadcast's lifetime. #[derive(Clone)] pub struct Producer { // Held behind an Arc so each track born from this broadcast can inherit a shared @@ -248,12 +240,12 @@ impl Producer { /// Call it once the tracks a subscriber needs first (a catalog) exist, so the /// advertisement lands with them in place: peers act on it immediately. /// The origin's local cursor already enumerates the path from creation. - /// The peer route retracts on [`unannounce`](Self::unannounce), [`finish`](Self::finish), - /// [`abort`](Self::abort), or the last producer dropping. + /// The peer route retracts on [`unannounce`](Self::unannounce), [`close`](Self::close), + /// or the last producer dropping. /// /// Fails with [`Error::Closed`] on a standalone broadcast (one not created - /// through an origin, so there is nothing to announce into) or once the - /// origin's driver has been dropped. + /// through an origin, so there is nothing to announce into), once the broadcast + /// has closed, or once the origin's driver has been dropped. pub fn announce(&self, route: Route) -> Result<(), Error> { let mut announcer = self.alive.announcer.lock(); let announcer = announcer.as_mut().ok_or(Error::Closed)?; @@ -339,7 +331,7 @@ impl Producer { /// /// Subscribers wait on the name until it is accepted, so a reservation the producer /// ends up never filling has to be dropped or rejected. Ending the broadcast - /// ([`Self::finish`] or [`Self::abort`]) resolves whatever is left. + /// resolves whatever is left. pub fn reserve_track(&self, name: impl Into>) -> Result { let request = track::Request::new(self.info.clone(), name).with_stats(self.stats.clone()); self.state.lock().insert_track(request.weak())?; @@ -423,7 +415,11 @@ impl Producer { } } - /// Create a consumer that can subscribe to tracks in this broadcast. + /// Create a consumer of this one publisher's broadcast. + /// + /// A view of this broadcast object, not of its path: a new publisher at the same + /// path is never spliced into it, so it ends when this broadcast does. Go through + /// an origin for a consumer that should not care which publisher serves the path. pub fn consume(&self) -> Consumer { Consumer { info: self.info.clone(), @@ -433,51 +429,33 @@ impl Producer { } } - /// Cleanly finish the broadcast once you are done publishing. - /// - /// Marks the broadcast as deliberately finished so consumers observe a normal - /// end. Prefer this over dropping the producer: an accidental drop (see the note - /// on [`Producer`]) logs a warning, whereas `finish()` is silent. - /// - /// Ends the broadcast outright: consumers observe a normal end immediately and no - /// new tracks are served, whether or not other producer clones are still alive. - /// Existing tracks stay readable so consumers can drain what they already have. + /// End the broadcast for good, whether or not other clones are still alive. /// - /// A name that was reserved or requested but never served resolves with - /// [`Error::NotFound`]: nothing can fill it now, so its subscribers fail rather - /// than waiting on a [`track::Info`] that is never coming. + /// Retracts its announcement and local discovery; a later [`Self::announce`] fails + /// with [`Error::Closed`]. Tracks already handed out carry on and end with their + /// own finish or abort. Every later [`Consumer::track`], and every request still + /// waiting on a name nothing served, answers [`Error::Unroutable`]: the same answer + /// an origin gives for a path nobody publishes. /// - /// Borrows rather than consumes, matching [`track::Producer::finish`]. Finishing - /// declares the end, so it must not depend on the caller also surrendering the - /// handle. + /// Dropping the last producer does the same. Closing twice is a no-op. + pub fn close(&self) { + self.alive.close(); + } + + #[doc(hidden)] + #[deprecated(note = "use close(); a broadcast end carries no cause")] pub fn finish(&self) { { let mut state = self.state.lock(); - state.closing = true; - state.finished = true; - // A name that was reserved or requested but never served can't arrive now, - // and `Consumer::track` already answers `NotFound` for one asked about after - // this point. Say the same to whoever asked earlier. - state.reject_unserved(Error::NotFound); + if !state.closing { + state.finished = true; + } } - // Ending the broadcast is what consumers wait on, so signal it here rather - // than leaving it to the last handle drop. - let _ = self.alive.token.close(); - self.alive.retire(); + self.close(); } - /// Abort the broadcast, ending it for consumers with `err`. - /// - /// Like [`finish`](Self::finish) the end is immediate, whether or not other - /// producer clones are still alive, and existing tracks stay readable so - /// consumers can drain what they already have (an abort does not cascade into - /// the tracks), while a name nothing ever served resolves with `err` the same - /// way [`finish`](Self::finish) resolves it. Unlike a finish, consumers observe - /// `err` from [`Consumer::closed`], so the end reads as a failure rather than a - /// deliberate one. - /// - /// Consumes the producer: an abort is terminal. Errors if the broadcast was - /// already finished or aborted. + #[doc(hidden)] + #[deprecated(note = "use close(); a broadcast end carries no cause")] pub fn abort(self, err: Error) -> Result<(), Error> { { let mut state = self.state.lock(); @@ -501,8 +479,8 @@ impl Producer { } } -/// Ends the broadcast when the last [`Producer`] or [`Dynamic`] drops, closing the -/// liveness channel every [`Consumer`] watches. +/// Ends the broadcast on [`Producer::close`] or when the last [`Producer`] or +/// [`Dynamic`] drops, closing the liveness channel every [`Consumer`] watches. /// /// A refcount rather than a "am I the last one?" check inside `Drop`: that answer is /// a snapshot, and acting on it is exactly what invalidates it. @@ -510,7 +488,7 @@ struct Alive { token: kio::Producer<()>, state: kio::Shared, // The advertisement of the broadcast's exact path, owned here so it retracts - // with the broadcast: on finish, abort, or the last producer-side handle + // with the broadcast: on close, abort, or the last producer-side handle // dropping. `None` for a standalone broadcast. announcer: kio::Lock>, } @@ -531,6 +509,22 @@ impl Alive { } } + /// End the broadcast. See [`Producer::close`]. + fn close(&self) { + { + let mut state = self.state.lock(); + if std::mem::replace(&mut state.closing, true) { + return; + } + // A name that was reserved or requested but never served can't arrive now, + // and `Consumer::track` answers `Unroutable` for one asked about after this + // point. Say the same to whoever asked earlier. + state.reject_unserved(Error::Unroutable); + } + let _ = self.token.close(); + self.retire(); + } + /// End the broadcast's advertising for good: retract the standing advertisement /// and drop the announcer, so a later `announce` fails with `Closed`. fn retire(&self) { @@ -543,15 +537,7 @@ impl Alive { impl Drop for Alive { fn drop(&mut self) { - // Warn if the last exit wasn't an explicit finish(), since consumers will - // then see Error::Dropped (classically a GC-collected handle in a language - // binding that tears the stream down mid-publish). - if !self.state.read().closing { - tracing::warn!( - "broadcast::Producer dropped without finish(). Keep the producer alive while publishing, then call finish()." - ); - } - self.retire(); + self.close(); } } @@ -568,36 +554,22 @@ impl Producer { } /// A session-owned handle to a source broadcast created via -/// [`crate::origin::Producer::create_broadcast`]: [`Self::finish`] ends it -/// deliberately, while dropping the guard aborts it as [`Error::Dropped`] (a dead -/// session), so consumers observe the loss as an error. Shared by the lite and -/// IETF subscribers so the drop-vs-finish contract lives in one place. -pub(crate) struct SourceGuard { - // `Option` so `finish` can consume the producer while `Drop` aborts it. - producer: Option, -} +/// [`crate::origin::Producer::create_broadcast`], closing it on drop even while the +/// session's serve machines still hold its [`Dynamic`]. +/// +/// A peer's retraction and a dead session end the source the same way: a broadcast +/// carries no end cause past this hop. Shared by the lite and IETF subscribers. +pub(crate) struct SourceGuard(Producer); impl SourceGuard { pub fn new(producer: Producer) -> Self { - Self { - producer: Some(producer), - } - } - - /// End the source deliberately: the origin detaches it immediately, - /// unannouncing the path if it was the last. - pub fn finish(mut self) { - if let Some(producer) = self.producer.take() { - producer.finish(); - } + Self(producer) } } impl Drop for SourceGuard { fn drop(&mut self) { - if let Some(producer) = self.producer.take() { - let _ = producer.abort(Error::Dropped); - } + self.0.close(); } } @@ -608,6 +580,7 @@ impl Drop for SourceGuard { /// [`track::Request::accept`]s it with a concrete [`track::Info`] or /// [`track::Request::reject`]s it. Dropped when no longer needed; pending requests /// are automatically aborted. +#[derive(Clone)] pub struct Dynamic { info: Arc, // Keeps the broadcast alive while a handler exists (mirrors a producer). @@ -616,31 +589,51 @@ pub struct Dynamic { // Ingress stats scope, applied to the tracks this handler serves. Empty (no-op) // for an untagged broadcast. stats: stats::Scope, + // Declared after `alive` so it drops second: when this was the broadcast's last + // handle, `Alive` has already ended it and answered every queued request + // `Unroutable`, so the handler's own `Dropped` rejection finds nothing left. + _handler: Handler, +} + +/// Counts one live [`Dynamic`], rejecting the queued requests when the last one drops. +struct Handler(kio::Shared); + +impl Handler { + fn new(state: kio::Shared) -> Self { + state.lock().requests.add_handler(); + Self(state) + } } -impl Clone for Dynamic { +impl Clone for Handler { fn clone(&self) -> Self { - // Mirror `new`: count each live handle. Without this, deriving Clone would - // let `Drop` decrement past `new`'s single increment and prematurely flip - // the handler count to zero, causing future `track` calls to return `NotFound`. - self.state.lock().requests.add_handler(); + // Count each live handle, or dropping a clone would flip the handler count to + // zero and future `track` calls would return `NotFound`. + Self::new(self.0.clone()) + } +} - Self { - info: self.info.clone(), - alive: self.alive.clone(), - state: self.state.clone(), - stats: self.stats.clone(), +impl Drop for Handler { + fn drop(&mut self) { + // Decrement and reject under one lock, so a `track` call that saw a live + // handler through the same lock can't slip a request past the rejection. + let mut state = self.0.lock(); + if state.requests.remove_handler() { + // No handlers left to fulfill pending requests; reject them so consumers + // don't block forever on tracks nobody will serve. + for request in state.requests.drain_queued() { + request.reject(Error::Dropped); + } } } } impl Dynamic { fn new(info: Arc, alive: Arc, state: kio::Shared, stats: stats::Scope) -> Self { - state.lock().requests.add_handler(); - Self { info, alive, + _handler: Handler::new(state.clone()), state, stats, } @@ -653,9 +646,8 @@ impl Dynamic { /// Poll for the next consumer-requested track, without blocking. /// - /// Returns [`Error::Closed`] once the broadcast was deliberately ended - /// ([`Producer::finish`] or aborted), so a serving loop knows to stop and - /// release its handle. + /// Returns [`Error::Closed`] once the broadcast has ended, so a serving loop + /// knows to stop and release its handle. pub fn poll_requested_track(&mut self, waiter: &kio::Waiter) -> Poll> { let mut state = ready!(self.state.poll(waiter, |state| { if state.requests.has_queued() || state.closing { @@ -693,15 +685,14 @@ impl Dynamic { } } - /// Block until the broadcast is closed, by [`Producer::finish`], - /// [`Producer::abort`], or every producer dropping, returning the cause. + /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. + /// + /// Returns [`Error::Dropped`], or the error passed to the deprecated `abort`. pub async fn closed(&self) -> Error { kio::wait(|waiter| self.poll_closed(waiter)).await } - /// Poll until the broadcast closes; ready with the cause: the error passed to - /// [`Producer::abort`], or [`Error::Dropped`] for a [`Producer::finish`] or a - /// dropped producer (check [`Consumer::is_finished`] to tell those apart). + /// Poll-based variant of [`Self::closed`]. pub fn poll_closed(&self, waiter: &kio::Waiter) -> Poll { ready!(self.alive.token.poll_closed(waiter)); Poll::Ready(self.state.read().abort.clone().unwrap_or(Error::Dropped)) @@ -713,21 +704,6 @@ impl Dynamic { } } -impl Drop for Dynamic { - fn drop(&mut self) { - // Decrement and reject under one lock, so a `track` call that saw a live - // handler through the same lock can't slip a request past the rejection. - let mut state = self.state.lock(); - if state.requests.remove_handler() { - // No handlers left to fulfill pending requests; reject them so consumers - // don't block forever on tracks nobody will serve. - for request in state.requests.drain_queued() { - request.reject(Error::Dropped); - } - } - } -} - #[cfg(test)] use futures::FutureExt; @@ -747,6 +723,10 @@ impl Dynamic { } /// Subscribe to arbitrary broadcast/tracks. +/// +/// Its close signal means this broadcast object ended, not that the path went +/// offline: announcements say whether a path is live, and a new publisher may +/// announce the same path again. pub struct Consumer { info: Arc, // Broadcast liveness (read-only): watched for close. @@ -800,6 +780,8 @@ impl Consumer { } /// Get a handle to a track on this broadcast. + /// + /// Fails with [`Error::Unroutable`] once the broadcast has ended. pub fn track(&self, name: &str) -> Result { // Rebind the track to *this* handle's view of the broadcast, so a catalog track // resolves its relative references against the path we were handed out at rather @@ -810,16 +792,17 @@ impl Consumer { } fn track_inner(&self, name: &str) -> Result { - // A closed broadcast (every producer and handler gone) serves nothing. - if self.is_closed() { - return Err(Error::Dropped); - } - let mut state = self.state.lock(); + // An ended broadcast serves nothing new, not even a track it still has: + // the lookup answers what a fresh `request_broadcast` for the path would. + // Tracks already handed out are untouched. + if state.closing { + return Err(Error::Unroutable); + } + // A route-fed broadcast mints spliced logical tracks: they outlive any // session, and a route is asked (via the pending queue) to start serving. - let closing = state.closing; if let Some(spliced) = state.spliced.as_mut() { // An aborted logical track is a verdict from the sources attached at // the time, not a property of the name: a publisher that had not yet @@ -844,11 +827,6 @@ impl Consumer { producer.consume(), )); } - // A deliberately-ended broadcast serves nothing new; nothing drains the - // pending queue once the front is torn down. - if closing { - return Err(Error::NotFound); - } let name: Arc = name.into(); let producer = super::resume::Producer::new(); let consumer = producer.consume(); @@ -876,12 +854,6 @@ impl Consumer { return Ok(pending.consume()); } - // A deliberately-ended broadcast serves nothing new; existing tracks above - // stay readable so consumers can drain the cache. - if state.closing { - return Err(Error::NotFound); - } - // Allocate the name once and share the same Arc across the request, the // requests map, and the FIFO order. The request inherits the broadcast's // cache pool through its `Arc`, same as a producer-created track. @@ -913,32 +885,28 @@ impl Consumer { } } - /// Block until the broadcast is closed, by [`Producer::finish`], - /// [`Producer::abort`], or every producer dropping, and return the cause. + /// Block until the broadcast ends, by [`Producer::close`] or every producer dropping. /// - /// Returns the error passed to [`Producer::abort`], or [`Error::Dropped`] for a - /// [`Producer::finish`] or a dropped producer (check [`Self::is_finished`] to - /// tell those apart). + /// Returns [`Error::Dropped`], or the error passed to the deprecated `abort`. pub async fn closed(&self) -> Error { self.alive.closed().await; self.state.read().abort.clone().unwrap_or(Error::Dropped) } - /// Returns true if every [`Producer`] has been dropped. + /// Returns true once the broadcast has ended. pub fn is_closed(&self) -> bool { self.alive.is_closed() } - /// Whether the broadcast is on its way out: deliberately ended (finish/abort - /// marked, even while handles remain) or already fully closed. The origin's + /// Whether the broadcast has ended, observed under the state lock. The origin's /// dispatcher treats a rejection from such a source as imminent detach rather /// than a strike. pub(crate) fn is_closing(&self) -> bool { - self.is_closed() || self.state.read().closing + self.state.read().closing } - /// Whether the broadcast ended via a deliberate [`Producer::finish`], as opposed - /// to aborting or losing its producer. `false` while the broadcast is still live. + #[doc(hidden)] + #[deprecated(note = "a broadcast end carries no cause")] pub fn is_finished(&self) -> bool { self.state.read().finished } @@ -1182,7 +1150,7 @@ mod test { assert!(!demand.is_used()); // Every producer gone: both edges report the closure. - producer.finish(); + producer.close(); assert!(matches!(demand.used().await, Err(Error::Dropped))); assert!(matches!(demand.unused().await, Err(Error::Dropped))); } @@ -1299,31 +1267,47 @@ mod test { track1c.assert_not_closed(); } - /// `closed()` reports the cause: the abort error, or `Dropped` for a finish or - /// a dropped producer, with `is_finished` telling the latter two apart. + /// `close()` ends the broadcast for every clone at once, and a second close is a no-op. #[tokio::test] - async fn closed_cause() { - // Abort: the error comes through, and it isn't a finish. + async fn close_ends_every_clone() { + let producer = Info::new().produce(); + let clone = producer.clone(); + let consumer = producer.consume(); + + producer.close(); + assert!(matches!(consumer.closed().await, Error::Dropped)); + assert!(matches!(consumer.track("video"), Err(Error::Unroutable))); + assert!(matches!(clone.consume().track("video"), Err(Error::Unroutable))); + + clone.close(); + producer.close(); + } + + /// Dropping the last producer ends the broadcast exactly like `close()`. + #[tokio::test] + async fn drop_ends_like_close() { + let producer = Info::new().produce(); + let consumer = producer.consume(); + drop(producer); + assert!(matches!(consumer.closed().await, Error::Dropped)); + assert!(matches!(consumer.track("video"), Err(Error::Unroutable))); + } + + /// The deprecated end APIs keep their old causes until they are removed. + #[tokio::test] + #[allow(deprecated)] + async fn deprecated_end_causes() { let producer = Info::new().produce(); let consumer = producer.consume(); producer.abort(Error::Timeout).unwrap(); assert!(matches!(consumer.closed().await, Error::Timeout)); assert!(!consumer.is_finished()); - // Finish: a deliberate clean end. let producer = Info::new().produce(); let consumer = producer.consume(); producer.finish(); assert!(matches!(consumer.closed().await, Error::Dropped)); assert!(consumer.is_finished()); - - // Plain drop: neither aborted nor finished. - let producer = Info::new().produce(); - let consumer = producer.consume(); - // Deliberate for the test: exercises the accidental-drop path (warns). - drop(producer); - assert!(matches!(consumer.closed().await, Error::Dropped)); - assert!(!consumer.is_finished()); } #[tokio::test] @@ -1497,24 +1481,25 @@ mod test { } /// A reserved name nobody accepts is the parking case a publisher has to be able to - /// end. Ending the broadcast is where it does: `Consumer::track` already answers - /// `NotFound` for a name asked about after this point, so whoever asked earlier gets + /// end. Ending the broadcast is where it does: `Consumer::track` answers + /// `Unroutable` for a name asked about after this point, so whoever asked earlier gets /// the same answer instead of waiting on info that can never arrive. #[tokio::test] - async fn finish_resolves_a_reserved_name() { + async fn close_resolves_a_reserved_name() { let producer = Info::new().produce(); let consumer = producer.consume(); let _request = producer.reserve_track("track1").unwrap(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); } - /// An abort says why the broadcast ended, and an unserved name resolves with that - /// reason rather than a generic failure. + /// The deprecated abort says why the broadcast ended, and an unserved name resolves + /// with that reason. #[tokio::test] + #[allow(deprecated)] async fn abort_resolves_a_reserved_name_with_its_reason() { let producer = Info::new().produce(); let consumer = producer.consume(); @@ -1533,22 +1518,50 @@ mod test { /// A request still queued for a handler is the same parking case reached from the /// consumer side, so it ends the same way. #[tokio::test] - async fn finish_resolves_a_queued_request() { + async fn close_resolves_a_queued_request() { + let producer = Info::new().produce(); + let dynamic = producer.dynamic(); + let consumer = dynamic.consume(); + + let pending = subscribe_pending!(consumer, "track1"); + + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); + drop(dynamic); + } + + /// A queued request answers the same when the broadcast ends by its last handle + /// dropping, even when that handle is the `Dynamic` that would have served it. + #[tokio::test] + async fn dropping_the_last_handle_resolves_a_queued_request() { + let dynamic = Info::new().produce().dynamic(); + let consumer = dynamic.consume(); + + let pending = subscribe_pending!(consumer, "track1"); + + drop(dynamic); + assert!(matches!(pending.await, Err(Error::Unroutable))); + } + + /// With a producer still alive, losing the last handler is not the broadcast ending: + /// the queued request fails as `Dropped`. + #[tokio::test] + async fn dropping_the_last_handler_resolves_a_queued_request_dropped() { let producer = Info::new().produce(); let dynamic = producer.dynamic(); let consumer = dynamic.consume(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); drop(dynamic); + assert!(matches!(pending.await, Err(Error::Dropped))); + producer.close(); } /// A request a handler already took parks the same way if the handler never answers /// it, so the sweep has to reach that one too. #[tokio::test] - async fn finish_resolves_a_request_a_handler_never_answered() { + async fn close_resolves_a_request_a_handler_never_answered() { let producer = Info::new().produce(); let mut dynamic = producer.dynamic(); let consumer = dynamic.consume(); @@ -1556,17 +1569,17 @@ mod test { let pending = subscribe_pending!(consumer, "track1"); let _request = dynamic.requested_track().await.unwrap(); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); drop(dynamic); } /// A reverse fetch can install the track metadata before the live request is - /// accepted, but it does not create a live publisher. Finishing the broadcast + /// accepted, but it does not create a live publisher. Closing the broadcast /// must still reject that name so an arrival-order subscriber does not park on /// backfill that is deliberately absent from its queue. #[tokio::test] - async fn finish_resolves_an_unaccepted_track_with_fetched_info() { + async fn close_resolves_an_unaccepted_track_with_fetched_info() { let producer = Info::new().produce(); let consumer = producer.consume(); @@ -1580,24 +1593,25 @@ mod test { pending_fetch.await.unwrap(); let mut subscriber = track.subscribe(None).await.unwrap(); - producer.finish(); - assert!(matches!(subscriber.recv_group().await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(subscriber.recv_group().await, Err(Error::Unroutable))); let stale = request.accept(None); assert!(stale.append_group().is_err()); } /// Ending the broadcast doesn't cascade into a track someone is publishing: it keeps - /// its cache and its publisher decides when it ends. + /// its cache and its publisher decides when it ends. Only a new lookup is refused. #[tokio::test] - async fn finish_spares_a_served_track() { + async fn close_spares_a_served_track() { let producer = Info::new().produce(); let consumer = producer.consume(); let track = producer.create_track("track1", None).unwrap(); let mut subscriber = consumer.track("track1").unwrap().subscribe(None).await.unwrap(); - producer.finish(); + producer.close(); + assert!(matches!(consumer.track("track1"), Err(Error::Unroutable))); track.append_group().unwrap(); subscriber.assert_group(); @@ -1606,22 +1620,22 @@ mod test { /// The publisher may still be holding the `track::Request` for a name the broadcast /// just gave up on. Accepting it afterwards must not resurrect the track, or a - /// subscriber that was told `NotFound` could be contradicted by a later one. + /// subscriber that was told `Unroutable` could be contradicted by a later one. #[tokio::test] - async fn finish_leaves_a_stale_reservation_inert() { + async fn close_leaves_a_stale_reservation_inert() { let producer = Info::new().produce(); let consumer = producer.consume(); let request = producer.reserve_track("track1").unwrap(); let pending = subscribe_pending!(consumer, "track1"); - producer.finish(); - assert!(matches!(pending.await, Err(Error::NotFound))); + producer.close(); + assert!(matches!(pending.await, Err(Error::Unroutable))); let track = request.accept(None); assert!(track.append_group().is_err()); let mut subscriber = track.subscribe(None); - assert!(matches!(subscriber.recv_group().await, Err(Error::NotFound))); + assert!(matches!(subscriber.recv_group().await, Err(Error::Unroutable))); assert!(consumer.track("track1").is_err()); } @@ -1638,7 +1652,7 @@ mod test { drop(request); assert!(matches!(pending.await, Err(Error::Dropped))); - producer.finish(); + producer.close(); } /// `track::Request::reject` carries its reason the same way, which is what lets a @@ -1653,7 +1667,7 @@ mod test { request.reject(Error::NotFound); assert!(matches!(pending.await, Err(Error::NotFound))); - producer.finish(); + producer.close(); } /// The interleave every unused-driven teardown has to survive: a wire subscriber @@ -1690,7 +1704,7 @@ mod test { assert!(track.abort_unused(Error::Cancel).is_ok()); assert!(matches!(consumer.track("video"), Err(Error::NotFound))); - producer.finish(); + producer.close(); } #[test] @@ -1703,6 +1717,6 @@ mod test { track.clone().abort(Error::Cancel).unwrap(); assert!(!track.is_used()); assert!(track.abort_unused(Error::Cancel).is_ok()); - producer.finish(); + producer.close(); } } diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 2f7ab0f77b..1fe8bb4747 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -1126,11 +1126,8 @@ impl Producer { /// polled. Register a [`broadcast::Producer::dynamic`] handler right away, so /// the first consumer finds the tracks it serves. /// - /// End the broadcast with [`broadcast::Producer::finish`]; dropping it - /// without finishing also works, but logs a warning. Either way the path - /// closes once it was the last source; an unfinished drop additionally aborts - /// the spliced tracks with an error, so consumers observe a failure rather - /// than a clean end. + /// End the broadcast with [`broadcast::Producer::close`] or by dropping it; + /// either way the path closes once it was the last source. /// /// Fails with [`Error::Unauthorized`] if `path` is outside the prefixes this /// producer may publish under (after [`scope`](Self::scope)), @@ -1161,7 +1158,7 @@ impl Producer { // The broadcast is a route table entry at its exact path from the start, // so requests resolve to it and the newest publisher at a path wins; // local cursors see it immediately. The entry lives as long as the - // broadcast: its announcer drops on finish, abort, or the last handle. + // broadcast: its announcer drops on close, abort, or the last handle. let announcing = Announcing { hop: self.hop, shared: self.shared.clone(), @@ -2032,7 +2029,7 @@ async fn run_front(task: FrontTask) { // subscriptions already in flight): dropping their producers // leaves each reader on the copy it was spliced from, ending // when and as that copy ends. - broadcast.finish(); + broadcast.close(); broadcast.release_spliced(err.clone()); for (_, mut io) in tracks.drain() { // Nothing in flight: unread, never spliced, or only a warm cache. @@ -3789,7 +3786,7 @@ mod tests { broadcast.announce(Route::default()).unwrap(); announced.assert_next_wait(); peer.assert_next_active("room/alice"); - broadcast.finish(); + broadcast.close(); announced.assert_next_ended("room/alice"); peer.assert_next_ended("room/alice"); assert!(matches!(broadcast.announce(Route::default()), Err(Error::Closed))); @@ -5147,9 +5144,9 @@ mod tests { assert!(again.is_clone(&resolved)); // Losing one source keeps the front alive; losing both closes it. - first.finish(); + first.close(); settle(|| consumer.get_broadcast("room/alice").is_some()).await; - second.finish(); + second.close(); settle(|| consumer.get_broadcast("room/alice").is_none()).await; // The path is free again for a fresh broadcast. @@ -5185,7 +5182,7 @@ mod tests { group.finish().unwrap(); track.finish().unwrap(); drop(track); - broadcast.finish(); + broadcast.close(); let mut group = next_group(&mut subscription) .await @@ -5230,7 +5227,7 @@ mod tests { // ANNOUNCE_END overtakes the track's end: the route is retracted, and the front // has acted on it, before the track's last group and end arrive. - source.finish(); + source.close(); drop(server); settle(|| resolved.is_closed()).await; let mut group = track.append_group().unwrap(); @@ -5460,7 +5457,7 @@ mod tests { // The incumbent leaving exhausts the table: the refusal is never retried. drop(track); - first.finish(); + first.close(); assert!(matches!(subscription.recv_group().await, Err(Error::Unsupported))); } @@ -5659,7 +5656,7 @@ mod tests { // Publishing through the nested view lands where the root says. let broadcast = nested.create_broadcast("room/chat/live").unwrap(); assert!(producer.consume().get_broadcast("room/chat/live").is_some()); - broadcast.finish(); + broadcast.close(); } #[test] diff --git a/rs/moq-net/tests/broadcast_close.rs b/rs/moq-net/tests/broadcast_close.rs new file mode 100644 index 0000000000..584f7f77b0 --- /dev/null +++ b/rs/moq-net/tests/broadcast_close.rs @@ -0,0 +1,72 @@ +//! `broadcast::Producer::close()` ends a broadcast the same way for a local consumer +//! and a remote one: the handle closes, every new track lookup answers +//! [`Error::Unroutable`](moq_net::Error::Unroutable), a fresh request for the path +//! answers the same, and the broadcast can never be announced again. + +mod support; + +use std::time::Duration; + +use moq_net::{Error, Hop, Version}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// Close a published broadcast and check what `reader`'s origin answers afterwards. +async fn close_then_lookup(publisher: moq_net::origin::Producer, reader: moq_net::origin::Producer) { + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let _track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let consumer = reader.consume(); + tokio::time::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let handle = tokio::time::timeout(TIMEOUT, consumer.request_broadcast("bcast")) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + handle.track("video").expect("a live broadcast serves its track"); + + broadcast.close(); + + tokio::time::timeout(TIMEOUT, handle.closed()) + .await + .expect("the handle never closed"); + assert!(matches!(handle.track("video"), Err(Error::Unroutable))); + assert!(matches!(handle.track("audio"), Err(Error::Unroutable))); + assert!(matches!( + consumer.request_broadcast("bcast").await, + Err(Error::Unroutable) + )); + assert!(matches!(broadcast.announce(Default::default()), Err(Error::Closed))); +} + +#[tokio::test] +async fn close_ends_a_local_consumer() { + tokio::time::pause(); + let origin = produce_origin(1); + close_then_lookup(origin.clone(), origin).await; +} + +#[tokio::test] +async fn close_ends_a_remote_consumer() { + tokio::time::pause(); + for version in ["moq-lite-05", "moq-transport-14", "moq-transport-19"] { + let publisher = produce_origin(1); + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.clone()); + options.client_subscribe = Some(subscriber.clone()); + let _pair = connect_mock(options).await; + + close_then_lookup(publisher, subscriber).await; + } +} diff --git a/rs/moq-net/tests/finished_broadcast_mock.rs b/rs/moq-net/tests/finished_broadcast_mock.rs index 30fcefe404..dc64c0506c 100644 --- a/rs/moq-net/tests/finished_broadcast_mock.rs +++ b/rs/moq-net/tests/finished_broadcast_mock.rs @@ -90,7 +90,7 @@ async fn round(finish_broadcast: bool) -> (Vec>, Option) track.finish().unwrap(); drop(track); if finish_broadcast { - broadcast.finish(); + broadcast.close(); } // The session and both origins stay up until the reader is done. diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index e274431079..9a37392103 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -1549,9 +1549,9 @@ impl Cluster { } } - // Deliberate shutdown: finishing the registration retracts the route. + // Deliberate shutdown: closing the registration retracts the route. if let Some(registration) = self_registration.as_mut() { - registration.finish(); + registration.close(); } Ok(()) } diff --git a/rs/moq-rtc/src/server/mod.rs b/rs/moq-rtc/src/server/mod.rs index 3585f15b8a..a23bc1cd6f 100644 --- a/rs/moq-rtc/src/server/mod.rs +++ b/rs/moq-rtc/src/server/mod.rs @@ -86,10 +86,10 @@ impl AcceptedSession { } _ = cancel => { tracing::debug!(role = self.role, "webrtc session terminated by DELETE"); - // A deliberate end: finish the broadcast so the origin + // A deliberate end: close the broadcast so the origin // unannounces it immediately. if let Some(broadcast) = self.broadcast.take() { - broadcast.finish(); + broadcast.close(); } Ok(()) } diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 4e987e63e1..1a0572b5ad 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -485,11 +485,11 @@ async fn client_handshake(stream: &mut S) -> anyhow::Result> /// An active pull: the moq-mux FLV importer publishing into the origin. Mirrors the /// server's publisher; either [`Self::finish`] or dropping it unannounces the -/// path, the former without the dropped-without-finish warning. +/// path. struct Publisher { importer: FlvImport, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -522,16 +522,17 @@ impl Publisher { fn finish(&mut self) -> anyhow::Result<()> { self.importer.finish()?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } /// Abort the published tracks with `err` so subscribers see the real cause /// (the remote dropped, a protocol error) rather than a generic `Error::Dropped`. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 250eac7c5e..a5a2767a8b 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -1284,11 +1284,11 @@ async fn run_handshake(stream: &mut S, peer: SocketAddr) -> anyhow::R /// An active publish: the moq-mux FLV importer, which owns the origin-created /// [`BroadcastProducer`](moq_net::broadcast::Producer) it publishes into. /// Either [`Self::finish`] or dropping it closes the broadcast and unannounces -/// the path, the former without the dropped-without-finish warning. +/// the path. struct Publisher { importer: FlvImport, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -1326,7 +1326,7 @@ impl Publisher { /// the broadcast so the origin unannounces it immediately. fn finish(&mut self) -> anyhow::Result<()> { self.importer.finish()?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } @@ -1334,9 +1334,10 @@ impl Publisher { /// (the client disconnected, a protocol error) rather than a generic /// `Error::Dropped` from the importer being dropped. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index baafbb3ef1..8e6ccf2abc 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -21,14 +21,14 @@ use crate::Result; /// transport packets and retains any partial trailing packet internally for the /// next call (the same pattern `moq-cli import ... stdin ts` uses against stdin). /// Either [`Self::finish`] or dropping the publisher ends the broadcast and -/// unannounces the path, the former without the dropped-without-finish warning. +/// unannounces the path. pub struct Publisher { // TS carries undecoded elementary streams (SCTE-35, teletext, DVB AC-3, ...) // verbatim, so the importer uses the `mpegts` catalog extension rather than the // media-only `()`, which would route those PIDs to `Stream::Ignored` and drop them. importer: ts::Import, - // A clone of the importer's producer, so a deliberate end can finish() the - // broadcast (prompt unannounce) even though the importer owns it. + // A clone of the importer's producer, so an end can close the broadcast + // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, } @@ -65,16 +65,17 @@ impl Publisher { /// the broadcast so the origin unannounces it immediately. pub fn finish(&mut self) -> Result<()> { self.importer.finish().map_err(moq_mux::Error::from)?; - self.broadcast.finish(); + self.broadcast.close(); Ok(()) } /// Abort the published tracks with `err` so subscribers see the real cause /// (the SRT caller dropped, a demux error) rather than a generic `Error::Dropped`. /// - /// Consumes the publisher: the broadcast is done. + /// Consumes the publisher and closes the broadcast. pub fn abort(self, err: moq_net::Error) { self.importer.abort(err); + self.broadcast.close(); } } diff --git a/rs/moq-stats/src/produce.rs b/rs/moq-stats/src/produce.rs index ce706424d4..59cc13b2fd 100644 --- a/rs/moq-stats/src/produce.rs +++ b/rs/moq-stats/src/produce.rs @@ -852,12 +852,11 @@ impl GroupPublisher { self.sessions.adopt_parked(&self.broadcast, &mut self.requested); } - /// Deliberately end the broadcast: finish every pair, then the broadcast - /// itself, so teardown emits no dropped-without-finish warnings. + /// Deliberately end the broadcast: finish every pair, then close the broadcast. fn finish(mut self) { self.traffic.finish(); self.sessions.finish(); - self.broadcast.finish(); + self.broadcast.close(); } } diff --git a/rs/moq-tokio/examples/chat.rs b/rs/moq-tokio/examples/chat.rs index 2e78dc5881..98241f9e5d 100644 --- a/rs/moq-tokio/examples/chat.rs +++ b/rs/moq-tokio/examples/chat.rs @@ -83,10 +83,8 @@ async fn run_broadcast(origin: moq_net::origin::Producer) -> anyhow::Result<()> // Sleep before exiting and closing the broadcast. tokio::time::sleep(tokio::time::Duration::from_secs(10)).await; - // Cleanly close the broadcast so subscribers see a normal end. Dropping the - // producer works too, but closing is explicit (and dropping it by accident - // while still publishing is a common bug this makes obvious). - broadcast.finish(); + // End the broadcast. Dropping the producer does the same; closing is explicit. + broadcast.close(); Ok(()) } diff --git a/rs/moq-tokio/examples/clock.rs b/rs/moq-tokio/examples/clock.rs index 42912d89c1..c792195209 100644 --- a/rs/moq-tokio/examples/clock.rs +++ b/rs/moq-tokio/examples/clock.rs @@ -84,9 +84,8 @@ async fn main() -> anyhow::Result<()> { _ = clock.run() => Ok(()), }; - // Cleanly close the broadcast on exit so subscribers see a normal end - // rather than Error::Dropped. - broadcast.finish(); + // End the broadcast on exit. + broadcast.close(); result } Command::Subscribe => { diff --git a/rs/moq-tokio/src/origin.rs b/rs/moq-tokio/src/origin.rs index b2911dfba5..ac350fff45 100644 --- a/rs/moq-tokio/src/origin.rs +++ b/rs/moq-tokio/src/origin.rs @@ -39,7 +39,7 @@ mod tests { assert_eq!(update.prefix.as_str(), "cam"); assert!(update.kind.is_active()); - broadcast.finish(); + broadcast.close(); let update = announced.next().await.expect("retraction"); assert!(!update.kind.is_active()); } diff --git a/rs/moq-tokio/tests/broadcast.rs b/rs/moq-tokio/tests/broadcast.rs index ddc1db8b78..ce93e3ec47 100644 --- a/rs/moq-tokio/tests/broadcast.rs +++ b/rs/moq-tokio/tests/broadcast.rs @@ -776,7 +776,7 @@ async fn broadcast_moq_lite_06_announce_lifecycle() { // Unannounce: retracted by announce id on the wire. Dropping the announcement // retracts the route; the broadcast's own end is independent. - second.finish(); + second.close(); let update = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "second"); assert!(!update.kind.is_active(), "expected retraction"); @@ -791,7 +791,7 @@ async fn broadcast_moq_lite_06_announce_lifecycle() { // Replace the route at "first": retract the original (retiring its announce // id on the wire), then announce the same path again (assigning a fresh id). // Await the retraction first so the events cannot coalesce away. - first.finish(); + first.close(); let update = next_announce(&mut announcements).await; assert_eq!(update.prefix.as_str(), "first"); assert!(!update.kind.is_active(), "expected the replaced retraction"); diff --git a/rs/moq-transcode/src/lib.rs b/rs/moq-transcode/src/lib.rs index a9cda09c0d..4884b0b342 100644 --- a/rs/moq-transcode/src/lib.rs +++ b/rs/moq-transcode/src/lib.rs @@ -190,7 +190,7 @@ impl Transcoder { tasks.shutdown().await; derived.finish()?; - output.finish(); + output.close(); Ok(()) } } From 144166a97306b5ffd250cb72c1237d77c31efc4a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 11:16:07 -0700 Subject: [PATCH 3/4] test(net): routed broadcasts close without a cause Co-Authored-By: Claude Opus 5.5 --- js/net/src/origin.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/js/net/src/origin.test.ts b/js/net/src/origin.test.ts index e23a5d0bd8..c94cc92bb1 100644 --- a/js/net/src/origin.test.ts +++ b/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(); From 53195dfa3914d15145c21f25050fd5ce1af33e63 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 24 Sep 2026 17:08:37 -0700 Subject: [PATCH 4/4] fix(net): claim the end and the finished flag in one locked step A deprecated finish() racing abort() on another clone could set finished, then lose the end to abort, leaving is_finished() true beside an abort cause. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/broadcast.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index f07809f3e6..0b34ead84a 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -445,13 +445,7 @@ impl Producer { #[doc(hidden)] #[deprecated(note = "use close(); a broadcast end carries no cause")] pub fn finish(&self) { - { - let mut state = self.state.lock(); - if !state.closing { - state.finished = true; - } - } - self.close(); + self.alive.end(true); } #[doc(hidden)] @@ -511,11 +505,18 @@ impl Alive { /// End the broadcast. See [`Producer::close`]. fn close(&self) { + self.end(false); + } + + /// End the broadcast, recording the deprecated `finished` flag in the same locked + /// transition that claims the end, so a racing `abort` can't win after it's set. + fn end(&self, finished: bool) { { let mut state = self.state.lock(); if std::mem::replace(&mut state.closing, true) { return; } + state.finished = finished; // A name that was reserved or requested but never served can't arrive now, // and `Consumer::track` answers `Unroutable` for one asked about after this // point. Say the same to whoever asked earlier.