From ea3be7e4ff8a65842d987596b973185172ca4a0b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 10:26:08 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m0/broadcast-sync.md Quest-Claim: 71cde651b74cb86e From 3ecb031265fc891edabf8781a84ab1f29221b5bc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 10:57:20 -0800 Subject: [PATCH 2/7] feat(scene): broadcast sources play in sync on the publisher's timestamps Co-Authored-By: Claude Opus 5.5 --- README.md | 14 +- quest/m0/README.md | 1 - quest/m0/broadcast-sync.md | 34 ----- quest/m0/sync-clap.md | 4 - rs/moq-scene/src/lib.rs | 1 + rs/moq-scene/src/playout.rs | 266 +++++++++++++++++++++++++++++++++ rs/moq-scene/src/source.rs | 289 ++++++++++++++++++++++++++++++++---- rs/moq-studio/src/ui.rs | 7 + 8 files changed, 546 insertions(+), 70 deletions(-) delete mode 100644 quest/m0/broadcast-sync.md create mode 100644 rs/moq-scene/src/playout.rs diff --git a/README.md b/README.md index a94fa40a..20fd1e68 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,16 @@ failover replaces the first relay's copy once that copy ends. It plays the best video and audio its catalog lists, switches when the catalog changes, and comes back on its own when the publisher restarts. +A broadcast source, a guest's included, plays on its publisher's timestamps, +so its sound and picture stay in sync through network jitter and stalls. Its +audio plays as far behind what arrives as the network's jitter needs, measured +as it goes, the same delay `moq play --delay auto` holds, up to 2 s. Each +picture waits for the sound stamped with it, and the canvas shows the last one +meanwhile. A stall longer than that delay pauses both and starts them again +together, and a burst after one skips ahead to the delay. Without audio, the +picture plays 100 ms behind. Sources shows how far behind each broadcast +plays. + ### Audio mixes Audio from every source is mixed whether or not its scene is live. A @@ -170,7 +180,9 @@ studio remembers for you rather than in the scene file. Use headphones: a microphone that hears the speakers feeds the mix back into itself. A source's **delay** holds it back in every mix, up to 2 s, so a microphone -lines up with a camera that lags it. +lines up with a camera that lags it. A broadcast's picture already follows its +sound, so its delay holds the sound back from its own picture: use it only for +a publisher whose sound leads their picture. Each mix is advanced on the program clock, and each output reads its mix on its own, so an output that falls behind skips ahead in its own audio rather diff --git a/quest/m0/README.md b/quest/m0/README.md index 4eea230c..9e6b718e 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -91,7 +91,6 @@ repository. - [Studio on macOS with capture granted](/quest/m0/app-macos.md) - camera, screen, and audio capture run on a Mac, not only Linux - [Studio on Windows](/quest/m0/app-windows.md) - the studio runs on Windows, not only Linux - [Audio source tracks](/quest/m0/audio-source-tracks.md) - microphones and system audio get their own tracks, for isolated recording and replay -- [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - subscribed broadcasts and guests play audio and video on the publisher's timestamps, with `moq play`'s delay - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker - [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request - [Keyframes and stingers](/quest/m0/animation.md) - keyframed item properties with easing, and stinger transitions diff --git a/quest/m0/broadcast-sync.md b/quest/m0/broadcast-sync.md deleted file mode 100644 index 8c1646b2..00000000 --- a/quest/m0/broadcast-sync.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] Broadcast sources play in sync - -## Goal - -A subscribed MoQ broadcast in `moq-studio`, guests included, plays its audio -and video on the publisher's timestamps, so lip sync holds through network -jitter and stalls. Today each track plays at its own live edge, so a stall -can leave audio and video apart. - -## Plan - -Decided 2026-10-26 (maintainer): - -- **Scope:** broadcast sources only. Local devices keep the manual - per-source `delay`, which [Clap-to-measure](/quest/m0/sync-clap.md) sets - from a clap. -- **Delay:** reuse `moq play`'s timing, the `--delay auto` target from - moq-audio's jitter estimate (`rs/moq-audio/src/jitter.rs`, - `doc/concept/audio-jitter.md`), rather than a second algorithm or a manual - setting. As in `moq play`, the estimate is the broadcast's audio track's, - and video follows it rather than keeping its own, so one delay governs the - whole broadcast. A broadcast with no audio keeps video's own short delay. - Show each source's delay in the UI. - -Use the timing `moq play` shares rather than copying it: moq-mux's -`timeline` module, pinned here (decided 2026-11-02, maintainer). The video -side holds frames until their audio-aligned time, and a source still repeats -its last frame rather than stalling the canvas. This replaces the live-edge behavior the studio app PR chose, where -replaying a publisher's timestamp gaps pushed audio behind its own video. Here -video waits for audio instead. Update `README.md`. - -## Related - -- [Audio jitter target](https://github.com/moq-dev/moq/blob/main/quest/m1/audio-jitter-target/README.md) - the delay estimate this reuses diff --git a/quest/m0/sync-clap.md b/quest/m0/sync-clap.md index 915f631b..cdb20d9f 100644 --- a/quest/m0/sync-clap.md +++ b/quest/m0/sync-clap.md @@ -22,7 +22,3 @@ The studio sees a source's audio only through its strip's level meter, read once a frame, which is too coarse to time a clap. Timing it needs the source's samples, or an onset, with their position on the bus clock, which is moq-audio work: land it there first and pin it. - -## Related - -- [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - broadcast sources align on the publisher's timestamps instead diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 4cbd9693..0e72c245 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -36,6 +36,7 @@ mod edit; mod effect; mod lut; mod overlay; +mod playout; pub mod publish; pub mod replay; pub mod source; diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs new file mode 100644 index 00000000..66a5ed94 --- /dev/null +++ b/rs/moq-scene/src/playout.rs @@ -0,0 +1,266 @@ +//! When a broadcast source's audio sounds and its pictures show: on its +//! publisher's timestamps, so they stay in sync through jitter and stalls. +//! +//! A broadcast's audio and video share one [`timeline`] clock, as they do in +//! `moq play`: audio owns the clock's anchor while it plays, and video shows +//! each picture once the clock says it is due. Without audio, video paces +//! itself [`VIDEO_DELAY`] behind its arrivals. +//! +//! `moq play` holds its jitter buffer in the speaker. The bus holds only +//! [`BUS_LATENCY`] of a source, so a [`Speaker`] holds the rest: each frame +//! waits until it is due, the playout target moq-audio's decoder estimates +//! after it arrives, and is written the bus's latency ahead of when it sounds. + +use std::time::Duration; + +use kio::clock::Instant; +use moq_mux::timeline; + +/// How far a broadcast without audio plays behind its arrivals, as under +/// `moq play --delay auto`. With audio, video follows the audio's target. +pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); + +/// The longest playout target, the most lateness the jitter estimate can +/// represent, and so how long a stalled group is waited on before it is +/// skipped, as under `moq play --delay auto`. A shorter wait would cap the +/// estimate at it, since a skipped group is never measured. +pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); + +/// What the bus holds between a channel write and the mix that plays it. +/// +/// moq-audio's `bus::LATENCY`, which is private; `a_write_sounds_the_bus_latency_later` +/// pins the two together. +pub(crate) const BUS_LATENCY: Duration = Duration::from_millis(50); + +/// How far over its target the queue may run before an arrival is skipped to +/// land back on it, as `moq play`'s speaker allows: one frame of cadence. +const SLACK: Duration = Duration::from_millis(20); + +/// How far apart two frames' stamps may be and still continue each other: a +/// millisecond-stamped ingest rounds each one. +const CONTIGUOUS: Duration = Duration::from_millis(1); + +/// Places one broadcast audio track's frames on the bus clock, and reports +/// where they sound to the broadcast's clock, which video follows. +/// +/// Dropping it hands the clock's anchor back to video. +pub(crate) struct Speaker { + clock: timeline::Producer, + /// The last frame placed: when it finishes sounding, and the media time it + /// ends at. + tail: Option<(Instant, Duration)>, +} + +/// What to do with a frame [`Speaker::place`] was given. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Place { + /// Write it to the bus at `at`. `moved` says the clock's anchor moved, so + /// every picture waiting for its moment is due at another one now. + Write { at: Instant, moved: bool }, + /// Drop it: more is queued than the target needs, after a burst or a + /// target that fell, so the media skips ahead to land back on the target. + Skip, +} + +impl Speaker { + /// Place audio on `clock`, taking its anchor from video on the first frame. + pub(crate) fn new(clock: timeline::Producer) -> Self { + Self { clock, tail: None } + } + + /// Place a decoded frame covering `duration` of media from `start`, + /// arriving at `now`, given the playout `target` moq-audio estimated. + /// + /// A frame continues the one before, after silence for any hole in the + /// media between them. A frame that arrives too late for that, because the + /// network stalled longer than the target, re-buffers instead: it sounds the + /// whole target after it arrived, and the clock re-pins on it, so video + /// waits along with the sound rather than running ahead of it. + pub(crate) fn place(&mut self, start: Duration, duration: Duration, target: Duration, now: Instant) -> Place { + let target = target.min(MAX_DELAY); + let end = start.saturating_add(duration); + // The bus rides out a write up to half its latency late without a dropout. + let follows = self.tail.and_then(|(sounds, media)| { + let begins = sounds + start.saturating_sub(media); + let rewound = start.saturating_add(CONTIGUOUS) < media; + let dry = begins < now + BUS_LATENCY / 2; + (!rewound && !dry).then_some((begins, sounds)) + }); + let begins = match follows { + Some((begins, sounds)) => { + if begins.saturating_duration_since(now) > target + BUS_LATENCY + SLACK { + self.tail = Some((sounds, end)); + return Place::Skip; + } + begins + } + None => { + let begins = now + target + BUS_LATENCY; + match self.tail { + Some((sounds, _)) => { + self.clock.audio_restarted(); + // Never over audio still queued, after a rewind. + begins.max(sounds) + } + None => begins, + } + } + }; + let sounds = begins + duration; + self.tail = Some((sounds, end)); + let audio = timeline::Audio { + end, + buffered: sounds.saturating_duration_since(now), + target, + }; + let moved = self.clock.audio(audio, now); + Place::Write { + at: begins - BUS_LATENCY, + moved, + } + } +} + +impl Drop for Speaker { + fn drop(&mut self) { + self.clock.audio_stopped(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const FRAME: Duration = Duration::from_millis(20); + const TARGET: Duration = Duration::from_millis(100); + + fn at(millis: u64) -> Instant { + Instant::ZERO + Duration::from_secs(10) + Duration::from_millis(millis) + } + + fn ms(millis: u64) -> Duration { + Duration::from_millis(millis) + } + + fn stamp(millis: u64) -> moq_net::Timestamp { + moq_net::Timestamp::from_millis(millis).unwrap() + } + + fn written(place: Place) -> Instant { + match place { + Place::Write { at, .. } => at, + Place::Skip => panic!("skipped"), + } + } + + /// The first frame waits out the target, later ones follow it on their + /// stamps however unevenly they arrive within it, and a picture is due when + /// the sound stamped with it plays: the bus's latency after its write. + #[test] + fn video_is_due_when_its_audio_sounds() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let mut speaker = Speaker::new(clock.clone()); + assert_eq!(written(speaker.place(ms(1_000), FRAME, TARGET, at(0))), at(100)); + assert_eq!(clock.due(stamp(1_000)), Some(at(150))); + // Late within the target, and early: both keep their place. + assert_eq!(written(speaker.place(ms(1_020), FRAME, TARGET, at(90))), at(120)); + assert_eq!(written(speaker.place(ms(1_040), FRAME, TARGET, at(91))), at(140)); + assert_eq!(clock.due(stamp(1_040)), Some(at(190))); + assert_eq!(clock.consume().target(), Some(TARGET)); + + // A stopped track hands the clock back to video. + drop(speaker); + assert_eq!(clock.consume().target(), None); + } + + /// A stall longer than the target re-buffers: the next frame sounds the + /// target after it arrived, and the pictures stamped with it wait as long. + #[test] + fn a_stall_holds_video_back_with_the_sound() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let mut speaker = Speaker::new(clock.clone()); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + let place = speaker.place(ms(1_020), FRAME, TARGET, at(500)); + assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(600))); + assert_eq!(clock.due(stamp(1_020)), Some(at(650))); + } + + /// Lost media plays as the silence it was, so the pictures after it keep + /// their moments; a hole past the target skips instead, and the clock + /// re-pins on the frame after it. + #[test] + fn a_hole_plays_as_silence_up_to_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let mut speaker = Speaker::new(clock.clone()); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + let place = speaker.place(ms(1_060), FRAME, TARGET, at(40)); + assert_eq!( + place, + Place::Write { + at: at(160), + moved: false + } + ); + + // 1 s lost, with only the target queued: skipped back onto the target. + assert_eq!(speaker.place(ms(2_080), FRAME, TARGET, at(60)), Place::Skip); + let place = speaker.place(ms(2_100), FRAME, TARGET, at(61)); + assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(180))); + assert_eq!(clock.due(stamp(2_100)), Some(at(230))); + } + + /// A burst queues no more than the target and a frame of slack: the rest + /// is skipped, and play goes on from what arrives next. + #[test] + fn a_burst_skips_past_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let mut speaker = Speaker::new(clock.clone()); + let placed = (0..10) + .filter_map( + |index| match speaker.place(ms(1_000 + index * 20), FRAME, TARGET, at(0)) { + Place::Write { at, .. } => Some(at), + Place::Skip => None, + }, + ) + .collect::>(); + assert_eq!(placed, [at(100), at(120)]); + let place = speaker.place(ms(1_200), FRAME, TARGET, at(20)); + assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(140))); + assert_eq!(clock.due(stamp(1_200)), Some(at(190))); + } + + /// The bus plays a channel's first write, and each one after it on time, + /// [`BUS_LATENCY`] after it was written. + #[test] + fn a_write_sounds_the_bus_latency_later() { + let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let rate = u64::from(bus.config().sample_rate); + let strip = moq_audio::bus::Strip::default(); + let mut input = moq_audio::encode::Input::new(bus.config().sample_rate, bus.config().layout); + input.format = moq_audio::Format::F32; + let mut channel = bus.channel(input, strip.clone()).unwrap(); + let mix = bus.mix(); + mix.route(&strip); + let mut reader = mix.reader(); + + let block = rate as usize / 100; + driver.advance(block); + let written = block as u64; + let ones = vec![1.0f32; 2 * block]; + channel.write(bytemuck::cast_slice(&ones)).unwrap(); + + let sounded = loop { + driver.advance(block); + let std::task::Poll::Ready(Some(block)) = reader.poll_read(&kio::Waiter::noop()) else { + panic!("an advance publishes a block"); + }; + if let Some(index) = block.data.chunks(2).position(|frame| frame[0] != 0.0) { + break block.position + index as u64; + } + }; + assert_eq!( + Duration::from_micros((sounded - written) * 1_000_000 / rate), + BUS_LATENCY + ); + } +} diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 9466a7a2..a895fc1e 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -8,7 +8,8 @@ //! With [`Context::publish`] set, each local source that shows a picture also //! publishes it as its own track, web pages aside; see [`crate::publish`]. -use std::collections::BTreeMap; +use std::cell::{Cell, RefCell}; +use std::collections::{BTreeMap, VecDeque}; use std::future::Future; use std::path::{Path, PathBuf}; use std::pin::Pin; @@ -19,12 +20,14 @@ use hang::catalog::{AudioConfig, Reference, VideoConfig}; use kio::clock::Instant; use moq_audio::bus::{Bus, Mix, Strip}; use moq_mux::catalog::Stream; -use tokio::sync::watch; +use moq_mux::timeline; +use tokio::sync::{Notify, watch}; use crate::Error; use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; +use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -114,6 +117,19 @@ impl Stall { /// One instance of a broadcast being decoded. type Playing<'a> = Pin> + 'a>>; +/// The playout clock one instance of a broadcast's audio and video share. +struct Clock { + timeline: timeline::Producer, + /// Woken when audio moves the anchor, so video works out again when its + /// next picture is due. + moved: Notify, +} + +/// Decoded pictures a broadcast source holds until they are due. The rest wait +/// undecoded in the subscription, as in `moq play`, so a deep playout target +/// holds no more decoder output than this. +const QUEUED: usize = 3; + /// What a running source is doing. #[derive(Clone, Debug, PartialEq, Eq)] #[non_exhaustive] @@ -287,6 +303,8 @@ pub struct Feed { played: Playback, /// A replay source's next replay. plays: Option>, + /// A broadcast source's playout clock. + clock: Option, } impl Feed { @@ -369,6 +387,14 @@ impl Feed { (since >= STALL).then(|| stall(since)) } + /// How far a live broadcast source plays behind what arrives, to ride out + /// network jitter: its audio's playout target, which video follows, or + /// without audio, video's own short delay. `None` for any other source. + pub fn delay(&self) -> Option { + let clock = self.clock.as_ref()?; + (self.status() == Status::Live).then(|| clock.target().unwrap_or(VIDEO_DELAY)) + } + /// The source's gain, mute, delay, and meter in every mix, if it has sound. pub fn strip(&self) -> Option<&Strip> { self.strip.as_ref() @@ -404,6 +430,7 @@ pub struct Driver { track: Option, played: Playback, plays: Option>, + clock: Option, } /// Where a source's own track goes, its claim on the studio catalog's listing @@ -671,6 +698,7 @@ impl Driver { let mut instances = moq_mux::follow::Instances::new(&self.context.origin, path, None)?; set(&self.status, Status::Offline); + let mut clock = self.clock.clone().expect("a broadcast source has a clock"); let mut playing: Option> = None; loop { tokio::select! { @@ -681,7 +709,9 @@ impl Driver { broadcast: Some(broadcast.clone()), ..Played::default() }; - playing = Some(Box::pin(self.play(broadcast, path))); + // Each instance's timestamps are its own. + clock = clock.reset(VIDEO_DELAY); + playing = Some(Box::pin(self.play(broadcast, path, clock.clone()))); } Ok(moq_mux::follow::Event::End) => { playing = None; @@ -703,17 +733,27 @@ impl Driver { /// Decode one instance of a broadcast: the first video and audio renditions /// each catalog offers, switching when a catalog update changes them, until - /// the catalog ends. - async fn play(&self, broadcast: moq_net::broadcast::Consumer, path: &str) -> Result<(), Error> { + /// the catalog ends. Both play on `clock`. + async fn play( + &self, + broadcast: moq_net::broadcast::Consumer, + path: &str, + clock: timeline::Producer, + ) -> Result<(), Error> { let format = moq_mux::catalog::CatalogFormat::detect(path).unwrap_or_default(); let mut catalogs = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await?; + let clock = Clock { + timeline: clock, + moved: Notify::new(), + }; // A track name always means the same content, so only a new track // restarts a decoder. That is the track an entry resolves to, not its // catalog ID: a publisher may keep the ID and point it at a replacement // track, as the JS publisher does on a codec change. A metadata-only - // update, such as a jitter estimate, leaves playback alone. - let mut video: Option = None; + // update, such as a jitter estimate, leaves playback alone. Video also + // restarts when audio comes or goes, which changes how long it waits. + let mut video: Option<(String, bool)> = None; let mut audio: Option = None; let mut video_playing: Playing<'_> = Box::pin(std::future::pending()); let mut audio_playing: Playing<'_> = Box::pin(std::future::pending()); @@ -724,28 +764,33 @@ impl Driver { let Some(catalog) = catalog? else { return Ok(()); }; - let next = pick(catalog.video.enabled())?; - let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + let next_video = pick(catalog.video.enabled())?; + let next_audio = pick(catalog.audio.enabled())?; + let track = next_video + .as_ref() + .map(|(id, config)| (Reference::track(config, id).to_string(), next_audio.is_some())); if track != video { video = track; let mut played = self.played.lock().unwrap(); - played.video = next.clone(); + played.video = next_video.clone(); // The newest picture was the old track's, a moment the // new one may not hold. played.newest = None; drop(played); - video_playing = match next { - Some((id, config)) => Box::pin(self.play_video(&broadcast, id, config)), + // Video waits for its audio as long as audio's target can + // grow; alone, it waits only its own delay. + let max = if next_audio.is_some() { MAX_DELAY } else { VIDEO_DELAY }; + video_playing = match next_video { + Some((id, config)) => Box::pin(self.play_video(&broadcast, id, config, &clock, max)), None => Box::pin(std::future::pending()), }; } - let next = pick(catalog.audio.enabled())?; - let track = next.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); + let track = next_audio.as_ref().map(|(id, config)| Reference::track(config, id).to_string()); if track != audio { audio = track; - self.played.lock().unwrap().audio = next.clone(); - audio_playing = match next { - Some((id, config)) => Box::pin(self.play_audio(&broadcast, id, config)), + self.played.lock().unwrap().audio = next_audio.clone(); + audio_playing = match next_audio { + Some((id, config)) => Box::pin(self.play_audio(&broadcast, id, config, &clock)), None => Box::pin(std::future::pending()), }; } @@ -763,30 +808,64 @@ impl Driver { } } + /// Show each picture once `clock` says it is due, the newest due one when + /// several are, holding a few decoded ones for their moments. The canvas + /// keeps showing the last picture meanwhile. async fn play_video( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::VideoConfig, + clock: &Clock, + max: Duration, ) -> Result<(), Error> { let mut options = moq_video::decode::Options::new(); options.start = moq_video::decode::Start::Latest; options.decoder.output = self.context.output; + options.delay = moq_video::decode::Delay { + min: Some(VIDEO_DELAY), + max: Some(max), + }; let mut consumer = moq_video::decode::Consumer::new(broadcast, &config, name, options).await?; - while let Some(read) = consumer.read().await? { - if let moq_video::decode::Read::Frame(frame) = read { + let schedule = clock.timeline.consume(); + let mut queue = VecDeque::::new(); + let mut ended = false; + loop { + let now = kio::clock::now(); + if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), now) { + let frame = queue.drain(..=index).next_back().expect("an index into the queue"); self.played.lock().unwrap().newest = Some(frame.timestamp); self.show(Picture::Frame(frame)); } + if ended && queue.is_empty() { + return Ok(()); + } + let due = queue.front().and_then(|frame| schedule.due(frame.timestamp)); + tokio::select! { + read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { + Some(moq_video::decode::Read::Frame(frame)) => { + clock.timeline.video(frame.timestamp, kio::clock::now()); + queue.push_back(frame); + } + Some(moq_video::decode::Read::Gap { .. } | moq_video::decode::Read::End { .. }) => { + clock.timeline.video_restarted(); + } + None => ended = true, + }, + () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} + () = clock.moved.notified() => {} + } } - Ok(()) } + /// Play the broadcast's sound into the bus on its timestamps, holding the + /// jitter estimate moq-audio's decoder makes, and anchor `clock` on it. async fn play_audio( &self, broadcast: &moq_net::broadcast::Consumer, name: String, config: hang::catalog::AudioConfig, + clock: &Clock, ) -> Result<(), Error> { let Some(strip) = self.strip.clone() else { return Ok(()); @@ -794,6 +873,10 @@ impl Driver { let bus = self.context.bus.config().clone(); let mut options = moq_audio::decode::Options::new(); options.start = moq_audio::decode::Start::Latest; + options.delay = moq_audio::decode::Delay { + min: None, + max: Some(MAX_DELAY), + }; options.output.format = moq_audio::Format::F32; options.output.sample_rate = Some(bus.sample_rate); options.output.layout = Some(bus.layout); @@ -802,17 +885,50 @@ impl Driver { let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); input.format = moq_audio::Format::F32; let mut channel = self.context.bus.channel(input, strip)?; - // Gaps and encoder boundaries need nothing. Like a picture, which shows - // on arrival, the bus plays a source at its live edge on the program - // clock: a stall plays as silence while it lasts, and a burst after it - // is bounded by the bus's buffer. Replaying the publisher's - // timeline, holes included, would put the audio behind the live edge - // and its own video. - while let Some(read) = consumer.read().await? { - if let moq_audio::decode::Read::Frame(frame) = read { - channel.write(&frame.data)?; + let mut speaker = Speaker::new(clock.timeline.clone()); + let stride = bus.layout.channels() as usize * size_of::(); + let rate = f64::from(bus.sample_rate); + + // Frames wait here for their moment, bounded by the target the speaker + // holds, while every arrival is read at once: the jitter estimate times + // each one as `read` hands it over. + let queue = RefCell::new(VecDeque::new()); + let queued = Notify::new(); + let received = Cell::new(false); + let receive = async { + while let Some(next) = consumer.read().await? { + let moq_audio::decode::Read::Frame(frame) = next else { + // A hole or an encoder restart shows in the stamps that follow. + continue; + }; + let duration = Duration::from_secs_f64((frame.data.len() / stride) as f64 / rate); + let place = speaker.place(frame.timestamp.into(), duration, consumer.delay(), kio::clock::now()); + if let Place::Write { at, moved } = place { + if moved { + clock.moved.notify_one(); + } + queue.borrow_mut().push_back((at, frame)); + queued.notify_one(); + } } - } + received.set(true); + queued.notify_one(); + Ok::<_, Error>(()) + }; + let write = async { + loop { + let next = queue.borrow_mut().pop_front(); + match next { + Some((at, frame)) => { + kio::clock::sleep_until(at).await; + channel.write(&frame.data)?; + } + None if received.get() => return Ok::<_, Error>(()), + None => queued.notified().await, + } + } + }; + tokio::try_join!(receive, write)?; Ok(()) } @@ -1027,6 +1143,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { }); let played = Playback::default(); + let clock = matches!(source, Source::Broadcast { .. }).then(|| timeline::Producer::new(VIDEO_DELAY)); let (plays, replays) = match source { Source::Replay => { let (plays, replays) = watch::channel(Next::default()); @@ -1045,6 +1162,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track: track.as_ref().map(|track| track.status.consume()), played: played.clone(), plays, + clock: clock.as_ref().map(timeline::Producer::consume), }; let driver = Driver { name: name.to_string(), @@ -1058,6 +1176,7 @@ fn open(name: &str, source: &Source, context: &Context) -> (Feed, Driver) { track, played, plays: replays, + clock, }; (feed, driver) } @@ -1568,6 +1687,116 @@ mod tests { .await; } + /// A live broadcast at `room/guest` with a video track, and an audio one + /// with `audio`, writing nothing until the test does. + async fn live( + origin: &moq_net::origin::Producer, + audio: bool, + ) -> ( + moq_net::origin::Publication, + moq_video::encode::Producer, + moq_video::encode::Encoder, + Option, + ) { + let publication = origin.create_broadcast("room/guest").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); + config.kind = moq_video::encode::Kind::Software; + let rendition = config.probe().await.unwrap(); + let reserved = catalog.reserve().with_interval(config.gop.duration(config.framerate)); + let video = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); + let encoder = moq_video::encode::Encoder::new(&config).unwrap(); + let audio = audio.then(|| { + let input = moq_audio::encode::Input::default(); + let mut options = moq_audio::encode::Options::default(); + options.config = moq_audio::encode::Config::from_input(moq_audio::encode::Codec::Pcm, &input); + moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options).unwrap() + }); + publication.announce(moq_net::origin::Route::default()).unwrap(); + (publication, video, encoder, audio) + } + + /// Encode and publish one gray picture stamped `at`. + fn picture(producer: &mut moq_video::encode::Producer, encoder: &mut moq_video::encode::Encoder, at: u64) { + let gray = vec![0x80; 64 * 64 * 4]; + let surface = moq_video::Surface::rgba(&gray, moq_video::Size::new(64, 64)).unwrap(); + let frame = moq_video::Frame::new(surface, moq_net::Timestamp::from_millis(at).unwrap()); + producer.publish(&encoder.encode(&frame).unwrap()).unwrap(); + } + + /// The picture a feed shows next, once it shows one. + async fn shown(feed: &Feed) -> Duration { + until(|| match feed.take() { + Some(Picture::Frame(frame)) => Some(frame.timestamp.into()), + Some(_) => panic!("a broadcast shows only frames"), + None => None, + }) + .await + } + + /// Without audio to follow, a broadcast's picture shows [`VIDEO_DELAY`] + /// after it arrives, not on arrival. + #[tokio::test(start_paused = true)] + async fn a_picture_waits_for_its_moment() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let (_publication, mut video, mut encoder, _) = live(&origin, false).await; + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut video, &mut encoder, 1_000); + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(stamp)).await; + assert_eq!(due, arrived + VIDEO_DELAY); + assert!(feed.take().is_none(), "shown on arrival"); + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(stamp)); + assert_eq!(feed.delay(), Some(VIDEO_DELAY)); + }) + .await; + } + + /// A picture waits for the sound stamped with it, which plays the jitter + /// estimate after it arrives, rather than showing on its own delay. + #[tokio::test(start_paused = true)] + async fn a_picture_waits_for_its_sound() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let (_publication, mut video, mut encoder, audio) = live(&origin, true).await; + let mut audio = audio.unwrap(); + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + let silence = vec![0u8; 960 * 2 * size_of::()]; + let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + audio.write(&moq_audio::Frame::new(silence.into(), stamp)).unwrap(); + picture(&mut video, &mut encoder, 1_000); + let target = until(|| clock.target()).await; + assert!(target + crate::playout::BUS_LATENCY > VIDEO_DELAY, "{target:?}"); + let due = arrived + target + crate::playout::BUS_LATENCY; + until(|| (clock.due(stamp) == Some(due)).then_some(())).await; + until_time(due - Duration::from_millis(1)).await; + assert!(feed.take().is_none(), "shown before its sound"); + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(stamp)); + assert_eq!(feed.delay(), Some(target)); + }) + .await; + } + /// Dropping a feed stops its driver, which is what releases a device. #[tokio::test] async fn a_dropped_feed_stops_its_driver() { diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 1225d859..2b7a12da 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1325,6 +1325,13 @@ impl Panels { status(ui, &track); }); } + // A broadcast publishes no track either, and shows how far + // behind its publisher it plays. + None if let Some(delay) = feed.and_then(Feed::delay) => { + ui.weak(format!("{} ms behind", delay.as_millis())).on_hover_text( + "Plays this far behind what arrives, to ride out the network's jitter; the picture waits for its sound", + ); + } None => match feed.and_then(Feed::page).and_then(|page| page.paint()) { Some(Paint::Gpu) => { ui.weak("GPU").on_hover_text("The page is copied on the GPU"); From ecc9cae833a7f9ddccc665b26e01cebd9280baf7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 11:20:00 -0800 Subject: [PATCH 3/7] fix(scene): keep a broadcast's bus channel fed, and hold pictures for their sound Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 240 +++++++++++++++++++++++++----------- rs/moq-scene/src/source.rs | 43 ++++--- 2 files changed, 193 insertions(+), 90 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 66a5ed94..790515a4 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -3,17 +3,18 @@ //! //! A broadcast's audio and video share one [`timeline`] clock, as they do in //! `moq play`: audio owns the clock's anchor while it plays, and video shows -//! each picture once the clock says it is due. Without audio, video paces -//! itself [`VIDEO_DELAY`] behind its arrivals. +//! each picture once the clock says it is due and its sound has played. +//! Without audio, video paces itself [`VIDEO_DELAY`] behind its arrivals. //! //! `moq play` holds its jitter buffer in the speaker. The bus holds only -//! [`BUS_LATENCY`] of a source, so a [`Speaker`] holds the rest: each frame -//! waits until it is due, the playout target moq-audio's decoder estimates -//! after it arrives, and is written the bus's latency ahead of when it sounds. +//! [`BUS_LATENCY`] of a source, so a [`Speaker`] holds the rest: it places each +//! frame the playout target moq-audio's decoder estimates after it arrives, +//! and a [`Writer`] writes it the bus's latency ahead of then. use std::time::Duration; use kio::clock::Instant; +use moq_audio::bus; use moq_mux::timeline; /// How far a broadcast without audio plays behind its arrivals, as under @@ -26,9 +27,10 @@ pub(crate) const VIDEO_DELAY: Duration = Duration::from_millis(100); /// estimate at it, since a skipped group is never measured. pub(crate) const MAX_DELAY: Duration = Duration::from_secs(2); -/// What the bus holds between a channel write and the mix that plays it. +/// What a bus channel holds between a write and the mix that plays it, while +/// it is kept fed. /// -/// moq-audio's `bus::LATENCY`, which is private; `a_write_sounds_the_bus_latency_later` +/// moq-audio's `bus::LATENCY`, which is private; `a_write_sounds_when_placed` /// pins the two together. pub(crate) const BUS_LATENCY: Duration = Duration::from_millis(50); @@ -36,8 +38,8 @@ pub(crate) const BUS_LATENCY: Duration = Duration::from_millis(50); /// land back on it, as `moq play`'s speaker allows: one frame of cadence. const SLACK: Duration = Duration::from_millis(20); -/// How far apart two frames' stamps may be and still continue each other: a -/// millisecond-stamped ingest rounds each one. +/// How far apart two stamps may be and still meet: a millisecond-stamped +/// ingest rounds each one. const CONTIGUOUS: Duration = Duration::from_millis(1); /// Places one broadcast audio track's frames on the bus clock, and reports @@ -54,9 +56,8 @@ pub(crate) struct Speaker { /// What to do with a frame [`Speaker::place`] was given. #[derive(Debug, PartialEq, Eq)] pub(crate) enum Place { - /// Write it to the bus at `at`. `moved` says the clock's anchor moved, so - /// every picture waiting for its moment is due at another one now. - Write { at: Instant, moved: bool }, + /// Play it from this moment. + Play(Instant), /// Drop it: more is queued than the target needs, after a burst or a /// target that fell, so the media skips ahead to land back on the target. Skip, @@ -79,7 +80,8 @@ impl Speaker { pub(crate) fn place(&mut self, start: Duration, duration: Duration, target: Duration, now: Instant) -> Place { let target = target.min(MAX_DELAY); let end = start.saturating_add(duration); - // The bus rides out a write up to half its latency late without a dropout. + // A write half the bus's latency late still finds sound queued ahead of + // it, however coarsely the bus advances. let follows = self.tail.and_then(|(sounds, media)| { let begins = sounds + start.saturating_sub(media); let rewound = start.saturating_add(CONTIGUOUS) < media; @@ -113,11 +115,8 @@ impl Speaker { buffered: sounds.saturating_duration_since(now), target, }; - let moved = self.clock.audio(audio, now); - Place::Write { - at: begins - BUS_LATENCY, - moved, - } + self.clock.audio(audio, now); + Place::Play(begins) } } @@ -127,6 +126,69 @@ impl Drop for Speaker { } } +/// Whether a picture stamped `timestamp` may show at `now` on `schedule`: +/// while audio plays, only once the sound stamped with it has played, so a +/// stall in the audio alone holds the picture too. +pub(crate) fn heard(schedule: &timeline::Consumer, timestamp: moq_net::Timestamp, now: Instant) -> bool { + schedule.target().is_none() + || schedule + .playhead(now) + .is_some_and(|head| Duration::from(timestamp) <= head + CONTIGUOUS) +} + +/// Writes placed frames into a bus channel, so each sounds when it was placed. +/// +/// A channel plays a write [`BUS_LATENCY`] later only while it holds that much: +/// one that ran dry plays the next write at once. So the writer keeps track of +/// when what it wrote runs out, and puts silence ahead of a frame for however +/// much of the latency has drained. +pub(crate) struct Writer { + channel: bus::Channel, + /// Bytes per sample frame of the channel's `f32` input. + stride: usize, + rate: u32, + /// When what was written finishes sounding, or `None` before the first + /// write, which the channel's own priming holds back the full latency. + runs_out: Option, + silence: Vec, +} + +impl Writer { + /// Write into a new `channel`, taking interleaved `f32` at `rate` in `channels`. + pub(crate) fn new(channel: bus::Channel, rate: u32, channels: u32) -> Self { + Self { + channel, + stride: channels as usize * size_of::(), + rate, + runs_out: None, + silence: Vec::new(), + } + } + + /// Write `data` at `now` to sound at `sounds`, or as soon after as it can. + /// Call it no sooner than [`BUS_LATENCY`] ahead. + pub(crate) fn write(&mut self, sounds: Instant, data: &[u8], now: Instant) -> Result<(), moq_audio::Error> { + let free = self.runs_out.unwrap_or(now + BUS_LATENCY).max(now); + let gap = sounds.saturating_duration_since(free).min(BUS_LATENCY); + let gap = self.frames(gap); + if gap > 0 { + self.silence.resize(gap * self.stride, 0); + self.channel.write(&self.silence)?; + } + self.channel.write(data)?; + self.runs_out = Some(free + self.duration(gap + data.len() / self.stride)); + Ok(()) + } + + fn frames(&self, duration: Duration) -> usize { + (duration.as_secs_f64() * f64::from(self.rate)).round() as usize + } + + fn duration(&self, frames: usize) -> Duration { + Duration::from_secs_f64(frames as f64 / f64::from(self.rate)) + } +} + #[cfg(test)] mod tests { use super::*; @@ -146,25 +208,18 @@ mod tests { moq_net::Timestamp::from_millis(millis).unwrap() } - fn written(place: Place) -> Instant { - match place { - Place::Write { at, .. } => at, - Place::Skip => panic!("skipped"), - } - } - - /// The first frame waits out the target, later ones follow it on their - /// stamps however unevenly they arrive within it, and a picture is due when - /// the sound stamped with it plays: the bus's latency after its write. + /// The first frame waits out the target and the bus's latency, later ones + /// follow it on their stamps however unevenly they arrive within it, and a + /// picture is due when the sound stamped with it plays. #[test] fn video_is_due_when_its_audio_sounds() { let clock = timeline::Producer::new(VIDEO_DELAY); let mut speaker = Speaker::new(clock.clone()); - assert_eq!(written(speaker.place(ms(1_000), FRAME, TARGET, at(0))), at(100)); + assert_eq!(speaker.place(ms(1_000), FRAME, TARGET, at(0)), Place::Play(at(150))); assert_eq!(clock.due(stamp(1_000)), Some(at(150))); // Late within the target, and early: both keep their place. - assert_eq!(written(speaker.place(ms(1_020), FRAME, TARGET, at(90))), at(120)); - assert_eq!(written(speaker.place(ms(1_040), FRAME, TARGET, at(91))), at(140)); + assert_eq!(speaker.place(ms(1_020), FRAME, TARGET, at(90)), Place::Play(at(170))); + assert_eq!(speaker.place(ms(1_040), FRAME, TARGET, at(91)), Place::Play(at(190))); assert_eq!(clock.due(stamp(1_040)), Some(at(190))); assert_eq!(clock.consume().target(), Some(TARGET)); @@ -173,6 +228,25 @@ mod tests { assert_eq!(clock.consume().target(), None); } + /// A picture whose sound has not arrived waits for it, even once due: the + /// audio stalled, and the picture stalls with it. + #[test] + fn a_picture_waits_for_its_sound_to_play() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let schedule = clock.consume(); + assert!( + heard(&schedule, stamp(5_000), at(0)), + "without audio, only the deadline counts" + ); + let mut speaker = Speaker::new(clock.clone()); + speaker.place(ms(1_000), FRAME, TARGET, at(0)); + assert!(!heard(&schedule, stamp(1_000), at(148))); + assert!(heard(&schedule, stamp(1_000), at(150))); + assert!(clock.due(stamp(1_100)).is_some_and(|due| due < at(500))); + assert!(heard(&schedule, stamp(1_020), at(500))); + assert!(!heard(&schedule, stamp(1_100), at(500)), "shown past its sound"); + } + /// A stall longer than the target re-buffers: the next frame sounds the /// target after it arrived, and the pictures stamped with it wait as long. #[test] @@ -180,8 +254,7 @@ mod tests { let clock = timeline::Producer::new(VIDEO_DELAY); let mut speaker = Speaker::new(clock.clone()); speaker.place(ms(1_000), FRAME, TARGET, at(0)); - let place = speaker.place(ms(1_020), FRAME, TARGET, at(500)); - assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(600))); + assert_eq!(speaker.place(ms(1_020), FRAME, TARGET, at(500)), Place::Play(at(650))); assert_eq!(clock.due(stamp(1_020)), Some(at(650))); } @@ -193,19 +266,12 @@ mod tests { let clock = timeline::Producer::new(VIDEO_DELAY); let mut speaker = Speaker::new(clock.clone()); speaker.place(ms(1_000), FRAME, TARGET, at(0)); - let place = speaker.place(ms(1_060), FRAME, TARGET, at(40)); - assert_eq!( - place, - Place::Write { - at: at(160), - moved: false - } - ); + assert_eq!(speaker.place(ms(1_060), FRAME, TARGET, at(40)), Place::Play(at(210))); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); // 1 s lost, with only the target queued: skipped back onto the target. assert_eq!(speaker.place(ms(2_080), FRAME, TARGET, at(60)), Place::Skip); - let place = speaker.place(ms(2_100), FRAME, TARGET, at(61)); - assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(180))); + assert_eq!(speaker.place(ms(2_100), FRAME, TARGET, at(61)), Place::Play(at(230))); assert_eq!(clock.due(stamp(2_100)), Some(at(230))); } @@ -216,51 +282,79 @@ mod tests { let clock = timeline::Producer::new(VIDEO_DELAY); let mut speaker = Speaker::new(clock.clone()); let placed = (0..10) - .filter_map( - |index| match speaker.place(ms(1_000 + index * 20), FRAME, TARGET, at(0)) { - Place::Write { at, .. } => Some(at), - Place::Skip => None, - }, - ) + .map(|index| speaker.place(ms(1_000 + index * 20), FRAME, TARGET, at(0))) + .filter(|place| *place != Place::Skip) .collect::>(); - assert_eq!(placed, [at(100), at(120)]); - let place = speaker.place(ms(1_200), FRAME, TARGET, at(20)); - assert!(matches!(place, Place::Write { at: written, moved: true } if written == at(140))); + assert_eq!(placed, [Place::Play(at(150)), Place::Play(at(170))]); + assert_eq!(speaker.place(ms(1_200), FRAME, TARGET, at(20)), Place::Play(at(190))); assert_eq!(clock.due(stamp(1_200)), Some(at(190))); } - /// The bus plays a channel's first write, and each one after it on time, - /// [`BUS_LATENCY`] after it was written. + /// A frame sounds when placed: the first, one that follows it, and one + /// written after the channel ran dry, which the bus would otherwise play at + /// once, ahead of its picture. #[test] - fn a_write_sounds_the_bus_latency_later() { + fn a_write_sounds_when_placed() { let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); - let rate = u64::from(bus.config().sample_rate); + let config = bus.config().clone(); + let rate = u64::from(config.sample_rate); let strip = moq_audio::bus::Strip::default(); - let mut input = moq_audio::encode::Input::new(bus.config().sample_rate, bus.config().layout); + let mut input = moq_audio::encode::Input::new(config.sample_rate, config.layout); input.format = moq_audio::Format::F32; - let mut channel = bus.channel(input, strip.clone()).unwrap(); + let channel = bus.channel(input, strip.clone()).unwrap(); + let mut writer = Writer::new(channel, config.sample_rate, config.layout.channels()); let mix = bus.mix(); mix.route(&strip); let mut reader = mix.reader(); - let block = rate as usize / 100; - driver.advance(block); - let written = block as u64; - let ones = vec![1.0f32; 2 * block]; - channel.write(bytemuck::cast_slice(&ones)).unwrap(); - - let sounded = loop { - driver.advance(block); - let std::task::Poll::Ready(Some(block)) = reader.poll_read(&kio::Waiter::noop()) else { - panic!("an advance publishes a block"); - }; - if let Some(index) = block.data.chunks(2).position(|frame| frame[0] != 0.0) { - break block.position + index as u64; + // The bus clock, as an instant, and a frame of `value` on it. + let block = rate / 100; + let clock = |frames: u64| Instant::ZERO + Duration::from_micros(frames * 1_000_000 / rate); + let frame = |value: f32| bytemuck::cast_slice::(&vec![value; 2 * 960]).to_vec(); + let mut heard = Vec::new(); + let mut position = 0; + let mut advance = |until: u64, heard: &mut Vec<(u64, f32)>| { + while position < until { + driver.advance(block as usize); + position += block; + let std::task::Poll::Ready(Some(read)) = reader.poll_read(&kio::Waiter::noop()) else { + panic!("an advance publishes a block"); + }; + let mut last = heard.last().map_or(0.0, |(_, value)| *value); + for (index, sample) in read.data.chunks(2).enumerate() { + if sample[0] != last { + last = sample[0]; + heard.push((read.position + index as u64, last)); + } + } } }; + + advance(block, &mut heard); + let sounds = clock(block) + BUS_LATENCY; + writer.write(sounds, &frame(1.0), clock(block)).unwrap(); + // The next one, on time. + advance(block + 960, &mut heard); + writer.write(sounds + FRAME, &frame(2.0), clock(block + 960)).unwrap(); + // Long after both ran out. + advance(rate, &mut heard); + writer + .write(clock(rate) + BUS_LATENCY, &frame(3.0), clock(rate)) + .unwrap(); + advance(2 * rate, &mut heard); + + let latency = rate * BUS_LATENCY.as_millis() as u64 / 1_000; + let first = block + latency; + let after = rate + latency; assert_eq!( - Duration::from_micros((sounded - written) * 1_000_000 / rate), - BUS_LATENCY + heard, + [ + (first, 1.0), + (first + 960, 2.0), + (first + 1_920, 0.0), + (after, 3.0), + (after + 960, 0.0) + ] ); } } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index a895fc1e..bfaa3581 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -27,7 +27,7 @@ use crate::Error; use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; -use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY}; +use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -120,9 +120,9 @@ type Playing<'a> = Pin> + 'a>>; /// The playout clock one instance of a broadcast's audio and video share. struct Clock { timeline: timeline::Producer, - /// Woken when audio moves the anchor, so video works out again when its - /// next picture is due. - moved: Notify, + /// Woken when audio places a frame, which may move the anchor or let a + /// picture waiting for its sound show. + placed: Notify, } /// Decoded pictures a broadcast source holds until they are due. The rest wait @@ -744,7 +744,7 @@ impl Driver { let mut catalogs = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await?; let clock = Clock { timeline: clock, - moved: Notify::new(), + placed: Notify::new(), }; // A track name always means the same content, so only a new track @@ -832,7 +832,13 @@ impl Driver { let mut ended = false; loop { let now = kio::clock::now(); - if let Some(index) = schedule.present(queue.iter().map(|frame| frame.timestamp), now) { + // Counted first: `present` holds the clock's lock while it iterates. + let heard = queue + .iter() + .take_while(|frame| crate::playout::heard(&schedule, frame.timestamp, now)) + .count(); + let ready = queue.iter().take(heard).map(|frame| frame.timestamp); + if let Some(index) = schedule.present(ready, now) { let frame = queue.drain(..=index).next_back().expect("an index into the queue"); self.played.lock().unwrap().newest = Some(frame.timestamp); self.show(Picture::Frame(frame)); @@ -840,7 +846,11 @@ impl Driver { if ended && queue.is_empty() { return Ok(()); } - let due = queue.front().and_then(|frame| schedule.due(frame.timestamp)); + // A picture due but not yet heard waits for audio to place more. + let due = queue + .front() + .and_then(|frame| schedule.due(frame.timestamp)) + .filter(|due| *due > now); tokio::select! { read = consumer.read(), if !ended && queue.len() < QUEUED => match read? { Some(moq_video::decode::Read::Frame(frame)) => { @@ -853,7 +863,7 @@ impl Driver { None => ended = true, }, () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} - () = clock.moved.notified() => {} + () = clock.placed.notified() => {} } } } @@ -884,7 +894,8 @@ impl Driver { let mut input = moq_audio::encode::Input::new(bus.sample_rate, bus.layout); input.format = moq_audio::Format::F32; - let mut channel = self.context.bus.channel(input, strip)?; + let channel = self.context.bus.channel(input, strip)?; + let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); let mut speaker = Speaker::new(clock.timeline.clone()); let stride = bus.layout.channels() as usize * size_of::(); let rate = f64::from(bus.sample_rate); @@ -903,11 +914,9 @@ impl Driver { }; let duration = Duration::from_secs_f64((frame.data.len() / stride) as f64 / rate); let place = speaker.place(frame.timestamp.into(), duration, consumer.delay(), kio::clock::now()); - if let Place::Write { at, moved } = place { - if moved { - clock.moved.notify_one(); - } - queue.borrow_mut().push_back((at, frame)); + if let Place::Play(sounds) = place { + clock.placed.notify_one(); + queue.borrow_mut().push_back((sounds, frame)); queued.notify_one(); } } @@ -919,9 +928,9 @@ impl Driver { loop { let next = queue.borrow_mut().pop_front(); match next { - Some((at, frame)) => { - kio::clock::sleep_until(at).await; - channel.write(&frame.data)?; + Some((sounds, frame)) => { + kio::clock::sleep_until(sounds - crate::playout::BUS_LATENCY).await; + writer.write(sounds, &frame.data, kio::clock::now())?; } None if received.get() => return Ok::<_, Error>(()), None => queued.notified().await, From 39a0d5ff68e583f9c423b7bcb8e719b8aef03a27 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 11:32:08 -0800 Subject: [PATCH 4/7] fix(scene): write a broadcast's silence and sound as one, and wake video when audio stops Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 78 ++++++++++++++++++++++++++----------- rs/moq-scene/src/source.rs | 13 +++---- 2 files changed, 61 insertions(+), 30 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 790515a4..f37a0b64 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -16,6 +16,7 @@ use std::time::Duration; use kio::clock::Instant; use moq_audio::bus; use moq_mux::timeline; +use tokio::sync::Notify; /// How far a broadcast without audio plays behind its arrivals, as under /// `moq play --delay auto`. With audio, video follows the audio's target. @@ -46,8 +47,11 @@ const CONTIGUOUS: Duration = Duration::from_millis(1); /// where they sound to the broadcast's clock, which video follows. /// /// Dropping it hands the clock's anchor back to video. -pub(crate) struct Speaker { +pub(crate) struct Speaker<'a> { clock: timeline::Producer, + /// Woken on every frame placed and when the speaker stops, so pictures + /// held for their sound show once it plays, or once there is none. + changed: &'a Notify, /// The last frame placed: when it finishes sounding, and the media time it /// ends at. tail: Option<(Instant, Duration)>, @@ -63,10 +67,15 @@ pub(crate) enum Place { Skip, } -impl Speaker { - /// Place audio on `clock`, taking its anchor from video on the first frame. - pub(crate) fn new(clock: timeline::Producer) -> Self { - Self { clock, tail: None } +impl<'a> Speaker<'a> { + /// Place audio on `clock`, taking its anchor from video on the first frame, + /// and wake `changed` whenever that may let a picture show. + pub(crate) fn new(clock: timeline::Producer, changed: &'a Notify) -> Self { + Self { + clock, + changed, + tail: None, + } } /// Place a decoded frame covering `duration` of media from `start`, @@ -116,13 +125,15 @@ impl Speaker { target, }; self.clock.audio(audio, now); + self.changed.notify_one(); Place::Play(begins) } } -impl Drop for Speaker { +impl Drop for Speaker<'_> { fn drop(&mut self) { self.clock.audio_stopped(); + self.changed.notify_one(); } } @@ -150,7 +161,9 @@ pub(crate) struct Writer { /// When what was written finishes sounding, or `None` before the first /// write, which the channel's own priming holds back the full latency. runs_out: Option, - silence: Vec, + /// Silence and the frame after it, written as one, so the channel pads + /// behind the frame rather than between them. + pending: Vec, } impl Writer { @@ -161,7 +174,7 @@ impl Writer { stride: channels as usize * size_of::(), rate, runs_out: None, - silence: Vec::new(), + pending: Vec::new(), } } @@ -171,12 +184,17 @@ impl Writer { let free = self.runs_out.unwrap_or(now + BUS_LATENCY).max(now); let gap = sounds.saturating_duration_since(free).min(BUS_LATENCY); let gap = self.frames(gap); - if gap > 0 { - self.silence.resize(gap * self.stride, 0); - self.channel.write(&self.silence)?; - } - self.channel.write(data)?; - self.runs_out = Some(free + self.duration(gap + data.len() / self.stride)); + self.pending.clear(); + self.pending.resize(gap * self.stride, 0); + self.pending.extend_from_slice(data); + self.channel.write(&self.pending)?; + let runs_out = free + self.duration(self.pending.len() / self.stride); + // A channel left holding a quarter of its latency or less pads itself + // back up to it behind the write, as moq-audio's ring does. + self.runs_out = Some(match runs_out.saturating_duration_since(now) <= BUS_LATENCY / 4 { + true => now + BUS_LATENCY, + false => runs_out, + }); Ok(()) } @@ -214,7 +232,8 @@ mod tests { #[test] fn video_is_due_when_its_audio_sounds() { let clock = timeline::Producer::new(VIDEO_DELAY); - let mut speaker = Speaker::new(clock.clone()); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); assert_eq!(speaker.place(ms(1_000), FRAME, TARGET, at(0)), Place::Play(at(150))); assert_eq!(clock.due(stamp(1_000)), Some(at(150))); // Late within the target, and early: both keep their place. @@ -238,7 +257,8 @@ mod tests { heard(&schedule, stamp(5_000), at(0)), "without audio, only the deadline counts" ); - let mut speaker = Speaker::new(clock.clone()); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); speaker.place(ms(1_000), FRAME, TARGET, at(0)); assert!(!heard(&schedule, stamp(1_000), at(148))); assert!(heard(&schedule, stamp(1_000), at(150))); @@ -252,7 +272,8 @@ mod tests { #[test] fn a_stall_holds_video_back_with_the_sound() { let clock = timeline::Producer::new(VIDEO_DELAY); - let mut speaker = Speaker::new(clock.clone()); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); speaker.place(ms(1_000), FRAME, TARGET, at(0)); assert_eq!(speaker.place(ms(1_020), FRAME, TARGET, at(500)), Place::Play(at(650))); assert_eq!(clock.due(stamp(1_020)), Some(at(650))); @@ -264,7 +285,8 @@ mod tests { #[test] fn a_hole_plays_as_silence_up_to_the_target() { let clock = timeline::Producer::new(VIDEO_DELAY); - let mut speaker = Speaker::new(clock.clone()); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); speaker.place(ms(1_000), FRAME, TARGET, at(0)); assert_eq!(speaker.place(ms(1_060), FRAME, TARGET, at(40)), Place::Play(at(210))); assert_eq!(clock.due(stamp(1_060)), Some(at(210))); @@ -280,7 +302,8 @@ mod tests { #[test] fn a_burst_skips_past_the_target() { let clock = timeline::Producer::new(VIDEO_DELAY); - let mut speaker = Speaker::new(clock.clone()); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); let placed = (0..10) .map(|index| speaker.place(ms(1_000 + index * 20), FRAME, TARGET, at(0))) .filter(|place| *place != Place::Skip) @@ -290,9 +313,9 @@ mod tests { assert_eq!(clock.due(stamp(1_200)), Some(at(190))); } - /// A frame sounds when placed: the first, one that follows it, and one - /// written after the channel ran dry, which the bus would otherwise play at - /// once, ahead of its picture. + /// A frame sounds when placed: the first, one that follows it, and ones + /// written after the channel ran dry, on time or late, which the bus would + /// otherwise play at once or pad behind, apart from their pictures. #[test] fn a_write_sounds_when_placed() { let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); @@ -342,10 +365,17 @@ mod tests { .write(clock(rate) + BUS_LATENCY, &frame(3.0), clock(rate)) .unwrap(); advance(2 * rate, &mut heard); + // Run dry and written late, with less than the latency to go. + advance(2 * rate + 1_920, &mut heard); + writer + .write(clock(2 * rate) + BUS_LATENCY, &frame(4.0), clock(2 * rate + 1_920)) + .unwrap(); + advance(3 * rate, &mut heard); let latency = rate * BUS_LATENCY.as_millis() as u64 / 1_000; let first = block + latency; let after = rate + latency; + let late = 2 * rate + latency; assert_eq!( heard, [ @@ -353,7 +383,9 @@ mod tests { (first + 960, 2.0), (first + 1_920, 0.0), (after, 3.0), - (after + 960, 0.0) + (after + 960, 0.0), + (late, 4.0), + (late + 960, 0.0) ] ); } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index bfaa3581..b97bb027 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -120,9 +120,9 @@ type Playing<'a> = Pin> + 'a>>; /// The playout clock one instance of a broadcast's audio and video share. struct Clock { timeline: timeline::Producer, - /// Woken when audio places a frame, which may move the anchor or let a - /// picture waiting for its sound show. - placed: Notify, + /// Woken when audio places a frame or stops, which may move the anchor or + /// let a picture waiting for its sound show. + changed: Notify, } /// Decoded pictures a broadcast source holds until they are due. The rest wait @@ -744,7 +744,7 @@ impl Driver { let mut catalogs = moq_mux::catalog::Consumer::<()>::new(&broadcast, format).await?; let clock = Clock { timeline: clock, - placed: Notify::new(), + changed: Notify::new(), }; // A track name always means the same content, so only a new track @@ -863,7 +863,7 @@ impl Driver { None => ended = true, }, () = kio::clock::sleep_until(due.unwrap_or(now)), if due.is_some() => {} - () = clock.placed.notified() => {} + () = clock.changed.notified() => {} } } } @@ -896,7 +896,7 @@ impl Driver { input.format = moq_audio::Format::F32; let channel = self.context.bus.channel(input, strip)?; let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); - let mut speaker = Speaker::new(clock.timeline.clone()); + let mut speaker = Speaker::new(clock.timeline.clone(), &clock.changed); let stride = bus.layout.channels() as usize * size_of::(); let rate = f64::from(bus.sample_rate); @@ -915,7 +915,6 @@ impl Driver { let duration = Duration::from_secs_f64((frame.data.len() / stride) as f64 / rate); let place = speaker.place(frame.timestamp.into(), duration, consumer.delay(), kio::clock::now()); if let Place::Play(sounds) = place { - clock.placed.notify_one(); queue.borrow_mut().push_back((sounds, frame)); queued.notify_one(); } From 25be31cea076a0e0929cca3978ec4bafbe6255f6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 13:17:03 -0800 Subject: [PATCH 5/7] fix(scene): skip broadcast audio that overlaps the queue, and restart video pacing on a new track Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 64 +++++++++++++-------- rs/moq-scene/src/source.rs | 111 +++++++++++++++++++++++++++++------- 2 files changed, 128 insertions(+), 47 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index f37a0b64..07b8a9cf 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -85,35 +85,30 @@ impl<'a> Speaker<'a> { /// media between them. A frame that arrives too late for that, because the /// network stalled longer than the target, re-buffers instead: it sounds the /// whole target after it arrived, and the clock re-pins on it, so video - /// waits along with the sound rather than running ahead of it. + /// waits along with the sound rather than running ahead of it. A frame + /// that would queue past the target, or that overlaps media already queued, + /// is skipped, so the queue never holds more than the target. pub(crate) fn place(&mut self, start: Duration, duration: Duration, target: Duration, now: Instant) -> Place { let target = target.min(MAX_DELAY); let end = start.saturating_add(duration); - // A write half the bus's latency late still finds sound queued ahead of - // it, however coarsely the bus advances. - let follows = self.tail.and_then(|(sounds, media)| { - let begins = sounds + start.saturating_sub(media); - let rewound = start.saturating_add(CONTIGUOUS) < media; - let dry = begins < now + BUS_LATENCY / 2; - (!rewound && !dry).then_some((begins, sounds)) - }); - let begins = match follows { - Some((begins, sounds)) => { - if begins.saturating_duration_since(now) > target + BUS_LATENCY + SLACK { - self.tail = Some((sounds, end)); + let begins = match self.tail { + None => now + target + BUS_LATENCY, + Some((sounds, media)) => { + let follows = sounds + start.saturating_sub(media); + // A write half the bus's latency late still finds sound queued + // ahead of it, however coarsely the bus advances. + if follows < now + BUS_LATENCY / 2 { + self.clock.audio_restarted(); + now + target + BUS_LATENCY + } else if start.saturating_add(CONTIGUOUS) < media + || follows.saturating_duration_since(now) > target + BUS_LATENCY + SLACK + { + // Media overlapping what is queued, or past the target, would + // queue without bound however fast it arrives. + self.tail = Some((sounds, media.max(end))); return Place::Skip; - } - begins - } - None => { - let begins = now + target + BUS_LATENCY; - match self.tail { - Some((sounds, _)) => { - self.clock.audio_restarted(); - // Never over audio still queued, after a rewind. - begins.max(sounds) - } - None => begins, + } else { + follows } } }; @@ -313,6 +308,25 @@ mod tests { assert_eq!(clock.due(stamp(1_200)), Some(at(190))); } + /// Media overlapping what is queued, stamped the same or a millisecond + /// apart however long each frame is, is skipped rather than queued behind + /// it; once the queue runs out, a rewound timeline plays from the target. + #[test] + fn overlapping_media_queues_no_more_than_the_target() { + let clock = timeline::Producer::new(VIDEO_DELAY); + let changed = Notify::new(); + let mut speaker = Speaker::new(clock.clone(), &changed); + for index in 0..1_000 { + for start in [ms(1_000), ms(1_000 + index)] { + if let Place::Play(begins) = speaker.place(start, FRAME, TARGET, at(0)) { + assert!(begins <= at(0) + TARGET + BUS_LATENCY + SLACK, "{index}: {begins:?}"); + } + } + } + assert_eq!(speaker.place(ms(500), FRAME, TARGET, at(1_000)), Place::Play(at(1_150))); + assert_eq!(clock.due(stamp(500)), Some(at(1_150))); + } + /// A frame sounds when placed: the first, one that follows it, and ones /// written after the channel ran dry, on time or late, which the bus would /// otherwise play at once or pad behind, apart from their pictures. diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 60a9b7bb..23dead6c 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -868,6 +868,8 @@ impl Driver { clock: &Clock, max: Duration, ) -> Result<(), Error> { + // A replacement track's stamps need not continue the old one's. + clock.timeline.video_restarted(); let mut options = moq_video::decode::Options::new(); options.start = moq_video::decode::Start::Latest; options.decoder.output = self.context.output; @@ -1744,26 +1746,35 @@ mod tests { .await; } - /// A live broadcast at `room/guest` with a video track, and an audio one - /// with `audio`, writing nothing until the test does. - async fn live( - origin: &moq_net::origin::Producer, - audio: bool, - ) -> ( - moq_net::origin::Publication, - moq_video::encode::Producer, - moq_video::encode::Encoder, - Option, - ) { - let publication = origin.create_broadcast("room/guest").unwrap(); - let mut broadcast = publication.broadcast().clone(); - let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + /// A live broadcast at `room/guest`, writing nothing until the test does. + struct Live { + _publication: moq_net::origin::Publication, + broadcast: moq_net::broadcast::Producer, + catalog: moq_mux::catalog::Producer, + video: moq_video::encode::Producer, + encoder: moq_video::encode::Encoder, + audio: Option, + } + + /// A video track on its own timestamp base. + async fn rendition( + broadcast: &moq_net::broadcast::Producer, + timebase: moq_mux::catalog::Timebase, + ) -> (moq_video::encode::Producer, moq_video::encode::Encoder) { let mut config = moq_video::encode::Config::new(64, 64, moq_video::FrameRate::new(30, 1).unwrap()); config.kind = moq_video::encode::Kind::Software; let rendition = config.probe().await.unwrap(); - let reserved = catalog.reserve().with_interval(config.gop.duration(config.framerate)); + let reserved = timebase.reserve().with_interval(config.gop.duration(config.framerate)); let video = moq_video::encode::Producer::new(broadcast.clone(), reserved, rendition).unwrap(); - let encoder = moq_video::encode::Encoder::new(&config).unwrap(); + (video, moq_video::encode::Encoder::new(&config).unwrap()) + } + + /// A broadcast with a video track, and an audio one with `audio`. + async fn live(origin: &moq_net::origin::Producer, audio: bool) -> Live { + let publication = origin.create_broadcast("room/guest").unwrap(); + let mut broadcast = publication.broadcast().clone(); + let catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let (video, encoder) = rendition(&broadcast, catalog.timebase()).await; let audio = audio.then(|| { let input = moq_audio::encode::Input::default(); let mut options = moq_audio::encode::Options::default(); @@ -1771,7 +1782,14 @@ mod tests { moq_audio::encode::Producer::new(&mut broadcast, catalog.clone(), input, &options).unwrap() }); publication.announce(moq_net::origin::Route::default()).unwrap(); - (publication, video, encoder, audio) + Live { + _publication: publication, + broadcast, + catalog, + video, + encoder, + audio, + } } /// Encode and publish one gray picture stamped `at`. @@ -1799,7 +1817,7 @@ mod tests { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); - let (_publication, mut video, mut encoder, _) = live(&origin, false).await; + let mut live = live(&origin, false).await; let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); start(&sources, drivers).await; @@ -1808,7 +1826,7 @@ mod tests { until_time(kio::clock::now()).await; let arrived = kio::clock::now(); - picture(&mut video, &mut encoder, 1_000); + picture(&mut live.video, &mut live.encoder, 1_000); let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); let due = until(|| clock.due(stamp)).await; assert_eq!(due, arrived + VIDEO_DELAY); @@ -1820,6 +1838,55 @@ mod tests { .await; } + /// A video track replaced by one stamped earlier starts its own delay over, + /// rather than showing every picture on arrival against the old track's. + #[tokio::test(start_paused = true)] + async fn a_replacement_track_keeps_its_delay() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (context, origin) = context(dir.path()); + let mut live = live(&origin, false).await; + let mut sources = Sources::new(context); + let drivers = sources.sync(&collection()); + start(&sources, drivers).await; + let feed = sources.get("guest").unwrap(); + let clock = feed.clock.clone().unwrap(); + picture(&mut live.video, &mut live.encoder, 1_000); + let first = moq_net::Timestamp::from_millis(1_000).unwrap(); + let due = until(|| clock.due(first)).await; + until_time(due).await; + assert_eq!(shown(feed).await, Duration::from(first)); + + // Its stamps sit half a second before the first track's clock reading. + let timebase = live.catalog.timebase(); + timebase + .place( + moq_net::Timestamp::ZERO, + kio::clock::wall() - Duration::from_millis(500), + ) + .unwrap(); + let (video, mut encoder) = rendition(&live.broadcast, timebase).await; + drop(std::mem::replace(&mut live.video, video)); + until_time(kio::clock::now()).await; + let arrived = kio::clock::now(); + picture(&mut live.video, &mut encoder, 0); + let moved = until(|| match feed.take() { + Some(Picture::Frame(frame)) => panic!("{:?} shown on arrival", frame.timestamp), + _ => clock.due(first).filter(|at| *at != due), + }) + .await; + assert!(moved > due, "the replacement re-anchored behind the first track"); + until_time(arrived + VIDEO_DELAY - Duration::from_millis(1)).await; + assert!(feed.take().is_none(), "shown before its delay"); + until_time(arrived + VIDEO_DELAY).await; + assert!( + shown(feed).await < Duration::from(first), + "stamped before the first track" + ); + }) + .await; + } + /// A picture waits for the sound stamped with it, which plays the jitter /// estimate after it arrives, rather than showing on its own delay. #[tokio::test(start_paused = true)] @@ -1827,8 +1894,7 @@ mod tests { local(async { let dir = tempfile::tempdir().unwrap(); let (context, origin) = context(dir.path()); - let (_publication, mut video, mut encoder, audio) = live(&origin, true).await; - let mut audio = audio.unwrap(); + let mut live = live(&origin, true).await; let mut sources = Sources::new(context); let drivers = sources.sync(&collection()); start(&sources, drivers).await; @@ -1839,8 +1905,9 @@ mod tests { let arrived = kio::clock::now(); let silence = vec![0u8; 960 * 2 * size_of::()]; let stamp = moq_net::Timestamp::from_millis(1_000).unwrap(); + let audio = live.audio.as_mut().unwrap(); audio.write(&moq_audio::Frame::new(silence.into(), stamp)).unwrap(); - picture(&mut video, &mut encoder, 1_000); + picture(&mut live.video, &mut live.encoder, 1_000); let target = until(|| clock.target()).await; assert!(target + crate::playout::BUS_LATENCY > VIDEO_DELAY, "{target:?}"); let due = arrived + target + crate::playout::BUS_LATENCY; From d21fda148f7bb344d99a86bc0bc6e51336ef3014 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 14:55:45 -0800 Subject: [PATCH 6/7] fix(scene): defer the clock past a skip, drop audio written after its moment, and correct the stall docs Co-Authored-By: Claude Opus 5.5 --- README.md | 5 +-- rs/moq-scene/src/playout.rs | 63 ++++++++++++++++++++++++++++++++----- 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 512003df..36bc9a67 100644 --- a/README.md +++ b/README.md @@ -158,8 +158,9 @@ so its sound and picture stay in sync through network jitter and stalls. Its audio plays as far behind what arrives as the network's jitter needs, measured as it goes, the same delay `moq play --delay auto` holds, up to 2 s. Each picture waits for the sound stamped with it, and the canvas shows the last one -meanwhile. A stall longer than that delay pauses both and starts them again -together, and a burst after one skips ahead to the delay. Without audio, the +meanwhile. A sound stall longer than that delay pauses the picture too and +starts both again together; a picture stall leaves the sound playing. A burst +after a stall skips ahead to the delay. Without audio, the picture plays 100 ms behind. Sources shows how far behind each broadcast plays. diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 07b8a9cf..767cd55e 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -55,6 +55,10 @@ pub(crate) struct Speaker<'a> { /// The last frame placed: when it finishes sounding, and the media time it /// ends at. tail: Option<(Instant, Duration)>, + /// When the media after a skip starts sounding. The clock holds one + /// mapping, and the audio queued ahead of the skip plays on the old one + /// until then, so frames placed before it are not reported yet. + skipped: Option, } /// What to do with a frame [`Speaker::place`] was given. @@ -75,6 +79,7 @@ impl<'a> Speaker<'a> { clock, changed, tail: None, + skipped: None, } } @@ -99,6 +104,7 @@ impl<'a> Speaker<'a> { // ahead of it, however coarsely the bus advances. if follows < now + BUS_LATENCY / 2 { self.clock.audio_restarted(); + self.skipped = None; now + target + BUS_LATENCY } else if start.saturating_add(CONTIGUOUS) < media || follows.saturating_duration_since(now) > target + BUS_LATENCY + SLACK @@ -106,6 +112,7 @@ impl<'a> Speaker<'a> { // Media overlapping what is queued, or past the target, would // queue without bound however fast it arrives. self.tail = Some((sounds, media.max(end))); + self.skipped = Some(sounds); return Place::Skip; } else { follows @@ -114,6 +121,10 @@ impl<'a> Speaker<'a> { }; let sounds = begins + duration; self.tail = Some((sounds, end)); + if self.skipped.is_some_and(|skipped| now < skipped) { + return Place::Play(begins); + } + self.skipped = None; let audio = timeline::Audio { end, buffered: sounds.saturating_duration_since(now), @@ -173,12 +184,16 @@ impl Writer { } } - /// Write `data` at `now` to sound at `sounds`, or as soon after as it can. - /// Call it no sooner than [`BUS_LATENCY`] ahead. + /// Write `data` at `now` to sound at `sounds`. Call it no sooner than + /// [`BUS_LATENCY`] ahead. Written late, the part whose moment has passed is + /// dropped, so the rest still sounds on schedule, and so does what follows. pub(crate) fn write(&mut self, sounds: Instant, data: &[u8], now: Instant) -> Result<(), moq_audio::Error> { let free = self.runs_out.unwrap_or(now + BUS_LATENCY).max(now); - let gap = sounds.saturating_duration_since(free).min(BUS_LATENCY); - let gap = self.frames(gap); + let late = self.frames(free.saturating_duration_since(sounds)) * self.stride; + let Some(data) = data.get(late..).filter(|data| !data.is_empty()) else { + return Ok(()); + }; + let gap = self.frames(sounds.saturating_duration_since(free).min(BUS_LATENCY)); self.pending.clear(); self.pending.resize(gap * self.stride, 0); self.pending.extend_from_slice(data); @@ -287,8 +302,17 @@ mod tests { assert_eq!(clock.due(stamp(1_060)), Some(at(210))); // 1 s lost, with only the target queued: skipped back onto the target. + // The audio queued ahead of the skip keeps its pictures' moments until + // the media after it starts to sound. assert_eq!(speaker.place(ms(2_080), FRAME, TARGET, at(60)), Place::Skip); - assert_eq!(speaker.place(ms(2_100), FRAME, TARGET, at(61)), Place::Play(at(230))); + let schedule = clock.consume(); + for index in 0..9 { + let place = speaker.place(ms(2_100 + index * 20), FRAME, TARGET, at(61 + index * 20)); + assert_eq!(place, Place::Play(at(230 + index * 20))); + assert_eq!(clock.due(stamp(1_060)), Some(at(210)), "{index}"); + assert!(!heard(&schedule, stamp(2_100), at(61 + index * 20)), "{index}"); + } + assert_eq!(speaker.place(ms(2_280), FRAME, TARGET, at(241)), Place::Play(at(410))); assert_eq!(clock.due(stamp(2_100)), Some(at(230))); } @@ -304,7 +328,14 @@ mod tests { .filter(|place| *place != Place::Skip) .collect::>(); assert_eq!(placed, [Place::Play(at(150)), Place::Play(at(170))]); - assert_eq!(speaker.place(ms(1_200), FRAME, TARGET, at(20)), Place::Play(at(190))); + // What arrives next follows the queue, and the clock takes its media + // once the first of it sounds. + for index in 0..9 { + let place = speaker.place(ms(1_200 + index * 20), FRAME, TARGET, at(20 + index * 20)); + assert_eq!(place, Place::Play(at(190 + index * 20))); + assert_eq!(clock.due(stamp(1_000)), Some(at(150)), "{index}"); + } + assert_eq!(speaker.place(ms(1_380), FRAME, TARGET, at(200)), Place::Play(at(370))); assert_eq!(clock.due(stamp(1_200)), Some(at(190))); } @@ -329,7 +360,8 @@ mod tests { /// A frame sounds when placed: the first, one that follows it, and ones /// written after the channel ran dry, on time or late, which the bus would - /// otherwise play at once or pad behind, apart from their pictures. + /// otherwise play at once or pad behind, apart from their pictures. One + /// written after its moment loses what has passed, or all of it. #[test] fn a_write_sounds_when_placed() { let (bus, mut driver) = moq_audio::bus::Bus::new(Default::default()).unwrap(); @@ -385,6 +417,19 @@ mod tests { .write(clock(2 * rate) + BUS_LATENCY, &frame(4.0), clock(2 * rate + 1_920)) .unwrap(); advance(3 * rate, &mut heard); + // Run dry and written after its moment began, then after it ended. + writer + .write( + clock(3 * rate) - Duration::from_millis(10), + &frame(5.0), + clock(3 * rate), + ) + .unwrap(); + advance(4 * rate, &mut heard); + writer + .write(clock(4 * rate) - FRAME, &frame(6.0), clock(4 * rate)) + .unwrap(); + advance(5 * rate, &mut heard); let latency = rate * BUS_LATENCY.as_millis() as u64 / 1_000; let first = block + latency; @@ -399,7 +444,9 @@ mod tests { (after, 3.0), (after + 960, 0.0), (late, 4.0), - (late + 960, 0.0) + (late + 960, 0.0), + (3 * rate, 5.0), + (3 * rate + 480, 0.0) ] ); } From 4e434eede75b7c7ff1c70d2b3946299a0b79c216 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 15:17:27 -0800 Subject: [PATCH 7/7] fix(scene): commit a broadcast's clock past a skip when it sounds, not on the next arrival Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/playout.rs | 55 +++++++++++++++++++++++++++++-------- rs/moq-scene/src/source.rs | 32 +++++++++++++++------ 2 files changed, 67 insertions(+), 20 deletions(-) diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 767cd55e..7ec59038 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -57,8 +57,12 @@ pub(crate) struct Speaker<'a> { tail: Option<(Instant, Duration)>, /// When the media after a skip starts sounding. The clock holds one /// mapping, and the audio queued ahead of the skip plays on the old one - /// until then, so frames placed before it are not reported yet. + /// until then, so frames placed before it are reported only once it comes: + /// see [`commit`](Self::commit). skipped: Option, + /// The newest frame placed but not reported yet: when it finishes + /// sounding, the media time it ends at, and the target it was placed on. + unreported: Option<(Instant, Duration, Duration)>, } /// What to do with a frame [`Speaker::place`] was given. @@ -80,6 +84,7 @@ impl<'a> Speaker<'a> { changed, tail: None, skipped: None, + unreported: None, } } @@ -105,6 +110,7 @@ impl<'a> Speaker<'a> { if follows < now + BUS_LATENCY / 2 { self.clock.audio_restarted(); self.skipped = None; + self.unreported = None; now + target + BUS_LATENCY } else if start.saturating_add(CONTIGUOUS) < media || follows.saturating_duration_since(now) > target + BUS_LATENCY + SLACK @@ -121,10 +127,32 @@ impl<'a> Speaker<'a> { }; let sounds = begins + duration; self.tail = Some((sounds, end)); - if self.skipped.is_some_and(|skipped| now < skipped) { - return Place::Play(begins); + self.unreported = Some((sounds, end, target)); + if self.skipped.is_none_or(|skipped| skipped <= now) { + self.report(now); } + Place::Play(begins) + } + + /// When [`commit`](Self::commit) has a report to make, after a skip. + pub(crate) fn deferred(&self) -> Option { + self.unreported.and(self.skipped) + } + + /// Report the frames placed since a skip, once the media after it sounds at + /// `now`, whether or not another frame has arrived since. + pub(crate) fn commit(&mut self, now: Instant) { + if self.deferred().is_some_and(|skipped| skipped <= now) { + self.report(now); + } + } + + /// Tell the clock where the newest frame placed sounds, as of `now`. + fn report(&mut self, now: Instant) { self.skipped = None; + let Some((sounds, end, target)) = self.unreported.take() else { + return; + }; let audio = timeline::Audio { end, buffered: sounds.saturating_duration_since(now), @@ -132,7 +160,6 @@ impl<'a> Speaker<'a> { }; self.clock.audio(audio, now); self.changed.notify_one(); - Place::Play(begins) } } @@ -306,14 +333,17 @@ mod tests { // the media after it starts to sound. assert_eq!(speaker.place(ms(2_080), FRAME, TARGET, at(60)), Place::Skip); let schedule = clock.consume(); - for index in 0..9 { - let place = speaker.place(ms(2_100 + index * 20), FRAME, TARGET, at(61 + index * 20)); - assert_eq!(place, Place::Play(at(230 + index * 20))); - assert_eq!(clock.due(stamp(1_060)), Some(at(210)), "{index}"); - assert!(!heard(&schedule, stamp(2_100), at(61 + index * 20)), "{index}"); - } - assert_eq!(speaker.place(ms(2_280), FRAME, TARGET, at(241)), Place::Play(at(410))); + assert_eq!(speaker.place(ms(2_100), FRAME, TARGET, at(61)), Place::Play(at(230))); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); + assert!(!heard(&schedule, stamp(2_100), at(61))); + // It sounds whether or not anything arrives after it. + assert_eq!(speaker.deferred(), Some(at(230))); + speaker.commit(at(229)); + assert_eq!(clock.due(stamp(1_060)), Some(at(210))); + speaker.commit(at(230)); + assert_eq!(speaker.deferred(), None); assert_eq!(clock.due(stamp(2_100)), Some(at(230))); + assert!(heard(&schedule, stamp(2_100), at(230))); } /// A burst queues no more than the target and a frame of slack: the rest @@ -329,7 +359,7 @@ mod tests { .collect::>(); assert_eq!(placed, [Place::Play(at(150)), Place::Play(at(170))]); // What arrives next follows the queue, and the clock takes its media - // once the first of it sounds. + // once the first of it sounds, on arrival or not. for index in 0..9 { let place = speaker.place(ms(1_200 + index * 20), FRAME, TARGET, at(20 + index * 20)); assert_eq!(place, Place::Play(at(190 + index * 20))); @@ -337,6 +367,7 @@ mod tests { } assert_eq!(speaker.place(ms(1_380), FRAME, TARGET, at(200)), Place::Play(at(370))); assert_eq!(clock.due(stamp(1_200)), Some(at(190))); + assert_eq!(speaker.deferred(), None); } /// Media overlapping what is queued, stamped the same or a millisecond diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 23dead6c..5c7ca861 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -947,7 +947,7 @@ impl Driver { input.format = moq_audio::Format::F32; let channel = self.context.bus.channel(input, strip)?; let mut writer = Writer::new(channel, bus.sample_rate, bus.layout.channels()); - let mut speaker = Speaker::new(clock.timeline.clone(), &clock.changed); + let speaker = RefCell::new(Speaker::new(clock.timeline.clone(), &clock.changed)); let stride = bus.layout.channels() as usize * size_of::(); let rate = f64::from(bus.sample_rate); @@ -964,7 +964,10 @@ impl Driver { continue; }; let duration = Duration::from_secs_f64((frame.data.len() / stride) as f64 / rate); - let place = speaker.place(frame.timestamp.into(), duration, consumer.delay(), kio::clock::now()); + let now = kio::clock::now(); + let place = speaker + .borrow_mut() + .place(frame.timestamp.into(), duration, consumer.delay(), now); if let Place::Play(sounds) = place { queue.borrow_mut().push_back((sounds, frame)); queued.notify_one(); @@ -974,14 +977,27 @@ impl Driver { queued.notify_one(); Ok::<_, Error>(()) }; + // Writes each frame its bus latency ahead, and tells the clock about + // the frames after a skip once they sound. let write = async { loop { - let next = queue.borrow_mut().pop_front(); - match next { - Some((sounds, frame)) => { - kio::clock::sleep_until(sounds - crate::playout::BUS_LATENCY).await; - writer.write(sounds, &frame.data, kio::clock::now())?; - } + let now = kio::clock::now(); + speaker.borrow_mut().commit(now); + let write = queue + .borrow() + .front() + .map(|(sounds, _)| *sounds - crate::playout::BUS_LATENCY); + if write.is_some_and(|at| at <= now) { + let (sounds, frame) = queue.borrow_mut().pop_front().expect("peeked"); + writer.write(sounds, &frame.data, now)?; + continue; + } + let wake = write.into_iter().chain(speaker.borrow().deferred()).min(); + match wake { + Some(at) => tokio::select! { + () = kio::clock::sleep_until(at) => {} + () = queued.notified() => {} + }, None if received.get() => return Ok::<_, Error>(()), None => queued.notified().await, }