From 0d033aa3da9d48ed621b6376b4dd4e2ac979a2b7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 23:14:04 -0700 Subject: [PATCH 1/3] chore(quest): claim quest/m1/broadcast-clock.md Quest-Claim: 4c9bf704a218d58f From 956f7d32ada87cd63b947887c45cc96ecab64eb8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 15:21:14 -0800 Subject: [PATCH 2/3] fix(mux)!: stamp every broadcast track from the clock it was created under A moq-mux broadcast stamped media on its Clock but the hang and MSF catalog groups and the HLS records with the ambient Timestamp::now(), and the Clock itself read the ambient clock. A call from another runtime or clock context could then move a track's timestamps backwards. kio::clock::Clock::current() captures the ambient clock (host, tokio runtime, installed manual clock, or simulator run) and keeps reading it from anywhere. moq_mux::Clock holds one, so it is Clone and no longer Copy. The catalog's hang and MSF groups and the HLS records are stamped on the broadcast clock, so they share the media timeline. hls::Producer::new takes the clock to stamp on. The moq-ffi container test now writes through the exported entry point, which enters the FFI worker, and checks the catalog lands on the media timeline. Finishes quest/m1/broadcast-clock.md. Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/kio.md | 5 + doc/lib/rs/moq-mux.md | 4 +- quest/m1/README.md | 1 - quest/m1/broadcast-clock.md | 45 ---- rs/kio/src/clock/captured.rs | 217 +++++++++++++++++++ rs/kio/src/clock/installed.rs | 5 + rs/kio/src/clock/mod.rs | 4 + rs/kio/src/clock/runtime.rs | 12 + rs/kio/src/clock/sim.rs | 22 ++ rs/moq-audio/src/encode/capture.rs | 2 +- rs/moq-ffi/src/media.rs | 35 ++- rs/moq-hls/src/export/archive_tests.rs | 2 +- rs/moq-hls/src/export/mod.rs | 13 +- rs/moq-hls/src/server/routes.rs | 2 +- rs/moq-mux/src/catalog/data.rs | 3 +- rs/moq-mux/src/catalog/producer.rs | 213 ++++++++++++++---- rs/moq-mux/src/clock.rs | 87 +++++--- rs/moq-mux/src/container/fmp4/import_test.rs | 2 +- rs/moq-mux/src/container/producer.rs | 4 +- rs/moq-mux/src/hls.rs | 56 +++-- rs/moq-transcode/src/catalog.rs | 7 +- rs/moq-video/src/encode/producer.rs | 5 +- 22 files changed, 571 insertions(+), 175 deletions(-) delete mode 100644 quest/m1/broadcast-clock.md create mode 100644 rs/kio/src/clock/captured.rs diff --git a/doc/lib/rs/kio.md b/doc/lib/rs/kio.md index 551b21bded1..0a70ca5e6ca 100644 --- a/doc/lib/rs/kio.md +++ b/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 diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 1ba340321d0..1ed006443c9 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -24,7 +24,9 @@ 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 +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 (`/` for the codec importers) while the old track ends; everything else, diff --git a/quest/m1/README.md b/quest/m1/README.md index ec184dd6023..6093272248d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -146,7 +146,6 @@ per-frame CPU growth since #5926. - [Exact timestamps in every binding](/quest/m1/binding-timestamps.md) - bindings carry `{value, scale}` timestamps instead of lossy microseconds - [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 diff --git a/quest/m1/broadcast-clock.md b/quest/m1/broadcast-clock.md deleted file mode 100644 index 2e39a443f6f..00000000000 --- a/quest/m1/broadcast-clock.md +++ /dev/null @@ -1,45 +0,0 @@ -# [S] A broadcast stamps everything from one clock - -## Goal - -Every group a moq-mux broadcast writes, its media and also its hang and MSF -catalog groups and HLS records, is stamped from the broadcast's own clock, -captured once when it is created. None reads the ambient `Timestamp::now()`, -so a call made from another runtime or clock context cannot move a track's -timestamps backwards. - -## Plan - -Decided 2026-10-29 (maintainer), a follow-up of -[moq-dev/moq#7161](https://github.com/moq-dev/moq/pull/7161): one clock for -the whole broadcast is the fix, not stamping with `max(now, last)`, which -would hide the second clock. - -Found 2026-10-29: media is stamped on the broadcast timeline -(`moq_mux::Clock`, held by the catalog producer), while the catalog's hang and -MSF groups (`rs/moq-mux/src/catalog/producer.rs`) and the HLS record track -(`rs/moq-mux/src/hls.rs`) stamp with ambient `Timestamp::now()`, which reads -whichever runtime or `Manual` clock is current and anchors on its origin. The -`Clock` itself also reads its instant from the ambient clock. In moq-ffi's -`containers_of_one_source_keep_their_relative_media_time`, the broadcast is -built under a paused runtime while some calls enter the FFI worker's -runtime, so one catalog track got groups from two anchors and was refused -with `TimestampMismatch` (moq-net checks group order per track). - -- Find every stamp a broadcast writes that reads the ambient clock, and route - it through the broadcast's clock. -- The catalog's timestamps change meaning if they move onto the media - timeline. Check that against - [Address groups by timestamp](/quest/m1/time-addressing/README.md) and - record the choice. -- Fix the moq-ffi test's mixed clock contexts, or make it prove the new - guarantee. - -Tests: a broadcast written from two clock contexts keeps every track's -timestamps non-decreasing. - -Public API: none expected; record any change. Wire: none. - -## Related - -- [Address groups by timestamp](/quest/m1/time-addressing/README.md) - defines what a group's timestamp means diff --git a/rs/kio/src/clock/captured.rs b/rs/kio/src/clock/captured.rs new file mode 100644 index 00000000000..6410fdbabdf --- /dev/null +++ b/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()); + }); + } +} diff --git a/rs/kio/src/clock/installed.rs b/rs/kio/src/clock/installed.rs index f4757354e44..222e255fcec 100644 --- a/rs/kio/src/clock/installed.rs +++ b/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 { diff --git a/rs/kio/src/clock/mod.rs b/rs/kio/src/clock/mod.rs index bd16fd7e327..802e7f7c274 100644 --- a/rs/kio/src/clock/mod.rs +++ b/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; diff --git a/rs/kio/src/clock/runtime.rs b/rs/kio/src/clock/runtime.rs index 97e7a15f70a..0beec1e1403 100644 --- a/rs/kio/src/clock/runtime.rs +++ b/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 { diff --git a/rs/kio/src/clock/sim.rs b/rs/kio/src/clock/sim.rs index ffd9dee61c2..4cdb20b036d 100644 --- a/rs/kio/src/clock/sim.rs +++ b/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 { + current().map(|runtime| runtime.id) +} + +/// Run `id`, which the calling thread must be inside. +fn expect_run(id: u64) -> Rc { + 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()) diff --git a/rs/moq-audio/src/encode/capture.rs b/rs/moq-audio/src/encode/capture.rs index 066f31df8ff..f160750205b 100644 --- a/rs/moq-audio/src/encode/capture.rs +++ b/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(); diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 3da7421872b..e6dde5fe490 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -1872,26 +1872,11 @@ mod timebase_tests { .unwrap(); tokio::time::advance(Duration::from_secs(1)).await; let first = clock.now().as_micros(); - // Drive the private importers on the paused clock, without entering the unpaused FFI worker. - chunks - .inner - .lock() - .unwrap() - .as_mut() - .unwrap() - .import - .decode(&fragment(&source, 3600, 1)) - .unwrap(); + // Each write enters the unpaused FFI worker, whose ambient clock is the host's. The + // broadcast still stamps on the paused clock it was built under. + chunks.write(fragment(&source, 3600, 1).to_vec()).unwrap(); tokio::time::advance(Duration::from_secs(5)).await; - audio_chunks - .inner - .lock() - .unwrap() - .as_mut() - .unwrap() - .import - .decode(&fragment(&audio, 3602, 2)) - .unwrap(); + audio_chunks.write(fragment(&audio, 3602, 2).to_vec()).unwrap(); let mut readers = Vec::new(); let snapshot = catalog.snapshot(); let containers = @@ -1930,6 +1915,18 @@ mod timebase_tests { first + if shared { 2_000_000 } else { 5_000_000 }, "shared={shared}" ); + + // The catalog the second write published is on the same timeline as the media. + let mut catalogs = consumer + .track(hang::catalog::Catalog::DEFAULT_NAME) + .unwrap() + .subscribe(moq_net::track::Subscription::default()) + .await + .unwrap() + .ordered(); + let mut group = catalogs.next_group().await.unwrap().unwrap(); + let stamp = group.read_frame().await.unwrap().unwrap().timestamp.unwrap(); + assert_eq!(stamp.as_micros(), first + 5_000_000, "shared={shared}"); } } diff --git a/rs/moq-hls/src/export/archive_tests.rs b/rs/moq-hls/src/export/archive_tests.rs index 2184c22c3c5..39c23cd136c 100644 --- a/rs/moq-hls/src/export/archive_tests.rs +++ b/rs/moq-hls/src/export/archive_tests.rs @@ -149,7 +149,7 @@ impl Recording { .unwrap(); let source = Source { media, - records: moq_mux::hls::Producer::new(track.clone()), + records: moq_mux::hls::Producer::new(track.clone(), moq_mux::Clock::new()), track, record: 0, }; diff --git a/rs/moq-hls/src/export/mod.rs b/rs/moq-hls/src/export/mod.rs index 789ef2334c9..05da992c727 100644 --- a/rs/moq-hls/src/export/mod.rs +++ b/rs/moq-hls/src/export/mod.rs @@ -1061,7 +1061,7 @@ mod tests { let track = broadcast .create_track(hang::hls::default_name(name), moq_mux::hls::Producer::info()) .unwrap(); - (track.clone(), moq_mux::hls::Producer::new(track)) + (track.clone(), moq_mux::hls::Producer::new(track, moq_mux::Clock::new())) }) .collect(); for sequence in 0..4 { @@ -1279,7 +1279,7 @@ mod tests { let track = broadcast .create_track(hang::hls::default_name(name), moq_mux::hls::Producer::info()) .unwrap(); - moq_mux::hls::Producer::new(track) + moq_mux::hls::Producer::new(track, moq_mux::Clock::new()) }); (origin, broadcast, catalog, records) } @@ -1414,7 +1414,7 @@ mod tests { let track = broadcast .create_track(hang::hls::default_name(name), moq_mux::hls::Producer::info()) .unwrap(); - (track.clone(), moq_mux::hls::Producer::new(track)) + (track.clone(), moq_mux::hls::Producer::new(track, moq_mux::Clock::new())) }); for sequence in 0..4 { video0.push(&gop(sequence)).unwrap(); @@ -2154,6 +2154,7 @@ mod tests { broadcast .create_track(hang::hls::default_name("video0"), moq_mux::hls::Producer::info()) .unwrap(), + moq_mux::Clock::new(), ); let media = Records::write_media(&broadcast, "video0", 3); for sequence in 0..3 { @@ -2603,7 +2604,7 @@ mod tests { let track = broadcast .create_track(hang::hls::default_name("video0"), moq_mux::hls::Producer::info()) .unwrap(); - (track.clone(), moq_mux::hls::Producer::new(track)) + (track.clone(), moq_mux::hls::Producer::new(track, moq_mux::Clock::new())) } /// Write three keyframes of `payload` into `broadcast`'s `video0` track, 2s apart from @@ -3438,7 +3439,7 @@ mod tests { .broadcast .create_track(hang::hls::default_name(track), moq_mux::hls::Producer::info()) .unwrap(); - (track.clone(), moq_mux::hls::Producer::new(track)) + (track.clone(), moq_mux::hls::Producer::new(track, moq_mux::Clock::new())) } /// Publish `name`'s media: one keyframe group per [`gop`] record `0..groups`, so a cursor @@ -3651,7 +3652,7 @@ mod tests { let track = broadcast .create_track(hang::hls::default_name(name), moq_mux::hls::Producer::info()) .unwrap(); - (track.clone(), moq_mux::hls::Producer::new(track)) + (track.clone(), moq_mux::hls::Producer::new(track, moq_mux::Clock::new())) }; let (_video0_track, mut video0) = publish("video0"); let (video1_track, mut video1) = publish("video1"); diff --git a/rs/moq-hls/src/server/routes.rs b/rs/moq-hls/src/server/routes.rs index 6390d794b43..5ff5e6caf27 100644 --- a/rs/moq-hls/src/server/routes.rs +++ b/rs/moq-hls/src/server/routes.rs @@ -582,7 +582,7 @@ mod tests { }) .unwrap(); let track = broadcast.create_track(name, moq_mux::hls::Producer::info()).unwrap(); - let mut records = moq_mux::hls::Producer::new(track.clone()); + let mut records = moq_mux::hls::Producer::new(track.clone(), moq_mux::Clock::new()); for sequence in 0..4 { let record = hang::hls::Record::new(sequence, sequence, sequence * 2_000); records.push(&record).unwrap(); diff --git a/rs/moq-mux/src/catalog/data.rs b/rs/moq-mux/src/catalog/data.rs index 7279138e414..a0783866366 100644 --- a/rs/moq-mux/src/catalog/data.rs +++ b/rs/moq-mux/src/catalog/data.rs @@ -88,8 +88,9 @@ impl Listing { if clock != self.clock { self.estimator.discontinuity(); } + let now = clock.now(); self.clock = clock; - self.record_at(clock.now(), bytes, captured) + self.record_at(now, bytes, captured) } /// [`record`](Self::record) at a chosen time, which a test needs since the broadcast clock only diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 5741ef5981b..10f705878a8 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -215,22 +215,27 @@ impl Declared { } } -/// The producer's `clock` under a root `clock` of `advertised`: re-based on a changed mapping while -/// nothing has `anchored` the clock, and refused once something has. Clearing it is always refused. -fn adopt(clock: crate::Clock, anchored: bool, advertised: Option) -> crate::Result { +/// The producer's `clock` re-based on a root `clock` of `advertised`, or `None` when that is the +/// mapping it already has. A changed mapping is adopted while nothing has `anchored` the clock and +/// refused once something has. Clearing it is always refused. +fn adopt( + clock: &crate::Clock, + anchored: bool, + advertised: Option, +) -> crate::Result> { if advertised == clock.wall() { - return Ok(clock); + return Ok(None); } let refuse = |why: &str| crate::Error::UnmappableTimestamp(format!("root clock {advertised:?} {why}")); let wall = crate::Clock::exact(advertised.ok_or_else(|| refuse("clears the producer's clock"))?)?; // The same mapping in other units needs no host time, which may be unset or behind PTS zero. if Some(wall) == clock.wall() { - return Ok(clock); + return Ok(None); } if anchored { return Err(refuse("changes the producer's clock after it was fixed")); } - crate::Clock::adopted(wall) + Ok(Some(clock.adopted(wall)?)) } /// Take the shared state, ignoring a poisoned lock. @@ -325,15 +330,18 @@ impl Outputs { std::borrow::Cow::Owned(catalog) } - /// Emit the catalog to hang and MSF. Neither writes a catalog identical to its last. - fn emit(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { + /// Emit the catalog to hang and MSF, stamped now on `clock`. Neither writes a catalog identical + /// to its last. + fn emit(&mut self, catalog: &Catalog, msf: &mut Msf, clock: &crate::Clock) -> crate::Result<()> { let catalog = self.advertise(catalog); let catalog = catalog.as_ref(); // Every consumer refuses a catalog that fails this, so refuse it before it reaches one. catalog.check_renditions()?; + // The media timeline, so the catalog in effect at a media timestamp is the one stamped at or + // before it, whichever thread or runtime publishes. + let at = clock.now(); // One snapshot per group while deltas are disabled. - self.hang - .update(moq_net::Timed::from(catalog).at(moq_net::Timestamp::now()))?; + self.hang.update(moq_net::Timed::from(catalog).at(at))?; // `abort` below is what keeps the tracks consistent if this fails after the hang track was // updated. @@ -341,7 +349,7 @@ impl Outputs { return Ok(()); }; let mut group = self.msf_track.append_group()?; - group.write_frame(moq_net::Timestamp::now(), msf)?; + group.write_frame(at, msf)?; group.finish()?; Ok(()) @@ -517,10 +525,11 @@ impl Producer { // Resolved before any track exists, so a refused seed leaves the broadcast untouched. let anchored = config.clock.is_some(); let mut catalog = config.catalog; + // An unseeded catalog runs on the configured clock, or a fresh one. + let clock = config.clock.unwrap_or_default(); let clock = match catalog.clock { - // An unseeded catalog runs on the configured clock, or a fresh one. - None => config.clock.unwrap_or_default(), - seeded => adopt(config.clock.unwrap_or_default(), anchored, seeded)?, + None => clock, + seeded => adopt(&clock, anchored, seeded)?.unwrap_or(clock), }; catalog.clock = clock.wall(); @@ -537,7 +546,10 @@ impl Producer { let json_config = moq_json::snapshot::Config::default().with_deltas(false); let hang = moq_json::snapshot::Producer::new(hang_track.clone(), json_config); - let hls = config.hls.as_ref().map(|_| crate::hls::Tracks::new(broadcast)); + let hls = config + .hls + .as_ref() + .map(|_| crate::hls::Tracks::new(broadcast, clock.clone())); Ok(Self { outputs: Outputs { @@ -650,18 +662,18 @@ impl Producer { /// The broadcast's shared clock: the timeline producers stamp against and the wall mapping /// advertised at the catalog root. /// - /// Copies share the timeline, so handing them to concurrent producers keeps every track on + /// Clones share the timeline, so handing them to concurrent producers keeps every track on /// one clock. Taking the clock fixes it: an importer whose first frame arrives later shifts its - /// timestamps onto this clock rather than moving it, so a copy never goes stale. + /// timestamps onto this clock rather than moving it, so a clone never goes stale. pub fn clock(&self) -> crate::Clock { let mut state = take(&self.current); state.anchored = true; - state.clock + state.clock.clone() } /// The clock as it stands, without fixing it: for readers that follow it on every use. pub(super) fn current_clock(&self) -> crate::Clock { - take(&self.current).clock + take(&self.current).clock.clone() } /// The config `name` holds in `C`'s section, cloned without copying the rest of the catalog. @@ -694,7 +706,7 @@ impl Producer { } let offset = if guard.state.anchored { - let clock = guard.state.clock; + let clock = &guard.state.clock; let at = match wall { Some(wall) => clock.reading(wall)?, None => clock.now().as_micros() as i128, @@ -702,17 +714,19 @@ impl Producer { super::Offset::from_micros(at - pts.as_micros() as i128)? } else { let since = std::time::Duration::from_micros(u64::try_from(pts.as_micros()).unwrap_or(u64::MAX)); + // Moved, not replaced: it keeps reading the clock the broadcast was created under. let clock = match wall { - Some(wall) => crate::Clock::placed(since, wall)?, - None => crate::Clock::arrival(since)?, + Some(wall) => guard.state.clock.placed(since, wall)?, + None => guard.state.clock.arrival(since)?, }; - guard.state.clock = clock; + let advertised = clock.wall(); + guard.retime(clock); guard.state.anchored = true; match wall { // A placement can precede every track. It stages no publish, so the catalog first // goes out listing them rather than carrying a clock alone. - Some(_) => guard.state.catalog.clock = clock.wall(), - None => guard.clock = clock.wall(), + Some(_) => guard.state.catalog.clock = advertised, + None => guard.clock = advertised, } super::Offset::default() }; @@ -882,7 +896,7 @@ impl Producer { state.estimates.fold(); // Emitted under the lock, as a guard's publish is, since the MSF state belongs to it. let state = &mut *state; - if let Err(err) = self.outputs.emit(&state.catalog, &mut state.msf) { + if let Err(err) = self.outputs.emit(&state.catalog, &mut state.msf, &state.clock) { tracing::error!(%err, "failed to publish the catalog, aborting its tracks"); self.outputs.abort(&err); state.closed = Some(err); @@ -1032,7 +1046,7 @@ impl Producer { return; } let state = &mut *state; - if let Err(err) = outputs.emit(&state.catalog, &mut state.msf) { + if let Err(err) = outputs.emit(&state.catalog, &mut state.msf, &state.clock) { tracing::error!(%err, "failed to publish an HLS declaration, aborting catalog tracks"); outputs.abort(&err); state.closed = Some(err); @@ -1189,6 +1203,14 @@ impl Guard<'_, E> { self.publish() } + /// Move the producer onto `clock`, a re-basing of the one it has, before anything fixed it. + fn retime(&mut self, clock: crate::Clock) { + if let Some(hls) = &self.outputs.hls { + hls.retime(&clock); + } + self.state.clock = clock; + } + /// Publish a mutated catalog once, clearing the flag so it isn't published again. fn publish(&mut self) -> crate::Result<()> { if !self.updated { @@ -1197,16 +1219,12 @@ impl Guard<'_, E> { self.updated = false; // The root `clock` is the producer's clock: never advertise one it disagrees with. - match adopt(self.state.clock, self.state.anchored, self.state.catalog.clock) { - Ok(clock) => { - self.state.clock = clock; - self.state.catalog.clock = clock.wall(); - } - Err(err) => { - self.state.catalog.clock = self.state.clock.wall(); - return Err(err); - } + let adopted = adopt(&self.state.clock, self.state.anchored, self.state.catalog.clock); + if let Ok(Some(clock)) = &adopted { + self.retime(clock.clone()); } + self.state.catalog.clock = self.state.clock.wall(); + adopted?; { // Withhold every emit while still buffering the initial reserved set, or until a reserved @@ -1227,7 +1245,7 @@ impl Guard<'_, E> { // Folded only once it is on the wire, so a failed emit leaves the estimate held. let state = &mut *self.state; - self.outputs.emit(&state.catalog, &mut state.msf)?; + self.outputs.emit(&state.catalog, &mut state.msf, &state.clock)?; self.state.estimates.fold(); Ok(()) } @@ -2285,8 +2303,16 @@ mod test { let front = origin.create_broadcast("robot/front.hang").unwrap(); let rear = origin.create_broadcast("robot/rear.hang").unwrap(); let catalogs = [ - Producer::new(&mut front.broadcast().clone(), Config::default().with_clock(clock)).unwrap(), - Producer::new(&mut rear.broadcast().clone(), Config::default().with_clock(clock)).unwrap(), + Producer::new( + &mut front.broadcast().clone(), + Config::default().with_clock(clock.clone()), + ) + .unwrap(), + Producer::new( + &mut rear.broadcast().clone(), + Config::default().with_clock(clock.clone()), + ) + .unwrap(), ]; let mut broadcasts = [front, rear]; for (catalog, broadcast) in catalogs.iter().zip(&mut broadcasts) { @@ -2324,8 +2350,107 @@ mod test { } } + /// A broadcast built under one clock and written from others, a paused runtime a minute behind + /// it and a plain thread on the host's, stamps every track on the clock it was built under, so + /// none of them goes back: media, both catalogs, and the HLS records. + #[test] + fn a_broadcast_written_from_two_clock_contexts_never_rewinds() { + use hang::catalog::{JsonConfig, Mode}; + use std::time::Duration; + + let paused = || { + tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(true) + .build() + .unwrap() + }; + let (home, away) = (paused(), paused()); + let mut broadcast = moq_net::broadcast::Info::default().produce(); + let (mut catalog, mut video, readers) = home.block_on(async { + tokio::time::advance(Duration::from_secs(60)).await; + let catalog = Producer::new( + &mut broadcast, + Config::default().with_max_age(Duration::from_secs(3600)), + ) + .unwrap(); + let video = video0(&mut broadcast, &catalog, &[]); + let consumer = broadcast.consume(); + let mut readers = Vec::new(); + for name in [ + hang::Catalog::DEFAULT_NAME, + moq_msf::DEFAULT_NAME, + "video0", + "video0.hls", + ] { + let all = moq_net::track::Subscription::default().with_max_delay(None); + let track = consumer.track(name).unwrap().subscribe(all).await.unwrap(); + readers.push((name, track.ordered())); + } + (catalog, video, readers) + }); + + // One keyframe group, so one HLS segment, and one catalog change, so one hang catalog group. + let write = |catalog: &mut Producer, video: &mut crate::container::Producer<_, VideoConfig>, n: u64| { + let timestamp = catalog.timestamp(None).unwrap(); + video + .write(crate::container::Frame { + timestamp, + duration: None, + payload: Bytes::from_static(b"keyframe"), + keyframe: true, + }) + .unwrap(); + catalog + .mutate(|catalog| { + catalog + .json + .tracks + .insert(format!("json{n}"), JsonConfig::new(Mode::Snapshot)); + }) + .unwrap(); + }; + for n in 0..4 { + // Only the clock the broadcast was built under moves it. + home.block_on(tokio::time::advance(Duration::from_secs(4))); + match n % 2 { + 0 => away.block_on(async { write(&mut catalog, &mut video, n) }), + _ => std::thread::scope(|scope| { + scope.spawn(|| write(&mut catalog, &mut video, n)); + }), + } + } + away.block_on(async { + video.finish().unwrap(); + catalog.finish().unwrap(); + }); + + home.block_on(async { + for (name, mut track) in readers { + let (mut stamps, mut starts) = (Vec::new(), Vec::::new()); + while let Some(mut group) = track.next_group().await.unwrap() { + let first = stamps.len(); + while let Some(frame) = group.read_frame().await.unwrap() { + stamps.push(frame.timestamp.unwrap().as_micros()); + } + starts.extend(stamps.get(first).copied()); + } + assert!(stamps.is_sorted(), "{name} went back: {stamps:?}"); + // The clock read ten seconds when built, and only `home` advanced it, to 26. + let timeline = 10_000_000..=26_000_000; + assert!( + stamps.iter().all(|stamp| timeline.contains(stamp)), + "{name}: {stamps:?}" + ); + if name == "video0" { + assert_eq!(starts, [14, 18, 22, 26].map(|secs| secs * 1_000_000), "{name}"); + } + } + }); + } + /// How far `pts` lands from `clock`'s reading now, which is zero for a frame live on arrival. - fn lateness(clock: crate::Clock, pts: moq_net::Timestamp) -> std::time::Duration { + fn lateness(clock: &crate::Clock, pts: moq_net::Timestamp) -> std::time::Duration { let now = clock.now().as_micros(); std::time::Duration::from_micros(now.abs_diff(pts.as_micros()) as u64) } @@ -2350,7 +2475,7 @@ mod test { let first = moq_net::Timestamp::from_micros(3_600_000_000).unwrap(); let shifted = catalog.timebase().shift(first).unwrap(); assert!( - lateness(catalog.clock(), shifted) < Duration::from_secs(1), + lateness(&catalog.clock(), shifted) < Duration::from_secs(1), "{shifted:?}" ); assert_eq!(catalog.snapshot().clock.map(|clock| clock.wall), advertised); @@ -2388,7 +2513,7 @@ mod test { .timebase() .shift(moq_net::Timestamp::from_micros(0).unwrap()) .unwrap(); - assert!(lateness(clock, other) < Duration::from_secs(1), "{other:?}"); + assert!(lateness(&clock, other) < Duration::from_secs(1), "{other:?}"); assert_eq!( catalog.clock().wall(), clock.wall(), @@ -2409,7 +2534,7 @@ mod test { .unwrap(); assert_eq!(catalog.clock().wall(), clock.wall()); assert!( - lateness(clock, shifted) < std::time::Duration::from_secs(1), + lateness(&clock, shifted) < std::time::Duration::from_secs(1), "{shifted:?}" ); } @@ -2567,7 +2692,7 @@ mod test { let fixed = crate::Clock::at(kio::clock::now(), zero).unwrap(); seed.clock = Some(an_hour_ago(moq_net::Timescale::MILLI)); crate::clock::fake_host_wall(kio::clock::Wall::UNIX_EPOCH + Duration::from_secs(86_400)); - let config = Config::default().with_catalog(seed.clone()).with_clock(fixed); + let config = Config::default().with_catalog(seed.clone()).with_clock(fixed.clone()); let mut broadcast = moq_net::broadcast::Info::new().produce(); let mut catalog = Producer::new(&mut broadcast, config).unwrap(); assert_eq!(catalog.clock(), fixed); diff --git a/rs/moq-mux/src/clock.rs b/rs/moq-mux/src/clock.rs index 142ffeba6a4..d6f4e3aa0ab 100644 --- a/rs/moq-mux/src/clock.rs +++ b/rs/moq-mux/src/clock.rs @@ -1,9 +1,13 @@ //! The broadcast's shared clock: one monotonic timeline plus its wall mapping. //! -//! Create one [`Clock`] per broadcast and hand copies to every producer: because they share a +//! Create one [`Clock`] per broadcast and hand clones to every producer: because they share a //! timeline, frames captured at the same instant get the same timestamp, keeping concurrently -//! produced tracks (e.g. audio and video capture on separate threads) in sync. It is `Copy`, so -//! handing it out is cheap. Taking the catalog's clock fixes it, so a copy never goes stale. +//! produced tracks (e.g. audio and video capture on separate threads) in sync. Cloning is cheap. +//! Taking the catalog's clock fixes it, so a clone never goes stale. +//! +//! A clock reads the ambient clock it was created under (see [`kio::clock::Clock`]), wherever it +//! is read from later. A broadcast built under one runtime and written from another, or from a +//! plain thread, stays on one timeline instead of jumping between the two clocks. //! //! The clock also owns the broadcast's wall mapping, advertised at the catalog root as //! `clock: { wall, timescale }`: `wall` is the wall-clock time of PTS zero in @@ -31,7 +35,11 @@ fn moq_micros(wall: Wall) -> i128 { /// The host's wall clock, or `None` while it reads before the moq epoch (2020): unset. pub(crate) fn host_wall() -> Option { - let now = kio::clock::wall(); + set(kio::clock::wall()) +} + +/// `now`, or `None` while it reads before the moq epoch (2020): a wall clock not yet set. +fn set(now: Wall) -> Option { #[cfg(test)] let now = HOST_WALL.get().unwrap_or(now); (moq_micros(now) >= 0).then_some(now) @@ -67,11 +75,13 @@ fn wall_clock(wall: Wall) -> crate::Result { /// concurrently, e.g. an audio and a video capture running on separate /// threads, land on a single timeline. /// -/// Copies share the timeline and the wall mapping, so handing them to several producers keeps +/// Clones share the timeline and the wall mapping, so handing them to several producers keeps /// the broadcast on one clock. Two clocks are equal only when they are one timeline. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub struct Clock { - /// A monotonic instant, and what the clock read then in micros. + /// The ambient clock captured where this one was created, read wherever it is read. + source: kio::clock::Clock, + /// An instant on `source`, and what the clock read then in micros. instant: Instant, reading: u64, /// `None` when the host's wall clock was unset at construction. @@ -97,33 +107,38 @@ impl Clock { /// Without a wall mapping when the host's wall clock is unset, or reads a time the mapping can't /// name: media needs only the monotonic timeline. pub fn new() -> Self { + let source = kio::clock::Clock::current(); Self { - instant: kio::clock::now(), + instant: source.now(), reading: Self::LEAD.as_micros() as u64, - wall: host_wall().and_then(|now| wall_clock(now.checked_sub(Self::LEAD)?).ok()), + wall: set(source.wall()).and_then(|now| wall_clock(now.checked_sub(Self::LEAD)?).ok()), + source, } } /// Start a clock at an explicit monotonic epoch and wall time: PTS zero at both. /// /// The deterministic constructor: synthetic sources and fixtures pin both ends instead of - /// sampling. Refuses an unrepresentable wall. + /// sampling. `epoch` is an instant on the ambient clock, which the clock keeps reading wherever + /// it is read from. Refuses an unrepresentable wall. pub fn at(epoch: Instant, wall: Wall) -> crate::Result { Ok(Self { + source: kio::clock::Clock::current(), instant: epoch, reading: 0, wall: Some(wall_clock(wall)?), }) } - /// A clock that reads `since` now: a source whose first timestamp is `since` is live on arrival. + /// This clock moved to read `since` now: a source whose first timestamp is `since` is live on + /// arrival. /// /// Refuses a `since` so large that PTS zero lands before the moq epoch (2020), which the wall /// mapping cannot name. On a host whose wall clock is unset, the clock has no wall mapping. - pub(crate) fn arrival(since: Duration) -> crate::Result { - let instant = kio::clock::now(); + pub(crate) fn arrival(&self, since: Duration) -> crate::Result { + let instant = self.source.now(); let unmappable = || crate::Error::UnmappableTimestamp(format!("{since:?} puts PTS zero before 2020")); - let wall = match host_wall() { + let wall = match set(self.source.wall()) { Some(now) => { let zero = now.checked_sub(since).ok_or_else(unmappable)?; Some(wall_clock(zero).map_err(|_| unmappable())?) @@ -134,24 +149,25 @@ impl Clock { None => None, }; Ok(Self { + source: self.source.clone(), instant, reading: u64::try_from(since.as_micros()).map_err(|_| unmappable())?, wall, }) } - /// A clock on which `pts` is the instant `wall`: a source that knows when its timestamps + /// This clock moved so `pts` is the instant `wall`: a source that knows when its timestamps /// happened places them there rather than at their arrival. /// /// Refuses a mapping whose PTS zero lands before the moq epoch (2020) or after now, and a host /// whose wall clock is unset, which has no now to place it against. - pub(crate) fn placed(pts: Duration, wall: Wall) -> crate::Result { + pub(crate) fn placed(&self, pts: Duration, wall: Wall) -> crate::Result { let unmappable = |why: &str| crate::Error::UnmappableTimestamp(format!("{pts:?} at {wall:?} {why}")); let zero = wall .checked_sub(pts) .and_then(|zero| wall_clock(zero).ok()) .ok_or_else(|| unmappable("puts PTS zero before 2020"))?; - Self::mapped(zero, unmappable) + self.mapped(zero, unmappable) } /// A catalog root `clock` in [`TIMESCALE`](Self::TIMESCALE), the units [`wall`](Self::wall) @@ -169,26 +185,28 @@ impl Clock { Ok(hang::catalog::Clock::new(moq_net::Timestamp::from_micros(micros)?)?) } - /// A clock on an [`exact`](Self::exact) catalog root `clock`: a derivative republishing its - /// input's timestamps runs on its input's mapping. + /// This clock moved onto an [`exact`](Self::exact) catalog root `clock`: a derivative + /// republishing its input's timestamps runs on its input's mapping. /// /// Refuses a PTS zero after now, and a host whose wall clock is unset. - pub(crate) fn adopted(wall: hang::catalog::Clock) -> crate::Result { - Self::mapped(wall, |why| { + pub(crate) fn adopted(&self, wall: hang::catalog::Clock) -> crate::Result { + self.mapped(wall, |why| { crate::Error::UnmappableTimestamp(format!("root clock {wall:?} {why}")) }) } - /// A clock whose PTS zero is `zero`, anchored now at however long ago that was on the host. + /// This clock moved so its PTS zero is `zero`, anchored now at however long ago that was on the + /// host. /// /// No `Instant` arithmetic, so a PTS zero before this host booted still maps. - fn mapped(zero: hang::catalog::Clock, unmappable: impl Fn(&str) -> crate::Error) -> crate::Result { - let instant = kio::clock::now(); - let now = host_wall().ok_or_else(|| unmappable("needs the host wall clock, which is unset"))?; + fn mapped(&self, zero: hang::catalog::Clock, unmappable: impl Fn(&str) -> crate::Error) -> crate::Result { + let instant = self.source.now(); + let now = set(self.source.wall()).ok_or_else(|| unmappable("needs the host wall clock, which is unset"))?; // Saturating would stamp everything early by the skew without saying so. let reading = u64::try_from(moq_micros(now) - zero.wall.as_micros() as i128) .map_err(|_| unmappable("puts PTS zero after now"))?; Ok(Self { + source: self.source.clone(), instant, reading, wall: Some(zero), @@ -211,17 +229,18 @@ impl Clock { /// The current timestamp on this clock. pub fn now(&self) -> moq_net::Timestamp { // u128 -> u64 truncation is unreachable: u64 microseconds is ~584,000 years. - let elapsed = self.instant.elapsed().as_micros() as u64; + let elapsed = self.source.now().saturating_duration_since(self.instant).as_micros() as u64; moq_net::Timestamp::from_micros(self.reading + elapsed) .expect("an instant elapsed duration fits in a timestamp") } /// Map the instant a payload was captured (a datagram's arrival, a sensor read) onto this clock. /// - /// Refuses an instant ahead of now, which would claim the payload reached the transport before - /// it existed, and one before PTS zero, which no timestamp can name. + /// `at` is an instant on the ambient clock this one was created under, which is the host's + /// outside tests. Refuses an instant ahead of now, which would claim the payload reached the + /// transport before it existed, and one before PTS zero, which no timestamp can name. pub fn capture(&self, at: Instant) -> crate::Result { - if at > kio::clock::now() { + if at > self.source.now() { return Err(crate::Error::InvalidCapture); } let micros = match at.checked_duration_since(self.instant) { @@ -376,7 +395,7 @@ mod tests { async fn arrival_maps_the_first_timestamp_to_now() { kio::clock::sleep(Duration::from_secs(1)).await; let since = Duration::from_secs(10 * 3600); - let clock = Clock::arrival(since).unwrap(); + let clock = Clock::new().arrival(since).unwrap(); let now = clock.now(); assert!(now.as_micros() >= since.as_micros()); let actual = clock.wall_clock(now).unwrap(); @@ -390,7 +409,7 @@ mod tests { // PTS zero before 2020 cannot be named on the wire. assert!(matches!( - Clock::arrival(Duration::from_secs(100 * 365 * 86_400)), + Clock::new().arrival(Duration::from_secs(100 * 365 * 86_400)), Err(crate::Error::UnmappableTimestamp(_)) )); } @@ -414,13 +433,13 @@ mod tests { )); // An importer arriving an hour in still gets a timeline, and placing one has no wall to use. - assert_eq!(Clock::arrival(Duration::from_secs(3600)).unwrap().wall(), None); + assert_eq!(clock.arrival(Duration::from_secs(3600)).unwrap().wall(), None); assert!(matches!( - Clock::arrival(Duration::from_micros(MAX_SAFE_INTEGER + 1)), + clock.arrival(Duration::from_micros(MAX_SAFE_INTEGER + 1)), Err(crate::Error::UnmappableTimestamp(_)) )); assert!(matches!( - Clock::placed(Duration::from_secs(1), moq_epoch() + Duration::from_secs(60)), + clock.placed(Duration::from_secs(1), moq_epoch() + Duration::from_secs(60)), Err(crate::Error::UnmappableTimestamp(_)) )); diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index cf4192072c9..8feb3363323 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -2780,7 +2780,7 @@ async fn an_import_shifts_onto_an_explicit_clock() { let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); let clock = crate::Clock::new(); - let config = crate::catalog::Config::default().with_clock(clock); + let config = crate::catalog::Config::default().with_clock(clock.clone()); let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap(); let mut fmp4 = crate::container::fmp4::Import::new(broadcast, catalog.reserve()); diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index 420dff6f233..e7465ce816a 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -1741,7 +1741,7 @@ mod tests { #[tokio::test] async fn a_wholly_buffered_group_records_its_end() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = crate::hls::Tracks::new(&broadcast); + let tracks = crate::hls::Tracks::new(&broadcast, crate::Clock::new()); let recorder = tracks.track("video", crate::hls::Config::default()).unwrap(); let track = broadcast.create_track("video", timed()).unwrap(); let mut producer = Producer::new(track, Container::Legacy(crate::container::Kind::Video)) @@ -1760,7 +1760,7 @@ mod tests { #[tokio::test] async fn a_discontinuity_stops_the_segment_at_the_marker() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = crate::hls::Tracks::new(&broadcast); + let tracks = crate::hls::Tracks::new(&broadcast, crate::Clock::new()); let recorder = tracks.track("video", crate::hls::Config::default()).unwrap(); let track = broadcast.create_track("video", timed()).unwrap(); let mut producer = diff --git a/rs/moq-mux/src/hls.rs b/rs/moq-mux/src/hls.rs index a10420ea1e8..d929dbbace1 100644 --- a/rs/moq-mux/src/hls.rs +++ b/rs/moq-mux/src/hls.rs @@ -377,6 +377,11 @@ pub struct Producer { sink: moq_json::window::Producer, /// The record track, to abort. track: moq_net::track::Producer, + /// What each record is stamped on. A record's own pts can't be the stamp: a keyframe's pts may + /// dip below earlier content, and a group may not start before the one it follows. + clock: crate::Clock, + /// Whether a record was stamped on `clock`, which then can't be swapped for another. + stamped: bool, } impl Producer { @@ -391,12 +396,15 @@ impl Producer { .with_compress(true) } - /// Publish onto `track`, starting at record zero. - pub fn new(track: moq_net::track::Producer) -> Self { + /// Publish onto `track`, starting at record zero, stamping each record on `clock`: the + /// broadcast's [`Clock`](crate::Clock). + pub fn new(track: moq_net::track::Producer, clock: crate::Clock) -> Self { let config = moq_json::window::ProducerConfig::default().with_checkpoint_records(CHECKPOINT_RECORDS); Self { sink: moq_json::window::Producer::new(track.clone(), config), track, + clock, + stamped: false, } } @@ -410,8 +418,9 @@ impl Producer { }); } self.refresh()?; + self.stamped = true; self.sink - .push(moq_net::Timed::from(record).at(moq_net::Timestamp::now())) + .push(moq_net::Timed::from(record).at(self.clock.now())) .map_err(json) } @@ -523,6 +532,8 @@ struct State { /// recorder drops, so a re-enrolled track continues its numbering. tracks: BTreeMap>)>, finished: bool, + /// The broadcast clock every record is stamped on. + clock: crate::Clock, } /// A broadcast's HLS record tracks, one per enrolled rendition. @@ -536,17 +547,34 @@ pub(crate) struct Tracks { } impl Tracks { - /// Record tracks for tracks of `broadcast`. - pub fn new(broadcast: &moq_net::broadcast::Producer) -> Self { + /// Record tracks for tracks of `broadcast`, stamped on its `clock`. + pub fn new(broadcast: &moq_net::broadcast::Producer, clock: crate::Clock) -> Self { Self { broadcast: broadcast.clone(), state: Arc::new(Mutex::new(State { tracks: BTreeMap::new(), finished: false, + clock, })), } } + /// Stamp later records on `clock`, which the catalog moved to before anything fixed it. + /// + /// A record track that already stamped one keeps its clock: the move may read earlier, and a + /// track's timestamps never go back. + pub fn retime(&self, clock: &crate::Clock) { + let mut state = self.state.lock().unwrap(); + state.clock = clock.clone(); + for (_, live) in state.tracks.values() { + if let Some(output) = live.lock().unwrap().output.as_mut() + && !output.stamped + { + output.clock = clock.clone(); + } + } + } + /// Enroll the track `name`, its segments cut by `config`. /// /// Creates its record track, named by [`hang::hls::default_name`], on first enrollment. @@ -584,7 +612,7 @@ impl Tracks { let live = Arc::new(Mutex::new(Live { segmenter: segmenter(0), on_declare: None, - output: Some(Producer::new(track)), + output: Some(Producer::new(track, state.clock.clone())), recording: true, })); state @@ -1164,7 +1192,7 @@ mod test { #[tokio::test] async fn renditions_cut_alike_without_sharing_state() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut video = tracks.track("video", hinted()).unwrap(); let mut audio = tracks.track("audio", hinted()).unwrap(); let mut late = tracks.track("late", hinted()).unwrap(); @@ -1212,7 +1240,7 @@ mod test { #[test] fn an_unhinted_track_is_advertised_once_declared() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut recorder = tracks.track("video", Config::default()).unwrap(); recorder.frame(0, ms(0)); assert_eq!(tracks.hls("video"), None); @@ -1226,7 +1254,7 @@ mod test { #[tokio::test] async fn a_rejected_report_fails_the_record_track() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut recorder = tracks.track("video", hinted()).unwrap(); let hls = tracks.hls("video").unwrap(); let mut consumer = Consumer::subscribe(&broadcast.consume(), &hls).await.unwrap(); @@ -1248,7 +1276,7 @@ mod test { #[tokio::test] async fn re_enrolling_continues_the_numbering() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut recorder = tracks.track("video", hinted()).unwrap(); assert!(tracks.track("video", hinted()).is_err(), "its recorder is still alive"); recorder.frame(0, ms(0)); @@ -1275,7 +1303,7 @@ mod test { let broadcast = moq_net::broadcast::Info::new().produce(); let media = broadcast.create_track("video", None).unwrap(); let demand = media.demand(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut recorder = tracks.track("video", hinted()).unwrap(); recorder.lend(&media); assert_eq!(demand.wanted(), None, "nobody reads the records yet"); @@ -1303,7 +1331,7 @@ mod test { let broadcast = moq_net::broadcast::Info::new().produce(); let media = broadcast.create_track("video", None).unwrap(); let demand = media.demand(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); let mut recorder = tracks.track("video", hinted()).unwrap(); recorder.lend(&media); @@ -1330,14 +1358,14 @@ mod test { fn a_taken_record_track_name_fails_enrollment() { let broadcast = moq_net::broadcast::Info::new().produce(); let _taken = broadcast.create_track("video.hls", None).unwrap(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); assert!(tracks.track("video", Config::default()).is_err()); } #[test] fn enrollment_after_finish_is_refused() { let broadcast = moq_net::broadcast::Info::new().produce(); - let tracks = Tracks::new(&broadcast); + let tracks = Tracks::new(&broadcast, crate::Clock::new()); tracks.finish(); assert!(tracks.track("video", Config::default()).is_err()); } diff --git a/rs/moq-transcode/src/catalog.rs b/rs/moq-transcode/src/catalog.rs index 79d6a64541d..5ed28433e10 100644 --- a/rs/moq-transcode/src/catalog.rs +++ b/rs/moq-transcode/src/catalog.rs @@ -1013,8 +1013,11 @@ mod tests { tokio::time::advance(ago).await; let clock = moq_mux::Clock::at(moq_net::time::Instant::now() - ago, kio::clock::wall() - ago).unwrap(); let mut input = moq_net::broadcast::Info::default().produce(); - let source = - moq_mux::catalog::Producer::new(&mut input, moq_mux::catalog::Config::default().with_clock(clock)).unwrap(); + let source = moq_mux::catalog::Producer::new( + &mut input, + moq_mux::catalog::Config::default().with_clock(clock.clone()), + ) + .unwrap(); let mut output = moq_net::broadcast::Info::default().produce(); let mut derived = moq_mux::catalog::Producer::new(&mut output, moq_mux::catalog::Config::default()).unwrap(); diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index 885349fdfd8..ec80446f92a 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -1147,7 +1147,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 track = broadcast @@ -1169,6 +1169,7 @@ mod tests { let (opens, rx) = tokio::sync::mpsc::unbounded_channel(); let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); // Local: a capture stream is `!Send` on macOS, so each test runs on a `LocalSet`. + let capturing = clock.clone(); let task = tokio::task::spawn_local(async move { let mut source = Opens(rx); // Each subscription opens the camera, so a fixture can idle and restart it. @@ -1178,7 +1179,7 @@ mod tests { ..Capture::default() }; tokio::select! { - res = capture_loop(&mut producer, &mut source, &options, &clock) => res?, + res = capture_loop(&mut producer, &mut source, &options, &capturing) => res?, _ = stopped => {} } producer.finish() From e3850e01154a12aedbd5fb5ca1d719a0ac963c74 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 15:43:18 -0800 Subject: [PATCH 3/3] docs(mux): rewrap the broadcast clock paragraph Co-Authored-By: Claude Opus 5.5 --- doc/lib/rs/moq-mux.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/doc/lib/rs/moq-mux.md b/doc/lib/rs/moq-mux.md index 1ed006443c9..c7f17da59d4 100644 --- a/doc/lib/rs/moq-mux.md +++ b/doc/lib/rs/moq-mux.md @@ -26,9 +26,10 @@ 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. 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 +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 (`/` 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