feat(moq-mux)!: an export's later requests stay on the broadcast it resolved

doc/setup/upgrade.md

@@ -40,6 +40,16 @@ These land with the next breaking release, not the 2026-09-23 train.
   the same two flags. In Rust, moq-mux's `ts::Export::resume` is
   `ts::Export::follow`, `Source::returned` is gone, and `ts::Follower` follows a
   path's announcements the way both exports do.
+- **moq-mux's `Source` is built from the resolved catalog broadcast.**
+  `Source::new(origin, path)` is `Source::new(origin, broadcast)`: resolve the
+  broadcast first (`origin.request_broadcast(path, None)`), and its path is
+  where catalog references resolve from. `Source::broadcast()` returns that
+  broadcast without awaiting. Every later request for a path, siblings
+  included, stays on the broadcast the source first held there and fails once
+  it ends, rather than reaching a replacement; `bind` included. Build a new
+  `Source` to follow a new instance. moq-hls's `Broadcaster::new` no longer
+  awaits or fails, and a rendition whose sibling broadcast ends now ends its
+  playlist instead of rebinding.
 - **fMP4 export of Annex-B H.264 and H.265 inits from the catalog.** When the
   catalog codec string and dimensions are enough, `moq export fmp4` writes an
   `avc3` or `hev1` init segment before the first keyframe and leaves SPS, PPS,

quest/m0/broadcast-epoch/README.md

@@ -125,7 +125,6 @@ End+Start. The child quests own their focused regressions.
 
 - [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - a held broadcast or track cannot open unpinned wire work against its replacement
 - [Resume duplicates](/quest/m0/broadcast-epoch/resume-duplicates.md) - a delayed group is delivered once even beyond the old 1024-entry history
-- [Source pin](/quest/m0/broadcast-epoch/source-pin.md) - `Source` is built from the resolved catalog broadcast and every later request stays on it, so an epochless replacement never splices into the old program
 - [Catalog references](/quest/m0/broadcast-epoch/catalog-references.md) - players and exporters pin siblings consistently, with optional reference epochs and documented legacy best effort
 - [Publish catalog restart](/quest/m0/broadcast-epoch/publish-catalog-restart.md) - `@moq/publish` never reuses catalog group numbers under one name and epoch after a re-announce
 - [Bindings](/quest/m0/broadcast-epoch/bindings.md) - moq-ffi and every wrapper expose the epoch and let a publisher announce one

quest/m0/broadcast-epoch/catalog-references.md

@@ -68,14 +68,10 @@ upgrade notes inline. The [identity guide](/quest/m1/identity-guide.md)
 consolidates the behavior separately. Explain both the strong epoch-aware
 guarantee and the selected legacy best-effort behavior.
 
-## Required
-
-- [Source pin](/quest/m0/broadcast-epoch/source-pin.md) - exporters share their resolved catalog handle
-
 ## Related
 
+- [Source pin](https://github.com/moq-dev/moq/pull/5252) - exporters share their resolved catalog handle
 - [JS consume identity](https://github.com/moq-dev/moq/pull/5254) - JS resolves the actual serving identity
-
 - [Retired requests](/quest/m0/broadcast-epoch/retired-requests.md) - held handles cannot emit unpinned requests after retirement
 - [Catalog rendition IDs](/quest/m1/catalog-track-id.md) - the adjacent optional wire-track field
 - [Catalog track identity](/quest/m1/catalog-tracks.md) - catalog updates cannot redefine a track's content

quest/m0/broadcast-epoch/retired-requests.md

@@ -58,7 +58,7 @@ Update existing identity and request-lifetime documentation inline.
 
 ## Related
 
-- [Source pin](/quest/m0/broadcast-epoch/source-pin.md) - the exporter holds the exact catalog broadcast
+- [Source pin](https://github.com/moq-dev/moq/pull/5252) - the exporter holds the exact catalog broadcast
 - [JS consume identity](https://github.com/moq-dev/moq/pull/5254) - fresh consumers select the serving announcement's identity
 - [JS restart keeps the request](https://github.com/moq-dev/moq/pull/5199) - an existing resolved request stays sticky
 - [IETF epochs](/quest/m1/ietf-epochs.md) - explicit wire identity removes the epochless ambiguity

rs/moq-c/src/consume.rs

@@ -62,7 +62,7 @@ async fn resolve(
 	};
 
 	let origin = origin.ok_or_else(|| Error::UnresolvableBroadcast(reference.as_str().to_string()))?;
-	let source = moq_mux::Source::new(origin, &broadcast.info().path);
+	let source = moq_mux::Source::new(origin, broadcast);
 	Ok(source.resolve(Some(&reference.borrow())).await?)
 }
 

rs/moq-cli/src/main.rs

@@ -810,14 +810,14 @@ async fn run_stdout(consumer: moq_net::origin::Consumer, name: String, args: Sub
 	let catalog = args.catalog_format(&name);
 
 	// Confirm the broadcast is reachable and wait for it to be announced; `Subscribe` then
-	// resolves it (and any sibling broadcast a rendition's `broadcast` field references,
-	// e.g. "./source") through the origin.
+	// resolves it, and any sibling broadcast a rendition's `broadcast` field references
+	// (e.g. "./source") through the same origin.
 	consumer
 		.routed(&name)
 		.await
 		.ok_or_else(|| anyhow::anyhow!("origin closed before broadcast `{name}` was announced"))?;
 
-	Subscribe::new(consumer, &name, catalog, args).run().await
+	Subscribe::new(consumer, &name, catalog, args).await?.run().await
 }
 
 /// Run every endpoint until the first finishes (stdin EOF, SIGINT, SIGTERM, or an

rs/moq-cli/src/play/media.rs

@@ -69,7 +69,6 @@ impl<O: Output> Media<O> {
 	/// update is the same instance, which its subscriptions already ride out.
 	async fn play(self) -> anyhow::Result<()> {
 		let mut follow = self.origin.follow(&self.broadcast)?;
-		let source = moq_mux::Source::new(self.origin.clone(), &self.broadcast);
 		let mut playing = None;
 
 		loop {
@@ -97,7 +96,7 @@ impl<O: Output> Media<O> {
 					self.video.lock().unwrap().clear();
 					self.drained.notify_one();
 					self.output.send(Event::Wake);
-					playing = Some(Box::pin(self.play_broadcast(source.clone())));
+					playing = Some(Box::pin(self.play_broadcast()));
 				}
 				result = async { playing.as_mut().expect("guarded").await }, if playing.is_some() => {
 					playing = None;
@@ -111,13 +110,15 @@ impl<O: Output> Media<O> {
 		}
 	}
 
-	/// Play the broadcast at the path until its catalog and every track it
+	/// Play the broadcast now at the path until its catalog and every track it
 	/// started end.
-	async fn play_broadcast(&self, source: moq_mux::Source) -> anyhow::Result<()> {
-		let broadcast = source
-			.broadcast()
+	async fn play_broadcast(&self) -> anyhow::Result<()> {
+		let broadcast = self
+			.origin
+			.request_broadcast(&self.broadcast, None)
 			.await
 			.context("failed to subscribe to the broadcast")?;
+		let source = moq_mux::Source::new(self.origin.clone(), broadcast.clone());
 		let catalog = catalog::Consumer::<()>::new(&broadcast, self.args.catalog_format(&self.broadcast))
 			.await
 			.context("failed to subscribe to the catalog")?;

rs/moq-cli/src/publish.rs

@@ -749,14 +749,21 @@ pub(crate) mod tests {
 		// `catalog`, the producers, and `import` stay alive: the exporter subscribes to
 		// the retained tracks.
 		drain(
-			Export::with_ts(moq_mux::Source::new(origin.consume(), "cli"), CatalogFormat::Hang)
+			Export::with_ts(resolved(&origin, "cli").await, CatalogFormat::Hang)
 				.await
 				.unwrap()
 				.with_delay(RECORDING_MAX_AGE),
 		)
 		.await
 	}
 
+	/// A source for the broadcast at `path`, resolved through `origin`.
+	async fn resolved(origin: &moq_net::origin::Producer, path: &str) -> moq_mux::Source {
+		let consumer = origin.consume();
+		let broadcast = consumer.request_broadcast(path, None).await.unwrap();
+		moq_mux::Source::new(consumer, broadcast)
+	}
+
 	/// The export's delay, long enough to reach the ancillary tracks: these tests publish a
 	/// whole feed before exporting it, and the media tracks start at their newest group.
 	const RECORDING_MAX_AGE: std::time::Duration = Duration::from_secs(30);
@@ -800,7 +807,7 @@ pub(crate) mod tests {
 		// Subscribe side: the same `with_ts` call `run_ts` makes, re-emitting the
 		// ancillary streams verbatim.
 		let output = drain(
-			Export::with_ts(moq_mux::Source::new(origin.consume(), "cli"), format)
+			Export::with_ts(resolved(&origin, "cli").await, format)
 				.await
 				.unwrap()
 				.with_delay(RECORDING_MAX_AGE),

rs/moq-cli/src/subscribe.rs

@@ -253,18 +253,27 @@ pub struct Subscribe {
 }
 
 impl Subscribe {
-	/// Export the broadcast at `path` on `origin` with the resolved settings; [`run`](Self::run)
-	/// drives it.
-	pub fn new(origin: moq_net::origin::Consumer, path: &str, catalog: CatalogFormat, args: SubscribeArgs) -> Self {
+	/// Resolve the broadcast at `path` on `origin` to export it with the resolved settings;
+	/// [`run`](Self::run) drives it.
+	pub async fn new(
+		origin: moq_net::origin::Consumer,
+		path: &str,
+		catalog: CatalogFormat,
+		args: SubscribeArgs,
+	) -> anyhow::Result<Self> {
 		let path = moq_net::Path::new(path).to_owned();
-		let source = moq_mux::Source::new(origin.clone(), &path);
-		Self {
+		let broadcast = origin
+			.request_broadcast(&path, None)
+			.await
+			.with_context(|| format!("failed to resolve broadcast `{path}`"))?;
+		let source = moq_mux::Source::new(origin.clone(), broadcast);
+		Ok(Self {
 			origin,
 			path,
 			source,
 			catalog,
 			args,
-		}
+		})
 	}
 
 	/// Build the catalog stream, narrowed by the rendition selection flags. The

rs/moq-ffi/src/consumer.rs

@@ -124,7 +124,7 @@ impl MoqBroadcastConsumer {
 			.clone()
 			.ok_or_else(|| MoqError::UnresolvableBroadcast(reference.as_str().to_string()))?;
 
-		let source = moq_mux::Source::new(origin, &self.inner.info().path);
+		let source = moq_mux::Source::new(origin, self.inner.clone());
 		Ok(source.resolve(Some(&reference)).await?)
 	}
 }

rs/moq-hls/src/export/archive_tests.rs

@@ -316,8 +316,7 @@ impl Replay {
 			tokio::task::yield_now().await;
 		}
 
-		let source = moq_mux::Source::new(origin.consume(), "rec");
-		let broadcaster = Broadcaster::new(source, config).await.unwrap();
+		let broadcaster = Broadcaster::new(crate::resolved(&origin, "rec").await, config);
 		tokio::time::timeout(Duration::from_secs(5), broadcaster.ready())
 			.await
 			.expect("a rendition becomes playable");