diff --git a/doc/bin/relay/cluster.md b/doc/bin/relay/cluster.md index 5aba2c0ff5..f81469a506 100644 --- a/doc/bin/relay/cluster.md +++ b/doc/bin/relay/cluster.md @@ -38,7 +38,10 @@ groups it has open, overlapping the new one. A group neither route delivers is dropped once the readers' max delay 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. 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 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..2742ab690a 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -29,7 +29,11 @@ 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. 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 12828b9c49..2d2605fcb4 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -29,7 +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. +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/quest/m0/broadcast-epoch/README.md b/quest/m0/broadcast-epoch/README.md index 8f4ce2b0d0..4523e47390 100644 --- a/quest/m0/broadcast-epoch/README.md +++ b/quest/m0/broadcast-epoch/README.md @@ -72,7 +72,6 @@ timeout, and killing the newest epoch falls back to a still-live older one. - [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 - [Un-epoched takeover](/quest/m0/broadcast-epoch/unepoched-takeover.md) - a better route without an epoch takes over with a hard switch and an announcement, so a restarted lite-06 publisher reaches viewers -- [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 - [Remove `--hop`](/quest/m0/broadcast-epoch/hop-removal.md) - `moq` takes an optional `--epoch` instead of `--hop`, a plain publisher declares a random Hop ID, and the per-session hop stamp is gone 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..32c2c919ee 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -98,9 +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 (so a fast subscriber doesn't 404 in the gap -/// before the first RTP packet), registers a media session on the shared mux, -/// and returns the SDP answer plus an opaque `resource_id` for the WHIP +/// 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 /// `Location` header. The caller must run the returned [`Response`] to drive the /// RTP->MoQ session. /// @@ -118,12 +119,9 @@ 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. + // 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)?; - producer.announce(moq_net::origin::Route::default())?; - let handle = producer.clone(); let config = moq_mux::catalog::Config::default() .with_max_age(server.config().max_age) @@ -144,12 +142,22 @@ pub async fn accept( } let answer = rtc.sdp_api().accept_offer(offer).map_err(Error::rtc)?; + + // 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(); + 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); // 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 +193,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 d02708fde3..7fad7741a3 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 334d0b225e..ac26a9a287 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() {