feat(gateways): mint an epoch per ingest connection

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

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).

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.

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`

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

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

rs/moq-rtc/src/server/whip.rs

@@ -98,9 +98,10 @@ async fn delete(State(state): State<RouterState>, Path(path): Path<String>) -> 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");
+	}
+}

rs/moq-rtmp/src/dial.rs

@@ -290,15 +290,15 @@ impl<S: Stream> Client<S> {
 		self.work.push_back(request);
 		self.await_event(Direction::Play).await?;
 
-		// The stream key is the ingest credential (`rtmp://host/<app>/<key>`), 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/<app>/<key>`), 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<Self> {
+		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,
 		})
 	}

rs/moq-rtmp/src/server.rs

@@ -646,7 +646,7 @@ impl<S: Stream> Publish<S> {
 			.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<Self> {
-		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