From 8cd7086a2f319487890d9ca7e95058c719298b37 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:14:52 -0700 Subject: [PATCH 1/5] quest(broadcast-epoch): claim gateways From 6ad9e274042891d8ee799dd3bfc280cbbd97f7ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 6 Oct 2026 23:22:18 -0700 Subject: [PATCH 2/5] feat(gateways): mint an epoch per ingest connection RTMP, SRT, and WHIP ingest publish each incoming connection (and each RTMP pull) under a freshly minted epoch, so an encoder that reconnects under the same key while its stale connection lingers replaces the old broadcast at once instead of being resumed into it. The epoch is logged with each accepted publish. Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/cluster.md | 4 +- doc/bin/rtc.md | 4 +- doc/bin/rtmp.md | 5 +- doc/bin/srt.md | 5 +- quest/m0/broadcast-epoch/README.md | 1 - quest/m0/broadcast-epoch/gateways.md | 22 -------- quest/m0/broadcast-epoch/ts-restart.md | 2 +- rs/moq-rtc/src/server/whip.rs | 67 ++++++++++++++++++++++-- rs/moq-rtmp/src/dial.rs | 14 +++-- rs/moq-rtmp/src/server.rs | 72 ++++++++++++++++++++++++-- rs/moq-srt/src/ts.rs | 42 ++++++++++++++- 11 files changed, 194 insertions(+), 44 deletions(-) delete mode 100644 quest/m0/broadcast-epoch/gateways.md diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index eb169faf27..3dbd7a7251 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -38,7 +38,9 @@ groups it has open, overlapping the new one. A group neither route delivers is dropped once the readers' max age has passed it. A route through the subscribing peer itself is never used. A publisher whose groups restart, such as an encoder restarting from group 0, publishes under a new epoch, which replaces the old -broadcast instead of resuming it. A route without an epoch, including every +broadcast instead of resuming it. RTMP, SRT, and WHIP ingest mint one per +connection, so an encoder reconnecting to the same path replaces its stale +connection at once. A route without an epoch, including every route on moq-lite 06 and older or moq-transport, keeps its subscriptions until it goes. Seamless failover needs both: a publisher that announces an epoch, and moq-lite 07 on every link the route crosses, since epochs travel on nothing diff --git a/doc/bin/rtc.md b/doc/bin/rtc.md index 989a4d9c3a..7a66452847 100644 --- a/doc/bin/rtc.md +++ b/doc/bin/rtc.md @@ -29,7 +29,9 @@ opens the endpoint to browsers on other origins, and `--udp-bind` plus `--public-addr` pin one media port for firewalls. The listener is plain HTTP; put a TLS-terminating proxy in front for WHIP clients that require HTTPS. A fresh WHEP peer joins at the current group, so it starts at a keyframe without waiting for the next one. `DELETE` on the -resource URL tears a session down per the RFC. +resource URL tears a session down per the RFC. Each WHIP session publishes +under a fresh [epoch](/concept/moq-lite#publisher-epochs), so a publisher that +reconnects while its stale session lingers replaces it at once. Built on [str0m](https://crates.io/crates/str0m); the library is [`moq-rtc`](https://docs.rs/moq-rtc). diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md index c40a914d9c..a1a13078c1 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -29,7 +29,10 @@ the [`moq-rtmp`](https://docs.rs/moq-rtmp) library, which hands you each publish or play request to accept, map to a path, or reject. The CLI listener is unauthenticated; firewall it. -Each push is its own broadcast. Import publishes the encoder's own timestamps +Each push or pull is its own broadcast, under a fresh +[epoch](/concept/moq-lite#publisher-epochs): an encoder that reconnects while +its stale connection is still open replaces it at once, so viewers move to the +new push instead of stalling on the old one. Import publishes the encoder's own timestamps and anchors the catalog clock on the first frame, so it names the wall time the push arrived. A group starting before the previous group's start, such as an encoder restarting its timestamps mid-push, ends that push with an error. diff --git a/doc/bin/srt.md b/doc/bin/srt.md index 12828b9c49..2f6cf5fd92 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -29,7 +29,10 @@ moq --connect https://relay.example.com/anon --broadcast event.hang import srt - ``` Import publishes the feed's own PTS and anchors the catalog clock on its first -frame, as [`import ts`](/bin/cli) does. +frame, as [`import ts`](/bin/cli) does. Each connection is its own broadcast, +under a fresh [epoch](/concept/moq-lite#publisher-epochs): an encoder that +reconnects while its stale connection is still open replaces it at once, so +viewers move to the new feed instead of stalling on the old one. A multi-program feed is refused, as with `import ts`, unless `--program` picks one: `--program 2` imports program 2 alone, and `--program all` diff --git a/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index e8d4b4cd75..e76bbbee36 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -69,7 +69,6 @@ timeout, and killing the newest epoch falls back to a still-live older one. ## Required - [Apps](/quest/m0/broadcast-epoch/apps.md) - moq-cli, the browser publish and watch components, and demo/web restart into a new epoch and reset on the switch -- [Gateways](/quest/m0/broadcast-epoch/gateways.md) - RTMP, SRT, and WHIP ingest mint an epoch per incoming connection, so an encoder reconnect is a clean takeover - [TS restart](/quest/m0/broadcast-epoch/ts-restart.md) - a signalled backward TS discontinuity finishes the broadcast and continues the same input under a fresh epoch - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one - [GStreamer](/quest/m0/broadcast-epoch/gst.md) - moqsink publishes each run under a fresh epoch diff --git a/quest/m0/broadcast-epoch/gateways.md b/quest/m0/broadcast-epoch/gateways.md deleted file mode 100644 index fef1bca1a7..0000000000 --- a/quest/m0/broadcast-epoch/gateways.md +++ /dev/null @@ -1,22 +0,0 @@ -# [S] Ingest gateways mint an epoch per connection - -## Goal - -RTMP, SRT, and WHIP ingest publish each incoming connection under a fresh -epoch. An encoder that reconnects to the same stream key becomes a clean -takeover at once. Without it, the reconnect is resumed into the stale -connection's broadcast and stalls viewers until its group sequence catches -up. - -## Plan - -Mint an epoch per incoming connection and announce it on the broadcast's -route (`moq-rtmp`, `moq-srt`, `moq-rtc`, and the relay wiring); nothing mints -by default. Test a reconnect under the -same key with the stale connection still open. Update `doc/bin/relay/` where it -describes ingest paths. - - -## Related - -- [TS restart](/quest/m0/broadcast-epoch/ts-restart.md) - a signalled restart inside one connection, which this quest's per-connection epoch does not cover diff --git a/quest/m0/broadcast-epoch/ts-restart.md b/quest/m0/broadcast-epoch/ts-restart.md index 75561176ec..1633ab9f0e 100644 --- a/quest/m0/broadcast-epoch/ts-restart.md +++ b/quest/m0/broadcast-epoch/ts-restart.md @@ -8,7 +8,7 @@ When a TS feed rewinds its time base and signals it the rest of the same input as a new broadcast under a fresh epoch, in the same process and connection. This covers an encoder restart or source switch behind a gateway that keeps its connection up, and a looping playout server, none of -which makes a new connection for [Gateways](/quest/m0/broadcast-epoch/gateways.md) +which makes a new connection for the ingest gateways to turn into an epoch. An unsignalled rewind stays fatal, as #4543 decided. ## Plan diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index bf08db2b1b..def3148677 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -98,8 +98,9 @@ async fn delete(State(state): State, Path(path): Path) -> S /// server's) lets the embedder publish through a *scoped* origin, so the publish /// scope is enforced by moq-net exactly as for a native session; the bundled /// [`router`] passes the server's own (unauthenticated) producer. It parses the -/// offer, registers the broadcast (so a fast subscriber doesn't 404 in the gap -/// before the first RTP packet), registers a media session on the shared mux, +/// offer, registers the broadcast under a fresh epoch (so a fast subscriber +/// doesn't 404 in the gap before the first RTP packet, and a reconnecting +/// encoder replaces a stale session's broadcast at once), registers a media session on the shared mux, /// and returns the SDP answer plus an opaque `resource_id` for the WHIP /// `Location` header. The caller must run the returned [`Response`] to drive the /// RTP->MoQ session. @@ -120,9 +121,10 @@ pub async fn accept( // Create the broadcast on the publish origin before negotiating, so a // fast subscriber doesn't see a 404 in the gap between the SDP answer - // and the first RTP packet. - let producer = publisher.create_broadcast(&broadcast)?; - producer.announce(moq_net::origin::Route::default())?; + // and the first RTP packet. Each offer is its own publisher instance, so a + // reconnect replaces a stale session's broadcast instead of resuming into it. + let epoch = moq_net::Epoch::mint(); + let producer = publisher.publish(&broadcast, moq_net::origin::Route::default().with_epoch(epoch.clone()))?; let handle = producer.clone(); let config = moq_mux::catalog::Config::default() @@ -150,6 +152,7 @@ pub async fn accept( // Register before returning so a DELETE that races the first packet still // finds the session; Response::run unregisters itself when it ends. let cancel = server.register_session(resource_id.clone()); + tracing::info!(%broadcast, %epoch, "whip publish accepted"); Ok(Response { resource_id: resource_id.clone(), @@ -185,3 +188,57 @@ fn status_for(err: &Error) -> StatusCode { _ => StatusCode::INTERNAL_SERVER_ERROR, } } + +#[cfg(test)] +mod tests { + use str0m::{ + format::Codec, + media::{Direction, MediaKind}, + }; + + use super::*; + + /// An encoder's offer to send one Opus track. + fn offer() -> String { + let mut rtc = session::rtc_with_codecs(&[Codec::Opus]); + let mut api = rtc.sdp_api(); + api.add_media(MediaKind::Audio, Direction::SendOnly, None, None, None); + let (offer, _pending) = api.apply().expect("an offer"); + offer.to_sdp_string() + } + + /// An encoder reconnecting to the same broadcast while its stale session is still + /// open replaces the stale broadcast at once: each offer is its own epoch, so + /// viewers re-request instead of stalling on the old one. + #[tokio::test] + async fn a_reconnect_replaces_the_stale_session() { + let server = Server::new(crate::server::Config::default()); + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let _stale_session = accept(&server, &origin, "live/cam0", &offer()).await.unwrap(); + let stale = consumer.request_broadcast("live/cam0").await.unwrap(); + let mut catalog = stale + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let _fresh_session = accept(&server, &origin, "live/cam0", &offer()).await.unwrap(); + // The session binds a real UDP socket, so this runs on the wall clock. + let ended = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + Ok(None) => panic!("the stale broadcast ended cleanly"), + Err(err) => return err, + } + } + }) + .await + .expect("the stale viewer stalled"); + assert!(matches!(ended, moq_net::Error::Unroutable), "{ended:?}"); + let fresh = consumer.request_broadcast("live/cam0").await.unwrap(); + assert!(!fresh.is_clone(&stale), "viewers reach the reconnected session"); + } +} diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 1a0572b5ad..5fa04d029d 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -290,15 +290,15 @@ impl Client { self.work.push_back(request); self.await_event(Direction::Play).await?; - // The stream key is the ingest credential (`rtmp://host//`), so the - // app and broadcast path stand in for it. - tracing::info!(app = %self.app, %path, "rtmp play accepted by remote"); - let config = moq_mux::catalog::Config::default() .with_max_age(self.import_max_age) .with_bandwidth(self.import_bandwidth.clone()); let mut publisher = Publisher::new(origin, path.as_str(), config)?; + // The stream key is the ingest credential (`rtmp://host//`), so the + // app and broadcast path stand in for it. + tracing::info!(app = %self.app, %path, epoch = %publisher.epoch, "rtmp play accepted by remote"); + let result = self.pull_media(&mut publisher).await; match &result { // Clean end: flush so the final groups close cleanly before unannouncing. @@ -491,12 +491,15 @@ struct Publisher { // A clone of the importer's producer, so an end can close the broadcast // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, + // This pull's publisher instance, so pulling again replaces it rather than resuming into it. + epoch: moq_net::Epoch, } impl Publisher { fn new(origin: &origin::Producer, path: &str, config: moq_mux::catalog::Config) -> anyhow::Result { + let epoch = moq_net::Epoch::mint(); let mut broadcast = origin - .publish(path, moq_net::origin::Route::default()) + .publish(path, moq_net::origin::Route::default().with_epoch(epoch.clone())) .map_err(|err| anyhow::anyhow!("broadcast '{path}' could not be published: {err}"))?; let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; let handle = broadcast.clone(); @@ -506,6 +509,7 @@ impl Publisher { importer.decode(&flv::file_header())?; Ok(Self { importer, + epoch, broadcast: handle, }) } diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 12b003db56..950c1f7fe7 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -646,7 +646,7 @@ impl Publish { .map_err(|e| anyhow::anyhow!("rtmp accept publish: {e:?}"))?; self.work.extend(results); - tracing::info!(peer = %self.peer, %path, "rtmp publish accepted"); + tracing::info!(peer = %self.peer, %path, epoch = %publisher.epoch, "rtmp publish accepted"); let result = pump( &mut self.stream, @@ -1387,13 +1387,17 @@ struct Publisher { // A clone of the importer's producer, so an end can close the broadcast // (prompt unannounce) even though the importer owns it. broadcast: moq_net::broadcast::Producer, + // This connection's publisher instance, so an encoder reconnecting under the + // same key replaces this broadcast at once instead of resuming into it. + epoch: moq_net::Epoch, } impl Publisher { - /// Open a broadcast at `path` and prime the importer with the FLV file - /// header, so subsequent tags decode against an initialized demuxer. + /// Open a broadcast at `path` under a fresh epoch and prime the importer with + /// the FLV file header, so subsequent tags decode against an initialized demuxer. fn new(origin: &origin::Producer, path: &str, config: moq_mux::catalog::Config) -> anyhow::Result { - let mut broadcast = origin.publish(path, moq_net::origin::Route::default())?; + let epoch = moq_net::Epoch::mint(); + let mut broadcast = origin.publish(path, moq_net::origin::Route::default().with_epoch(epoch.clone()))?; let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; let handle = broadcast.clone(); let mut importer = FlvImport::new(broadcast, catalog.reserve()); @@ -1404,6 +1408,7 @@ impl Publisher { Ok(Self { importer, broadcast: handle, + epoch, }) } @@ -2119,6 +2124,65 @@ mod tests { client.abort(); } + /// An encoder reconnecting under the same key while its stale connection is still + /// open replaces the stale broadcast at once: each connection is its own epoch, so + /// viewers re-request instead of stalling on the old one. + #[tokio::test] + async fn a_reconnect_replaces_the_stale_publish() { + let mut server = Server::bind("127.0.0.1:0".parse().unwrap()).await.unwrap(); + let addr = server.local_addr().unwrap(); + let origin = moq_tokio::origin::spawn(); + let consumer = origin.consume(); + let server_task = tokio::spawn(async move { + while let Some(request) = server.accept().await { + let Request::Publish(publish) = request else { + panic!("expected a publish request"); + }; + let origin = origin.clone(); + tokio::spawn(async move { publish.accept(&origin, "live/cam0").await }); + } + }); + + let stale_client = tokio::spawn(async move { + run_client(TcpStream::connect(addr).await.unwrap(), ClientMode::Publish).await; + }); + tokio::time::timeout(Duration::from_secs(5), consumer.routed("live/cam0")) + .await + .expect("stale publish timed out") + .unwrap(); + let stale = consumer.request_broadcast("live/cam0").await.unwrap(); + let mut catalog = stale + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let fresh_client = tokio::spawn(async move { + run_client(TcpStream::connect(addr).await.unwrap(), ClientMode::Publish).await; + }); + let ended = tokio::time::timeout(Duration::from_secs(5), async { + loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + Ok(None) => panic!("the stale broadcast ended cleanly"), + Err(err) => return err, + } + } + }) + .await + .expect("the stale viewer stalled"); + assert!(matches!(ended, moq_net::Error::Unroutable), "{ended:?}"); + assert!(!stale_client.is_finished(), "the stale connection is still open"); + + let fresh = consumer.request_broadcast("live/cam0").await.unwrap(); + assert!(!fresh.is_clone(&stale), "viewers reach the reconnected publish"); + + stale_client.abort(); + fresh_client.abort(); + server_task.abort(); + } + /// The connect `_result` should advertise the enhanced-RTMP codecs we ingest /// plus our `capsEx` (multitrack), so an enhanced encoder knows what it may /// send. Drives a raw client through the handshake, sends `connect` (itself diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index 8935c3501c..e8f879bcbe 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -72,10 +72,15 @@ impl Publisher { config: moq_mux::catalog::Config, program: Option, ) -> Result { + // Each connection is its own publisher instance, so an encoder reconnecting + // under the same stream id replaces the stale broadcast instead of resuming into + // it. `ts::Programs` mints one per program likewise. + let mut epoch = None; let importer = match program { Some(Program::All) => Importer::All(Box::new(ts::Programs::new(origin.clone(), path, config))), Some(Program::One(_)) | None => { - let mut broadcast = origin.publish(path, moq_net::origin::Route::default())?; + let route = moq_net::origin::Route::default().with_epoch(epoch.insert(moq_net::Epoch::mint()).clone()); + let mut broadcast = origin.publish(path, route)?; let config = config.with_catalog(moq_mux::catalog::hang::Catalog::::default()); let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config)?; let mut import = ts::Import::new(broadcast.clone(), catalog.reserve()); @@ -88,7 +93,7 @@ impl Publisher { } } }; - tracing::info!(%path, ?program, "publishing ingest broadcast"); + tracing::info!(%path, ?program, epoch = epoch.as_ref().map(tracing::field::display), "publishing ingest broadcast"); Ok(Self { importer, @@ -448,6 +453,39 @@ mod tests { assert_eq!(refused.map(|refused| refused.programs.as_slice()), Some(&[1, 2][..])); } + /// A caller reconnecting under the same stream id while its stale connection is + /// still open replaces the stale broadcast at once: each connection is its own + /// epoch, so viewers re-request instead of stalling on the old one. + #[tokio::test(start_paused = true)] + async fn a_reconnect_replaces_the_stale_connection() { + let origin = produce_origin(); + let consumer = origin.consume(); + let _stale = Publisher::new(&origin, "ingest", Default::default(), None).unwrap(); + let stale = consumer.request_broadcast("ingest").await.unwrap(); + let mut catalog = stale + .track(hang::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(None) + .await + .unwrap(); + + let _fresh = Publisher::new(&origin, "ingest", Default::default(), None).unwrap(); + let ended = tokio::time::timeout(Duration::from_secs(1), async { + loop { + match catalog.recv_group().await { + Ok(Some(_)) => continue, + Ok(None) => panic!("the stale broadcast ended cleanly"), + Err(err) => return err, + } + } + }) + .await + .expect("the stale viewer stalled"); + assert!(matches!(ended, moq_net::Error::Unroutable), "{ended:?}"); + let fresh = consumer.request_broadcast("ingest").await.unwrap(); + assert!(!fresh.is_clone(&stale), "viewers reach the reconnected caller"); + } + /// `Program::One` publishes the chosen program alone on the ingest's path. #[tokio::test(start_paused = true)] async fn publisher_imports_one_program() { From 469e67b3c32e6e7058abb4555fecf1e4c5fc32f4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 09:12:52 -0700 Subject: [PATCH 3/5] fix(rtc): publish the WHIP epoch only after the offer is accepted A failed offer no longer replaces a live session's broadcast. Docs now say a stale subscription ends with Unroutable and the next subscribe reaches the new connection, and SRT mints an epoch per broadcast for --program all. Co-Authored-By: Claude Opus 5.5 --- doc/bin/rtmp.md | 5 +++-- doc/bin/srt.md | 9 +++++---- rs/moq-rtc/src/server/whip.rs | 34 ++++++++++++++++++---------------- 3 files changed, 26 insertions(+), 22 deletions(-) diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md index a1a13078c1..2742ab690a 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -31,8 +31,9 @@ is unauthenticated; firewall it. Each push or pull is its own broadcast, under a fresh [epoch](/concept/moq-lite#publisher-epochs): an encoder that reconnects while -its stale connection is still open replaces it at once, so viewers move to the -new push instead of stalling on the old one. Import publishes the encoder's own timestamps +its stale connection is still open replaces it at once. Subscriptions to the +stale push end with `Unroutable` instead of stalling, and a viewer's next +subscribe reaches the new push. Import publishes the encoder's own timestamps and anchors the catalog clock on the first frame, so it names the wall time the push arrived. A group starting before the previous group's start, such as an encoder restarting its timestamps mid-push, ends that push with an error. diff --git a/doc/bin/srt.md b/doc/bin/srt.md index 2f6cf5fd92..2d2605fcb4 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -29,10 +29,11 @@ moq --connect https://relay.example.com/anon --broadcast event.hang import srt - ``` Import publishes the feed's own PTS and anchors the catalog clock on its first -frame, as [`import ts`](/bin/cli) does. Each connection is its own broadcast, -under a fresh [epoch](/concept/moq-lite#publisher-epochs): an encoder that -reconnects while its stale connection is still open replaces it at once, so -viewers move to the new feed instead of stalling on the old one. +frame, as [`import ts`](/bin/cli) does. Each connection publishes under a fresh +[epoch](/concept/moq-lite#publisher-epochs) per broadcast: an encoder that +reconnects while its stale connection is still open replaces it at once. +Subscriptions to the stale feed end with `Unroutable` instead of stalling, and +a viewer's next subscribe reaches the new feed. A multi-program feed is refused, as with `import ts`, unless `--program` picks one: `--program 2` imports program 2 alone, and `--program all` diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index def3148677..c0355d5bbf 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -98,10 +98,10 @@ async fn delete(State(state): State, Path(path): Path) -> S /// server's) lets the embedder publish through a *scoped* origin, so the publish /// scope is enforced by moq-net exactly as for a native session; the bundled /// [`router`] passes the server's own (unauthenticated) producer. It parses the -/// offer, registers the broadcast under a fresh epoch (so a fast subscriber +/// offer, registers a media session on the shared mux, publishes the +/// broadcast under a fresh epoch once negotiation succeeds (so a fast subscriber /// doesn't 404 in the gap before the first RTP packet, and a reconnecting -/// encoder replaces a stale session's broadcast at once), registers a media session on the shared mux, -/// and returns the SDP answer plus an opaque `resource_id` for the WHIP +/// encoder replaces a stale session's broadcast at once), and returns the SDP answer plus an opaque `resource_id` for the WHIP /// `Location` header. The caller must run the returned [`Response`] to drive the /// RTP->MoQ session. /// @@ -119,19 +119,6 @@ pub async fn accept( let offer = sdp::parse_offer(offer)?; let broadcast = broadcast.as_path(); - // Create the broadcast on the publish origin before negotiating, so a - // fast subscriber doesn't see a 404 in the gap between the SDP answer - // and the first RTP packet. Each offer is its own publisher instance, so a - // reconnect replaces a stale session's broadcast instead of resuming into it. - let epoch = moq_net::Epoch::mint(); - let producer = publisher.publish(&broadcast, moq_net::origin::Route::default().with_epoch(epoch.clone()))?; - - let handle = producer.clone(); - let config = moq_mux::catalog::Config::default() - .with_max_age(server.config().max_age) - .with_bandwidth(server.config().bandwidth.clone()); - let sink = Box::new(IngestSink::new(producer, config)?); - // Register a session on the shared media mux: known ICE credentials (so the // demux routes this peer's STUN by ufrag), an inbox to read datagrams from, // and a guard the session task holds for its lifetime (unregisters on exit). @@ -146,6 +133,21 @@ pub async fn accept( } let answer = rtc.sdp_api().accept_offer(offer).map_err(Error::rtc)?; + + // Publish only once negotiation succeeded, so a failed offer leaves a live + // session's broadcast alone, but before returning the answer, so a fast + // subscriber doesn't see a 404 in the gap before the first RTP packet. Each + // offer is its own publisher instance, so a reconnect replaces a stale + // session's broadcast instead of resuming into it. + let epoch = moq_net::Epoch::mint(); + let producer = publisher.publish(&broadcast, moq_net::origin::Route::default().with_epoch(epoch.clone()))?; + + let handle = producer.clone(); + let config = moq_mux::catalog::Config::default() + .with_max_age(server.config().max_age) + .with_bandwidth(server.config().bandwidth.clone()); + let sink = Box::new(IngestSink::new(producer, config)?); + let resource_id = sdp::new_resource_id(); let session = session::Session::ingest(rtc, mux.socket(), mux.candidates().to_vec(), inbound, sink); From d4aa5072589dc4fa7a0674bee82ceda8d71d1293 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 10:28:28 -0700 Subject: [PATCH 4/5] fix(rtc): scope-check a WHIP offer before any mux work The broadcast is created hidden up front, so an out-of-scope offer is refused before any mux or SDP work, and announced under its epoch only after the offer is accepted. Co-Authored-By: Claude Opus 5.5 --- rs/moq-rtc/src/server/whip.rs | 21 ++++++++++++--------- 1 file changed, 12 insertions(+), 9 deletions(-) diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index c0355d5bbf..32c2c919ee 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -98,7 +98,7 @@ async fn delete(State(state): State, Path(path): Path) -> S /// server's) lets the embedder publish through a *scoped* origin, so the publish /// scope is enforced by moq-net exactly as for a native session; the bundled /// [`router`] passes the server's own (unauthenticated) producer. It parses the -/// offer, registers a media session on the shared mux, publishes the +/// offer, registers a media session on the shared mux, announces the /// broadcast under a fresh epoch once negotiation succeeds (so a fast subscriber /// doesn't 404 in the gap before the first RTP packet, and a reconnecting /// encoder replaces a stale session's broadcast at once), and returns the SDP answer plus an opaque `resource_id` for the WHIP @@ -119,6 +119,15 @@ pub async fn accept( let offer = sdp::parse_offer(offer)?; let broadcast = broadcast.as_path(); + // Create the broadcast hidden up front, so an out-of-scope offer is refused before + // any mux work. It announces only once negotiation succeeds (below). + let producer = publisher.create_broadcast(&broadcast)?; + let handle = producer.clone(); + let config = moq_mux::catalog::Config::default() + .with_max_age(server.config().max_age) + .with_bandwidth(server.config().bandwidth.clone()); + let sink = Box::new(IngestSink::new(producer, config)?); + // Register a session on the shared media mux: known ICE credentials (so the // demux routes this peer's STUN by ufrag), an inbox to read datagrams from, // and a guard the session task holds for its lifetime (unregisters on exit). @@ -134,19 +143,13 @@ pub async fn accept( let answer = rtc.sdp_api().accept_offer(offer).map_err(Error::rtc)?; - // Publish only once negotiation succeeded, so a failed offer leaves a live + // Announce only once negotiation succeeded, so a failed offer leaves a live // session's broadcast alone, but before returning the answer, so a fast // subscriber doesn't see a 404 in the gap before the first RTP packet. Each // offer is its own publisher instance, so a reconnect replaces a stale // session's broadcast instead of resuming into it. let epoch = moq_net::Epoch::mint(); - let producer = publisher.publish(&broadcast, moq_net::origin::Route::default().with_epoch(epoch.clone()))?; - - let handle = producer.clone(); - let config = moq_mux::catalog::Config::default() - .with_max_age(server.config().max_age) - .with_bandwidth(server.config().bandwidth.clone()); - let sink = Box::new(IngestSink::new(producer, config)?); + handle.announce(moq_net::origin::Route::default().with_epoch(epoch.clone()))?; let resource_id = sdp::new_resource_id(); let session = session::Session::ingest(rtc, mux.socket(), mux.candidates().to_vec(), inbound, sink); From 1655a282483940ef43dfa822548f578c33852e1d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 10:29:41 -0700 Subject: [PATCH 5/5] docs(relay): note a cross-gateway reconnect trusts the gateways' clocks Co-Authored-By: Claude Opus 5.5 --- doc/bin/relay/cluster.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index 3dbd7a7251..429a2d8ccb 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -40,7 +40,8 @@ itself is never used. A publisher whose groups restart, such as an encoder restarting from group 0, publishes under a new epoch, which replaces the old broadcast instead of resuming it. RTMP, SRT, and WHIP ingest mint one per connection, so an encoder reconnecting to the same path replaces its stale -connection at once. A route without an epoch, including every +connection at once. Epochs order by their creation time, so a reconnect to a +different gateway assumes the two gateways' clocks roughly agree. A route without an epoch, including every route on moq-lite 06 and older or moq-transport, keeps its subscriptions until it goes. Seamless failover needs both: a publisher that announces an epoch, and moq-lite 07 on every link the route crosses, since epochs travel on nothing