fix(mux)!: a broadcast stamps everything from one clock

doc/lib/rs/kio.md

@@ -41,6 +41,11 @@ and `Instant::from_std()` convert at a third-party API that takes
 `std::time::Instant`, through the runtime clock's origin. That is exact on a
 running clock, and keeps only durations on paused tokio or the simulator.
 
+`kio::clock::Clock::current()` captures the ambient clock, and the handle keeps
+reading that clock from any thread or runtime. Stamp with it when one timeline
+is shared by calls that may come from different contexts. A moq-mux broadcast
+does this: it stamps every track on the clock it was created under.
+
 ## Caller-driven code
 
 A state machine polled as `poll(now, waiter)` never reads a clock. It owns a

doc/lib/rs/moq-mux.md

@@ -24,9 +24,12 @@ publish the source's own timestamps, mapped to the wall time the first frame
 arrived. The catalog clock never moves once something uses it: an importer that
 starts after a data track, a capture, another importer, or a pinned clock
 shifts its timestamps onto that clock instead, landing its first frame at now.
-A `catalog::Timebase` gives renditions on one timestamp base one shared shift. A
-stream whose decoder identity changes, such as an SPS with a new profile or level
-or an audio header with a new sample rate, continues under a new track name
+The media, the hang and MSF catalog groups, and the HLS records are all stamped
+on that one clock. It keeps reading the clock the broadcast was created under,
+even when a write comes from another runtime or thread. A `catalog::Timebase`
+gives renditions on one timestamp base one shared shift. A stream whose decoder
+identity changes, such as an SPS with a new profile or level or an audio header
+with a new sample rate, continues under a new track name
 (`<name>/<n>` for the codec importers) while the old track ends; everything else,
 such as the coded size, updates the rendition in place. A
 timeline that rewinds ends the import. The MPEG-TS import ends at a clock step

quest/m1/README.md

@@ -145,7 +145,6 @@ per-frame CPU growth since #5926.
 - [moq-c at parity](/quest/m1/c/README.md) - the hand-written `moq-c` ships at the cut with every m1 binding feature moq-ffi gains
 - [Bindings see a moq-transport run end as an end](/quest/m1/ffi-ietf-run-end.md) - an IETF publisher ending its run reaches moq-ffi consumers as an end, not `Internal`
 - [Byte-stream renditions form a ladder](/quest/m1/stream-ladder.md) - moq-ffi stream producers link as `lower` and disable unfundable rungs like frame producers
-- [A broadcast stamps everything from one clock](/quest/m1/broadcast-clock.md) - catalog, MSF, and HLS record groups are stamped from the broadcast's clock, never the ambient one
 - [moq export ends when the broadcast ends](/quest/m1/export-broadcast-end.md) - the export finishes on its own, proven without a give-up timeout
 - [A relay answers a SUBSCRIBE on a copy it kept after a fetch](/quest/m1/relay-fetched-copy-subscribe.md) - the downstream SUBSCRIBE_OK waits for the upstream edge instead of hanging
 - [A job's exact route continues its claim's readers](/quest/m1/announce-takeover.md) - a job's route inherits its claim's identity, replacing the ANNOUNCE_TAKEOVER predecessor design

rs/kio/src/clock/captured.rs

@@ -0,0 +1,217 @@
+//! The ambient clock, captured so it reads the same from anywhere.
+
+use crate::clock::{Instant, Wall};
+
+/// The ambient clock as it was when [`current`](Self::current) captured it.
+///
+/// It keeps reading that clock from any thread, runtime, or clock context, so stamps taken
+/// through it share one timeline wherever they are taken. The ambient [`now`](crate::clock::now)
+/// reads whichever clock the caller happens to be in, and instants from different clocks don't
+/// compare. Clones read the same clock.
+#[derive(Clone)]
+pub struct Clock(Source);
+
+#[derive(Clone)]
+enum Source {
+	/// The host clock every unpaused runtime and plain thread shares.
+	Host,
+	/// A tokio runtime's own clock, which differs from the host's while it is paused.
+	#[cfg(feature = "runtime")]
+	Runtime(tokio::runtime::Handle),
+	/// A manual clock installed on a tokio runtime.
+	#[cfg(feature = "manual")]
+	Installed(crate::clock::Installed),
+	/// A simulator run, by id: only a thread inside that run can read it.
+	#[cfg(feature = "sim")]
+	Sim(u64),
+}
+
+impl Clock {
+	/// Capture the ambient clock, in the order [`now`](crate::clock::now) picks it.
+	///
+	/// An entered [`Deadlines`](crate::clock::Deadlines) queue is a view of the clock it is
+	/// advanced from, so the capture is that clock: the installed or simulated one, else the host's.
+	pub fn current() -> Self {
+		let entered = super::deadline::entered().is_some();
+		#[cfg(feature = "manual")]
+		if entered && let Some(clock) = super::deadline::installed() {
+			return Self(Source::Installed(clock));
+		}
+		#[cfg(feature = "sim")]
+		if let Some(id) = super::sim::run_id() {
+			return Self(Source::Sim(id));
+		}
+		if entered {
+			return Self(Source::Host);
+		}
+		#[cfg(feature = "manual")]
+		if let Some(clock) = super::installed::current() {
+			return Self(Source::Installed(clock));
+		}
+		#[cfg(feature = "runtime")]
+		if let Ok(runtime) = tokio::runtime::Handle::try_current() {
+			return Self(Source::Runtime(runtime));
+		}
+		Self(Source::Host)
+	}
+
+	/// The current instant on this clock.
+	///
+	/// # Panics
+	///
+	/// When captured inside a simulator run and read outside it.
+	pub fn now(&self) -> Instant {
+		match &self.0 {
+			Source::Host => super::runtime::host_now(),
+			#[cfg(feature = "runtime")]
+			Source::Runtime(runtime) => {
+				let _entered = runtime.enter();
+				super::runtime::now()
+			}
+			#[cfg(feature = "manual")]
+			Source::Installed(clock) => clock.now(),
+			#[cfg(feature = "sim")]
+			Source::Sim(id) => super::sim::now_in(*id),
+		}
+	}
+
+	/// The current wall time on this clock.
+	///
+	/// # Panics
+	///
+	/// When captured inside a simulator run and read outside it, or when the host's wall clock
+	/// reads before 1970.
+	pub fn wall(&self) -> Wall {
+		match &self.0 {
+			// Paused tokio does not pause wall time.
+			Source::Host => super::runtime::wall(),
+			#[cfg(feature = "runtime")]
+			Source::Runtime(_) => super::runtime::wall(),
+			#[cfg(feature = "manual")]
+			Source::Installed(clock) => clock.wall(),
+			#[cfg(feature = "sim")]
+			Source::Sim(id) => super::sim::wall_in(*id),
+		}
+	}
+}
+
+/// Two captures are equal when they read the same clock.
+impl PartialEq for Clock {
+	fn eq(&self, other: &Self) -> bool {
+		match (&self.0, &other.0) {
+			(Source::Host, Source::Host) => true,
+			#[cfg(feature = "runtime")]
+			(Source::Runtime(a), Source::Runtime(b)) => a.id() == b.id(),
+			#[cfg(feature = "manual")]
+			(Source::Installed(a), Source::Installed(b)) => a.same(b),
+			#[cfg(feature = "sim")]
+			(Source::Sim(a), Source::Sim(b)) => a == b,
+			#[allow(unreachable_patterns, reason = "only the host clock exists without features")]
+			_ => false,
+		}
+	}
+}
+
+impl Eq for Clock {}
+
+impl std::fmt::Debug for Clock {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		match &self.0 {
+			Source::Host => f.write_str("Host"),
+			#[cfg(feature = "runtime")]
+			Source::Runtime(runtime) => f.debug_tuple("Runtime").field(&runtime.id()).finish(),
+			#[cfg(feature = "manual")]
+			Source::Installed(clock) => std::fmt::Debug::fmt(clock, f),
+			#[cfg(feature = "sim")]
+			Source::Sim(id) => f.debug_tuple("Sim").field(id).finish(),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::time::Duration;
+
+	use super::Clock;
+
+	fn paused() -> tokio::runtime::Runtime {
+		tokio::runtime::Builder::new_current_thread()
+			.enable_time()
+			.start_paused(true)
+			.build()
+			.unwrap()
+	}
+
+	/// A clock captured in a paused runtime keeps reading it from another runtime and a plain
+	/// thread, where the ambient clock is a different one.
+	#[test]
+	fn a_paused_runtime_reads_the_same_everywhere() {
+		let first = paused();
+		let (clock, at) = first.block_on(async {
+			tokio::time::advance(Duration::from_secs(60)).await;
+			let clock = Clock::current();
+			(clock.clone(), clock.now())
+		});
+		assert_eq!(clock.now(), at, "the paused clock did not move");
+
+		let second = paused();
+		second.block_on(async {
+			tokio::time::advance(Duration::from_secs(3600)).await;
+			assert_eq!(clock.now(), at, "another runtime's time is not this clock's");
+			assert_ne!(Clock::current(), clock);
+		});
+		std::thread::scope(|scope| {
+			scope.spawn(|| assert_eq!(clock.now(), at));
+		});
+
+		first.block_on(async {
+			tokio::time::advance(Duration::from_secs(5)).await;
+			assert_eq!(Clock::current(), clock);
+		});
+		assert_eq!(clock.now(), at + Duration::from_secs(5));
+	}
+
+	/// An installed manual clock is read wherever its capture goes.
+	#[test]
+	fn an_installed_clock_reads_the_same_everywhere() {
+		let runtime = tokio::runtime::Builder::new_current_thread()
+			.enable_time()
+			.build()
+			.unwrap();
+		let manual = crate::clock::Manual::new().install(runtime.handle());
+		let clock = runtime.block_on(async { Clock::current() });
+		manual.advance(Duration::from_secs(7));
+		assert_eq!(clock.now(), manual.now());
+		assert_eq!(clock.wall(), manual.wall());
+		paused().block_on(async { assert_eq!(clock.now(), manual.now()) });
+	}
+
+	/// A capture inside the simulator reads simulated time, and one on a plain thread reads the
+	/// host's even from inside the simulator.
+	#[test]
+	fn the_simulator_and_the_host_stay_apart() {
+		let host = Clock::current();
+		crate::clock::sim::run(async move {
+			let sim = Clock::current();
+			crate::clock::sleep(Duration::from_secs(10)).await;
+			assert_eq!(sim.now(), crate::clock::Instant::ZERO + Duration::from_secs(10));
+			assert_eq!(sim.wall(), crate::clock::Manual::WALL_START + Duration::from_secs(10));
+			assert!(host.now() > crate::clock::Instant::ZERO + Duration::from_secs(1 << 31));
+			assert_ne!(sim, host);
+		});
+	}
+
+	/// A queue entered on a thread is read through the clock it is advanced from.
+	#[test]
+	fn an_entered_queue_captures_its_clock() {
+		crate::clock::sim::run(async {
+			let deadlines = crate::clock::Deadlines::new(crate::clock::now());
+			crate::clock::sim::attach(&deadlines);
+			let clock = {
+				let _entered = deadlines.enter();
+				Clock::current()
+			};
+			assert_eq!(clock, Clock::current());
+		});
+	}
+}

rs/kio/src/clock/installed.rs

@@ -120,6 +120,11 @@ impl Installed {
 		self.0.lock().unwrap().clock.step_wall(wall);
 	}
 
+	/// Whether `other` is a clone of this clock.
+	pub(crate) fn same(&self, other: &Self) -> bool {
+		Arc::ptr_eq(&self.0, &other.0)
+	}
+
 	pub(crate) fn anchor(&self) -> (Instant, Wall) {
 		let mut state = self.0.lock().unwrap();
 		if let Some(anchor) = state.anchor {

rs/kio/src/clock/mod.rs

@@ -13,6 +13,8 @@
 //!   simulator inside `sim::run`, and a [`Deadlines`] queue on a thread that
 //!   [entered](Deadlines::enter) it, which is how a runtime that reads the host
 //!   clock itself (an io_uring worker) becomes a backend.
+//! - [`Clock::current`] captures the ambient clock, so stamps taken through it
+//!   keep one timeline whichever thread or runtime takes them.
 //!
 //! The `runtime` feature arms timers on tokio's clock. Without it, reads fall back
 //! to std, and waiting on the host clock panics.
@@ -32,6 +34,7 @@
 //! The `serde` feature adds `serde::UnixSeconds`, matching existing integer
 //! Unix-second JSON encodings for wall times.
 
+mod captured;
 mod deadline;
 #[cfg(feature = "manual")]
 mod installed;
@@ -46,6 +49,7 @@ pub mod sim;
 mod timer;
 mod wall;
 
+pub use captured::Clock;
 pub use deadline::{Deadline, Deadlines, Entered};
 #[cfg(feature = "manual")]
 pub use installed::Installed;

rs/kio/src/clock/runtime.rs

@@ -73,6 +73,18 @@ fn real_now() -> Instant {
 	Instant::from_origin(LEAD + std::time::Instant::now().saturating_duration_since(origin.into_std()))
 }
 
+/// The host clock every unpaused runtime and plain thread shares, whichever runtime reads it.
+#[cfg(feature = "runtime")]
+pub(crate) fn host_now() -> Instant {
+	real_now()
+}
+
+/// Without the runtime there are no paused runtimes, so the clock is the host's.
+#[cfg(not(feature = "runtime"))]
+pub(crate) fn host_now() -> Instant {
+	now()
+}
+
 /// The std instant the runtime clock counts from, as [`read`] picks it.
 #[cfg(feature = "runtime")]
 fn std_base() -> std::time::Instant {

rs/kio/src/clock/sim.rs

@@ -233,6 +233,28 @@ fn seed_from_env() -> u64 {
 	}
 }
 
+/// The run the calling thread is inside, if any.
+pub(crate) fn run_id() -> Option<u64> {
+	current().map(|runtime| runtime.id)
+}
+
+/// Run `id`, which the calling thread must be inside.
+fn expect_run(id: u64) -> Rc<Runtime> {
+	current()
+		.filter(|runtime| runtime.id == id)
+		.expect("a clock captured in a kio::clock::sim::run read outside it")
+}
+
+/// The simulated instant of run `id`.
+pub(crate) fn now_in(id: u64) -> Instant {
+	expect_run(id).now()
+}
+
+/// The simulated wall time of run `id`.
+pub(crate) fn wall_in(id: u64) -> Wall {
+	expect_run(id).time.borrow().clock.wall()
+}
+
 /// Whether the calling thread is inside [`run`].
 pub fn is_running() -> bool {
 	CURRENT.with(|current| current.borrow().is_some())

rs/moq-audio/src/encode/capture.rs

@@ -2104,7 +2104,7 @@ mod tests {
 				let mut broadcast = moq_net::broadcast::Info::new().produce();
 				let consumer = broadcast.consume();
 				let config = moq_mux::catalog::Config::default()
-					.with_clock(clock)
+					.with_clock(clock.clone())
 					.with_max_age(RETAIN);
 				let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap();
 				let mut options = Capture::default();