diff --git a/quest/m1/README.md b/quest/m1/README.md index e81237726e..ae573fe6e2 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -35,7 +35,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Broadcast close](/quest/m1/broadcast-close/README.md) - `close()` is the one way to end a broadcast in every language, a permanent retraction that leaves in-flight tracks alone - [Relay peer set](/quest/m1/relay-peer-set.md) - a wire consumer tells a client hop from a peer hop, and every mesh credential can mark a peer - [CLI import clock](/quest/m1/cli-import-clock.md) - fMP4, TS, and FLV imports publish on the shared broadcast clock across restarts -- [Native clock fixtures](/quest/m1/native-clock-fixtures.md) - CI drives native capture through clock edge cases and asserts the published timestamps - [CLI inspection](/quest/m1/cli-inspect/README.md) - `moq ls` lists what is live and `moq fetch` reads a group over MoQ, and a guide shows how to inspect a relay - [JS caught up](/quest/m1/js-announce-caught-up.md) - @moq/net's announce consumer says when the initial set has landed, like Rust - [Bindings caught up](/quest/m1/announce-live-bindings.md) - moq-ffi, libmoq, and every wrapper yield the same flat announce event, `Live` included diff --git a/quest/m1/cli-import-clock.md b/quest/m1/cli-import-clock.md index ccdf7f764b..0aca2b75f7 100644 --- a/quest/m1/cli-import-clock.md +++ b/quest/m1/cli-import-clock.md @@ -30,5 +30,4 @@ archive playback. Update the import docs. ## Related -- [Native clock fixtures](/quest/m1/native-clock-fixtures.md) - the same scenarios through native capture - [GStreamer clock](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - separate source adapter diff --git a/quest/m1/native-clock-fixtures.md b/quest/m1/native-clock-fixtures.md deleted file mode 100644 index 88b158e028..0000000000 --- a/quest/m1/native-clock-fixtures.md +++ /dev/null @@ -1,19 +0,0 @@ -# [S] Native capture proves the broadcast clock in CI - -## Goal - -Per-PR CI drives the native video and audio capture publishers through clock -edge cases and asserts the published timestamps: simultaneous A/V, a late -first frame, a restart to zero, a restart after idle, a system-wall -adjustment, and retained archive playback. Anything they catch is fixed here. - -## Plan - -Native video already maps the device timeline onto `catalog.clock()` at open, -and native audio stamps arrival on it. The fixtures exercise publisher -integration with a synthetic device source and an injected clock, rather than -only the clock helper. No new clock API or catalog representation. - -## Related - -- [CLI import clock](/quest/m1/cli-import-clock.md) - the same scenarios through `moq import` diff --git a/quest/m2/README.md b/quest/m2/README.md index 1a0426d872..dc2461246f 100644 --- a/quest/m2/README.md +++ b/quest/m2/README.md @@ -82,5 +82,7 @@ upstream release waits in [m4](/quest/m4/README.md). - [Windows capture parity](/quest/m2/capture-windows.md) - system audio and screen cursor capture with a settled app-capture policy - [Linux capture parity](/quest/m2/capture-linux.md) - Wayland window/system-audio capture with explicit display-selection and app-capture limits - [Plan capture ergonomics](/quest/m2/capture-ergonomics.md) - scope independent crop and audio mixing quests +- [Capture clock source](/quest/m2/capture-clock-source.md) - capture publishers stamp on the catalog's clock, with no separate clock to pass, on dev +- [Audio capture time](/quest/m2/audio-capture-time.md) - native audio stamps a buffer's capture instant, not when the driver reads it - [X11 capture transport](/quest/m2/x11-capture-shm.md) - move X11 capture to shared memory and RandR events instead of a per-frame socket copy - [Capture frame buffers](/quest/m2/capture-frame-buffers.md) - stop rebuilding a full-frame buffer every tick in the X11 and Windows backends diff --git a/quest/m2/audio-capture-time.md b/quest/m2/audio-capture-time.md new file mode 100644 index 0000000000..b39ee0c5d7 --- /dev/null +++ b/quest/m2/audio-capture-time.md @@ -0,0 +1,20 @@ +# [S] Native audio stamps acquisition, not arrival + +## Goal + +Native audio capture stamps a buffer at the instant its first sample was +captured, like native video stamps a frame's acquisition. Today the driver +stamps the broadcast clock when it reads the buffer, so audio lands at least +one device buffer (plus callback and queue latency) later than video captured +at the same instant. + +## Plan + +cpal reports `InputCallbackInfo::timestamp().capture` per callback; carry it +through `capture::Samples` and map it onto the broadcast clock once per open, +the way video maps its private capture timeline. Fall back to arrival where a +host reports nothing usable, and keep the reset-on-gap behavior. Measure the +skew before and after on at least one real device, and extend the moq-audio +clock fixtures so a synthetic buffer's capture instant is what publishes. + +Public API: none. Wire: none. diff --git a/quest/m2/capture-clock-source.md b/quest/m2/capture-clock-source.md new file mode 100644 index 0000000000..ddd4d10888 --- /dev/null +++ b/quest/m2/capture-clock-source.md @@ -0,0 +1,20 @@ +# [S] Capture publishers read the catalog's clock + +## Goal + +`moq_video::encode::publish_capture` and `moq_audio::encode::Publication` stamp +on the clock their catalog advertises, with no separate clock to pass. Today +each takes its own `moq_mux::Clock`, and `PublicationOptions::default()` builds +a fresh one, so a caller relying on the default publishes audio against a +mapping the catalog never advertised. `moq import capture` passes +`catalog.clock()` to both, which is the only correct value. + +## Plan + +Drop the `clock` parameter from video `publish_capture` and the `clock` field +from `PublicationOptions`, reading `catalog.clock()` instead. Update moq-cli and +any binding that forwards a clock. The clock fixtures in both crates already +pass the catalog's clock, so they keep grading the same path. + +Public API: breaking in published `moq-video` and `moq-audio`, so it targets +`dev`. Wire: none. diff --git a/quest/m2/js-discontinuity.md b/quest/m2/js-discontinuity.md index a342c79b4e..36e41aada4 100644 --- a/quest/m2/js-discontinuity.md +++ b/quest/m2/js-discontinuity.md @@ -21,4 +21,9 @@ and any other caller over, and keep data tracks skipping a sequence the way Rust's `discontinuity()` does if a JS data-track caller appears. Rust is untouched. +Match Rust's break, too: without an explicit end, `discontinuity()` closes the +group with no cadence-estimated duration marker. Whatever resumes can land +sooner than one estimated frame later (a capture swap), and a marker past it +reads as a rewind to every consumer. + Public API: breaking in published `@moq/hang`, so it targets `dev`. Wire: none. diff --git a/rs/justfile b/rs/justfile index 4469fd9330..bcb1832f27 100644 --- a/rs/justfile +++ b/rs/justfile @@ -495,9 +495,14 @@ check-changed $FILES $TEST="false": # moq-video's and moq-audio's device code is behind `capture`, which is off # by default, so the pass above never compiles it. Only nightly's - # `--all-features` did, and a break there lands on main. + # `--all-features` did, and a break there lands on main. With TEST=true the + # lint rides the test build, as in `check-test`. if [[ "$packages" == "ALL" ]] || just rs _wants-capture "$packages"; then - just rs capture + if [[ "$TEST" == true ]]; then + CARGO_BUILD_WARNINGS=deny just rs capture-test + else + just rs capture + fi fi # moq-relay's io_uring listener is behind `io-uring`, off by default, so the @@ -628,6 +633,13 @@ macos *args: capture *args: cargo clippy --locked -p moq-video -p moq-audio --all-targets --features moq-video/capture,moq-audio/capture {{ args }} -- -D warnings +# The capture publishers' tests (the clock fixtures among them) drive synthetic +# devices, so they need no hardware and run with every PR that reaches either crate. + +# Test the capture feature, which the default test pass skips. +capture-test *args: + just rs test -p moq-video -p moq-audio --features moq-video/capture,moq-audio/capture {{ args }} + # Same idea as `windows`/`macos`, for Android: moq-video's MediaCodec encoder and # decoder, its `Surface::HardwareBuffer` variant, and the `frame::android` module # are all `#[cfg(target_os = "android")]`, so a host-target check compiles none of @@ -886,6 +898,11 @@ test-changed $FILES: *) echo "rs: testing $(just rs _names "$packages")"; just rs test --no-tests=pass $packages ;; esac + # Mirrors `check-changed`: the capture tests are off the default feature set. + if [[ "$packages" == "ALL" ]] || just rs _wants-capture "$packages"; then + just rs capture-test + fi + # Compile and run the `/// ```` examples, which nextest skips. doctest *args: cargo test --locked --doc {{ args }} diff --git a/rs/moq-audio/src/encode/capture.rs b/rs/moq-audio/src/encode/capture.rs index e166e19f1a..33fc3f428f 100644 --- a/rs/moq-audio/src/encode/capture.rs +++ b/rs/moq-audio/src/encode/capture.rs @@ -2041,4 +2041,268 @@ mod tests { assert_eq!(drops.load(Ordering::SeqCst), 1); } + + /// Clock fixtures: the real publication driver, fed by synthetic microphones against a + /// pinned broadcast clock, graded on the timestamps a subscriber reads back. + /// + /// Capture stamps a buffer when the driver reads it, so each expectation is bracketed by + /// the broadcast clock just before the fixture delivers the buffer and just after the + /// published packet is read back. + mod clock { + use std::time::{Duration, Instant, SystemTime}; + + use super::*; + + /// Retain every fixture group, so a slow runner never evicts one before it is read. + const RETAIN: Duration = Duration::from_secs(600); + + type Samples = kio::Queue>; + type Track = moq_mux::container::Consumer; + + struct Fixture { + epoch: Instant, + clock: moq_mux::Clock, + catalog: moq_mux::catalog::Producer, + consumer: moq_net::broadcast::Consumer, + publication: Publication, + task: tokio::task::JoinHandle>, + } + + impl Fixture { + /// Run a publication over `opens` on a broadcast whose clock began `behind` ago, at `wall`. + fn start(behind: Duration, wall: SystemTime, opens: impl IntoIterator) -> Self { + let epoch = Instant::now() + .checked_sub(behind) + .expect("a monotonic clock that far back"); + let clock = moq_mux::Clock::at(epoch, wall).unwrap(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = moq_mux::catalog::Config::default() + .with_clock(clock) + .with_max_age(RETAIN); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); + let mut options = PublicationOptions::default(); + options.encode.track = Some("audio".into()); + // The broadcast's own clock, as `moq import capture` hands it. + options.clock = catalog.clock(); + let (publication, driver) = + Publication::build(broadcast, catalog.clone(), options, Supervisor::exact()).unwrap(); + let task = tokio::spawn(driver.run_with(source(opens, false))); + Self { + epoch, + clock, + catalog, + consumer, + publication, + task, + } + } + + /// Subscribe to the audio track, which is what opens the microphone. + async fn subscribe(&mut self) -> Track { + let track = self + .consumer + .track("audio") + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_age(RETAIN)) + .await + .unwrap(); + wait_for(&mut self.publication, Status::Live).await; + moq_mux::container::Consumer::new( + track, + moq_mux::container::legacy::Wire(moq_mux::container::Kind::Audio), + ) + } + + /// `instant` on the broadcast clock, in microseconds. + fn at(&self, instant: Instant) -> u64 { + u64::try_from(instant.duration_since(self.epoch).as_micros()).unwrap() + } + + /// Deliver one Opus frame of stereo audio and read back the first packet not in + /// `seen`, asserting it is stamped while the driver held the buffer. + async fn deliver(&self, samples: &Samples, track: &mut Track, seen: &[u64]) -> u64 { + let pushed = self.at(Instant::now()); + samples + .try_push(Ok(capture::Samples::plain(vec![0.1; 1920], false))) + .unwrap(); + let published = loop { + let packet = track.read().await.unwrap().expect("a published packet"); + let timestamp = u64::try_from(packet.timestamp.as_micros()).unwrap(); + if !seen.contains(×tamp) { + break timestamp; + } + }; + let read = self.at(Instant::now()); + assert!( + (pushed..=read).contains(&published), + "published {published}us, delivered within {pushed}..={read}us on the broadcast clock" + ); + published + } + + /// Stop the publication, as dropping its last control does. + async fn finish(self) -> (moq_mux::catalog::Producer, moq_net::broadcast::Consumer) { + drop(self.publication); + self.task.await.unwrap().unwrap(); + (self.catalog, self.consumer) + } + } + + /// A microphone whose first buffer arrives long after the broadcast began stamps it + /// then, rather than restarting the broadcast at zero. + #[tokio::test] + async fn a_late_first_buffer_publishes_its_arrival() { + let (samples, input) = stream(None); + let mut fixture = Fixture::start(Duration::from_secs(5), SystemTime::now(), [Open::Stream(input)]); + let mut track = fixture.subscribe().await; + + let published = fixture.deliver(&samples, &mut track, &[]).await; + assert!(published >= 5_000_000, "{published}us restarted the broadcast at zero"); + fixture.finish().await; + } + + /// A device that fails and reopens counts its samples from zero again. The broadcast + /// continues forward from the reopen instead of rewinding to the old epoch. + #[tokio::test] + async fn a_device_restart_continues_forward() { + let (first, failing) = stream(None); + let (second, reopened) = stream(None); + let mut fixture = Fixture::start( + Duration::from_secs(1), + SystemTime::now(), + [Open::Stream(failing), Open::Stream(reopened)], + ); + let mut track = fixture.subscribe().await; + + let before = fixture.deliver(&first, &mut track, &[]).await; + first + .try_push(Err(capture::Failure::retry(Error::Capture("unplugged".into())))) + .unwrap(); + wait_for(&mut fixture.publication, Status::Failed).await; + // The supervisor backs off before reopening, so the restart lands at least that late. + let after = fixture.deliver(&second, &mut track, &[before]).await; + let backoff = u64::try_from(RETRY_MIN.as_micros()).unwrap(); + assert!( + after >= before + backoff, + "{after}us rewound or collapsed the {backoff}us backoff" + ); + fixture.finish().await; + } + + /// Releasing the microphone while nobody listens keeps the broadcast clock running: + /// the buffer after a resume lands after the real idle gap. + #[tokio::test] + async fn a_restart_after_idle_keeps_the_gap() { + let idle = Duration::from_millis(300); + let (first, input) = stream(None); + let (second, resumed) = stream(None); + let mut fixture = Fixture::start( + Duration::from_secs(1), + SystemTime::now(), + [Open::Stream(input), Open::Stream(resumed)], + ); + + let mut track = fixture.subscribe().await; + let before = fixture.deliver(&first, &mut track, &[]).await; + drop(track); + wait_for(&mut fixture.publication, Status::Waiting).await; + + tokio::time::sleep(idle).await; + + let mut track = fixture.subscribe().await; + let after = fixture.deliver(&second, &mut track, &[before]).await; + assert!( + after - before >= u64::try_from(idle.as_micros()).unwrap(), + "the {idle:?} idle gap collapsed to {}us", + after - before + ); + fixture.finish().await; + } + + /// The wall mapping is pinned when the broadcast clock is built. A system clock stepped + /// an hour since then retimes neither the published timestamps nor the advertised mapping. + #[tokio::test] + async fn a_system_wall_adjustment_retimes_nothing() { + // Whole seconds, so the advertised mapping holds it exactly. + let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(); + let wall = SystemTime::UNIX_EPOCH + Duration::from_secs(now.as_secs() - 3600); + let (samples, input) = stream(None); + let mut fixture = Fixture::start(Duration::from_secs(1), wall, [Open::Stream(input)]); + let advertised = fixture.catalog.snapshot().clock; + assert_eq!(advertised, Some(fixture.clock.wall())); + + let mut track = fixture.subscribe().await; + let published = fixture.deliver(&samples, &mut track, &[]).await; + + // The catalog maps to walls at millisecond precision. + let mapped = advertised + .unwrap() + .wall_clock(moq_net::Timestamp::from_micros(published).unwrap()) + .unwrap(); + assert_eq!(mapped, wall + Duration::from_millis(published / 1000)); + assert_eq!(fixture.catalog.snapshot().clock, advertised); + fixture.finish().await; + } + + /// A recording replays what the live edge published: the archive's segment records + /// carry the live timestamps across an idle restart, with the idle gap left in. + #[tokio::test] + async fn retained_archive_playback_keeps_the_live_timestamps() { + let (first, input) = stream(None); + let (second, resumed) = stream(None); + let mut fixture = Fixture::start( + Duration::from_secs(1), + SystemTime::now(), + [Open::Stream(input), Open::Stream(resumed)], + ); + let mut live = Vec::new(); + let mut timeline = None; + + for samples in [&first, &second] { + let mut track = fixture.subscribe().await; + live.push(fixture.deliver(samples, &mut track, &live).await); + // The rendition, and with it the archive, registers once the input is discovered. + if timeline.is_none() { + let section = fixture + .catalog + .snapshot() + .archive + .expect("the audio track enrolls an archive"); + timeline = Some( + moq_mux::timeline::Consumer::<()>::subscribe(&fixture.consumer, §ion) + .await + .unwrap(), + ); + } + drop(track); + wait_for(&mut fixture.publication, Status::Waiting).await; + // Idle past the minimum segment, so each run is archived as its own segment. + tokio::time::sleep(moq_mux::timeline::DEFAULT_DURATION_MIN + Duration::from_millis(100)).await; + } + + let (catalog, _consumer) = fixture.finish().await; + catalog.timeline().finish().unwrap(); + let mut timeline = timeline.unwrap(); + let mut archived = Vec::new(); + while let Some(event) = timeline.next().await.unwrap() { + match event { + moq_mux::timeline::Event::Push { entry, .. } => archived.push(entry), + other => panic!("unexpected timeline event {other:?}"), + } + } + + assert_eq!(archived.len(), live.len(), "one segment per capture run: {archived:?}"); + for (entry, live) in archived.iter().zip(&live) { + // The archive keeps millisecond precision. + assert_eq!(entry.pts.as_micros() / 1000, u128::from(*live / 1000), "{archived:?}"); + assert!(entry.tracks.contains_key("audio"), "{archived:?}"); + } + let first = &archived[0]; + assert!( + archived[1].pts.as_micros() >= first.pts.as_micros() + first.duration.as_micros(), + "the resumed segment overlaps the one before it: {archived:?}" + ); + } + } } diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index ba4af3a2c5..fa18792c30 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -490,6 +490,12 @@ where /// must be a keyframe. An explicit bound before the last ordered video frame /// returns [`InvalidEnd`](super::InvalidEnd) without flushing or closing the group. pub fn cut(&mut self, end: Option) -> crate::Result<()> { + let marker_at = end.or_else(|| self.estimated_end()); + self.close(end, marker_at) + } + + /// Close the current group, ending a video track's group with a duration marker at `marker_at`. + fn close(&mut self, end: Option, marker_at: Option) -> crate::Result<()> { if self.container.kind() == Kind::Video && !self.reordered && let Some((end, previous)) = end.zip(self.previous_timestamp) @@ -503,8 +509,6 @@ where self.estimator.cut(end); self.claim(); - let marker_at = end.or_else(|| self.estimated_end()); - // Tell the timeline where this group's content stops: the duration marker when we // write one, else the caller's bound, else the furthest point we wrote. if let Some(recorder) = self.recorder.as_mut() @@ -603,11 +607,12 @@ where /// an empty payload is data. The next [`write`](Self::write) opens the group after the /// marker and must continue forward from the live edge; it cannot rewind. /// - /// To bound the closing group's final frame, [`cut(end)`](Self::cut) before calling this; - /// the open group is closed either way (an unbounded [`cut`](Self::cut) here is a no-op - /// after yours). + /// To bound the closing group's final frame, [`cut(end)`](Self::cut) before calling this. + /// Otherwise the open group closes without a duration marker: what resumes may land sooner + /// than one estimated frame later (a capture that reopens at once), and a guessed end past + /// it would read as a rewind to every consumer. pub fn discontinuity(&mut self) -> crate::Result<()> { - self.cut(None)?; + self.close(None, None)?; // Nothing is measured across the break: the frames still open on this side have no end, and // the gap to the far side is not a frame duration. self.estimator.discontinuity(); @@ -1488,4 +1493,28 @@ mod tests { let groups = collect_payloads(consumer).await; assert_eq!(groups.last().unwrap().last(), Some(&(1_060_000, 0))); } + + /// A capture that reopens at once resumes sooner than one frame after the break. Guessing + /// the closing group's end from its cadence would put a duration marker past the resumed + /// keyframe, which a consumer reads as a rewind and refuses. + #[tokio::test] + async fn a_prompt_resume_after_a_discontinuity_is_not_a_rewind() { + let track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video)); + let subscriber = track.subscribe(replay()); + let mut producer = Producer::new(track, Container::Legacy(crate::container::Kind::Video)); + producer.write(frame(1_000_000, true)).unwrap(); + producer.write(frame(1_040_000, false)).unwrap(); + producer.discontinuity().unwrap(); + // Resumed 10ms later, inside the 40ms cadence measured before the break. + producer.write(frame(1_050_000, true)).unwrap(); + producer.finish().unwrap(); + + let mut consumer = + crate::container::Consumer::new(subscriber, Container::Legacy(crate::container::Kind::Video)); + let mut timestamps = Vec::new(); + while let Some(frame) = consumer.read().await.unwrap() { + timestamps.push(frame.timestamp.as_micros()); + } + assert_eq!(timestamps, [1_000_000, 1_040_000, 1_050_000]); + } } diff --git a/rs/moq-video/src/capture/channel.rs b/rs/moq-video/src/capture/channel.rs index 1a56b944bb..9e32692bb4 100644 --- a/rs/moq-video/src/capture/channel.rs +++ b/rs/moq-video/src/capture/channel.rs @@ -27,18 +27,29 @@ pub(super) struct FrameChannel { struct State { frame: Option, #[cfg(any(target_os = "linux", target_os = "windows", test))] - native_anchor: Option<(Timestamp, Timestamp)>, + native: Option, closed: bool, error: Option, } +/// How a device's own timeline maps onto this stream's. +#[cfg(any(target_os = "linux", target_os = "windows", test))] +#[derive(Clone, Copy)] +struct Native { + /// A device timestamp and the local time it was anchored to. + source: Timestamp, + local: Timestamp, + /// The previous device timestamp, which the next must exceed to keep the anchor. + last: Timestamp, +} + impl FrameChannel { pub(super) fn new() -> Arc { Arc::new(Self { state: Mutex::new(State { frame: None, #[cfg(any(target_os = "linux", target_os = "windows", test))] - native_anchor: None, + native: None, closed: false, error: None, }), @@ -53,7 +64,7 @@ impl FrameChannel { self.push_at(frame, Instant::now()); } - fn push_at(&self, surface: Surface, captured: Instant) { + pub(super) fn push_at(&self, surface: Surface, captured: Instant) { let micros = captured.saturating_duration_since(self.epoch).as_micros(); let micros = u64::try_from(micros).unwrap_or(u64::MAX); let frame = Frame::new(surface, Timestamp::from_micros(micros).expect("capture timestamp fits")); @@ -61,9 +72,13 @@ impl FrameChannel { } /// Map a device-local timestamp into this stream's private timeline. The - /// source epoch never escapes: its first sample is anchored to acquisition. + /// source epoch never escapes: its first sample is anchored to arrival. /// Only the blocking-device pump feeds native timestamps, so it is gated like /// `pump` plus `cfg(test)` for the mapping test below. + /// + /// A device timeline that steps back or stalls (a driver restarting its clock + /// at zero, or one reporting a constant) re-anchors that sample to arrival, so + /// the stream never rewinds or repeats a timestamp it already delivered. #[cfg(any(target_os = "linux", target_os = "windows", test))] pub(super) fn push_native(&self, surface: Surface, source: Timestamp) { let local = self.now(); @@ -71,11 +86,19 @@ impl FrameChannel { if state.closed { return; } - let (source_anchor, local_anchor) = *state.native_anchor.get_or_insert((source, local)); + let anchor = match state.native { + Some(native) if source > native.last => native, + _ => Native { + source, + local, + last: source, + }, + }; let timestamp = source - .checked_sub(source_anchor) - .and_then(|elapsed| local_anchor.checked_add(elapsed)) + .checked_sub(anchor.source) + .and_then(|elapsed| anchor.local.checked_add(elapsed)) .unwrap_or(local); + state.native = Some(Native { last: source, ..anchor }); state.frame = Some(Frame::new(surface, timestamp)); drop(state); self.notify.notify_one(); @@ -256,4 +279,42 @@ mod tests { assert_eq!(second.as_micros() - first.as_micros(), 33_367); assert!(first.as_micros() < 9_000_000); } + + /// A device clock that restarts at zero mid-stream must not rewind the stream's + /// timeline: the sample re-anchors to arrival and the device's spacing resumes from there. + #[tokio::test] + async fn native_timestamps_reanchor_when_the_device_clock_restarts() { + let chan = FrameChannel::new(); + let us = |micros| Timestamp::from_micros(micros).unwrap(); + + chan.push_native(frame(1), us(0)); + chan.recv().await.unwrap().unwrap(); + // Real time passes with the device clock, so the mapping stays at or behind arrival. + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + chan.push_native(frame(2), us(20_000)); + let before = chan.recv().await.unwrap().unwrap().timestamp; + + chan.push_native(frame(3), us(0)); + let restarted = chan.recv().await.unwrap().unwrap().timestamp; + assert!(restarted >= before, "{restarted:?} rewound behind {before:?}"); + + chan.push_native(frame(4), us(33_000)); + let next = chan.recv().await.unwrap().unwrap().timestamp; + assert_eq!(next.as_micros() - restarted.as_micros(), 33_000); + } + + /// A driver that reports one constant timestamp must not stamp every frame + /// identically: each sample falls back to its arrival. + #[tokio::test] + async fn a_stalled_device_clock_falls_back_to_arrival() { + let chan = FrameChannel::new(); + let constant = Timestamp::from_micros(0).unwrap(); + + chan.push_native(frame(1), constant); + let first = chan.recv().await.unwrap().unwrap().timestamp; + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + chan.push_native(frame(2), constant); + let second = chan.recv().await.unwrap().unwrap().timestamp; + assert!(second > first, "a stalled device clock repeated {first:?}"); + } } diff --git a/rs/moq-video/src/capture/mod.rs b/rs/moq-video/src/capture/mod.rs index 25fe9784c5..f39ff45389 100644 --- a/rs/moq-video/src/capture/mod.rs +++ b/rs/moq-video/src/capture/mod.rs @@ -364,6 +364,46 @@ impl Stream { } } +/// A test-driven capture device: fixtures push frames at explicit acquisition +/// instants or device timestamps, standing in for a camera behind a [`Stream`]. +#[cfg(test)] +pub(crate) struct Synthetic { + chan: Arc, +} + +#[cfg(test)] +impl Synthetic { + /// Open a synthetic device with the given geometry, returning it and the stream reading it. + pub(crate) fn open(size: crate::Size, framerate: Rate) -> (Self, Stream) { + let chan = FrameChannel::new(); + let stream = Stream::new( + chan.clone(), + size.width, + size.height, + Some(framerate), + "synthetic".to_string(), + None, + Box::new(()), + ); + (Self { chan }, stream) + } + + /// Deliver a frame acquired at `captured`. + pub(crate) fn push_at(&self, surface: crate::frame::Surface, captured: std::time::Instant) { + self.chan.push_at(surface, captured); + } + + /// Deliver a frame stamped on the device's own timeline, like V4L2 and Media Foundation. + pub(crate) fn push_native(&self, surface: crate::frame::Surface, source: moq_net::Timestamp) { + self.chan.push_native(surface, source); + } + + /// End the stream, as a device that stops delivering does. + pub(crate) fn close(&self) { + self.chan.close(); + } +} + /// Open the capture source described by `config`. pub async fn open(config: &Config) -> Result { match &config.source { diff --git a/rs/moq-video/src/encode/producer.rs b/rs/moq-video/src/encode/producer.rs index c6daa5c500..86353d907b 100644 --- a/rs/moq-video/src/encode/producer.rs +++ b/rs/moq-video/src/encode/producer.rs @@ -322,7 +322,7 @@ pub async fn publish_capture( let mut producer = Producer::new(broadcast, catalog, rendition)?; let demand = producer.demand(); - let result = capture_loop(&mut producer, &demand, &capture, &encode, &clock).await; + let result = capture_loop(&mut producer, &demand, &mut DeviceSource, &capture, &encode, &clock).await; // This runs only when the loop ends on its own (the track is usually already // going away by then); a Ctrl+C cancels the future before this point, since @@ -358,6 +358,23 @@ fn assert_publish_capture_send( is_send(&publish_capture(broadcast, catalog, capture, encode, clock)); } +/// Where the capture loop opens its camera. Kept apart from the device backends so +/// the clock fixtures can drive the real loop from a synthetic source. +#[cfg(feature = "capture")] +trait CaptureSource { + async fn open(&mut self, config: &capture::Config) -> Result; +} + +#[cfg(feature = "capture")] +struct DeviceSource; + +#[cfg(feature = "capture")] +impl CaptureSource for DeviceSource { + async fn open(&mut self, config: &capture::Config) -> Result { + capture::open(config).await + } +} + /// The live rate control state: the estimate source paired with the policy tracking /// it. `None` once it has *retired*, which is the only thing absence means now that /// every encoder has a share to read: an allocator with nothing to divide grants @@ -473,9 +490,10 @@ async fn wait_capture( /// so their joins return promptly unless the underlying device or encoder is /// itself wedged. #[cfg(feature = "capture")] -async fn capture_loop( +async fn capture_loop( producer: &mut Producer, demand: &moq_net::track::Demand, + source: &mut S, capture: &capture::Config, encode: &Options, clock: &moq_mux::Clock, @@ -495,7 +513,7 @@ async fn capture_loop( } // Open the camera and an encoder sized to its negotiated mode. - let Some(mut camera) = wait_capture(producer, demand, capture::open(capture)).await? else { + let Some(mut camera) = wait_capture(producer, demand, source.open(capture)).await? else { continue; }; // Capture timestamps use a private monotonic timeline. Sample both clocks @@ -858,4 +876,343 @@ mod tests { assert_eq!(config.coded_width, Some(320)); assert_eq!(config.coded_height, Some(240)); } + + /// Clock fixtures: the real capture loop, fed by a synthetic camera against a pinned + /// broadcast clock, graded on the timestamps a subscriber reads back. + /// + /// Each expectation is the acquisition instant measured on the broadcast clock. The loop + /// samples the broadcast clock and then the camera's timeline when it opens a camera, so a + /// published timestamp may land up to `SAMPLING` early, never late. + #[cfg(all(feature = "capture", feature = "openh264"))] + mod clock { + use std::time::{Duration, Instant, SystemTime}; + + use super::*; + use crate::capture::Synthetic; + + /// How early a mapped timestamp may land: the gap between the loop's two clock samples. + const SAMPLING: Duration = Duration::from_millis(250); + /// Rounding slack on the late side: each clock reading truncates to a microsecond. + const ROUNDING: u64 = 2; + /// Retain every fixture group, so a slow runner never evicts one before it is read. + const RETAIN: Duration = Duration::from_secs(600); + + /// Hands the loop one fixture-supplied stream per camera open. + struct Opens(tokio::sync::mpsc::UnboundedReceiver); + + impl CaptureSource for Opens { + async fn open(&mut self, _config: &capture::Config) -> Result { + self.0 + .recv() + .await + .ok_or_else(|| Error::SourceUnavailable("the fixture stopped opening cameras".to_string())) + } + } + + struct Fixture { + epoch: Instant, + clock: moq_mux::Clock, + catalog: moq_mux::catalog::Producer, + consumer: moq_net::broadcast::Consumer, + _broadcast: moq_net::broadcast::Producer, + opens: tokio::sync::mpsc::UnboundedSender, + stop: Option>, + task: tokio::task::JoinHandle>, + } + + impl Fixture { + /// Start the capture loop on a broadcast whose clock began `behind` ago, at `wall`. + async fn start(behind: Duration, wall: SystemTime) -> Self { + let epoch = Instant::now() + .checked_sub(behind) + .expect("a monotonic clock that far back"); + let clock = moq_mux::Clock::at(epoch, wall).unwrap(); + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let config = moq_mux::catalog::Config::default() + .with_clock(clock) + .with_max_age(RETAIN); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap(); + let track = broadcast + .create_track( + "video", + catalog.track_info(hang::catalog::PRIORITY.video).with_max_age(RETAIN), + ) + .unwrap(); + + let mut probe = Config::new(320, 240, crate::Rate::new(30, 1).unwrap()); + probe.kind = encoder::Kind::Software; + let mut producer = Producer::with_track(track, catalog.clone(), probe.probe().await.unwrap()).unwrap(); + let demand = producer.demand(); + + let (opens, rx) = tokio::sync::mpsc::unbounded_channel(); + let (stop, stopped) = tokio::sync::oneshot::channel::<()>(); + let task = tokio::spawn(async move { + let mut source = Opens(rx); + let options = Options { + kind: encoder::Kind::Software, + ..Options::default() + }; + let config = capture::Config::default(); + tokio::select! { + res = capture_loop(&mut producer, &demand, &mut source, &config, &options, &clock) => res?, + _ = stopped => {} + } + producer.finish() + }); + + Self { + epoch, + clock, + catalog, + consumer, + _broadcast: broadcast, + opens, + stop: Some(stop), + task, + } + } + + /// Subscribe to the video track, which is what opens the camera. + async fn subscribe(&self) -> moq_mux::container::Consumer { + let snapshot = self.catalog.snapshot(); + let (name, rendition) = snapshot.video.renditions.iter().next().expect("the probed rendition"); + let container = moq_mux::catalog::hang::Container::try_from(rendition).unwrap(); + let track = self + .consumer + .track(name) + .unwrap() + .subscribe(moq_net::track::Subscription::default().with_max_age(RETAIN)) + .await + .unwrap(); + moq_mux::container::Consumer::new(track, container) + } + + /// Plug in the camera the loop opens next. + fn camera(&self) -> Synthetic { + let (camera, stream) = Synthetic::open(crate::Size::new(320, 240), crate::Rate::new(30, 1).unwrap()); + self.opens.send(stream).unwrap(); + camera + } + + /// `instant` on the broadcast clock, in microseconds. + fn at(&self, instant: Instant) -> u64 { + u64::try_from(instant.duration_since(self.epoch).as_micros()).unwrap() + } + + /// Stop the loop and finalize the track, as a clean end of capture does. + async fn finish(mut self) -> (moq_mux::catalog::Producer, moq_net::broadcast::Consumer) { + let _ = self.stop.take().expect("finished once").send(()); + self.task.await.unwrap().unwrap(); + (self.catalog, self.consumer) + } + + /// A frame acquired at `captured` publishes at that instant on the broadcast clock. + fn assert_acquired(&self, published: u64, captured: Instant) { + let exact = self.at(captured); + let early = u64::try_from(SAMPLING.as_micros()).unwrap(); + assert!( + published + early >= exact && published <= exact + ROUNDING, + "published {published}us, acquired at {exact}us on the broadcast clock" + ); + } + } + + fn surface() -> crate::frame::Surface { + crate::frame::Surface::I420(crate::frame::I420 { + width: 320, + height: 240, + data: vec![0x80; 320 * 240 * 3 / 2], + color: None, + }) + } + + fn us(micros: u64) -> Timestamp { + Timestamp::from_micros(micros).unwrap() + } + + async fn read(track: &mut moq_mux::container::Consumer) -> u64 { + let frame = track.read().await.unwrap().expect("a published frame"); + u64::try_from(frame.timestamp.as_micros()).unwrap() + } + + /// Read the next frame not already in `seen`: a resubscription replays retained groups first. + async fn read_new( + track: &mut moq_mux::container::Consumer, + seen: &[u64], + ) -> u64 { + loop { + let timestamp = read(track).await; + if !seen.contains(×tamp) { + return timestamp; + } + } + } + + /// A camera whose first frame arrives long after the broadcast began stamps it at its + /// acquisition: not zero, and not the later instant the loop dequeued it. + #[tokio::test] + async fn a_late_first_frame_publishes_its_acquisition() { + let fixture = Fixture::start(Duration::from_secs(5), SystemTime::now()).await; + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + + let captured = Instant::now(); + // Delivered well after acquisition: dequeue time must not leak into the timestamp. + tokio::time::sleep(Duration::from_millis(50)).await; + camera.push_at(surface(), captured); + let published = read(&mut track).await; + + assert!(published >= 4_000_000, "{published}us restarted the broadcast at zero"); + fixture.assert_acquired(published, captured); + fixture.finish().await; + } + + /// A device clock that restarts at zero, mid-stream or across a reopen, continues the + /// broadcast forward with the device's spacing instead of rewinding it. + #[tokio::test] + async fn a_device_clock_restart_continues_forward() { + let fixture = Fixture::start(Duration::from_secs(1), SystemTime::now()).await; + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + + // The device numbers from zero, and real time keeps pace with it. + camera.push_native(surface(), us(0)); + let first = read(&mut track).await; + tokio::time::sleep(Duration::from_millis(40)).await; + camera.push_native(surface(), us(40_000)); + let second = read(&mut track).await; + assert_eq!(second - first, 40_000, "the device's spacing survives"); + + // The device restarts its clock without the stream ending. + camera.push_native(surface(), us(0)); + let restarted = read(&mut track).await; + assert!(restarted >= second, "{restarted}us rewound behind {second}us"); + tokio::time::sleep(Duration::from_millis(40)).await; + camera.push_native(surface(), us(40_000)); + let resumed = read(&mut track).await; + assert_eq!(resumed - restarted, 40_000, "the device's spacing resumes"); + + // The device goes away and comes back numbering from zero again. + camera.close(); + let camera = fixture.camera(); + let pushed = Instant::now(); + camera.push_native(surface(), us(0)); + let reopened = read(&mut track).await; + let arrived = fixture.at(Instant::now()); + assert!(reopened >= resumed, "{reopened}us rewound across the reopen"); + let early = u64::try_from(SAMPLING.as_micros()).unwrap(); + assert!(reopened + early >= fixture.at(pushed) && reopened <= arrived + ROUNDING); + fixture.finish().await; + } + + /// Releasing the camera while nobody watches keeps the broadcast clock running: the + /// frame after a resume lands after the real idle gap, at its own acquisition. + #[tokio::test] + async fn a_restart_after_idle_keeps_the_gap() { + let idle = Duration::from_millis(300); + let fixture = Fixture::start(Duration::from_secs(1), SystemTime::now()).await; + + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + let captured = Instant::now(); + camera.push_at(surface(), captured); + let before = read(&mut track).await; + fixture.assert_acquired(before, captured); + drop(track); + drop(camera); + + tokio::time::sleep(idle).await; + + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + let captured = Instant::now(); + camera.push_at(surface(), captured); + let after = read_new(&mut track, &[before]).await; + fixture.assert_acquired(after, captured); + assert!( + after - before >= u64::try_from(idle.as_micros()).unwrap(), + "the {idle:?} idle gap collapsed to {}us", + after - before + ); + fixture.finish().await; + } + + /// The wall mapping is pinned when the broadcast clock is built. A system clock stepped + /// an hour since then retimes neither the published timestamps nor the advertised mapping. + #[tokio::test] + async fn a_system_wall_adjustment_retimes_nothing() { + // Whole seconds, so the advertised mapping holds it exactly. + let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(); + let wall = SystemTime::UNIX_EPOCH + Duration::from_secs(now.as_secs() - 3600); + let fixture = Fixture::start(Duration::from_secs(1), wall).await; + let advertised = fixture.catalog.snapshot().clock; + assert_eq!(advertised, Some(fixture.clock.wall())); + + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + let captured = Instant::now(); + camera.push_at(surface(), captured); + let published = read(&mut track).await; + + // Timestamps follow the monotonic epoch and map to walls under the pinned mapping. + fixture.assert_acquired(published, captured); + let mapped = advertised.unwrap().wall_clock(us(published)).unwrap(); + // The catalog maps to walls at millisecond precision. + assert_eq!(mapped, wall + Duration::from_millis(published / 1000)); + assert_eq!(fixture.catalog.snapshot().clock, advertised); + fixture.finish().await; + } + + /// A recording replays what the live edge published: the archive's segment records + /// carry the live timestamps across an idle restart, with the idle gap left in. + #[tokio::test] + async fn retained_archive_playback_keeps_the_live_timestamps() { + let fixture = Fixture::start(Duration::from_secs(1), SystemTime::now()).await; + let section = fixture + .catalog + .snapshot() + .archive + .expect("the video track enrolls an archive"); + let mut timeline = moq_mux::timeline::Consumer::<()>::subscribe(&fixture.consumer, §ion) + .await + .unwrap(); + + let mut live = Vec::new(); + for _ in 0..2 { + let mut track = fixture.subscribe().await; + let camera = fixture.camera(); + let captured = Instant::now(); + camera.push_at(surface(), captured); + let published = read_new(&mut track, &live).await; + fixture.assert_acquired(published, captured); + live.push(published); + drop(track); + // Idle past the minimum segment, so each run is archived as its own segment. + tokio::time::sleep(moq_mux::timeline::DEFAULT_DURATION_MIN + Duration::from_millis(100)).await; + } + + let (catalog, _consumer) = fixture.finish().await; + catalog.timeline().finish().unwrap(); + let mut archived = Vec::new(); + while let Some(event) = timeline.next().await.unwrap() { + match event { + moq_mux::timeline::Event::Push { entry, .. } => archived.push(entry), + other => panic!("unexpected timeline event {other:?}"), + } + } + + assert_eq!(archived.len(), live.len(), "one segment per capture run: {archived:?}"); + for (entry, live) in archived.iter().zip(&live) { + // The archive keeps millisecond precision. + assert_eq!(entry.pts.as_micros() / 1000, u128::from(*live / 1000), "{archived:?}"); + assert!(entry.tracks.contains_key("video"), "{archived:?}"); + } + let first = &archived[0]; + assert!( + archived[1].pts.as_micros() >= first.pts.as_micros() + first.duration.as_micros(), + "the resumed segment overlaps the one before it: {archived:?}" + ); + } + } }