From 254624b247017c0cbd2be3703baa3eb3d1c457fc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 22:58:50 -0700 Subject: [PATCH 01/14] claim: quest/m1/restart-followers From 096b53bdeb83c28b80aa0c525886a4773ef97732 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:18:13 -0700 Subject: [PATCH 02/14] fix(cli): fail a non-TS export when another instance replaces it A stdout export other than TS read the old instance's sticky subscriptions until its session closed, where they end cleanly, so an epochless GOAWAY drain or publisher replacement exited 0 as if the broadcast had finished. fMP4, MKV, FLV, and the Annex-B sinks now watch the path from before it resolves and exit 1 once a replacement's old instance ends, as TS does without --stitch. Only TS can mark a break in its output. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 3 +- rs/moq-cli/src/subscribe.rs | 152 +++++++++++++++++++++++++++++++++++- 2 files changed, 153 insertions(+), 2 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index d127621494c..13341abcda0 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -393,7 +393,8 @@ Without `--stitch` the export stays on the old publisher while it is up. Only lite-07 sessions carry epochs, so on the default version every return is a replacement. An export that fails while the broadcast stays up, such as on a codec TS cannot carry, exits 1 without lingering. Only `ts` can mark a break, -so the other formats refuse `--linger`. +so the other formats refuse `--linger`, and exit 1 at a replacement once the old +publisher ends. ## Debugging diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 9f9579d76f1..86cedf87189 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -253,6 +253,9 @@ pub struct Subscribe { source: moq_mux::Source, catalog: CatalogFormat, args: SubscribeArgs, + /// Watching for a replacement since before the path resolved. TS follows the path + /// itself, through `container::ts::Follower`. + replacement: Option, } impl Subscribe { @@ -265,6 +268,10 @@ impl Subscribe { args: SubscribeArgs, ) -> anyhow::Result { let path = moq_net::Path::new(path).to_owned(); + let replacement = match args.format { + SubscribeFormat::Ts => None, + _ => Some(Replacement::watch(&origin, &path)?), + }; let broadcast = origin .request_broadcast(&path, None) .await @@ -276,6 +283,7 @@ impl Subscribe { source, catalog, args, + replacement, }) } @@ -287,7 +295,14 @@ impl Subscribe { } /// Write the broadcast to stdout until it ends. - pub async fn run(self) -> anyhow::Result<()> { + pub async fn run(mut self) -> anyhow::Result<()> { + match self.replacement.take() { + Some(replacement) => replacement.guard(self.export()).await, + None => self.export().await, + } + } + + async fn export(self) -> anyhow::Result<()> { match self.args.format { SubscribeFormat::Fmp4 => self.run_fmp4().await, SubscribeFormat::Mkv => self.run_mkv().await, @@ -489,3 +504,138 @@ impl Subscribe { Ok(()) } } + +/// Watches a path for another publisher instance taking it from the one resolved there. +/// +/// Only TS can mark a break in its output, so the other formats end at a replacement. +struct Replacement { + follow: moq_net::announce::Follow, + path: moq_net::PathOwned, +} + +impl Replacement { + /// Watch `path` from the route serving it now. Build it before resolving the path, so a + /// replacement racing the resolution is caught rather than folded into the first start. + fn watch(origin: &moq_net::origin::Consumer, path: &moq_net::PathOwned) -> anyhow::Result { + let mut follow = origin.follow(path)?; + // The cursor replays the routes on hand at once: take the start naming the one serving + // the path now, so the next restart marks another instance. + let _ = follow.poll_next(&moq_net::kio::Waiter::noop()); + Ok(Self { + follow, + path: path.clone(), + }) + } + + /// Run `export` until it ends, failing it if another instance took the path first. + /// + /// Subscriptions are sticky, so the export rides the old instance out before failing, as + /// TS does without `--stitch`. + async fn guard(mut self, export: impl Future>) -> anyhow::Result<()> { + let replaced = async { + while let Some(event) = self.follow.next().await { + if let moq_net::announce::Event::Restart(_) = event { + return; + } + } + // The origin closed, so nothing can replace the broadcast any more. + std::future::pending().await + }; + + tokio::pin!(export); + tokio::select! { + end = &mut export => return end, + () = replaced => {} + } + let path = &self.path; + tracing::info!(%path, "broadcast replaced, exporting the old instance until it ends"); + let replaced = + format!("another publisher instance replaced broadcast `{path}`; only `export ts --stitch` can follow it"); + match export.await { + Ok(()) => Err(anyhow::anyhow!(replaced)), + Err(err) => Err(err.context(replaced)), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Announce a broadcast at `room` under `epoch`, returning what keeps it up. + fn announce(origin: &moq_net::origin::Producer, epoch: &moq_net::Epoch) -> moq_net::broadcast::Producer { + origin + .publish("room", moq_net::origin::Route::default().with_epoch(epoch.clone())) + .unwrap() + } + + /// Follow `room`, once it is announced, for an export that runs until `end` sends. + async fn export( + origin: &moq_net::origin::Producer, + ) -> ( + tokio::sync::oneshot::Sender<()>, + tokio::task::JoinHandle>, + ) { + let consumer = origin.consume(); + consumer.routed("room").await.unwrap(); + let replacement = Replacement::watch(&consumer, &"room".into()).unwrap(); + let (end, ended) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(replacement.guard(async move { + ended.await.unwrap(); + Ok(()) + })); + (end, task) + } + + /// Another instance taking the path, as every return of an epochless route is, keeps the + /// export on the old one until it ends, then fails rather than reading as a finish. + #[tokio::test] + async fn a_replacement_fails_the_export_once_it_ends() { + tokio::time::pause(); + let origin = moq_tokio::origin::spawn(); + let _old = announce(&origin, &moq_net::Epoch::mint()); + let (end, task) = export(&origin).await; + + let _new = announce(&origin, &moq_net::Epoch::mint()); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + assert!(!task.is_finished(), "the export left the old instance before it ended"); + + end.send(()).unwrap(); + let err = task.await.unwrap().expect_err("a replaced export ended cleanly"); + assert!(err.to_string().contains("replaced broadcast `room`"), "{err:#}"); + } + + /// The same epoch is the same content, so the export carries on. + #[tokio::test] + async fn the_same_epoch_does_not_fail_the_export() { + tokio::time::pause(); + let origin = moq_tokio::origin::spawn(); + let epoch = moq_net::Epoch::mint(); + let _old = announce(&origin, &epoch); + let (end, task) = export(&origin).await; + + let _new = announce(&origin, &epoch); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + + end.send(()).unwrap(); + task.await.unwrap().unwrap(); + } + + /// A broadcast that goes offline ends the export as it always has, so what returns + /// after it is no replacement. + #[tokio::test] + async fn a_return_after_an_end_is_no_replacement() { + tokio::time::pause(); + let origin = moq_tokio::origin::spawn(); + let old = announce(&origin, &moq_net::Epoch::mint()); + let (end, task) = export(&origin).await; + + drop(old); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + let _new = announce(&origin, &moq_net::Epoch::mint()); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + + end.send(()).unwrap(); + task.await.unwrap().unwrap(); + } +} From 35c37ab373ce7d167fee5101d1fecd599e25e0cd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:24:33 -0700 Subject: [PATCH 03/14] feat(ffi): follow a broadcast's instances through announced_broadcast `MoqAnnouncedBroadcast::available()` resolved the path once, and calling it again handed back whatever served the path now. A binding consumer had to watch the prefix's announcements itself to notice a restart, and none did, so an epochless GOAWAY drain or publisher replacement left its tracks ending cleanly as if the broadcast had finished. Each `available()` now resolves the next publisher instance: the first call the one serving the path, later calls the one that takes over (a restart) or that returns after an end. The first call behaves as before. Every language wrapper forwards it, so only their docs change. Co-Authored-By: Claude Opus 5.5 --- dart/moq/lib/src/client.dart | 3 +- doc/lib/cpp/index.md | 4 + go/wrapper/client.go | 4 +- go/wrapper/origin.go | 13 +- .../jvmAndAndroidMain/kotlin/dev/moq/Moq.kt | 5 +- py/moq-rs/moq/client.py | 2 +- py/moq-rs/moq/origin.py | 15 ++- rs/moq-ffi/src/origin.rs | 61 ++++++--- rs/moq-ffi/src/test.rs | 118 ++++++++++++++++++ swift/Sources/Moq/Origin.swift | 10 +- 10 files changed, 201 insertions(+), 34 deletions(-) diff --git a/dart/moq/lib/src/client.dart b/dart/moq/lib/src/client.dart index e3119ca4153..0f80f763594 100644 --- a/dart/moq/lib/src/client.dart +++ b/dart/moq/lib/src/client.dart @@ -170,7 +170,8 @@ final class Moq { AnnounceOptions options = const AnnounceOptions(), }) => session.consume().announced(config: options._ffi); - /// Wait for a broadcast announced at exactly [path]. + /// Follow the broadcast at exactly [path]: each `available()` resolves the + /// next publisher instance serving it, such as a restarted publisher. AnnouncedBroadcast announcedBroadcast(String path) => session.consume().announcedBroadcast(path: path); diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 2b5e1beae87..60711ace4f8 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -106,6 +106,10 @@ if (auto catalog = ok(catalogs->next().get())) { // std::nullopt once the cata } ``` +Call `available()` again to wait for the next publisher instance at that path, such as a +restarted publisher, and drop what was read from the previous one then: its tracks end +when its session does. + `MediaTrackConsumer` reads in decode order. For arrival-order reads, decode each group from `TrackConsumer::recv_group` with `moq::MediaGroupConsumer::init(group, container)`; frames carry container media time as they land. diff --git a/go/wrapper/client.go b/go/wrapper/client.go index caf04525848..c75ee29aaca 100644 --- a/go/wrapper/client.go +++ b/go/wrapper/client.go @@ -280,8 +280,8 @@ func (c *Client) Announced(options AnnounceOptions) (*AnnounceConsumer, error) { return c.consumer.Announced(options) } -// AnnouncedBroadcast waits for a route covering path, then resolves the -// broadcast there. +// AnnouncedBroadcast follows path, resolving each publisher instance that serves +// it. func (c *Client) AnnouncedBroadcast(path string) (*AnnouncedBroadcast, error) { return c.consumer.AnnouncedBroadcast(path) } diff --git a/go/wrapper/origin.go b/go/wrapper/origin.go index 73bd7815be9..088e539b6be 100644 --- a/go/wrapper/origin.go +++ b/go/wrapper/origin.go @@ -147,8 +147,8 @@ func (o *OriginConsumer) Announced(options AnnounceOptions) (*AnnounceConsumer, return &AnnounceConsumer{inner: inner}, nil } -// AnnouncedBroadcast waits for a route covering an exact path, then resolves -// the broadcast there. +// AnnouncedBroadcast follows an exact path, resolving each publisher instance +// that serves it. func (o *OriginConsumer) AnnouncedBroadcast(path string) (*AnnouncedBroadcast, error) { inner, err := o.inner.AnnouncedBroadcast(path) if err != nil { @@ -195,12 +195,17 @@ func (a *AnnounceConsumer) Cancel() { a.inner.Cancel() } -// AnnouncedBroadcast awaits a specific broadcast becoming available. +// AnnouncedBroadcast follows a specific broadcast across its publisher instances. type AnnouncedBroadcast struct { inner *ffi.MoqAnnouncedBroadcast } -// Available blocks until the broadcast is available and returns its consumer. +// Available blocks until the next publisher instance serves the broadcast and +// returns its consumer. The first call resolves the instance serving the path once +// one is announced. Each later call waits for another to take over (a restarted +// publisher, or a replacement session) or for the broadcast to return after it went +// offline. Drop what was read from the previous instance then: its tracks end when +// its session does. func (a *AnnouncedBroadcast) Available(ctx context.Context) (*BroadcastConsumer, error) { inner, err := bridge.CallHandle(ctx, a.inner.Cancel, a.inner.Available) if err != nil { diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt index a09a3ddfb72..c6eb6b46973 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt @@ -33,10 +33,11 @@ class Moq internal constructor( session.consume().announced(config) /** - * Await a route covering exactly [path], then resolve the broadcast there. + * Follow exactly [path], resolving each publisher instance that serves it. * * Unlike [requestBroadcast] this waits indefinitely for a future - * announcement. Cancel the returned handle to stop waiting. + * announcement, and each later `available()` waits for the next instance, + * such as a restarted publisher. Cancel the returned handle to stop waiting. */ fun announcedBroadcast(path: String): AnnouncedBroadcast = session.consume().announcedBroadcast(path) diff --git a/py/moq-rs/moq/client.py b/py/moq-rs/moq/client.py index 3636aa26460..a7eccdb4129 100644 --- a/py/moq-rs/moq/client.py +++ b/py/moq-rs/moq/client.py @@ -154,7 +154,7 @@ def announced(self, prefix: str = "", *, filter: str | None = None, hidden: bool return self._require_consumer().announced(prefix, filter=filter, hidden=hidden) def announced_broadcast(self, path: str) -> AnnouncedBroadcast: - """Await announcement of the broadcast at exactly ``path``. + """Follow the broadcast at exactly ``path`` across its publisher instances. See :meth:`OriginConsumer.announced_broadcast`. """ diff --git a/py/moq-rs/moq/origin.py b/py/moq-rs/moq/origin.py index c5bdd575d6b..00e83ba2e18 100644 --- a/py/moq-rs/moq/origin.py +++ b/py/moq-rs/moq/origin.py @@ -81,10 +81,11 @@ def cancel(self) -> None: class AnnouncedBroadcast: - """Awaitable that resolves once a route covers a specific path. + """Follows a specific path across the publisher instances that serve it. ``await`` it (or call :meth:`available`) to get the :class:`BroadcastConsumer` - once the broadcast becomes available. Usable as an async context manager. + once the broadcast becomes available, and again for each instance that takes over. + Usable as an async context manager. """ def __init__(self, inner: MoqAnnouncedBroadcast) -> None: @@ -97,7 +98,13 @@ async def __aexit__(self, *exc) -> None: self.cancel() async def available(self) -> BroadcastConsumer: - """Await a covering route and return the resolved broadcast consumer.""" + """Await the next publisher instance serving the path and return its consumer. + + The first call resolves the instance serving the path once one is announced. Each + later call waits for another to take over (a restarted publisher, or a replacement + session) or for the broadcast to return after it went offline. Drop what was read + from the previous instance then: its tracks end when its session does. + """ return BroadcastConsumer(await self._inner.available()) def __await__(self): @@ -181,7 +188,7 @@ def announced(self, prefix: str = "", *, filter: str | None = None, hidden: bool return AnnounceConsumer(self._inner.announced(MoqAnnounceConfig(prefix=prefix, filter=filter, hidden=hidden))) def announced_broadcast(self, path: str) -> AnnouncedBroadcast: - """Await the broadcast at ``path``, resolving once something can serve it. + """Follow the broadcast at ``path``, resolving each publisher instance that serves it. A broadcast created on this origin resolves once it is announced, like a remote one. diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 191813ac8e6..9d1d7b74567 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -202,21 +202,42 @@ impl Announced { } } -/// Waits for a route covering one exact path, then resolves it. +/// Follows one exact path, resolving each publisher instance that serves it. struct AnnouncedBroadcast { origin: moq_net::origin::Consumer, path: moq_net::PathOwned, + follow: moq_net::announce::Follow, + /// An instance serves the path that was not handed out yet. + unseen: bool, driver: Background, } impl AnnouncedBroadcast { async fn available(&mut self) -> Result, MoqError> { - // `routed_broadcast` rides out the churn between something covering the path - // and the path actually resolving (failover, an advertise-only announce - // racing its handler). let broadcast = self .driver - .watch(async { Ok(self.origin.routed_broadcast(&self.path).await?) }) + .watch(async { + // Apply every change on hand before resolving, so a burst of them hands out + // only the instance serving the path now. + loop { + let event = match self.follow.poll_next(&kio::Waiter::noop()) { + std::task::Poll::Ready(event) => event, + std::task::Poll::Pending if self.unseen => break, + std::task::Poll::Pending => self.follow.next().await, + }; + match event.ok_or(MoqError::Closed)? { + moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, + moq_net::announce::Event::End(_) => self.unseen = false, + // The same instance over another route, which its subscriptions ride out. + moq_net::announce::Event::Update(_) => {} + } + } + self.unseen = false; + // `routed_broadcast` rides out the churn between something covering the path + // and the path actually resolving (failover, an advertise-only announce + // racing its handler). + Ok(self.origin.routed_broadcast(&self.path).await?) + }) .await?; Ok(Arc::new(MoqBroadcastConsumer::routed( broadcast, @@ -287,7 +308,7 @@ impl From for MoqAnnounceEvent { } } -/// Waits for a specific broadcast to be announced. +/// Follows a specific broadcast across the publisher instances that serve it. #[derive(uniffi::Object)] pub struct MoqAnnouncedBroadcast { task: Task, @@ -448,7 +469,7 @@ impl MoqOriginConsumer { })) } - /// Resolve the broadcast at `path`, waiting until something can serve it. + /// Follow the broadcast at `path`, resolving each publisher instance that serves it. /// /// This is how you resolve a path right after connecting: announcements arrive over the /// session after it opens, so `request_broadcast` on its own races them. A @@ -459,19 +480,20 @@ impl MoqOriginConsumer { self.driver.check()?; let path = moq_net::Path::new(&path).to_owned(); - // Probe the permission eagerly so an unreachable path fails here, rather than - // surfacing later as a `Closed` the caller can't tell from the origin ending. - self.inner - .scope(&path, &moq_net::Patterns::from(moq_net::Pattern::all()))?; + // Following checks the permission eagerly, so an unreachable path fails here rather + // than surfacing later as a `Closed` the caller can't tell from the origin ending. + let follow = self.inner.follow(&path)?; Ok(Arc::new(MoqAnnouncedBroadcast { task: Task::new(AnnouncedBroadcast { - // The wait runs on the *unrooted* cursor. `announced_broadcast` narrows with - // `scope`, which leaves the root alone, so the broadcast is handed out named by - // its full path. Rooting the cursor at `path` would name it "" instead, making - // it its own root, and a catalog's `../sibling` reference would read as escaping. + // The wait runs on the *unrooted* cursor, so the broadcast is handed out named + // by its full path. Rooting the cursor at `path` would name it "" instead, + // making it its own root, and a catalog's `../sibling` reference would read as + // escaping. origin: self.inner.clone(), path, + follow, + unseen: false, driver: self.driver.clone(), }), })) @@ -603,9 +625,14 @@ impl MoqAnnounceConsumer { #[uniffi::export] impl MoqAnnouncedBroadcast { - /// Wait until the broadcast is announced. Returns `Closed` if cancelled or the origin is closed. + /// Wait for the next publisher instance serving the broadcast. Returns `Closed` if cancelled + /// or the origin is closed. /// - /// Its end arrives as a [`MoqAnnounceEvent::End`] on the origin's announcements. + /// The first call resolves the instance serving the path, once one is announced. Each later + /// call waits for another to take over, as a restarted publisher or a replacement session + /// does, or for the broadcast to return after it went offline. Drop everything read from the + /// previous instance then and read the new one: the old one's tracks end when its session + /// does, and look finished. pub async fn available(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.available().await }).await } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 633ec70bba1..81418c57bb4 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4229,6 +4229,124 @@ async fn server_client_roundtrip() { server.cancel(); } +/// Subscribe to the first audio rendition in `broadcast`'s catalog. +async fn subscribe_audio(broadcast: &MoqBroadcastConsumer) -> Arc { + let catalogs = MoqMediaCatalogConsumer::subscribe(broadcast).await.unwrap(); + let catalog = tokio::time::timeout(TIMEOUT, catalogs.next()) + .await + .expect("timed out waiting for the catalog") + .unwrap() + .expect("expected a catalog"); + let (name, audio) = catalog.audio.iter().next().expect("an audio rendition"); + MoqMediaTrackConsumer::subscribe( + broadcast, + MoqMediaTrackConfig { + name: name.clone(), + container: audio.container.clone(), + subscription: None, + }, + ) + .await + .unwrap() +} + +/// Write `payload` and read through `track` until it arrives, past what a new subscription +/// starts with. +async fn roundtrip(media: &MoqMediaTrackProducer, track: &MoqMediaTrackConsumer, payload: &[u8], timestamp_us: u64) { + media + .write_frame(MoqFrame { + payload: payload.to_vec(), + timestamp_us: Some(timestamp_us), + }) + .unwrap(); + loop { + let frame = tokio::time::timeout(TIMEOUT, track.next()) + .await + .expect("timed out waiting for a frame") + .unwrap() + .expect("the track ended"); + if frame.payload == payload { + return; + } + } +} + +/// A replacement session from the same publisher re-announces every route while the old one +/// is still open, as a GOAWAY drain does. Without an epoch that is another instance: the +/// next `available()` hands it out, and reading it carries on past the old session's close. +#[tokio::test] +async fn announced_broadcast_follows_an_epochless_replacement() { + let server_origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let server = MoqServer::new(MoqServerConfig { + bind: Some("127.0.0.1:0".into()), + tls: localhost_tls(), + publish: Some(server_origin.clone()), + ..Default::default() + }) + .unwrap(); + let addr = tokio::time::timeout(TIMEOUT, server.listen()) + .await + .expect("listen timed out") + .expect("listen failed"); + let url = format!("https://{addr}"); + let accepting = server.clone(); + let accept = tokio::spawn(async move { + let mut sessions = Vec::new(); + while let Ok(Some(request)) = accepting.accept().await { + sessions.push(request.accept(None, None).await.expect("handshake failed")); + } + sessions + }); + + let broadcast = create_announced(&server_origin, "cam"); + let media = MoqMediaTrackProducer::audio( + &broadcast, + MoqMediaTarget::Named { name: None }, + audio_init(MoqAudioFormat::Opus, opus_head()), + ) + .unwrap(); + + let viewer = MoqOriginProducer::new(MoqOriginConfig::default()); + let connect = || async { + let client = MoqClient::new(MoqClientConfig { + tls: insecure_tls(), + bind: Some("127.0.0.1:0".into()), + consume: Some(viewer.clone()), + ..Default::default() + }) + .unwrap(); + tokio::time::timeout(TIMEOUT, client.connect(url.clone())) + .await + .expect("connect timed out") + .expect("connect failed") + }; + + let first = connect().await; + let followed = viewer.consume().announced_broadcast("cam".into()).unwrap(); + let old = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("timed out waiting for the broadcast") + .unwrap(); + let old_track = subscribe_audio(&old).await; + roundtrip(&media, &old_track, b"before", 1_000_000).await; + + let second = connect().await; + let fresh = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("the replacement was never handed out") + .unwrap(); + first.shutdown().await.unwrap(); + + let fresh_track = subscribe_audio(&fresh).await; + roundtrip(&media, &fresh_track, b"after", 2_000_000).await; + + media.finish().unwrap(); + broadcast.close().unwrap(); + second.shutdown().await.unwrap(); + server.cancel(); + drop(accept.await.unwrap()); +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits diff --git a/swift/Sources/Moq/Origin.swift b/swift/Sources/Moq/Origin.swift index 9bca84e20d1..081e2406f67 100644 --- a/swift/Sources/Moq/Origin.swift +++ b/swift/Sources/Moq/Origin.swift @@ -116,7 +116,7 @@ public final class OriginConsumer: Sendable { AnnounceConsumer(try ffi.announced(config: MoqAnnounceConfig(prefix: prefix, filter: filter, hidden: hidden))) } - /// Wait for a route covering an exact path, then resolve the broadcast there. + /// Follow an exact path, resolving each publisher instance that serves it. public func announcedBroadcast(path: String) throws -> AnnouncedBroadcast { AnnouncedBroadcast(try ffi.announcedBroadcast(path: path)) } @@ -171,8 +171,12 @@ public final class AnnouncedBroadcast: Sendable { self.ffi = ffi } - /// Suspend until a route covers the path, then resolve the broadcast there. - /// Throws `Closed` if cancelled or the origin closes first. + /// Suspend until the next publisher instance serves the path, then resolve it. + /// The first call resolves the instance serving the path once one is announced. + /// Each later call waits for another to take over (a restarted publisher, or a + /// replacement session) or for the broadcast to return after it went offline. + /// Drop what was read from the previous instance then: its tracks end when its + /// session does. Throws `Closed` if cancelled or the origin closes first. public func available() async throws -> BroadcastConsumer { BroadcastConsumer(try await ffi.available()) } From dac09f00b45bcbc5f9faf876337818e3888c9af6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:29:06 -0700 Subject: [PATCH 04/14] feat(c)!: moq_origin_announced_broadcast delivers every instance The callback delivered one broadcast handle and then its terminal 0, so a C consumer never learned that another publisher instance took the path. It now delivers a fresh handle for each instance serving the path, matching moq-ffi's `available()`, and its terminal only on cancel, an error, or the origin closing. The interop C client frees the handles it does not read. The upgrade guide notes this and the non-TS export change. Co-Authored-By: Claude Opus 5.5 --- doc/setup/upgrade.md | 10 ++++ rs/moq-c/src/api.rs | 11 ++-- rs/moq-c/src/origin.rs | 80 +++++++++++++++++++++--------- rs/moq-c/src/test.rs | 46 +++++++++++++++-- test/interop/clients/c/subscribe.c | 9 ++-- 5 files changed, 123 insertions(+), 33 deletions(-) diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 1afdc054f35..f57a7346ce3 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -426,6 +426,16 @@ These land with the next breaking release, not the 2026-09-23 train. decode time (RTMP, a browser) trails by the delay plus the video's send-ahead: 2 s at the new default, where `--delay 500ms` keeps the 1 s it had. moq-srt egress runs at its SRT latency or 1 s, whichever is longer. +- **`moq_origin_announced_broadcast` delivers every publisher instance.** Its + callback gets a new broadcast handle each time another instance takes over + the path or the broadcast returns, and its terminal `0` only on cancel. Free + each handle with `moq_consume_close` once its successor arrives. A binding's + `AnnouncedBroadcast.available()` works the same way, and its first call is + unchanged. +- **`moq export` fMP4, MKV, FLV, H.264, and H.265 exit 1 at a replacement.** + Where the export used to end cleanly once a replaced publisher's session + closed, it now fails. Restart it, or use `export ts --stitch` to follow the + replacement. ## Wire diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index ea76d27838f..0b477df5e6f 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -1819,7 +1819,7 @@ pub extern "C" fn moq_origin_announced_cancel(announced: u32) -> i32 { }) } -/// Consume a broadcast from an origin by path, waiting until something can serve it. +/// Follow a broadcast on an origin by path, delivering each publisher instance that serves it. /// /// Resolves against future announcements: it waits for the announcement to arrive (e.g. over the /// network) and then delivers the broadcast handle via `on_broadcast`. Use it right after @@ -1827,9 +1827,12 @@ pub extern "C" fn moq_origin_announced_cancel(announced: u32) -> i32 { /// reachable now, use [moq_origin_request] instead. A broadcast created on this origin /// resolves once it is announced, like a remote one. /// -/// `on_broadcast` is invoked with a positive broadcast handle once announced, then exactly once -/// more with a terminal code: `0` (the wait finished, including after -/// [moq_origin_announced_broadcast_cancel]) or a negative error. After the terminal (`<= 0`) callback, +/// `on_broadcast` is invoked with a positive broadcast handle once announced, and again with a +/// new handle each time another publisher instance takes over (a restarted publisher, or a +/// replacement session) or the broadcast returns after it went offline. Drop what was read +/// from the previous handle then: its tracks end when its session does. The wait ends with +/// exactly one terminal code: `0` (after [moq_origin_announced_broadcast_cancel], or the origin +/// closing once a handle was delivered) or a negative error. After the terminal (`<= 0`) callback, /// `on_broadcast` is never called again and `user_data` is never touched again, so release /// `user_data` there. The broadcast handle is usable with [moq_consume_catalog] / [moq_consume_track] /// and must be freed separately with [moq_consume_close]. diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index a4109492dd7..ec9c0a86131 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -242,11 +242,11 @@ impl Origin { Ok(()) } - /// Wait until the broadcast at `path` is announced, then deliver its handle via the callback. + /// Follow the broadcast at `path`, delivering each publisher instance's handle via the callback. /// - /// The callback fires the broadcast handle (> 0) once announced, then a terminal `0`. On error - /// or cancellation it fires a single terminal code (`0` on close, negative on error). Returns a - /// task handle for cancellation via [`Self::consume_announced_close`]. + /// The callback fires a broadcast handle (> 0) for each instance serving the path, until a + /// single terminal code (`0` on close, negative on error). Returns a task handle for + /// cancellation via [`Self::consume_announced_close`]. pub fn consume_announced(&mut self, origin: Id, path: String, on_broadcast: OnStatus) -> Result { let origin = self.active.get(origin).ok_or(Error::OriginNotFound)?; origin.driver.check()?; @@ -282,27 +282,61 @@ impl Origin { path: String, mut close: oneshot::Receiver<()>, ) -> Result<(), Error> { - // `routed_broadcast` rides out the churn between something covering the path - // and the path actually resolving (failover, an advertise-only announce - // racing its handler). `biased` so a pending close always wins. - let broadcast = tokio::select! { - biased; - _ = &mut close => return Ok(()), - resolved = consumer.routed_broadcast(path.as_str()) => match resolved { - Ok(broadcast) => broadcast, - // An unreachable path and a closed origin both mean no broadcast - // can ever arrive here. - Err(moq_net::Error::Unauthorized | moq_net::Error::Closed) => { - return Err(Error::BroadcastNotFound); - } - Err(err) => return Err(err.into()), - }, + // An unreachable path and a closed origin both mean no broadcast can ever arrive here. + let mut follow = match consumer.follow(path.as_str()) { + Ok(follow) => follow, + Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), + Err(err) => return Err(err.into()), }; + let mut delivered = false; - // Hold the lock only to buffer the broadcast; release it before the callback. - let broadcast_id = State::lock().consume.start(broadcast, Some(consumer))?; - callback.call(broadcast_id); - Ok(()) + // Deliver each publisher instance serving the path: the first, then each one that takes + // over or returns. `biased` so a pending close always wins. + loop { + // Apply every change on hand before resolving, so a burst of them delivers only the + // instance serving the path now. + let mut unseen = false; + loop { + let event = match follow.poll_next(&kio::Waiter::noop()) { + std::task::Poll::Ready(event) => event, + std::task::Poll::Pending if unseen => break, + std::task::Poll::Pending => tokio::select! { + biased; + _ = &mut close => return Ok(()), + event = follow.next() => event, + }, + }; + match event { + None if delivered => return Ok(()), + None => return Err(Error::BroadcastNotFound), + Some(moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_)) => unseen = true, + Some(moq_net::announce::Event::End(_)) => unseen = false, + // The same instance over another route, which its subscriptions ride out. + Some(moq_net::announce::Event::Update(_)) => {} + } + } + + // `routed_broadcast` rides out the churn between something covering the path + // and the path actually resolving (failover, an advertise-only announce + // racing its handler). + let broadcast = tokio::select! { + biased; + _ = &mut close => return Ok(()), + resolved = consumer.routed_broadcast(path.as_str()) => match resolved { + Ok(broadcast) => broadcast, + Err(moq_net::Error::Closed) if delivered => return Ok(()), + Err(moq_net::Error::Unauthorized | moq_net::Error::Closed) => { + return Err(Error::BroadcastNotFound); + } + Err(err) => return Err(err.into()), + }, + }; + + // Hold the lock only to buffer the broadcast; release it before the callback. + let broadcast_id = State::lock().consume.start(broadcast, Some(consumer.clone()))?; + callback.call(broadcast_id); + delivered = true; + } } /// Request the broadcast at `path`, delivering its handle once it can be served. diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index b4ce355b71a..b4bc496fd22 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -2932,7 +2932,7 @@ fn consume_announced_local() { // Start waiting before the broadcast exists: the announcement arrives afterwards. let cb = Callback::new(); let path = b"live"; - let _task = id(unsafe { + let task = id(unsafe { moq_origin_announced_broadcast( origin, path.as_ptr() as *const c_char, @@ -2947,9 +2947,10 @@ fn consume_announced_local() { let format = moq_audio_format::MOQ_AUDIO_FORMAT_OPUS; let media = id(publish_audio(broadcast, format, &init, None)); - // First the broadcast handle, then a terminal 0 once the wait finishes. + // The broadcast handle; the wait carries on for the next instance until cancelled. let consume = id(cb.recv()); - assert_eq!(cb.recv_terminal(), 0, "wait delivers terminal 0 after the handle"); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); // The delivered handle behaves like one from moq_origin_request. let catalog_cb = Callback::new(); @@ -3113,6 +3114,45 @@ fn consume_announced_close_cancels() { assert_eq!(moq_origin_close(origin), 0); } +/// Publish `path` on `origin` under a fresh epoch, as each run of a publisher does. The C ABI +/// announces without one, so this reaches past it. +fn publish_run(origin: u32, path: &str) -> moq_net::broadcast::Producer { + let _runtime = crate::ffi::RUNTIME.enter(); + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let origin = crate::Id::try_from(origin).unwrap(); + State::lock().origin.get(origin).unwrap().publish(path, route).unwrap() +} + +/// Another publisher instance taking the path delivers a fresh handle on the same wait. +#[test] +fn consume_announced_follows_a_restart() { + let origin = id(moq_origin_create()); + let cb = Callback::new(); + let path = b"live"; + let task = id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }); + + let first_run = publish_run(origin, "live"); + let first = id(cb.recv()); + let second_run = publish_run(origin, "live"); + let second = id(cb.recv()); + assert_ne!(first, second, "the restart delivered the old handle"); + + assert_eq!(moq_consume_close(first), 0); + assert_eq!(moq_consume_close(second), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((first_run, second_run)); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn video_publish_consume() { let origin = id(moq_origin_create()); diff --git a/test/interop/clients/c/subscribe.c b/test/interop/clients/c/subscribe.c index d960dc01acf..0bdea98d9dd 100644 --- a/test/interop/clients/c/subscribe.c +++ b/test/interop/clients/c/subscribe.c @@ -74,8 +74,9 @@ static void on_frame(void *ud, int32_t frame) { moq_consume_frame_free((uint32_t)frame); } -// Delivers the broadcast handle once it's announced, then once more with a -// terminal code (<= 0). Store the first positive handle and wake main. +// Delivers a broadcast handle for each publisher instance serving the path, then +// a terminal code (<= 0). Store the first positive handle and wake main; one frame +// is all this client reads, so a later instance's handle is freed unread. static void on_broadcast(void *ud, int32_t broadcast) { ctx_t *c = (ctx_t *)ud; if (broadcast <= 0) { // 0 = ended, negative = error @@ -83,11 +84,13 @@ static void on_broadcast(void *ud, int32_t broadcast) { return; } pthread_mutex_lock(&c->mu); - if (c->broadcast <= 0) { + int first = c->broadcast <= 0; + if (first) { c->broadcast = broadcast; pthread_cond_broadcast(&c->cv); } pthread_mutex_unlock(&c->mu); + if (!first) moq_consume_close((uint32_t)broadcast); } static void on_catalog(void *ud, int32_t catalog) { From 4db3ed43406f4e6c382103f5b56241514a4fb316 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:30:22 -0700 Subject: [PATCH 05/14] fix(interop): close the session when a C or C++ client fails The C client's fail() and the C++ client's fail() exited without closing their session, so every failing cell also left the relay to time its connection out. Both now close the session, bounded at five seconds, then exit 1. Co-Authored-By: Claude Opus 5.5 --- test/interop/clients/c/subscribe.c | 25 +++++++++++++++++++++---- test/interop/clients/cpp/main.cpp | 21 ++++++++++++++++++--- 2 files changed, 39 insertions(+), 7 deletions(-) diff --git a/test/interop/clients/c/subscribe.c b/test/interop/clients/c/subscribe.c index 0bdea98d9dd..44fd96570a5 100644 --- a/test/interop/clients/c/subscribe.c +++ b/test/interop/clients/c/subscribe.c @@ -121,19 +121,35 @@ static void on_catalog(void *ud, int32_t catalog) { moq_consume_catalog_free((uint32_t)catalog); } +// The context whose session a failure closes, once connected. +static ctx_t *failing; + // Report a failure and exit 1, for the paths where moq-c still holds &ctx. // // _exit rather than a return: it leaves main's frame (and so ctx) intact for the -// callbacks still pointing at it, which is what the failure paths want. Draining -// first would work too, but a client that has already failed should print its -// reason and go, not block for up to ten seconds on a teardown that may be exactly -// what is broken. +// callbacks still pointing at it, which is what the failure paths want. Only the +// session is closed first, so the relay sees a close rather than timing the +// connection out; a client that has already failed should not wait on the rest of +// a teardown that may be exactly what is broken. static _Noreturn void fail(const char *fmt, ...) { va_list ap; va_start(ap, fmt); vfprintf(stderr, fmt, ap); va_end(ap); fflush(stderr); + + ctx_t *c = failing; + failing = NULL; // a failure while closing exits at once + if (c && moq_session_close((uint32_t)c->session) == 0) { + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += 5; + pthread_mutex_lock(&c->mu); + while (!c->done_status) { + if (pthread_cond_timedwait(&c->cv, &c->mu, &deadline) != 0) break; // timed out + } + pthread_mutex_unlock(&c->mu); + } _exit(1); } @@ -209,6 +225,7 @@ int main(int argc, char **argv) { fprintf(stderr, "error: moq_session_connect failed: %d\n", c.session); return 1; } + failing = &c; struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 06b10cc6f43..ecc5a78ce0d 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #ifdef _WIN32 @@ -53,14 +54,26 @@ constexpr uint32_t AUDIO_FRAME_DURATION_US = 2'500; constexpr uint64_t AUDIO_BATCH_US = 20'000; constexpr size_t AUDIO_BATCH_SAMPLES = AUDIO_RATE * AUDIO_BATCH_US / 1'000'000; +// The open session, which a failure closes so the relay sees a close rather than timing the +// connection out. +std::shared_ptr open_session; + +[[noreturn]] void exit_failed() { + // Taken first, so a failure while closing cannot close it again. + if (auto session = std::exchange(open_session, nullptr)) { + session->shutdown().wait_for(std::chrono::seconds(5)); + } + std::exit(1); +} + [[noreturn]] void fail(const char *what, const moq::Error &error) { std::fprintf(stderr, "error: %s: %s\n", what, error.to_string().c_str()); - std::exit(1); + exit_failed(); } [[noreturn]] void fail(const char *what) { std::fprintf(stderr, "error: %s\n", what); - std::exit(1); + exit_failed(); } template @@ -91,12 +104,14 @@ std::shared_ptr connect(const std::string &url, Clock::time_point moq::ClientConfig config; config.tls.insecure = true; auto client = ok(moq::Client::init(config), "client init"); - return until(client->connect(url), deadline, "connect"); + open_session = until(client->connect(url), deadline, "connect"); + return open_session; } // Closes the session and waits for the close to reach the relay. `cancel` would return at // once, and stopping the runtime right after it can leave the relay to time the connection out. void close_session(const std::shared_ptr &session) { + open_session = nullptr; until(session->shutdown(), Clock::now() + std::chrono::seconds(5), "session shutdown"); } From daea6fd833c6d11b0ad7528f383ec9133b7722f9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:31:24 -0700 Subject: [PATCH 06/14] quest: finish restart-followers, split the sinks and OBS out The stdout exports, moq-ffi, moq-c, and the interop clients are done. JS already follows through `origin.request(path, { announced: true })`, and the interop subscribers read one frame, so they need no follow. What remains are moq-cli's server and push sinks and the OBS source, each its own quest. Co-Authored-By: Claude Opus 5.5 --- .../m0/broadcast-epoch/catalog-references.md | 2 +- quest/m1/README.md | 3 +- quest/m1/obs-restart-follow.md | 22 +++++++ quest/m1/restart-followers.md | 62 ------------------- quest/m1/sink-restart-follow.md | 35 +++++++++++ 5 files changed, 60 insertions(+), 64 deletions(-) create mode 100644 quest/m1/obs-restart-follow.md delete mode 100644 quest/m1/restart-followers.md create mode 100644 quest/m1/sink-restart-follow.md diff --git a/quest/m0/broadcast-epoch/catalog-references.md b/quest/m0/broadcast-epoch/catalog-references.md index 8c6ed71eec9..a65fbdb9342 100644 --- a/quest/m0/broadcast-epoch/catalog-references.md +++ b/quest/m0/broadcast-epoch/catalog-references.md @@ -53,7 +53,7 @@ retarget" rule above applies only to readers that follow no announcement. restarts its pump. - Exporters: a late first announcement starts the rendition and End ends it. Restart behavior belongs to - [restart followers](/quest/m1/restart-followers.md). + [sink restart follow](/quest/m1/sink-restart-follow.md). - Share this in `moq_mux` (a `Source::follow`-style API; the shape is open) rather than per reader. `Source::bind` binds once today, as `failed_binding_does_not_retry_a_later_publisher` asserts; that changes. diff --git a/quest/m1/README.md b/quest/m1/README.md index 2e859e5fe60..49b2e830b79 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -61,7 +61,8 @@ above them. - [Release builds](/quest/m0/release-link.md) - package artifacts can be built - [qmux credit](/quest/m0/qmux-credit.md) - dropped/stopped streams return credit and close is delivered -- [Restart followers](/quest/m1/restart-followers.md) - first-party consumers follow an epochless `Restart`, from a GOAWAY drain or a publisher replacement, by resolving the path again +- [Sink restart follow](/quest/m1/sink-restart-follow.md) - moq-cli's server and push sinks follow an epochless `Restart` where their output can mark it, and fail loud otherwise +- [OBS restart follow](/quest/m1/obs-restart-follow.md) - the OBS source keeps playing across a GOAWAY drain or a publisher replacement - [Cluster routing](/quest/m1/cluster-routing/README.md) - prove shared routing contracts, implement per-origin routes, and connect outbound local clusters - [P2P](/quest/m1/p2p/README.md) - application-selected peer sessions use shared priorities, routing, and same-epoch handover - [Routing guide](/quest/m1/routing-guide.md) - concepts and practical outbound federation/P2P setup diff --git a/quest/m1/obs-restart-follow.md b/quest/m1/obs-restart-follow.md new file mode 100644 index 00000000000..3015d7df061 --- /dev/null +++ b/quest/m1/obs-restart-follow.md @@ -0,0 +1,22 @@ +# [S] The OBS source follows a broadcast's restarts + +## Goal + +The OBS MoQ source keeps playing across an epochless GOAWAY drain or a +publisher replacement. When `AnnouncedBroadcast::available()` hands out the +next publisher instance, the source drops its catalog and track readers for +the old one and subscribes to the new one, without blanking for longer than +the new instance takes to deliver a keyframe. + +## Plan + +`cpp/obs/src/moq-source.cpp` calls `available()` once (`moq_source_on_connect`) +and treats a clean catalog end as final, so a replaced instance leaves the +source frozen once the old session closes. Since moq-ffi's `available()` now +resolves each instance in turn, re-arm it after every instance and rebuild the +`Connection`'s consumer, catalog and tracks on each one, reusing the decoders +when the codec config is unchanged. + +Cover it in `cpp/obs/test` with a replacement while the source plays. + +Public API: none. Wire: none. diff --git a/quest/m1/restart-followers.md b/quest/m1/restart-followers.md deleted file mode 100644 index 924814da6fd..00000000000 --- a/quest/m1/restart-followers.md +++ /dev/null @@ -1,62 +0,0 @@ -# [L] First-party consumers follow an epochless Restart - -## Goal - -Every first-party consumer that resolves a broadcast and keeps reading holds -the announce stream for its path, and on `Restart` drops its handles and -resolves the path again. On `Unroutable` it fails; it never retries a -request. A GOAWAY drain or a publisher replacement on an epochless route -then costs a long-lived consumer nothing but the restart. - -## Plan - -Moved to m1 on 2026-10-13 (maintainer): the WebSocket-to-QUIC upgrade is -removed, so an epochless Restart no longer follows -every connect; it remains for GOAWAY drains and publisher replacement. - -The origin side is done: a handle resolved through a peer keeps serving new -requests while the peer's epochless replacement is held, and refuses them -with `Unroutable` only once the `Restart` is deliverable (`Release` in `rs/moq-net/src/model/origin.rs`, -`rs/moq-net/tests/replacement.rs`). So a consumer that never follows `Restart` -works through the hold (300 ms by default), and its open subscriptions stay -sticky on the old session until it closes, where they end cleanly. A -consumer that opens a request after the hold, or reads past the old -session's close, needs this quest. - -Audit and fix: - -- Rust: `moq export` and `moq subscribe` (the interop subscriber), and any - other moq-cli command that resolves a broadcast and keeps reading. The TS - export's `Follower` (`rs/moq-mux/src/container/ts/follower.rs`) already - watches the path, but treats any epochless return as a replacement: - sticky until the old instance ends, then `Error::Replaced`, unless - stitching. The fmp4, mkv, h264, h265, and flv exports don't follow at all. - Decide whether an epochless `Restart` restarts the output (a new init - segment mid-stream), stitches like TS, or ends the export. - `moq play` already follows `Restart` (`rs/moq-cli/src/play/media.rs`). -- Bindings: moq-ffi's catalog and container consumers, as Go, C++, and - Python reach them, and moq-c's (`moq_origin_announced_broadcast`, - `moq_consume_catalog`). Prefer following once in moq-ffi and moq-c over - duplicating it in every language wrapper. New API goes through the - Cross-Package Sync table. -- JS: `js/net`'s origin owns restart following (`js/watch/src/broadcast.ts` - mirrors its current handle). Confirm the js-native interop subscriber, - which requests with `announced: true`, follows a Rust relay's `Restart`. -- The interop clients under `test/interop/clients`, where they resolve a - broadcast themselves. - -Regression: extend `rs/moq-net/tests/replacement.rs` (or the consumer's own -tests) so each follower reads past the old session's close after an -epochless replacement. Then run `just test interop --all`. - -If a failing client still leaves its connection to idle out, fix that -client's error path to close its session, as a separate commit. The C -client's `fail()` (`_exit` without `moq_session_close`) and the C++ client -on a consumer error both look suspect. - -Public API: consumer-follow API in moq-ffi and moq-c, if the audit needs it. -Wire: none. - -## Related - -- [Broadcast epochs](/quest/m0/broadcast-epoch/README.md) - the players that already follow `Restart` diff --git a/quest/m1/sink-restart-follow.md b/quest/m1/sink-restart-follow.md new file mode 100644 index 00000000000..8d7dac2d45c --- /dev/null +++ b/quest/m1/sink-restart-follow.md @@ -0,0 +1,35 @@ +# [M] Server and push sinks handle a broadcast's restarts + +## Goal + +Every moq-cli sink that resolves a broadcast and keeps serving it handles an +epochless `Restart` on purpose: it follows the new publisher instance where +its output can mark the break, and otherwise fails loud rather than ending as +if the broadcast had finished. + +## Plan + +The stdout exports already do this: TS follows with `--stitch`, and the other +formats exit 1 once the replaced instance ends (`Replacement` in +`rs/moq-cli/src/subscribe.rs`). Audit and fix the rest the same way: + +- `export hls`: `moq-hls`'s server resolves per request; check that a restart + moves new requests to the new instance and how a playlist marks the break + (`EXT-X-DISCONTINUITY`). Coordinate with + [HLS export failure signal](/quest/m1/hls-failure-signal.md). +- `export rtmp`, `export srt`, `export rtc`, both `--listen` (per-viewer + streams in `moq-rtmp`, `moq-srt`, `moq-rtc`) and `--connect` (one push). + SRT already rides the TS follower (`--linger`, `--stitch`); confirm its + listen side does too. FLV and WebRTC need a decision. +- `export archive`: a restart already fails a continued store under a new + epoch; decide what a mid-recording restart does. +- `transcode`: its source broadcast; following means restarting the ladder + on the new instance. + +Share one helper (the stdout exports' `Replacement`, or a `moq_mux` +`Source::follow`-style API per +[catalog references](/quest/m0/broadcast-epoch/catalog-references.md)) +rather than one rule per sink. Each fix lands with a test that replaces the +publisher while the sink runs. + +Public API: possibly a shared follow helper in `moq-mux`. Wire: none. From 705bc5c6b9167d32424eea6845437e5640a8a960 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 23:48:02 -0700 Subject: [PATCH 07/14] chore(dart): regenerate bindings for the announced_broadcast docs Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 92aba78c293..7aedba2c502 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -13710,7 +13710,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_available() != - 17719) { + 36710) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_cancel() != 63175) { @@ -13729,7 +13729,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_announced_broadcast() != - 8509) { + 56171) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqoriginconsumer_request_broadcast() != From ab634ffcc7317fe7e1762088fcbdd59a0a7162d0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:39:50 -0700 Subject: [PATCH 08/14] fix(ffi): keep an instance unseen until available() resolves it A foreign cancel only drops the available() future, so a caller retrying after a timeout keeps the same handle. The wait cleared its unseen flag before resolving, so a wait dropped while the path was covered but not yet served lost the instance, and the retry waited for another one. Co-Authored-By: Claude Opus 5.5 --- rs/moq-ffi/src/origin.rs | 8 +++++--- rs/moq-ffi/src/test.rs | 29 +++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 9d1d7b74567..b73f16cd932 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -232,11 +232,13 @@ impl AnnouncedBroadcast { moq_net::announce::Event::Update(_) => {} } } - self.unseen = false; // `routed_broadcast` rides out the churn between something covering the path // and the path actually resolving (failover, an advertise-only announce - // racing its handler). - Ok(self.origin.routed_broadcast(&self.path).await?) + // racing its handler). The instance stays unseen until it resolves: a foreign + // cancel only drops this future, and the caller may call again. + let broadcast = self.origin.routed_broadcast(&self.path).await?; + self.unseen = false; + Ok(broadcast) }) .await?; Ok(Arc::new(MoqBroadcastConsumer::routed( diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 2e554e90e1e..6740e30ee96 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4383,6 +4383,35 @@ async fn announced_broadcast_follows_an_epochless_replacement() { drop(accept.await.unwrap()); } +/// A foreign cancel only drops the `available()` future, so a caller retrying after a timeout +/// keeps the same handle. Dropped while the instance is covered but not yet served, the retry +/// must still hand that instance out rather than wait for another. +#[tokio::test] +async fn announced_broadcast_retries_a_dropped_wait() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, ""); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let dropped = tokio::spawn(async move { waiting.available().await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + dropped.abort(); + assert!(matches!(dropped.await, Err(err) if err.is_cancelled())); + + let retry = tokio::spawn(async move { followed.available().await }); + let served = MoqBroadcastProducer::new().unwrap(); + request.accept(&served).unwrap(); + tokio::time::timeout(TIMEOUT, retry) + .await + .expect("the retry waited for another instance") + .unwrap() + .unwrap(); + served.close().unwrap(); +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits From af01067b9571348edb19a2f21f17aaf6dfda0262 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:54:29 -0700 Subject: [PATCH 09/14] fix(cli): let a replacement on hand win over the export's end When the old instance ended before the guard ran again, the replacement and the finished export were both ready, and select! picked one at random, so a replaced export could exit 0. Bias the select toward the replacement. Co-Authored-By: Claude Opus 5.5 --- rs/moq-cli/src/subscribe.rs | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 7d8b21a4d16..559b54a2352 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -547,8 +547,10 @@ impl Replacement { tokio::pin!(export); tokio::select! { - end = &mut export => return end, + // A replacement on hand when the old instance ends still fails the export. + biased; () = replaced => {} + end = &mut export => return end, } let path = &self.path; tracing::info!(%path, "broadcast replaced, exporting the old instance until it ends"); @@ -624,6 +626,30 @@ mod tests { task.await.unwrap().unwrap(); } + /// The old instance can end before the guard runs again, leaving the replacement and the + /// finished export both on hand. The replacement must still fail it. `select!` picks among + /// ready branches at random unless biased, so try enough times to catch either order. + #[tokio::test] + async fn a_replacement_on_hand_at_the_end_fails_the_export() { + tokio::time::pause(); + for _ in 0..16 { + let origin = moq_tokio::origin::spawn(); + let _old = announce(&origin, &moq_net::Epoch::mint()); + let consumer = origin.consume(); + consumer.routed("room").await.unwrap(); + let replacement = Replacement::watch(&consumer, &"room".into()).unwrap(); + + let _new = announce(&origin, &moq_net::Epoch::mint()); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + + let err = replacement + .guard(std::future::ready(Ok(()))) + .await + .expect_err("a replaced export ended cleanly"); + assert!(err.to_string().contains("replaced broadcast `room`"), "{err:#}"); + } + } + /// A broadcast that goes offline ends the export as it always has, so what returns /// after it is no replacement. #[tokio::test] From 8897353a6e15dfa6485054416f8d7c1ca7c534c5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 19:54:37 -0700 Subject: [PATCH 10/14] fix(ffi,c): deliver an instance that takes over during resolution once routed_broadcast can resolve an instance that took over while it waited, with the restart announcing it still unread. The next wait took that restart and delivered the same instance again. Both follow loops now apply the changes on hand after resolving and resolve again if the serving instance moved. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/origin.rs | 72 ++++++++++++++++++++++++++++------------ rs/moq-c/src/test.rs | 40 ++++++++++++++++++++++ rs/moq-ffi/src/origin.rs | 64 +++++++++++++++++++++++------------ rs/moq-ffi/src/test.rs | 30 +++++++++++++++++ 4 files changed, 163 insertions(+), 43 deletions(-) diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index ec9c0a86131..9a408d08064 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -289,31 +289,32 @@ impl Origin { Err(err) => return Err(err.into()), }; let mut delivered = false; + let mut unseen = false; + // The origin closing ends the wait, cleanly once a handle was delivered. + let closed = |delivered: bool| match delivered { + true => Ok(()), + false => Err(Error::BroadcastNotFound), + }; // Deliver each publisher instance serving the path: the first, then each one that takes // over or returns. `biased` so a pending close always wins. loop { // Apply every change on hand before resolving, so a burst of them delivers only the // instance serving the path now. - let mut unseen = false; - loop { - let event = match follow.poll_next(&kio::Waiter::noop()) { - std::task::Poll::Ready(event) => event, - std::task::Poll::Pending if unseen => break, - std::task::Poll::Pending => tokio::select! { - biased; - _ = &mut close => return Ok(()), - event = follow.next() => event, - }, + if settle(&mut follow, &mut unseen).is_none() { + return closed(delivered); + } + if !unseen { + let event = tokio::select! { + biased; + _ = &mut close => return Ok(()), + event = follow.next() => event, + }; + let Some(event) = event else { + return closed(delivered); }; - match event { - None if delivered => return Ok(()), - None => return Err(Error::BroadcastNotFound), - Some(moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_)) => unseen = true, - Some(moq_net::announce::Event::End(_)) => unseen = false, - // The same instance over another route, which its subscriptions ride out. - Some(moq_net::announce::Event::Update(_)) => {} - } + apply(&mut unseen, event); + continue; } // `routed_broadcast` rides out the churn between something covering the path @@ -324,14 +325,20 @@ impl Origin { _ = &mut close => return Ok(()), resolved = consumer.routed_broadcast(path.as_str()) => match resolved { Ok(broadcast) => broadcast, - Err(moq_net::Error::Closed) if delivered => return Ok(()), - Err(moq_net::Error::Unauthorized | moq_net::Error::Closed) => { - return Err(Error::BroadcastNotFound); - } + Err(moq_net::Error::Closed) => return closed(delivered), + Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), Err(err) => return Err(err.into()), }, }; + // A change that landed while resolving may name the instance just resolved, or one + // after it. Resolve again rather than deliver a stale one, or the same one twice. + match settle(&mut follow, &mut unseen) { + None => return closed(delivered), + Some(true) => continue, + Some(false) => unseen = false, + } + // Hold the lock only to buffer the broadcast; release it before the callback. let broadcast_id = State::lock().consume.start(broadcast, Some(consumer.clone()))?; callback.call(broadcast_id); @@ -532,6 +539,27 @@ impl Origin { } } +/// Apply every change on hand to `unseen`, returning whether the instance serving the path +/// moved, or `None` once the origin closed. +fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool) -> Option { + let mut moved = false; + while let std::task::Poll::Ready(event) = follow.poll_next(&kio::Waiter::noop()) { + moved |= apply(unseen, event?); + } + Some(moved) +} + +/// Apply one change to `unseen`, returning whether the instance serving the path moved. +fn apply(unseen: &mut bool, event: moq_net::announce::Event) -> bool { + match event { + moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => *unseen = true, + moq_net::announce::Event::End(_) => *unseen = false, + // The same instance over another route, which its subscriptions ride out. + moq_net::announce::Event::Update(_) => return false, + } + true +} + #[cfg(test)] mod supervision_tests { use super::*; diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 121041e8073..30bb1e17670 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -3159,6 +3159,46 @@ fn consume_announced_follows_a_restart() { assert_eq!(moq_origin_close(origin), 0); } +/// Another instance taking over while the wait resolves is the one it delivers, so the restart +/// that announced it must not deliver it again. +#[test] +fn consume_announced_delivers_a_takeover_once() { + let origin = id(moq_origin_create()); + let dynamic = { + let _runtime = crate::ffi::RUNTIME.enter(); + let origin = crate::Id::try_from(origin).unwrap(); + let route = moq_net::origin::Route::default(); + State::lock().origin.get(origin).unwrap().dynamic("", route).unwrap() + }; + let cb = Callback::new(); + let path = b"live"; + let task = id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }); + + // The dynamic route covers the path, so the wait is on its answer when the exact path, a + // more specific route without an epoch, takes over. + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let takeover = publish_run(origin, "live"); + let delivered = id(cb.recv()); + assert!( + cb.rx.recv_timeout(Duration::from_millis(50)).is_err(), + "the takeover was delivered twice" + ); + + assert_eq!(moq_consume_close(delivered), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((takeover, dynamic)); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn video_publish_consume() { let origin = id(moq_origin_create()); diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index b73f16cd932..a2f795d99cd 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -214,31 +214,33 @@ struct AnnouncedBroadcast { impl AnnouncedBroadcast { async fn available(&mut self) -> Result, MoqError> { - let broadcast = self - .driver + let driver = self.driver.clone(); + let broadcast = driver .watch(async { - // Apply every change on hand before resolving, so a burst of them hands out - // only the instance serving the path now. loop { - let event = match self.follow.poll_next(&kio::Waiter::noop()) { - std::task::Poll::Ready(event) => event, - std::task::Poll::Pending if self.unseen => break, - std::task::Poll::Pending => self.follow.next().await, - }; - match event.ok_or(MoqError::Closed)? { - moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, - moq_net::announce::Event::End(_) => self.unseen = false, - // The same instance over another route, which its subscriptions ride out. - moq_net::announce::Event::Update(_) => {} + // Apply every change on hand before resolving, so a burst of them hands out + // only the instance serving the path now. + self.settle()?; + if !self.unseen { + let event = self.follow.next().await; + self.apply(event)?; + continue; + } + + // `routed_broadcast` rides out the churn between something covering the path + // and the path actually resolving (failover, an advertise-only announce + // racing its handler). The instance stays unseen until it resolves: a foreign + // cancel only drops this future, and the caller may call again. + let broadcast = self.origin.routed_broadcast(&self.path).await?; + + // A change that landed while resolving may name the instance just resolved, + // or one after it. Resolve again rather than hand out a stale one, or the + // same one twice. + if !self.settle()? { + self.unseen = false; + return Ok(broadcast); } } - // `routed_broadcast` rides out the churn between something covering the path - // and the path actually resolving (failover, an advertise-only announce - // racing its handler). The instance stays unseen until it resolves: a foreign - // cancel only drops this future, and the caller may call again. - let broadcast = self.origin.routed_broadcast(&self.path).await?; - self.unseen = false; - Ok(broadcast) }) .await?; Ok(Arc::new(MoqBroadcastConsumer::routed( @@ -247,6 +249,26 @@ impl AnnouncedBroadcast { self.driver.clone(), ))) } + + /// Apply every change on hand, returning whether the instance serving the path moved. + fn settle(&mut self) -> Result { + let mut moved = false; + while let std::task::Poll::Ready(event) = self.follow.poll_next(&kio::Waiter::noop()) { + moved |= self.apply(event)?; + } + Ok(moved) + } + + /// Apply one change, returning whether the instance serving the path moved. + fn apply(&mut self, event: Option) -> Result { + match event.ok_or(MoqError::Closed)? { + moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, + moq_net::announce::Event::End(_) => self.unseen = false, + // The same instance over another route, which its subscriptions ride out. + moq_net::announce::Event::Update(_) => return Ok(false), + } + Ok(true) + } } /// A route over a prefix, carried by a [`MoqAnnounceEvent`]. diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 6740e30ee96..86bcce1518d 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4412,6 +4412,36 @@ async fn announced_broadcast_retries_a_dropped_wait() { served.close().unwrap(); } +/// Another instance taking over while `available()` resolves is the one it hands out, so the +/// restart that announced it must not hand it out again. +#[tokio::test] +async fn announced_broadcast_hands_out_a_takeover_once() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, ""); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + // The dynamic route covers the path, so the wait is on its answer when the exact path, + // a more specific route without an epoch, takes over. + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + let takeover = create_announced(&origin, "live"); + tokio::time::timeout(TIMEOUT, first) + .await + .expect("the takeover never resolved") + .unwrap() + .unwrap(); + + let next = tokio::spawn(async move { followed.available().await }); + tokio::time::sleep(Duration::from_millis(50)).await; + assert!(!next.is_finished(), "the takeover was handed out twice"); + next.abort(); + takeover.close().unwrap(); +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits From 8cbda81b14ac61b974bf1a2233b96233405223d0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 20:08:44 -0700 Subject: [PATCH 11/14] fix(ffi,c): recognize a held takeover by the instance it resolves A takeover at the same prefix resolves at once while the origin holds its restart for the update hold, so settling the follower after resolving missed it and the held restart delivered the same instance again. Compare the resolved broadcast with the one delivered last instead, which covers a restart landing during resolution too, and drop the post-resolution settle. Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/origin.rs | 41 ++++++++++++++++++++-------------------- rs/moq-c/src/test.rs | 29 +++++++++++++++++++++------- rs/moq-ffi/src/origin.rs | 34 ++++++++++++++++++--------------- rs/moq-ffi/src/test.rs | 25 +++++++++++++++++------- 4 files changed, 79 insertions(+), 50 deletions(-) diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index 9a408d08064..f98cb488666 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -288,8 +288,9 @@ impl Origin { Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), Err(err) => return Err(err.into()), }; - let mut delivered = false; let mut unseen = false; + // The instance delivered last. + let mut last: Option = None; // The origin closing ends the wait, cleanly once a handle was delivered. let closed = |delivered: bool| match delivered { true => Ok(()), @@ -302,7 +303,7 @@ impl Origin { // Apply every change on hand before resolving, so a burst of them delivers only the // instance serving the path now. if settle(&mut follow, &mut unseen).is_none() { - return closed(delivered); + return closed(last.is_some()); } if !unseen { let event = tokio::select! { @@ -311,7 +312,7 @@ impl Origin { event = follow.next() => event, }; let Some(event) = event else { - return closed(delivered); + return closed(last.is_some()); }; apply(&mut unseen, event); continue; @@ -325,24 +326,25 @@ impl Origin { _ = &mut close => return Ok(()), resolved = consumer.routed_broadcast(path.as_str()) => match resolved { Ok(broadcast) => broadcast, - Err(moq_net::Error::Closed) => return closed(delivered), + Err(moq_net::Error::Closed) => return closed(last.is_some()), Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), Err(err) => return Err(err.into()), }, }; - // A change that landed while resolving may name the instance just resolved, or one - // after it. Resolve again rather than deliver a stale one, or the same one twice. - match settle(&mut follow, &mut unseen) { - None => return closed(delivered), - Some(true) => continue, - Some(false) => unseen = false, + unseen = false; + + // Resolving can run ahead of the follower: a takeover resolves at once while its + // restart is held, or lands while resolving. That restart then names the instance + // already delivered. + if last.as_ref().is_some_and(|last| last.is_clone(&broadcast)) { + continue; } + last = Some(broadcast.clone()); // Hold the lock only to buffer the broadcast; release it before the callback. let broadcast_id = State::lock().consume.start(broadcast, Some(consumer.clone()))?; callback.call(broadcast_id); - delivered = true; } } @@ -539,25 +541,22 @@ impl Origin { } } -/// Apply every change on hand to `unseen`, returning whether the instance serving the path -/// moved, or `None` once the origin closed. -fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool) -> Option { - let mut moved = false; +/// Apply every change on hand to `unseen`, or return `None` once the origin closed. +fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool) -> Option<()> { while let std::task::Poll::Ready(event) = follow.poll_next(&kio::Waiter::noop()) { - moved |= apply(unseen, event?); + apply(unseen, event?); } - Some(moved) + Some(()) } -/// Apply one change to `unseen`, returning whether the instance serving the path moved. -fn apply(unseen: &mut bool, event: moq_net::announce::Event) -> bool { +/// Apply one change to `unseen`. +fn apply(unseen: &mut bool, event: moq_net::announce::Event) { match event { moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => *unseen = true, moq_net::announce::Event::End(_) => *unseen = false, // The same instance over another route, which its subscriptions ride out. - moq_net::announce::Event::Update(_) => return false, + moq_net::announce::Event::Update(_) => {} } - true } #[cfg(test)] diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 30bb1e17670..6e2162d945e 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -3160,15 +3160,21 @@ fn consume_announced_follows_a_restart() { } /// Another instance taking over while the wait resolves is the one it delivers, so the restart -/// that announced it must not deliver it again. -#[test] -fn consume_announced_delivers_a_takeover_once() { +/// that announced it must not deliver it again. `dynamic_at` is where the route the wait first +/// resolves through covers the path: above it, the takeover's restart lands while resolving; +/// at it, the origin holds that restart for [`moq_net::origin::DEFAULT_UPDATE_HOLD`]. +fn consume_announced_delivers_a_takeover_once(dynamic_at: &str) { let origin = id(moq_origin_create()); let dynamic = { let _runtime = crate::ffi::RUNTIME.enter(); let origin = crate::Id::try_from(origin).unwrap(); let route = moq_net::origin::Route::default(); - State::lock().origin.get(origin).unwrap().dynamic("", route).unwrap() + State::lock() + .origin + .get(origin) + .unwrap() + .dynamic(dynamic_at, route) + .unwrap() }; let cb = Callback::new(); let path = b"live"; @@ -3182,13 +3188,12 @@ fn consume_announced_delivers_a_takeover_once() { ) }); - // The dynamic route covers the path, so the wait is on its answer when the exact path, a - // more specific route without an epoch, takes over. + // The wait is on the dynamic route's answer when the exact path takes over without an epoch. let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); let takeover = publish_run(origin, "live"); let delivered = id(cb.recv()); assert!( - cb.rx.recv_timeout(Duration::from_millis(50)).is_err(), + cb.rx.recv_timeout(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).is_err(), "the takeover was delivered twice" ); @@ -3199,6 +3204,16 @@ fn consume_announced_delivers_a_takeover_once() { assert_eq!(moq_origin_close(origin), 0); } +#[test] +fn consume_announced_delivers_a_takeover_once_while_resolving() { + consume_announced_delivers_a_takeover_once(""); +} + +#[test] +fn consume_announced_delivers_a_held_takeover_once() { + consume_announced_delivers_a_takeover_once("live"); +} + #[test] fn video_publish_consume() { let origin = id(moq_origin_create()); diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index a2f795d99cd..cfd4fc0dbe4 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -209,6 +209,8 @@ struct AnnouncedBroadcast { follow: moq_net::announce::Follow, /// An instance serves the path that was not handed out yet. unseen: bool, + /// The instance handed out last. + last: Option, driver: Background, } @@ -232,14 +234,16 @@ impl AnnouncedBroadcast { // racing its handler). The instance stays unseen until it resolves: a foreign // cancel only drops this future, and the caller may call again. let broadcast = self.origin.routed_broadcast(&self.path).await?; + self.unseen = false; - // A change that landed while resolving may name the instance just resolved, - // or one after it. Resolve again rather than hand out a stale one, or the - // same one twice. - if !self.settle()? { - self.unseen = false; - return Ok(broadcast); + // Resolving can run ahead of the follower: a takeover resolves at once while + // its restart is held, or lands while resolving. That restart then names the + // instance already handed out. + if self.last.as_ref().is_some_and(|last| last.is_clone(&broadcast)) { + continue; } + self.last = Some(broadcast.clone()); + return Ok(broadcast); } }) .await?; @@ -250,24 +254,23 @@ impl AnnouncedBroadcast { ))) } - /// Apply every change on hand, returning whether the instance serving the path moved. - fn settle(&mut self) -> Result { - let mut moved = false; + /// Apply every change on hand. + fn settle(&mut self) -> Result<(), MoqError> { while let std::task::Poll::Ready(event) = self.follow.poll_next(&kio::Waiter::noop()) { - moved |= self.apply(event)?; + self.apply(event)?; } - Ok(moved) + Ok(()) } - /// Apply one change, returning whether the instance serving the path moved. - fn apply(&mut self, event: Option) -> Result { + /// Apply one change to the path. + fn apply(&mut self, event: Option) -> Result<(), MoqError> { match event.ok_or(MoqError::Closed)? { moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, moq_net::announce::Event::End(_) => self.unseen = false, // The same instance over another route, which its subscriptions ride out. - moq_net::announce::Event::Update(_) => return Ok(false), + moq_net::announce::Event::Update(_) => {} } - Ok(true) + Ok(()) } } @@ -518,6 +521,7 @@ impl MoqOriginConsumer { path, follow, unseen: false, + last: None, driver: self.driver.clone(), }), })) diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 86bcce1518d..936a15cb7de 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4413,17 +4413,18 @@ async fn announced_broadcast_retries_a_dropped_wait() { } /// Another instance taking over while `available()` resolves is the one it hands out, so the -/// restart that announced it must not hand it out again. -#[tokio::test] -async fn announced_broadcast_hands_out_a_takeover_once() { +/// restart that announced it must not hand it out again. `dynamic_at` is where the route the +/// wait first resolves through covers the path: above it, the takeover's restart lands while +/// resolving; at it, the origin holds that restart for +/// [`moq_net::origin::DEFAULT_UPDATE_HOLD`]. +async fn announced_broadcast_hands_out_a_takeover_once(dynamic_at: &str) { let origin = MoqOriginProducer::new(MoqOriginConfig::default()); - let dynamic = serve(&origin, ""); + let dynamic = serve(&origin, dynamic_at); let followed = origin.consume().announced_broadcast("live".into()).unwrap(); let waiting = followed.clone(); let first = tokio::spawn(async move { waiting.available().await }); - // The dynamic route covers the path, so the wait is on its answer when the exact path, - // a more specific route without an epoch, takes over. + // The wait is on the dynamic route's answer when the exact path takes over without an epoch. let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) .await .expect("the wait never asked for the broadcast") @@ -4436,12 +4437,22 @@ async fn announced_broadcast_hands_out_a_takeover_once() { .unwrap(); let next = tokio::spawn(async move { followed.available().await }); - tokio::time::sleep(Duration::from_millis(50)).await; + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; assert!(!next.is_finished(), "the takeover was handed out twice"); next.abort(); takeover.close().unwrap(); } +#[tokio::test] +async fn announced_broadcast_hands_out_a_takeover_once_while_resolving() { + announced_broadcast_hands_out_a_takeover_once("").await; +} + +#[tokio::test] +async fn announced_broadcast_hands_out_a_held_takeover_once() { + announced_broadcast_hands_out_a_takeover_once("live").await; +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits From 504c4835ae0bc321e78db2ab07da90b1d17a08e2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:18:44 -0700 Subject: [PATCH 12/14] fix(cli): never fail a non-TS export on a route change alone A route change is never fatal: the export stays on the old instance while its publisher serves it, and once that route dies its tracks end with an error, which already fails the export. The replacement guard only turned a publisher's clean finish after a replacement into exit 1, so delete it, along with its doc and upgrade notes. The follow-up quests now state the same rule. Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 3 +- doc/setup/upgrade.md | 4 - quest/m1/obs-restart-follow.md | 4 +- quest/m1/sink-restart-follow.md | 13 +-- rs/moq-cli/src/subscribe.rs | 178 +------------------------------- 5 files changed, 11 insertions(+), 191 deletions(-) diff --git a/doc/bin/cli.md b/doc/bin/cli.md index eada00ff6e1..6a54f98ae64 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -405,8 +405,7 @@ Without `--stitch` the export stays on the old publisher while it is up. Only lite-07 sessions carry epochs, so on the default version every return is a replacement. An export that fails while the broadcast stays up, such as on a codec TS cannot carry, exits 1 without lingering. Only `ts` can mark a break, -so the other formats refuse `--linger`, and exit 1 at a replacement once the old -publisher ends. +so the other formats refuse `--linger`. ## Debugging diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 6a53be9922b..86b09986825 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -473,10 +473,6 @@ These land with the next breaking release, not the 2026-09-23 train. each handle with `moq_consume_close` once its successor arrives. A binding's `AnnouncedBroadcast.available()` works the same way, and its first call is unchanged. -- **`moq export` fMP4, MKV, FLV, H.264, and H.265 exit 1 at a replacement.** - Where the export used to end cleanly once a replaced publisher's session - closed, it now fails. Restart it, or use `export ts --stitch` to follow the - replacement. ## Wire diff --git a/quest/m1/obs-restart-follow.md b/quest/m1/obs-restart-follow.md index 3015d7df061..fe98a8f1123 100644 --- a/quest/m1/obs-restart-follow.md +++ b/quest/m1/obs-restart-follow.md @@ -11,8 +11,8 @@ the new instance takes to deliver a keyframe. ## Plan `cpp/obs/src/moq-source.cpp` calls `available()` once (`moq_source_on_connect`) -and treats a clean catalog end as final, so a replaced instance leaves the -source frozen once the old session closes. Since moq-ffi's `available()` now +and stops when its catalog ends, so a replaced instance leaves the source +stopped once the old route dies and its tracks end with an error. Since moq-ffi's `available()` now resolves each instance in turn, re-arm it after every instance and rebuild the `Connection`'s consumer, catalog and tracks on each one, reusing the decoders when the codec config is unchanged. diff --git a/quest/m1/sink-restart-follow.md b/quest/m1/sink-restart-follow.md index 8d7dac2d45c..06746a63e75 100644 --- a/quest/m1/sink-restart-follow.md +++ b/quest/m1/sink-restart-follow.md @@ -4,14 +4,16 @@ Every moq-cli sink that resolves a broadcast and keeps serving it handles an epochless `Restart` on purpose: it follows the new publisher instance where -its output can mark the break, and otherwise fails loud rather than ending as -if the broadcast had finished. +its output can mark the break. A route change alone is never fatal: a sink +that cannot follow stays on the old instance while its publisher serves it, +and fails once that route dies, since its subscriptions then end with an +error. ## Plan The stdout exports already do this: TS follows with `--stitch`, and the other -formats exit 1 once the replaced instance ends (`Replacement` in -`rs/moq-cli/src/subscribe.rs`). Audit and fix the rest the same way: +formats ride the old instance until its route dies and its tracks error. Audit +and fix the rest the same way: - `export hls`: `moq-hls`'s server resolves per request; check that a restart moves new requests to the new instance and how a playlist marks the break @@ -26,8 +28,7 @@ formats exit 1 once the replaced instance ends (`Replacement` in - `transcode`: its source broadcast; following means restarting the ladder on the new instance. -Share one helper (the stdout exports' `Replacement`, or a `moq_mux` -`Source::follow`-style API per +Share one helper (a `moq_mux` `Source::follow`-style API per [catalog references](/quest/m0/broadcast-epoch/catalog-references.md)) rather than one rule per sink. Each fix lands with a test that replaces the publisher while the sink runs. diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 559b54a2352..4d0fcd780ee 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -253,9 +253,6 @@ pub struct Subscribe { source: moq_mux::Source, catalog: CatalogFormat, args: SubscribeArgs, - /// Watching for a replacement since before the path resolved. TS follows the path - /// itself, through `container::ts::Follower`. - replacement: Option, } impl Subscribe { @@ -268,10 +265,6 @@ impl Subscribe { args: SubscribeArgs, ) -> anyhow::Result { let path = moq_net::Path::new(path).to_owned(); - let replacement = match args.format { - SubscribeFormat::Ts => None, - _ => Some(Replacement::watch(&origin, &path)?), - }; let broadcast = origin .request_broadcast(&path, None) .await @@ -283,7 +276,6 @@ impl Subscribe { source, catalog, args, - replacement, }) } @@ -295,14 +287,7 @@ impl Subscribe { } /// Write the broadcast to stdout until it ends. - pub async fn run(mut self) -> anyhow::Result<()> { - match self.replacement.take() { - Some(replacement) => replacement.guard(self.export()).await, - None => self.export().await, - } - } - - async fn export(self) -> anyhow::Result<()> { + pub async fn run(self) -> anyhow::Result<()> { match self.args.format { SubscribeFormat::Fmp4 => self.run_fmp4().await, SubscribeFormat::Mkv => self.run_mkv().await, @@ -507,164 +492,3 @@ impl Subscribe { Ok(()) } } - -/// Watches a path for another publisher instance taking it from the one resolved there. -/// -/// Only TS can mark a break in its output, so the other formats end at a replacement. -struct Replacement { - follow: moq_net::announce::Follow, - path: moq_net::PathOwned, -} - -impl Replacement { - /// Watch `path` from the route serving it now. Build it before resolving the path, so a - /// replacement racing the resolution is caught rather than folded into the first start. - fn watch(origin: &moq_net::origin::Consumer, path: &moq_net::PathOwned) -> anyhow::Result { - let mut follow = origin.follow(path)?; - // The cursor replays the routes on hand at once: take the start naming the one serving - // the path now, so the next restart marks another instance. - let _ = follow.poll_next(&moq_net::kio::Waiter::noop()); - Ok(Self { - follow, - path: path.clone(), - }) - } - - /// Run `export` until it ends, failing it if another instance took the path first. - /// - /// Subscriptions are sticky, so the export rides the old instance out before failing, as - /// TS does without `--stitch`. - async fn guard(mut self, export: impl Future>) -> anyhow::Result<()> { - let replaced = async { - while let Some(event) = self.follow.next().await { - if let moq_net::announce::Event::Restart(_) = event { - return; - } - } - // The origin closed, so nothing can replace the broadcast any more. - std::future::pending().await - }; - - tokio::pin!(export); - tokio::select! { - // A replacement on hand when the old instance ends still fails the export. - biased; - () = replaced => {} - end = &mut export => return end, - } - let path = &self.path; - tracing::info!(%path, "broadcast replaced, exporting the old instance until it ends"); - let replaced = - format!("another publisher instance replaced broadcast `{path}`; only `export ts --stitch` can follow it"); - match export.await { - Ok(()) => Err(anyhow::anyhow!(replaced)), - Err(err) => Err(err.context(replaced)), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - /// Announce a broadcast at `room` under `epoch`, returning what keeps it up. - fn announce(origin: &moq_net::origin::Producer, epoch: &moq_net::Epoch) -> moq_net::broadcast::Producer { - origin - .publish("room", moq_net::origin::Route::default().with_epoch(epoch.clone())) - .unwrap() - } - - /// Follow `room`, once it is announced, for an export that runs until `end` sends. - async fn export( - origin: &moq_net::origin::Producer, - ) -> ( - tokio::sync::oneshot::Sender<()>, - tokio::task::JoinHandle>, - ) { - let consumer = origin.consume(); - consumer.routed("room").await.unwrap(); - let replacement = Replacement::watch(&consumer, &"room".into()).unwrap(); - let (end, ended) = tokio::sync::oneshot::channel(); - let task = tokio::spawn(replacement.guard(async move { - ended.await.unwrap(); - Ok(()) - })); - (end, task) - } - - /// Another instance taking the path, as every return of an epochless route is, keeps the - /// export on the old one until it ends, then fails rather than reading as a finish. - #[tokio::test] - async fn a_replacement_fails_the_export_once_it_ends() { - tokio::time::pause(); - let origin = moq_tokio::origin::spawn(); - let _old = announce(&origin, &moq_net::Epoch::mint()); - let (end, task) = export(&origin).await; - - let _new = announce(&origin, &moq_net::Epoch::mint()); - tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; - assert!(!task.is_finished(), "the export left the old instance before it ended"); - - end.send(()).unwrap(); - let err = task.await.unwrap().expect_err("a replaced export ended cleanly"); - assert!(err.to_string().contains("replaced broadcast `room`"), "{err:#}"); - } - - /// The same epoch is the same content, so the export carries on. - #[tokio::test] - async fn the_same_epoch_does_not_fail_the_export() { - tokio::time::pause(); - let origin = moq_tokio::origin::spawn(); - let epoch = moq_net::Epoch::mint(); - let _old = announce(&origin, &epoch); - let (end, task) = export(&origin).await; - - let _new = announce(&origin, &epoch); - tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; - - end.send(()).unwrap(); - task.await.unwrap().unwrap(); - } - - /// The old instance can end before the guard runs again, leaving the replacement and the - /// finished export both on hand. The replacement must still fail it. `select!` picks among - /// ready branches at random unless biased, so try enough times to catch either order. - #[tokio::test] - async fn a_replacement_on_hand_at_the_end_fails_the_export() { - tokio::time::pause(); - for _ in 0..16 { - let origin = moq_tokio::origin::spawn(); - let _old = announce(&origin, &moq_net::Epoch::mint()); - let consumer = origin.consume(); - consumer.routed("room").await.unwrap(); - let replacement = Replacement::watch(&consumer, &"room".into()).unwrap(); - - let _new = announce(&origin, &moq_net::Epoch::mint()); - tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; - - let err = replacement - .guard(std::future::ready(Ok(()))) - .await - .expect_err("a replaced export ended cleanly"); - assert!(err.to_string().contains("replaced broadcast `room`"), "{err:#}"); - } - } - - /// A broadcast that goes offline ends the export as it always has, so what returns - /// after it is no replacement. - #[tokio::test] - async fn a_return_after_an_end_is_no_replacement() { - tokio::time::pause(); - let origin = moq_tokio::origin::spawn(); - let old = announce(&origin, &moq_net::Epoch::mint()); - let (end, task) = export(&origin).await; - - drop(old); - tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; - let _new = announce(&origin, &moq_net::Epoch::mint()); - tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; - - end.send(()).unwrap(); - task.await.unwrap().unwrap(); - } -} From 3377d142c0c484ec30c60cd0913d62525ca77fdd Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:18:56 -0700 Subject: [PATCH 13/14] test: a dead route errors the old instance's subscriptions Pin the rule that a route change alone is never fatal but a dead route is. After an epochless replacement, a subscription on the old instance keeps reading while its publisher serves it, and the old session closing ends it with an error, never cleanly: in moq-net directly and through a relay, inside and past the update hold; in js/net on lite and IETF, before and after the restart; and as an fMP4 export whose old publisher is interrupted right after the replacement announces. Co-Authored-By: Claude Opus 5.5 --- js/net/src/goaway-requests.test.ts | 106 ++++++++++++++++++++ rs/moq-cli/tests/integration/export.rs | 70 ++++++++++++- rs/moq-net/tests/integration/replacement.rs | 78 ++++++++++++++ 3 files changed, 249 insertions(+), 5 deletions(-) diff --git a/js/net/src/goaway-requests.test.ts b/js/net/src/goaway-requests.test.ts index f68978ef047..1e98339604e 100644 --- a/js/net/src/goaway-requests.test.ts +++ b/js/net/src/goaway-requests.test.ts @@ -283,3 +283,109 @@ test("announcing and subscribing after GOAWAY still open streams, on the adapter subscriber.close(); } }); + +/** + * A route change alone is never fatal, but a dead route is: once the replacement wins, the + * subscription already open on the draining session keeps going while the old publisher + * serves it, and that session closing ends the subscription with an error, never cleanly. + * The old session closes either before the `restart` reaches the viewer or after. + */ +async function routeDeath(kind: "lite" | "ietf", after: boolean): Promise { + const protocol = kind === "lite" ? ALPN_06 : ALPN.DRAFT_17; + const pair = createMockTransportPair(protocol); + const inject = kind === "ietf" ? captureFirstUni(pair.server) : undefined; + const goawayBytes = kind === "ietf" ? await encodeIetfGoaway() : undefined; + + const consume = new OriginProducer(); + const publish = new OriginProducer(); + const broadcast = publish.createBroadcast(Path.from("room")); + broadcast.announce(); + + const served: TrackProducer[] = []; + const serving = (async () => { + for (;;) { + const req = await wireOf(broadcast).requested(); + if (!req) break; + const track = req.accept({ timescale: Timescale.MILLI }); + track.writeFrame(textFrame("one")); + served.push(track); + } + })(); + + const [client, server] = await Promise.all([ + connect({ url, transport: pair.client, consume }), + accept({ transport: pair.server, url, publish: publish.consume() }), + ]); + + const watching = consume.announced(); + const request = consume.request(Path.from("room")); + let routes: ReturnType | undefined; + let replacement: Awaited> | undefined; + let replacementServer: Awaited> | undefined; + + try { + expect((await watching.next())?.kind).toBe("start"); + await waitUntil(() => request.active.peek() !== undefined); + const front = request.active.peek(); + if (!front) throw new Error("no route"); + const video = front.track("video").subscribe().ordered(); + expect(await video.readString()).toBe("one"); + + if (inject && goawayBytes) await inject.inject(goawayBytes); + else await sendLiteGoaway(pair.server); + await wireOf(client).goaway; + + routes = consume.announced(); + expect((await routes.next())?.kind).toBe("start"); + const pair2 = createMockTransportPair(protocol); + [replacement, replacementServer] = await Promise.all([ + connect({ url, transport: pair2.client, consume }), + accept({ transport: pair2.server, url, publish: publish.consume() }), + ]); + if (after) { + // Skip the drain cost reaching the route, if it lands after the start. + let event = await routes.next(); + while (event?.kind === "update") event = await routes.next(); + expect(event).toMatchObject({ prefix: Path.from("room"), kind: "restart" }); + } + + // The old publisher still serves the old instance, so its subscription carries on. + for (const producer of served) { + if (producer.closed.peek() === undefined) producer.writeFrame(textFrame("two")); + } + expect(await video.readString()).toBe("two"); + + // A draining close of a session still serving a live track gives up after its drain. + await server.close().catch(() => {}); + await expect( + (async () => { + for (;;) { + if ((await video.readString()) === undefined) return "ended cleanly"; + } + })(), + ).rejects.toThrow(); + video.close(); + } finally { + watching.close(); + routes?.close(); + request.close(); + replacement?.abort(); + replacementServer?.abort(); + client.abort(); + server.abort(); + broadcast.close(); + await serving; + publish.close(); + consume.close(); + } +} + +for (const kind of ["lite", "ietf"] as const) { + test(`${kind}: the draining session closing before the restart errors its subscription`, async () => { + await routeDeath(kind, false); + }); + + test(`${kind}: the draining session closing after the restart errors its subscription`, async () => { + await routeDeath(kind, true); + }); +} diff --git a/rs/moq-cli/tests/integration/export.rs b/rs/moq-cli/tests/integration/export.rs index 3465e5af161..3d958d68eb1 100644 --- a/rs/moq-cli/tests/integration/export.rs +++ b/rs/moq-cli/tests/integration/export.rs @@ -1,6 +1,7 @@ //! `moq export ts --linger` and `--stitch` over a real relay: the export rides out the same //! publisher instance leaving and coming back, switches to a replacement only when asked, and -//! exits with the verdict of the broadcast's last end. +//! exits with the verdict of the broadcast's last end. The other formats fail once the route +//! they read dies. #![cfg(unix)] use std::io::Cursor; @@ -246,15 +247,19 @@ fn import_as(relay: &Relay, global: &[&str], format: Format) -> (Child, JoinHand .stdin(Stdio::piped()) .spawn() .expect("spawn import"); - let mut stdin = child.stdin.take().expect("stdin"); - let feeding = tokio::spawn(async move { + let feeding = feed(child.stdin.take().expect("stdin"), format); + (child, feeding) +} + +/// Feed `format`'s clip into `stdin` at about real time, handing it back once it is all written. +fn feed(mut stdin: ChildStdin, format: Format) -> JoinHandle { + tokio::spawn(async move { for chunk in format.clip.chunks(format.chunk) { stdin.write_all(chunk).await.expect("write stdin"); tokio::time::sleep(Duration::from_millis(150)).await; } stdin - }); - (child, feeding) + }) } /// End the publisher the way an operator does, which drops the broadcast unfinished. @@ -709,3 +714,58 @@ async fn an_old_publisher_that_stays_up_keeps_the_export() { drop(new_stdin); wait(&mut new).await; } + +/// [`import_as`] that returns once the publisher has sent its announcement to the relay, which +/// it does once its input declares a track. +async fn import_announced(relay: &Relay, format: Format) -> (Child, JoinHandle) { + let mut child = moq(relay, &[]) + .args(["import", "--max-age", "1s", format.name]) + .env("RUST_LOG", "warn,moq_net::lite::publisher=debug") + .stdin(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn import"); + let feeding = feed(child.stdin.take().expect("stdin"), format); + + let mut stderr = BufReader::new(child.stderr.take().expect("stderr")).lines(); + let announced = async { + while let Some(line) = stderr.next_line().await.expect("read import stderr") { + let other = ["unannounce", "reannounce", "no longer", "cancelled", "withholding"]; + if line.contains("announce") && line.contains("demo") && !other.iter().any(|o| line.contains(o)) { + return; + } + } + panic!("the import exited before announcing"); + }; + tokio::time::timeout(TIMEOUT, announced) + .await + .expect("the import never announced"); + // Keep draining, so a full pipe never blocks the import. + tokio::spawn(async move { while let Ok(Some(_)) = stderr.next_line().await {} }); + (child, feeding) +} + +/// A route change alone never ends a non-TS export, which cannot mark a break, but the old +/// route dying does: its tracks end with an error rather than cleanly. The old publisher is +/// interrupted as soon as the replacement has announced, so the relay is usually still +/// holding the replacement's restart. +#[tokio::test] +async fn a_replaced_publisher_dying_fails_an_fmp4_export() { + let relay = relay().await; + let mut export = export_as(&relay, &[], FMP4, &[]); + let (old, feeding) = import_as(&relay, &[], FMP4); + export.output_past(0).await; + let old_stdin = feeding.await.unwrap(); + + let (mut new, replacing) = import_announced(&relay, FMP4).await; + interrupt(old, old_stdin).await; + + let (status, errors) = export.wait().await; + assert!( + !status.success(), + "the dead route ended the export cleanly; exporter stderr: {errors}" + ); + + drop(replacing.await.unwrap()); + wait(&mut new).await; +} diff --git a/rs/moq-net/tests/integration/replacement.rs b/rs/moq-net/tests/integration/replacement.rs index fa906a190cd..147c663aea3 100644 --- a/rs/moq-net/tests/integration/replacement.rs +++ b/rs/moq-net/tests/integration/replacement.rs @@ -170,3 +170,81 @@ async fn epoch_lite07() { .await .expect("timed out"); } + +/// A route change alone is never fatal, but a dead route is: once an epochless replacement +/// is announced, a subscription on the old instance keeps going while the old publisher +/// serves it, and its session closing ends that subscription with an error, never cleanly. +/// The viewer reads either straight from the publisher or through a relay, and the old +/// session closes after `wait`: inside the update hold, before the `Restart` is delivered, +/// or past it. +async fn route_death(version: &str, relayed: bool, wait: Duration) { + let version: Version = version.parse().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let viewer = produce_origin(3); + + let cam = publisher.publish("cam", origin::Route::default()).unwrap(); + let video = cam.create_track("video", timed()).unwrap(); + + let upstream = if relayed { &relay } else { &viewer }; + let old_session = link(version, &publisher, upstream).await; + let _downstream = match relayed { + true => Some(link(version, &relay, &viewer).await), + false => None, + }; + let mut announced = viewer.consume().announced(); + next_event(&mut announced).await; + + let old = viewer.consume().request_broadcast("cam", None).await.unwrap(); + let mut sticky = subscribe(&old, "video").await.unwrap(); + write(&video, "video:0"); + assert_eq!(read(&mut sticky).await, "video:0", "{version}"); + + let _replacement = link(version, &publisher, upstream).await; + moq_clock::sleep(wait).await; + write(&video, "video:1"); + assert_eq!( + read(&mut sticky).await, + "video:1", + "{version}: the replacement cut the old instance off while it served" + ); + + let _ = old_session.server.close().await; + let end = moq_clock::timeout(Duration::from_secs(10), async { + loop { + match sticky.recv_group().await { + Ok(Some(_)) => continue, + Ok(None) => return None, + Err(err) => return Some(err), + } + } + }) + .await + .expect("the dead route's subscription hung"); + assert!( + end.is_some(), + "{version} (relayed: {relayed}): the dead route's subscription ended cleanly" + ); +} + +#[moq_clock::test] +async fn route_death_within_hold_errors() { + for version in ["moq-lite-06", "moq-lite-07-wip"] { + for relayed in [false, true] { + moq_clock::timeout(TEST_TIMEOUT, route_death(version, relayed, WITHIN_HOLD)) + .await + .expect("timed out"); + } + } +} + +#[moq_clock::test] +async fn route_death_past_hold_errors() { + for version in ["moq-lite-06", "moq-lite-07-wip"] { + for relayed in [false, true] { + moq_clock::timeout(TEST_TIMEOUT, route_death(version, relayed, PAST_HOLD)) + .await + .expect("timed out"); + } + } +} From 0f396d707cf91f9f8a5679e8c09da4716cbd4ab9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 09:38:39 -0700 Subject: [PATCH 14/14] fix(ffi,c): hand an instance out again after an end or once freed The follower remembered the instance it handed out last as a full handle, so it kept that instance's front from retiring after the caller let go, and after an end it skipped the same instance coming back: a route above the path with the same epoch carries the front on, while the cursor reports the switch as an end and a start, a gap that ended every request on the old route. Remember the handle only while the caller holds it (a Weak in moq-ffi, the handle id in moq-c), and forget it at an end. Forgetting at an end alone would deliver an instance twice when an end and its start land while resolving, so when changes arrive while resolving, the path is resolved afresh once caught up. The available() doc now says the old instance's tracks end with an error unless this side closed its session. Co-Authored-By: Claude Opus 5.5 --- dart/moq_ffi/lib/src/moq.dart | 2 +- rs/moq-c/src/consume.rs | 7 ++ rs/moq-c/src/origin.rs | 66 ++++++++++------- rs/moq-c/src/test.rs | 129 ++++++++++++++++++++++++++++++++-- rs/moq-ffi/src/origin.rs | 64 ++++++++++------- rs/moq-ffi/src/test.rs | 110 +++++++++++++++++++++++++++-- 6 files changed, 318 insertions(+), 60 deletions(-) diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index cd2f4cb7d12..16541b8b35b 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -17762,7 +17762,7 @@ void _checkApiChecksums() { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_available() != - 36710) { + 47724) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } if (uniffi_moq_ffi_checksum_method_moqannouncedbroadcast_cancel() != 63175) { diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 48e2c6acbf4..733ace6302c 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -176,6 +176,13 @@ impl Consume { self.broadcast.insert(ConsumeBroadcast { broadcast, origin }) } + /// Whether the caller still holds `id` as a handle to `broadcast`. + pub fn holds(&self, id: Id, broadcast: &moq_net::broadcast::Consumer) -> bool { + self.broadcast + .get(id) + .is_some_and(|held| held.broadcast.is_clone(broadcast)) + } + pub fn catalog(&mut self, broadcast: Id, on_catalog: OnStatus) -> Result { let ConsumeBroadcast { broadcast, origin } = self.broadcast.get(broadcast).ok_or(Error::BroadcastNotFound)?.clone(); diff --git a/rs/moq-c/src/origin.rs b/rs/moq-c/src/origin.rs index f98cb488666..51c796f5b41 100644 --- a/rs/moq-c/src/origin.rs +++ b/rs/moq-c/src/origin.rs @@ -289,8 +289,10 @@ impl Origin { Err(err) => return Err(err.into()), }; let mut unseen = false; - // The instance delivered last. - let mut last: Option = None; + let mut delivered = false; + // The handle delivered last, until the caller frees it or the path goes offline. An id + // rather than the broadcast, so it never keeps an instance the caller freed from retiring. + let mut last: Option = None; // The origin closing ends the wait, cleanly once a handle was delivered. let closed = |delivered: bool| match delivered { true => Ok(()), @@ -302,8 +304,8 @@ impl Origin { loop { // Apply every change on hand before resolving, so a burst of them delivers only the // instance serving the path now. - if settle(&mut follow, &mut unseen).is_none() { - return closed(last.is_some()); + if settle(&mut follow, &mut unseen, &mut last).is_none() { + return closed(delivered); } if !unseen { let event = tokio::select! { @@ -312,9 +314,9 @@ impl Origin { event = follow.next() => event, }; let Some(event) = event else { - return closed(last.is_some()); + return closed(delivered); }; - apply(&mut unseen, event); + apply(&mut unseen, &mut last, event); continue; } @@ -326,24 +328,32 @@ impl Origin { _ = &mut close => return Ok(()), resolved = consumer.routed_broadcast(path.as_str()) => match resolved { Ok(broadcast) => broadcast, - Err(moq_net::Error::Closed) => return closed(last.is_some()), + Err(moq_net::Error::Closed) => return closed(delivered), Err(moq_net::Error::Unauthorized) => return Err(Error::BroadcastNotFound), Err(err) => return Err(err.into()), }, }; - unseen = false; - - // Resolving can run ahead of the follower: a takeover resolves at once while its - // restart is held, or lands while resolving. That restart then names the instance - // already delivered. - if last.as_ref().is_some_and(|last| last.is_clone(&broadcast)) { - continue; + // A change that came through while resolving may predate the answer, such as an end + // and a start for the instance it resolved. Resolve afresh once caught up. + match settle(&mut follow, &mut unseen, &mut last) { + None => return closed(delivered), + Some(true) => continue, + Some(false) => unseen = false, } - last = Some(broadcast.clone()); // Hold the lock only to buffer the broadcast; release it before the callback. - let broadcast_id = State::lock().consume.start(broadcast, Some(consumer.clone()))?; + let broadcast_id = { + let mut state = State::lock(); + // Resolving can run ahead of the follower: a takeover resolves at once while its + // restart is held, and that restart then names the instance already delivered. + if last.is_some_and(|last| state.consume.holds(last, &broadcast)) { + continue; + } + state.consume.start(broadcast, Some(consumer.clone()))? + }; + last = Some(broadcast_id); + delivered = true; callback.call(broadcast_id); } } @@ -541,22 +551,30 @@ impl Origin { } } -/// Apply every change on hand to `unseen`, or return `None` once the origin closed. -fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool) -> Option<()> { +/// Apply every change on hand, returning whether any came through, or `None` once the origin +/// closed. +fn settle(follow: &mut moq_net::announce::Follow, unseen: &mut bool, last: &mut Option) -> Option { + let mut changed = false; while let std::task::Poll::Ready(event) = follow.poll_next(&kio::Waiter::noop()) { - apply(unseen, event?); + changed |= apply(unseen, last, event?); } - Some(()) + Some(changed) } -/// Apply one change to `unseen`. -fn apply(unseen: &mut bool, event: moq_net::announce::Event) { +/// Apply one change to the path, returning whether it is one. +fn apply(unseen: &mut bool, last: &mut Option, event: moq_net::announce::Event) -> bool { match event { moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => *unseen = true, - moq_net::announce::Event::End(_) => *unseen = false, + // An end is a gap that ended every request on the old route, so whatever serves the path + // next is delivered, even the same instance coming back. + moq_net::announce::Event::End(_) => { + *unseen = false; + *last = None; + } // The same instance over another route, which its subscriptions ride out. - moq_net::announce::Event::Update(_) => {} + moq_net::announce::Event::Update(_) => return false, } + true } #[cfg(test)] diff --git a/rs/moq-c/src/test.rs b/rs/moq-c/src/test.rs index 6e2162d945e..dbf2e2048bc 100644 --- a/rs/moq-c/src/test.rs +++ b/rs/moq-c/src/test.rs @@ -3120,15 +3120,22 @@ fn consume_announced_close_cancels() { assert_eq!(moq_origin_close(origin), 0); } -/// Publish `path` on `origin` under a fresh epoch, as each run of a publisher does. The C ABI -/// announces without one, so this reaches past it. -fn publish_run(origin: u32, path: &str) -> moq_net::broadcast::Producer { +/// Publish `path` on `origin` through `route`, which the C ABI cannot set. +fn publish(origin: u32, path: &str, route: moq_net::origin::Route) -> moq_net::broadcast::Producer { let _runtime = crate::ffi::RUNTIME.enter(); - let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); let origin = crate::Id::try_from(origin).unwrap(); State::lock().origin.get(origin).unwrap().publish(path, route).unwrap() } +/// Publish `path` on `origin` under a fresh epoch, as each run of a publisher does. +fn publish_run(origin: u32, path: &str) -> moq_net::broadcast::Producer { + publish( + origin, + path, + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) +} + /// Another publisher instance taking the path delivers a fresh handle on the same wait. #[test] fn consume_announced_follows_a_restart() { @@ -3190,7 +3197,7 @@ fn consume_announced_delivers_a_takeover_once(dynamic_at: &str) { // The wait is on the dynamic route's answer when the exact path takes over without an epoch. let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); - let takeover = publish_run(origin, "live"); + let takeover = publish(origin, "live", moq_net::origin::Route::default()); let delivered = id(cb.recv()); assert!( cb.rx.recv_timeout(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).is_err(), @@ -3214,6 +3221,118 @@ fn consume_announced_delivers_a_held_takeover_once() { consume_announced_delivers_a_takeover_once("live"); } +/// Serve `prefix` on `origin` dynamically through `route`. +fn serve_dynamic(origin: u32, prefix: &str, route: moq_net::origin::Route) -> moq_net::origin::Dynamic { + let _runtime = crate::ffi::RUNTIME.enter(); + let origin = crate::Id::try_from(origin).unwrap(); + State::lock() + .origin + .get(origin) + .unwrap() + .dynamic(prefix, route) + .unwrap() +} + +/// Wait for the broadcast at `path` on `origin`, delivering each instance through `cb`. +fn consume_announced(origin: u32, path: &[u8], cb: &Callback) -> u32 { + id(unsafe { + moq_origin_announced_broadcast( + origin, + path.as_ptr() as *const c_char, + path.len(), + Some(channel_callback), + cb.ptr, + ) + }) +} + +/// An end is a gap that ended every request on the old route, even when a route above the path +/// with the same epoch carries the instance on. So the instance serving the path afterwards is +/// delivered again, though the caller still holds its handle. +#[test] +fn consume_announced_delivers_the_instance_again_after_an_end() { + let origin = id(moq_origin_create()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let run = publish(origin, "live", route.clone()); + let first = id(cb.recv()); + + let dynamic = serve_dynamic(origin, "", route); + { + let _runtime = crate::ffi::RUNTIME.enter(); + run.unannounce(); + } + let request = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let served = { + let _runtime = crate::ffi::RUNTIME.enter(); + let served = moq_net::broadcast::Info::new().produce(); + request.accept(served.consume()); + served + }; + let again = id(cb.recv()); + + assert_eq!(moq_consume_close(first), 0); + assert_eq!(moq_consume_close(again), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((run, dynamic, served)); + assert_eq!(moq_origin_close(origin), 0); +} + +/// A handle the caller freed no longer counts as delivered, so the restart that names its +/// instance delivers it again rather than leaving the caller with nothing. +#[test] +fn consume_announced_delivers_a_freed_takeover_again() { + let origin = id(moq_origin_create()); + let dynamic = serve_dynamic(origin, "live", moq_net::origin::Route::default()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + // Without an epoch, the origin holds the takeover's restart past its resolution. + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + let takeover = publish(origin, "live", moq_net::origin::Route::default()); + assert_eq!(moq_consume_close(id(cb.recv())), 0); + let again = id(cb.recv()); + + assert_eq!(moq_consume_close(again), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop((takeover, dynamic)); + assert_eq!(moq_origin_close(origin), 0); +} + +/// The route a wait resolves through leaving, and another instance arriving before it resolves, +/// reach the follower as an end and a start after the wait delivers the newcomer. That start +/// must not deliver it again. +#[test] +fn consume_announced_delivers_a_gap_while_resolving_once() { + let origin = id(moq_origin_create()); + let dynamic = serve_dynamic(origin, "", moq_net::origin::Route::default()); + let cb = Callback::new(); + let task = consume_announced(origin, b"live", &cb); + + let _unanswered = crate::ffi::RUNTIME.block_on(dynamic.requested_broadcast()).unwrap(); + { + let _runtime = crate::ffi::RUNTIME.enter(); + drop(dynamic); + } + std::thread::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2); + let arrival = publish(origin, "live", moq_net::origin::Route::default()); + let delivered = id(cb.recv()); + assert!( + cb.rx.recv_timeout(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).is_err(), + "the arrival was delivered twice" + ); + + assert_eq!(moq_consume_close(delivered), 0); + assert_eq!(moq_origin_announced_broadcast_cancel(task), 0); + assert_eq!(cb.recv_terminal(), 0, "cancel delivers terminal 0"); + drop(arrival); + assert_eq!(moq_origin_close(origin), 0); +} + #[test] fn video_publish_consume() { let origin = id(moq_origin_create()); diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index cfd4fc0dbe4..79ca3a595a5 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc, Weak}; use crate::consumer::MoqBroadcastConsumer; use crate::error::MoqError; @@ -209,15 +209,16 @@ struct AnnouncedBroadcast { follow: moq_net::announce::Follow, /// An instance serves the path that was not handed out yet. unseen: bool, - /// The instance handed out last. - last: Option, + /// The handle handed out last, until the caller drops it or the path goes offline. Weak, so + /// it never keeps an instance the caller let go of from retiring. + last: Weak, driver: Background, } impl AnnouncedBroadcast { async fn available(&mut self) -> Result, MoqError> { let driver = self.driver.clone(); - let broadcast = driver + driver .watch(async { loop { // Apply every change on hand before resolving, so a burst of them hands out @@ -234,43 +235,54 @@ impl AnnouncedBroadcast { // racing its handler). The instance stays unseen until it resolves: a foreign // cancel only drops this future, and the caller may call again. let broadcast = self.origin.routed_broadcast(&self.path).await?; + + // A change that came through while resolving may predate the answer, such as + // an end and a start for the instance it resolved. Resolve afresh once caught up. + if self.settle()? { + continue; + } self.unseen = false; // Resolving can run ahead of the follower: a takeover resolves at once while - // its restart is held, or lands while resolving. That restart then names the - // instance already handed out. - if self.last.as_ref().is_some_and(|last| last.is_clone(&broadcast)) { + // its restart is held, and that restart then names the instance handed out. + if self.last.upgrade().is_some_and(|last| last.inner.is_clone(&broadcast)) { continue; } - self.last = Some(broadcast.clone()); + let broadcast = Arc::new(MoqBroadcastConsumer::routed( + broadcast, + self.origin.clone(), + self.driver.clone(), + )); + self.last = Arc::downgrade(&broadcast); return Ok(broadcast); } }) - .await?; - Ok(Arc::new(MoqBroadcastConsumer::routed( - broadcast, - self.origin.clone(), - self.driver.clone(), - ))) + .await } - /// Apply every change on hand. - fn settle(&mut self) -> Result<(), MoqError> { + /// Apply every change on hand, returning whether any came through. + fn settle(&mut self) -> Result { + let mut changed = false; while let std::task::Poll::Ready(event) = self.follow.poll_next(&kio::Waiter::noop()) { - self.apply(event)?; + changed |= self.apply(event)?; } - Ok(()) + Ok(changed) } - /// Apply one change to the path. - fn apply(&mut self, event: Option) -> Result<(), MoqError> { + /// Apply one change to the path, returning whether it is one. + fn apply(&mut self, event: Option) -> Result { match event.ok_or(MoqError::Closed)? { moq_net::announce::Event::Start(_) | moq_net::announce::Event::Restart(_) => self.unseen = true, - moq_net::announce::Event::End(_) => self.unseen = false, + // An end is a gap that ended every request on the old route, so whatever serves the + // path next is handed out, even the same instance coming back. + moq_net::announce::Event::End(_) => { + self.unseen = false; + self.last = Weak::new(); + } // The same instance over another route, which its subscriptions ride out. - moq_net::announce::Event::Update(_) => {} + moq_net::announce::Event::Update(_) => return Ok(false), } - Ok(()) + Ok(true) } } @@ -521,7 +533,7 @@ impl MoqOriginConsumer { path, follow, unseen: false, - last: None, + last: Weak::new(), driver: self.driver.clone(), }), })) @@ -659,8 +671,8 @@ impl MoqAnnouncedBroadcast { /// The first call resolves the instance serving the path, once one is announced. Each later /// call waits for another to take over, as a restarted publisher or a replacement session /// does, or for the broadcast to return after it went offline. Drop everything read from the - /// previous instance then and read the new one: the old one's tracks end when its session - /// does, and look finished. + /// previous instance then and read the new one: the old one's tracks end when its route + /// dies, with an error unless this side closed that session. pub async fn available(&self) -> Result, MoqError> { self.task.run(|mut state| async move { state.available().await }).await } diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index f3671ce87f5..26fc2aa40d2 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -4653,9 +4653,9 @@ async fn announced_broadcast_retries_a_dropped_wait() { } /// Another instance taking over while `available()` resolves is the one it hands out, so the -/// restart that announced it must not hand it out again. `dynamic_at` is where the route the -/// wait first resolves through covers the path: above it, the takeover's restart lands while -/// resolving; at it, the origin holds that restart for +/// restart that announced it must not hand it out again while the caller holds it. `dynamic_at` +/// is where the route the wait first resolves through covers the path: above it, the takeover's +/// restart lands while resolving; at it, the origin holds that restart for /// [`moq_net::origin::DEFAULT_UPDATE_HOLD`]. async fn announced_broadcast_hands_out_a_takeover_once(dynamic_at: &str) { let origin = MoqOriginProducer::new(MoqOriginConfig::default()); @@ -4670,7 +4670,7 @@ async fn announced_broadcast_hands_out_a_takeover_once(dynamic_at: &str) { .expect("the wait never asked for the broadcast") .unwrap(); let takeover = create_announced(&origin, "live"); - tokio::time::timeout(TIMEOUT, first) + let _held = tokio::time::timeout(TIMEOUT, first) .await .expect("the takeover never resolved") .unwrap() @@ -4693,6 +4693,108 @@ async fn announced_broadcast_hands_out_a_held_takeover_once() { announced_broadcast_hands_out_a_takeover_once("live").await; } +/// An end is a gap that ended every request on the old route, even when a route above the path +/// with the same epoch carries the instance on. So the instance serving the path afterwards is +/// handed out again, though it is the one the caller still holds. +#[tokio::test] +async fn announced_broadcast_hands_out_the_instance_again_after_an_end() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + let route = MoqRoute { + epoch: Some(moq_mint_epoch()), + ..Default::default() + }; + let broadcast = origin.create_broadcast("live".into()).unwrap(); + broadcast.announce(route.clone()).unwrap(); + let first = tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("timed out waiting for the broadcast") + .unwrap(); + + let dynamic = origin.dynamic("".into(), route).unwrap(); + broadcast.unannounce().unwrap(); + let again = tokio::spawn(async move { followed.available().await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the instance never asked the route above the path") + .unwrap(); + let served = MoqBroadcastProducer::new().unwrap(); + request.accept(&served).unwrap(); + + let again = tokio::time::timeout(TIMEOUT, again) + .await + .expect("the instance was never handed out again") + .unwrap() + .unwrap(); + assert!( + again.inner().is_clone(first.inner()), + "the route above did not carry the instance on" + ); + served.close().unwrap(); + broadcast.close().unwrap(); +} + +/// The route a wait resolves through leaving, and another instance arriving before it resolves, +/// reach the follower as an end and a start after the wait hands the newcomer out. That start +/// must not hand it out again. +#[tokio::test] +async fn announced_broadcast_hands_out_a_gap_while_resolving_once() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, ""); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + dynamic.cancel(); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + let arrival = create_announced(&origin, "live"); + let _held = tokio::time::timeout(TIMEOUT, first) + .await + .expect("the arrival never resolved") + .unwrap() + .unwrap(); + + let next = tokio::spawn(async move { followed.available().await }); + tokio::time::sleep(moq_net::origin::DEFAULT_UPDATE_HOLD * 2).await; + assert!(!next.is_finished(), "the arrival was handed out twice"); + next.abort(); + arrival.close().unwrap(); +} + +/// A handle the caller dropped no longer counts as handed out, so the restart that names it +/// hands it out again rather than leaving the caller with nothing. +#[tokio::test] +async fn announced_broadcast_hands_out_a_dropped_takeover_again() { + let origin = MoqOriginProducer::new(MoqOriginConfig::default()); + let dynamic = serve(&origin, "live"); + let followed = origin.consume().announced_broadcast("live".into()).unwrap(); + + let waiting = followed.clone(); + let first = tokio::spawn(async move { waiting.available().await }); + let _unanswered = tokio::time::timeout(TIMEOUT, dynamic.requested_broadcast()) + .await + .expect("the wait never asked for the broadcast") + .unwrap(); + let takeover = create_announced(&origin, "live"); + drop( + tokio::time::timeout(TIMEOUT, first) + .await + .expect("the takeover never resolved") + .unwrap() + .unwrap(), + ); + + tokio::time::timeout(TIMEOUT, followed.available()) + .await + .expect("the dropped takeover was never handed out again") + .unwrap(); + takeover.close().unwrap(); +} + #[tokio::test] async fn server_client_roundtrip_auto_origin() { // Same shape as `server_client_roundtrip` but the client config omits