From 5c163757b6f54da9689fb240f7f606a42d49ba4d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 06:48:52 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m0/media-av-sync.md Quest-Claim: 1924528555cce7fd From c91680a32432862550e0650d309ca3e3b753a460 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:06:30 -0800 Subject: [PATCH 2/7] fix(studio): a media file's sound plays with its pictures A media source wrote each sound to the bus when it was due, and the bus sounds a write its 50 ms latency later, so the sound trailed the pictures in every output. It now writes through playout::Writer a bus latency ahead, as broadcast sources do, and hands each picture over then too, to a presenter that shows it at its moment. A play begins a bus latency after it decodes, and a loop hands its next pass over a bus latency before the one playing ends. Co-Authored-By: Claude Opus 5.5 --- quest/m0/README.md | 1 - quest/m0/media-av-sync.md | 20 -- rs/moq-scene/src/playout.rs | 4 +- rs/moq-scene/src/source.rs | 386 +++++++++++++++++++++++++++++------- rs/moq-studio/src/engine.rs | 14 +- 5 files changed, 318 insertions(+), 107 deletions(-) delete mode 100644 quest/m0/media-av-sync.md diff --git a/quest/m0/README.md b/quest/m0/README.md index 302d5fcf..fce9847e 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -102,7 +102,6 @@ repository. - [Browser sources on the GPU under Mesa](/quest/m0/cef-headless.md) - shared-texture pages paint on Intel and AMD under Wayland, and a page that cannot paint fails loudly - [Browser sources make no outside requests](/quest/m0/browser-offline.md) - no component updates, spellcheck downloads, or network time from the helper - [Unpublishing a scene ends its broadcast cleanly](/quest/m0/unpublish-clean.md) - viewers get a clean finish, not `Unroutable`, when a scene stops publishing -- [A media file's sound plays with its pictures](/quest/m0/media-av-sync.md) - media sources write a bus latency ahead, as broadcast sources do - [The Apple signing identity is available](/quest/m0/signing.md) - the Apple credentials packaging and the macOS camera need; Windows ships unsigned - [Packaging](/quest/m0/packaging.md) - a signed macOS app, a Windows installer, and a Flatpak for Linux - [The persistable portal restore token is pinned](/quest/m0/pin-portal-token.md) - moq-video's restore token, for Wayland sources diff --git a/quest/m0/media-av-sync.md b/quest/m0/media-av-sync.md deleted file mode 100644 index 7dda851c..00000000 --- a/quest/m0/media-av-sync.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] A media file's sound plays with its pictures - -## Goal - -A media file source's sound plays in step with its own pictures in every -output (program, published scenes, recordings, and Listen), instead of about -45 ms behind them. - -## Plan - -From [moq-dev/studio#81](https://github.com/moq-dev/studio/pull/81), picked -2026-11-07 (maintainer away under /afk, recommended option), m0 by the rule -that fixes join m0. - -Pictures show when they are due, but a media source writes its sound to the -bus when it is due, and the bus mixes it a bus latency (50 ms) later. -Broadcast sources already write a bus latency ahead (`playout::Writer`); -media sources should do the same. Test that a media file's picture change -and its sound onset leave an output within one bus block of each other, -measured on the media clock rather than with sleeps. diff --git a/rs/moq-scene/src/playout.rs b/rs/moq-scene/src/playout.rs index 7ec59038..12afc98b 100644 --- a/rs/moq-scene/src/playout.rs +++ b/rs/moq-scene/src/playout.rs @@ -9,7 +9,9 @@ //! `moq play` holds its jitter buffer in the speaker. The bus holds only //! [`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. +//! and a [`Writer`] writes it the bus's latency ahead of then. A media +//! source's sound goes through a [`Writer`] too, so it sounds with the +//! pictures it shows. use std::time::Duration; diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index e4c64660..7c592bbb 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -31,7 +31,7 @@ use crate::browser::Page; use crate::collection::{Collection, Source}; use crate::lut::Cube; use crate::media::Clip; -use crate::playout::{MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; +use crate::playout::{BUS_LATENCY, MAX_DELAY, Place, Speaker, VIDEO_DELAY, Writer}; use crate::publish::{Claim, Listing, Publish, Still}; use crate::replay::{MIN_SPEED, Take, Target}; @@ -131,11 +131,13 @@ pub struct Run { /// Which play it is: 0 for the one the source opens with, and one more /// for each [`Feed::play`] since. pub play: u64, - /// When its first picture or sound was due, or `None` until it is decoded. + /// When its first picture shows and sound sounds, or `None` until it is + /// decoded. Set a little ahead: the bus sounds a write 50 ms later, so a + /// play begins that long after it is decoded. pub begun: Option, - /// When its last picture or sound finished, a looping one's first pass - /// did, or [`Feed::stop`] stopped it, or `None` until then. A play cut off - /// by the next never ends. + /// When its last picture or sound finishes, a looping one's first pass + /// does, or [`Feed::stop`] stopped it, or `None` until known. A looping + /// one's is known 50 ms ahead. A play cut off by the next never ends. pub ended: Option, } @@ -1230,7 +1232,7 @@ impl Driver { let write = queue .borrow() .front() - .map(|(sounds, _)| *sounds - crate::playout::BUS_LATENCY); + .map(|(sounds, _)| *sounds - 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)?; @@ -1318,7 +1320,7 @@ impl Driver { moq_video::decode::Read::Frame(frame) if frame.timestamp >= take.end => break, moq_video::decode::Read::Frame(mut frame) => { kio::clock::sleep_until(at(frame.timestamp)).await; - self.encode(&mut own, &mut frame, config).await?; + self.encode(&mut own, &mut frame, config, kio::clock::now()).await?; self.show(Picture::Frame(frame)); } // A stretch the broadcast was offline, paused, or lost @@ -1383,7 +1385,8 @@ impl Driver { /// picture, silent, until the next. /// /// With `listing`, the pictures are also encoded onto the source's own - /// track as they show, while it is used, on one timeline across plays. + /// track, stamped when they show, while it is used, on one timeline across + /// plays. async fn media( &self, path: &Path, @@ -1404,7 +1407,7 @@ impl Driver { self.publish(crate::publish::Audio::new(listing, input)) }); Some(Sound { - channel: None, + writer: None, track, ended: None, }) @@ -1412,64 +1415,103 @@ impl Driver { _ => None, }; let mut own = Own::new(listing, kio::clock::now()); + let shows = Shows::default(); // The last channel on a strip to close drops what it has not played // yet, so a loop keeps its channel, sounding on, while a new cue drops - // it, cutting off what the play before still had queued; the next play - // takes a new one. Its pictures start a group there too, a cut to the - // clip's start. + // it, cutting off what the play before still had queued, pictures + // included; the next play takes a new one. Its pictures start a group + // there too, a cut to the clip's start. let cut = |sound: &mut Option>, own: &mut Own<'_, '_>| { if let Some(sound) = sound { - sound.channel = None; + sound.writer = None; sound.ended = None; } + shows.queue.borrow_mut().clear(); if let Some(video) = &mut own.track && let Err(err) = video.cut() { self.fail_encoding(&mut own.track, err); } }; - let mut cue = *cues.borrow_and_update(); - // Where a loop's next play starts: where the last one ended, so loops - // keep exact time whatever opening the decoders takes. - let mut next = None; - loop { - // Whether to rest until the next cue, rather than take it at once. - let rest = if cue.stopped { - true - } else { - if let (Some(sound), Some(strip)) = (&mut sound, &self.strip) - && sound.channel.is_none() - { - sound.channel = Some(self.channel(strip.clone())?); - } - let played = tokio::select! { - result = self.play_clip(&clip, sound.as_mut(), &mut own, next, cue.play) => Some(result?), - changed = cues.changed() => { - // The feed is gone, which stops this driver anyway. - if changed.is_err() { - return std::future::pending().await; + let plays = async { + let mut cue = *cues.borrow_and_update(); + // Where a loop's next play starts: where the last one ended, so + // loops keep exact time whatever opening the decoders takes. + let mut next = None; + loop { + // Whether to rest until the next cue, rather than take it at once. + let rest = if cue.stopped { + true + } else { + if let (Some(sound), Some(strip)) = (&mut sound, &self.strip) + && sound.writer.is_none() + { + let bus = self.context.bus.config(); + let channel = self.channel(strip.clone())?; + sound.writer = Some(Writer::new(channel, bus.sample_rate, bus.layout.channels())); + } + let play = async { + let end = self.play_clip(&clip, sound.as_mut(), &mut own, &shows, next, cue.play).await?; + // A loop's next play hands its first over now, a bus + // latency ahead, as this one did. + if !looping { + kio::clock::sleep_until(end).await; + } + Ok::<_, Error>(end) + }; + let played = tokio::select! { + result = play => Some(result?), + changed = cues.changed() => { + // The feed is gone, which stops this driver anyway. + if changed.is_err() { + return std::future::pending().await; + } + None + } + }; + if let Some(end) = played { + self.ran(cue.play, |run| { + run.ended.get_or_insert(end); + }); + next = Some(end); + if looping { + continue; } - None } + played.is_some() }; - if let Some(end) = played { - self.ran(cue.play, |run| { - run.ended.get_or_insert(end); - }); - next = Some(end); - if looping { - continue; + if rest { + set(&self.status, Status::Ended); + self.ended(&mut cues, &mut own).await; + } + cut(&mut sound, &mut own); + cue = *cues.borrow_and_update(); + next = None; + } + }; + // Shows each picture handed over at its moment, across plays, so a + // loop's next play can hand its first over before this one's last + // has shown. + let present = async { + loop { + let now = kio::clock::now(); + let due = shows.queue.borrow().front().map(|(due, ..)| *due); + match due { + Some(due) if due <= now => { + let (_, play, frame) = shows.queue.borrow_mut().pop_front().expect("peeked"); + self.show_play(play, Picture::Frame(frame)); } + Some(due) => tokio::select! { + () = kio::clock::sleep_until(due) => {} + () = shows.handed.notified() => {} + }, + None => shows.handed.notified().await, } - played.is_some() - }; - if rest { - set(&self.status, Status::Ended); - self.ended(&mut cues, &mut own).await; } - cut(&mut sound, &mut own); - cue = *cues.borrow_and_update(); - next = None; + }; + tokio::select! { + result = plays => result, + () = present => unreachable!("pictures show for as long as the source plays"), } } @@ -1521,16 +1563,24 @@ impl Driver { } } - /// Play `clip` once, each picture and sound when its moment comes, the - /// first at `begin`, or at once, as part of the feed's `play`. Returns - /// where it ends, once the last has played for as long as it lasts: a - /// sound for its samples, and a picture for as long as the one before it. - /// A clip that decodes to nothing fails. + /// Play `clip` once, each picture and sound at its moment, the first at + /// `begin`, or as soon as the bus can sound it, as part of the feed's + /// `play`. + /// + /// The bus sounds a write [`BUS_LATENCY`] later, so each sound is written + /// that far ahead of its moment, and each picture handed to `shows` then + /// too, to show at the same moment. + /// + /// Returns where it ends, once all is handed over and the bus latency + /// before the end has come, so a loop's next play hands its own first over + /// in time. A sound lasts for its samples, and the last picture as long as + /// the one before it. A clip that decodes to nothing fails. async fn play_clip( &self, clip: &Clip, mut sound: Option<&mut Sound<'_>>, own: &mut Own<'_, '_>, + shows: &Shows, begin: Option, play: u64, ) -> Result { @@ -1575,13 +1625,22 @@ impl Driver { if let (Some(frame), Some((_, config))) = (&picture, &video) { self.open(own, frame, config).await; } - let begin = begin.unwrap_or_else(kio::clock::now); + let begin = begin.unwrap_or_else(|| kio::clock::now() + BUS_LATENCY); self.ran(play, |run| { run.begun.get_or_insert(begin); }); let at = |timestamp: moq_net::Timestamp| begin + Duration::from(timestamp).saturating_sub(start); let end = std::cell::Cell::new(begin); let last = |until: Instant| end.set(end.get().max(until)); + // What is due at `due` is handed over a bus latency ahead. One due at + // once is handed over without a timer, which the timer wheel would + // round up to its next millisecond, a moment too late. + let ahead = async |due: Instant| { + let at = due - BUS_LATENCY; + if at > kio::clock::now() { + kio::clock::sleep_until(at).await; + } + }; let pictures = async { let Some((consumer, config)) = &mut video else { @@ -1590,11 +1649,12 @@ impl Driver { let mut previous = None; while let Some(mut frame) = picture.take() { let due = at(frame.timestamp); - kio::clock::sleep_until(due).await; + ahead(due).await; last(due + previous.map_or(Duration::ZERO, |previous| due - previous)); previous = Some(due); - self.encode(own, &mut frame, config).await?; - self.show_play(play, Picture::Frame(frame)); + self.encode(own, &mut frame, config, due).await?; + shows.queue.borrow_mut().push_back((due, play, frame)); + shows.handed.notify_one(); picture = next_picture(consumer).await?; } Ok::<_, Error>(()) @@ -1612,16 +1672,16 @@ impl Driver { let mut first = true; while let Some(frame) = samples.take() { let due = at(frame.timestamp); - kio::clock::sleep_until(due).await; + ahead(due).await; let length = (frame.data.len() / bytes) as u64 * 1_000_000_000 / u64::from(bus.sample_rate); last(due + Duration::from_nanos(length)); let gap = std::mem::take(&mut first) && sound.ended != Some(due); sound.ended = Some(due + Duration::from_nanos(length)); sound - .channel + .writer .as_mut() .expect("a channel while playing") - .write(&frame.data)?; + .write(due, &frame.data, kio::clock::now())?; if let Some(track) = &mut sound.track { let samples: Vec = bytemuck::pod_collect_to_vec(&frame.data); if let Err(err) = track.write(&samples, due, gap) { @@ -1635,8 +1695,9 @@ impl Driver { Ok::<_, Error>(()) }; tokio::try_join!(pictures, sounds)?; - kio::clock::sleep_until(end.get()).await; - Ok(end.get()) + let end = end.get(); + ahead(end).await; + Ok(end) } /// Open the source's own track for pictures like `frame`, which says their @@ -1652,18 +1713,20 @@ impl Driver { } /// Encode a played picture onto the source's own track, opened at the - /// first picture if not before. Stamped when it shows, on the timeline - /// `own` keeps. + /// first picture if not before. Stamped with when it `shows`, on the + /// timeline `own` keeps. async fn encode( &self, own: &mut Own<'_, '_>, frame: &mut moq_video::Frame, config: &VideoConfig, + shows: Instant, ) -> Result<(), Error> { self.open(own, frame, config).await; if let Some(video) = &mut own.track { - frame.timestamp = moq_net::Timestamp::from_micros(own.origin.elapsed().as_micros() as u64) - .map_err(moq_video::Error::from)?; + let stamp = shows.saturating_duration_since(own.origin); + frame.timestamp = + moq_net::Timestamp::from_micros(stamp.as_micros() as u64).map_err(moq_video::Error::from)?; if let Err(err) = video.encode(frame) { self.fail_encoding(&mut own.track, err); } @@ -1705,13 +1768,21 @@ impl Driver { /// A media source's sound: the bus channel it plays into, taken anew on a /// restart, and its own track. struct Sound<'a> { - channel: Option, + writer: Option, track: Option>, /// When the last sound written ends, or `None` after a break, such as a /// restart, so the next sound's track starts its timeline afresh. ended: Option, } +/// A media source's pictures, handed over a bus latency ahead of their moments +/// with their plays, and woken whenever one is. +#[derive(Default)] +struct Shows { + queue: RefCell>, + handed: Notify, +} + /// A played source's own track, opened at its first picture, and the timeline /// its pictures are stamped on: from `origin`, when they show. struct Own<'a, 'b> { @@ -2914,16 +2985,22 @@ mod media_tests { } /// The next picture `feed` shows, passing over the clear a restart puts up, - /// and when it showed. + /// and when it showed. The clock moves only to a play's beginning, which + /// comes a bus latency after its first picture decodes. async fn shown(feed: &Feed) -> (moq_net::Timestamp, Instant) { - let timestamp = until(|| match feed.next_picture() { - Some(Picture::Frame(frame)) => Some(frame.timestamp), - Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4] => None, - Some(_) => panic!("not a decoded picture"), - None => None, - }) - .await; - (timestamp, kio::clock::now()) + loop { + let next = until(|| match feed.next_picture() { + Some(Picture::Frame(frame)) => Some(Ok(frame.timestamp)), + Some(Picture::Rgba { pixels, .. }) if pixels == [0; 4] => None, + Some(_) => panic!("not a decoded picture"), + None => feed.run().begun.filter(|begun| *begun > kio::clock::now()).map(Err), + }) + .await; + match next { + Ok(timestamp) => return (timestamp, kio::clock::now()), + Err(begun) => until_time(begun).await, + } + } } /// Wait until `name` reads as `status`, without moving the clock. @@ -3184,6 +3261,161 @@ mod media_tests { .await; } + /// One bus block: how often the bus mixes, and so how finely a mix places + /// a sound in time. + const BLOCK: Duration = Duration::from_millis(10); + + /// The bus mixed along with a paused clock a block at a time, as an audio + /// clock would, keeping the first channel of what one mix heard. + struct Ear { + mixer: moq_audio::bus::Driver, + reader: moq_audio::bus::Reader, + /// When the clock read the bus's start, so a frame `n` in sounds `n` + /// samples after it. + origin: Instant, + at: Instant, + heard: Vec, + } + + impl Ear { + /// Start mixing now, a block of which the bus mixes at once, so a + /// channel keeps what is written to it from here on. + fn new(mixer: moq_audio::bus::Driver, mix: &Mix) -> Self { + let now = kio::clock::now(); + let mut ear = Self { + mixer, + reader: mix.reader(), + origin: now - BLOCK, + at: now, + heard: Vec::new(), + }; + ear.mix(); + ear + } + + /// Move the clock a block on, and mix that block. + async fn step(&mut self) { + self.at += BLOCK; + until_time(self.at).await; + self.mix(); + } + + fn mix(&mut self) { + self.mixer.advance(Self::frames(BLOCK)); + while let std::task::Poll::Ready(Some(block)) = self.reader.poll_read(&kio::Waiter::noop()) { + let channels = block.data.len() / block.frames(); + assert_eq!(block.position as usize, self.heard.len(), "a block skipped"); + self.heard.extend(block.data.iter().step_by(channels)); + } + } + + /// The frames the bus mixes over `duration`. + fn frames(duration: Duration) -> usize { + let rate = moq_audio::bus::Config::default().sample_rate; + (duration.as_nanos() * u128::from(rate) / 1_000_000_000) as usize + } + + /// When the mix first heard anything. + fn onset(&self) -> Option { + let index = self.heard.iter().position(|sample| sample.abs() > 0.01)?; + let rate = moq_audio::bus::Config::default().sample_rate; + Some(self.origin + Duration::from_secs_f64(index as f64 / f64::from(rate))) + } + + /// The loudest the mix heard over `duration` from `from`. + fn peak(&self, from: Instant, duration: Duration) -> f32 { + let start = Self::frames(from - self.origin); + let heard = &self.heard[start..start + Self::frames(duration)]; + heard.iter().fold(0.0, |peak, sample| peak.max(sample.abs())) + } + } + + /// A file's first picture shows as its first sound is heard, within a bus + /// block, rather than a bus latency ahead of it. + #[tokio::test(start_paused = true)] + async fn a_media_files_sound_plays_with_its_pictures() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (mut context, _origin) = context(dir.path()); + let (bus, mixer) = Bus::new(Default::default()).unwrap(); + context.bus = bus; + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("clip.mkv", CLIP, false, false)]); + let drivers = sources.sync(&collection); + let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); + let mut ear = Ear::new(mixer, &program); + start(&sources, drivers).await; + let feed = sources.get("clip.mkv").unwrap(); + + while !matches!(feed.next_picture(), Some(Picture::Frame(_))) { + ear.step().await; + } + // Seen at the step after it showed, which may be the block after, + // when the step woke before the picture's timer at the same moment. + let shown = ear.at; + let begun = feed.run().begun.unwrap(); + assert!( + begun <= shown && shown - begun <= BLOCK, + "the first picture shows when it begins: {:?} after", + shown - begun + ); + while ear.at < begun + Duration::from_millis(100) { + ear.step().await; + } + let onset = ear.onset().expect("the clip sounds"); + let apart = onset + .saturating_duration_since(begun) + .max(begun.saturating_duration_since(onset)); + assert!( + apart < BLOCK, + "its first sound is heard with its first picture: {onset:?} and {begun:?}" + ); + }) + .await; + } + + /// A looping file's sound runs on from its end into its start again, as + /// loud as the first time and as its pass begins, each pass heard from + /// when it begins. + #[tokio::test(start_paused = true)] + async fn a_looping_files_sound_runs_on() { + local(async { + let dir = tempfile::tempdir().unwrap(); + let (mut context, _origin) = context(dir.path()); + let (bus, mixer) = Bus::new(Default::default()).unwrap(); + context.bus = bus; + let mut sources = Sources::new(context); + let collection = media(dir.path(), &[("tone.webm", TONE, true, false)]); + let drivers = sources.sync(&collection); + let program = sources.mix(crate::AudioScene::DEFAULT).unwrap().clone(); + let mut ear = Ear::new(mixer, &program); + start(&sources, drivers).await; + let feed = sources.get("tone.webm").unwrap(); + + let ended = loop { + if let Some(ended) = feed.run().ended { + break ended; + } + ear.step().await; + }; + let begun = feed.run().begun.unwrap(); + while ear.at < ended + Duration::from_millis(100) { + ear.step().await; + } + // Past a block that either may be heard early by, and the start of + // the tone the codec fades in. + let into = Duration::from_millis(15); + let first = ear.peak(begun + into, Duration::from_millis(30)); + let again = ear.peak(ended + into, Duration::from_millis(30)); + assert!(first > 0.05, "the tone sounds: {first}"); + assert!( + again > first / 2.0, + "the loop starts again as loud: {again} after {first}" + ); + }) + .await; + } + /// A file of sound alone shows nothing, and goes silent at its end. One of /// pictures alone has no sound for a mix to play. #[tokio::test(start_paused = true)] diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 5ae01b27..d7244c48 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -3949,19 +3949,17 @@ mod tests { // Until then, the program's viewer heard nothing while the portrait's // heard the bed. assert!(bed < played, "the bed is heard before the cut"); - let (heard, before) = program.hears(quiet, played, LOUD).await; + // The viewer hears the tone as it plays: a media source writes each + // sound the bus's latency ahead, which places it within the block the + // bus mixes it in. + let (heard, before) = program.hears(quiet, played - AUDIO_PERIOD, LOUD).await; assert!( before < QUIET, "the program's audio scene holds the tone alone: {before}" ); - // The viewer hears the tone once the bus mixes it: a media source - // writes each sound when it is due, and the bus holds every source - // 50 ms (moq-audio's `bus::LATENCY`) before mixing it, give or take a - // block. - let mixed = Duration::from_millis(50) + AUDIO_PERIOD; assert!( - heard >= played && heard - played < mixed, - "the tone is heard as the bus mixes it: played {played:?}, heard {heard:?}" + heard.abs_diff(played) < AUDIO_PERIOD, + "the tone is heard as it plays: played {played:?}, heard {heard:?}" ); let (also, _) = tall.hears(quiet, played, LOUD).await; assert!( From 735b0a8bf135ea8e3d41737d2608b83f49267a63 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:21:38 -0800 Subject: [PATCH 3/7] test(studio): media tests wait for a play to begin A media play now begins a bus latency after its first picture decodes, so tests walk its pictures from when it began rather than from when it was asked. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 38 ++++++++++++++++++------------------- rs/moq-scene/src/source.rs | 9 ++++----- 2 files changed, 23 insertions(+), 24 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 9ff46f1f..576257d0 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -789,17 +789,20 @@ mod tests { } } - /// Walk a media source playing a 10 fps clip through `count` pictures, the - /// first due at `first`. The clock moves to each picture's moment and is - /// held while it decodes, encodes, and shows, so no wait counts on work in - /// threads the runtime cannot see, such as a decoder's or an encoder's, - /// finishing within any span of the clock. - async fn pictures(feed: &crate::source::Feed, first: kio::clock::Instant, count: u32) { + /// Walk a media source playing a 10 fps clip through `count` of its play's + /// pictures from the one at index `from`, returning when the play began. + /// The clock is held until the play begins, then moves to each picture's + /// moment and is held while it decodes, encodes, and shows, so no wait + /// counts on work in threads the runtime cannot see, such as a decoder's or + /// an encoder's, finishing within any span of the clock. + async fn pictures(feed: &crate::source::Feed, from: u32, count: u32) -> kio::clock::Instant { + let begun = until(|| feed.run().begun).await; let shown = || matches!(feed.next_picture(), Some(crate::source::Picture::Frame(_))).then_some(()); - for index in 0..count { - until_time(first + std::time::Duration::from_millis(100) * index).await; + for index in from..from + count { + until_time(begun + std::time::Duration::from_millis(100) * index).await; until(shown).await; } + begun } /// Yield until `f` holds of the local catalog's source listings. @@ -976,7 +979,7 @@ mod tests { tracks.push(consumer.track(name).unwrap().subscribe(subscription).await.unwrap()); } let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); - pictures(sources.get("bed").unwrap(), kio::clock::now(), 4).await; + pictures(sources.get("bed").unwrap(), 0, 4).await; until(|| names.iter().all(|name| written(name) > 0).then_some(())).await; sources.stop(); @@ -1564,8 +1567,8 @@ mod tests { let drivers = sources.sync(&collection); start(&sources, drivers).await; - // The first picture is due at once, so the clock stays put while - // the encoder opens. + // The pictures' track opens before the play begins, so the clock + // stays put while the encoder opens. let listed = until(|| { let listed = catalog.snapshot().ext.studio.sources.remove("clip"); listed.filter(|listed| !listed.video.renditions.is_empty()) @@ -1586,13 +1589,11 @@ mod tests { let mut track = moq_mux::container::Consumer::new(track, container); // Half way, five pictures in, it starts over, and plays to its end. - let first = kio::clock::now(); - pictures(feed, first, 5).await; + let first = pictures(feed, 0, 5).await; until_time(first + std::time::Duration::from_millis(450)).await; let cut = publish.catalog.clock().now(); feed.play().unwrap(); - let restarted = kio::clock::now(); - pictures(feed, restarted, 10).await; + let restarted = pictures(feed, 0, 10).await; until_time(restarted + std::time::Duration::from_millis(1_100)).await; until(|| (feed.status() == Status::Ended).then_some(())).await; @@ -1657,8 +1658,7 @@ mod tests { let written = |id: &str| publish.stats().get(id).map_or(0, |rendition| rendition.frames); let feed = sources.get("bed").unwrap(); - let first = kio::clock::now(); - pictures(feed, first, 4).await; + pictures(feed, 0, 4).await; until(|| (written(&audio) > 0).then_some(())).await; assert_eq!(written(&video), 0, "heard, not shown"); assert_eq!(sessions.open(), 0, "no encoder for the pictures"); @@ -1669,7 +1669,7 @@ mod tests { shown: ["bed".to_string()].into(), heard: ["bed".to_string()].into(), }); - pictures(feed, first + std::time::Duration::from_millis(400), 6).await; + pictures(feed, 4, 6).await; until(|| (written(&video) > 0).then_some(())).await; }) .await; @@ -1715,7 +1715,7 @@ mod tests { let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); let mut track = moq_mux::container::Consumer::new(track, container); // Two and a half loops of a second, each with half a second of sound. - pictures(sources.get("bed").unwrap(), kio::clock::now(), 25).await; + pictures(sources.get("bed").unwrap(), 0, 25).await; let now = kio::Waiter::noop(); let mut stamps = Vec::new(); let silences = until(|| { diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 7c592bbb..c3fc923e 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1229,10 +1229,7 @@ impl Driver { loop { let now = kio::clock::now(); speaker.borrow_mut().commit(now); - let write = queue - .borrow() - .front() - .map(|(sounds, _)| *sounds - BUS_LATENCY); + let write = queue.borrow().front().map(|(sounds, _)| *sounds - 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)?; @@ -1451,7 +1448,9 @@ impl Driver { sound.writer = Some(Writer::new(channel, bus.sample_rate, bus.layout.channels())); } let play = async { - let end = self.play_clip(&clip, sound.as_mut(), &mut own, &shows, next, cue.play).await?; + let end = self + .play_clip(&clip, sound.as_mut(), &mut own, &shows, next, cue.play) + .await?; // A loop's next play hands its first over now, a bus // latency ahead, as this one did. if !looping { From 08a6383f06bc62de49fe421a627c8ba46cb9978b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:32:35 -0800 Subject: [PATCH 4/7] test(studio): bound the cut's tone by how the bus places a first write Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/engine.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index d7244c48..9130447e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -3950,15 +3950,17 @@ mod tests { // heard the bed. assert!(bed < played, "the bed is heard before the cut"); // The viewer hears the tone as it plays: a media source writes each - // sound the bus's latency ahead, which places it within the block the - // bus mixes it in. - let (heard, before) = program.hears(quiet, played - AUDIO_PERIOD, LOUD).await; + // sound the bus's latency ahead. The bus holds a first write that long + // from the block it last mixed, so the tone may be heard up to a block + // early, or two while the audio clock's tick runs late. + let early = 2 * AUDIO_PERIOD; + let (heard, before) = program.hears(quiet, played - early, LOUD).await; assert!( before < QUIET, "the program's audio scene holds the tone alone: {before}" ); assert!( - heard.abs_diff(played) < AUDIO_PERIOD, + heard + early > played && heard < played + AUDIO_PERIOD, "the tone is heard as it plays: played {played:?}, heard {heard:?}" ); let (also, _) = tall.hears(quiet, played, LOUD).await; From dc44da45a669b71e841d758708f84729af938e9c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 07:54:27 -0800 Subject: [PATCH 5/7] fix(studio): a media source's own tracks are stamped when they play Handed over a bus latency ahead, a media source's own pictures and sound took their stamps from when they were encoded, 50 ms before they showed and sounded. The picture track now anchors its timeline at the source's origin, and the sound track counts a stamp ahead of the clock on from now, so both carry when they play. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/publish.rs | 41 +++++++++++++++++++++++++++++-------- rs/moq-scene/src/source.rs | 5 ++++- 2 files changed, 36 insertions(+), 10 deletions(-) diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 576257d0..55f9bb02 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -487,7 +487,7 @@ pub(crate) struct Video<'a> { config: moq_video::encode::Config, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in - /// microseconds, taken at the first frame. + /// microseconds, taken at the first frame unless [anchored](Self::anchor). epoch: Option, } @@ -521,6 +521,14 @@ impl<'a> Video<'a> { }) } + /// Start the private timeline at `origin` on the studio clock, rather than + /// where the first frame encodes, for a source that encodes its pictures + /// ahead of when they show. + pub(crate) fn anchor(&mut self, origin: kio::clock::Instant) -> Result<(), Error> { + self.epoch = Some(self.clock.capture(origin)?.as_micros()); + Ok(()) + } + /// Restamp `frame` onto the program's clock, and encode and publish it /// while the source is used: shown on program, or its track read. The /// encoder opens on the first frame used and closes while none is, marking @@ -667,10 +675,11 @@ impl<'a> Audio<'a> { }) } - /// Encode and publish `samples`, whose first was captured at `captured`, - /// while the source is used: heard on program, or its track read. `gap` - /// says sound was lost just before them. Leaving use marks a gap, and the - /// sound after one is stamped afresh, so the track's timeline holds it. + /// Encode and publish `samples`, whose first was captured or sounds at + /// `captured`, while the source is used: heard on program, or its track + /// read. `gap` says sound was lost just before them. Leaving use marks a + /// gap, and the sound after one is stamped afresh, so the track's timeline + /// holds it. pub(crate) fn write(&mut self, samples: &[f32], captured: kio::clock::Instant, gap: bool) -> Result<(), Error> { let producer = self.producer.as_mut().expect("open until it ends"); let used = self.on_program.borrow().heard.contains(&self.claim.name) || producer.demand().wanted().is_some(); @@ -684,9 +693,13 @@ impl<'a> Audio<'a> { producer.reanchor(); } self.encoding = true; - // A device may stamp a buffer a moment ahead of the clock, which the - // clock refuses as not yet captured. - let timestamp = self.clock.capture(captured.min(kio::clock::now()))?; + // A media source writes its sound ahead of when it sounds, and a device + // may stamp a buffer a moment ahead of the clock, which maps only what + // has passed: the rest is counted on from now. + let now = kio::clock::now(); + let ahead = captured.saturating_duration_since(now).as_micros(); + let timestamp = self.clock.capture(captured.min(now))?.as_micros() + ahead; + let timestamp = moq_net::Timestamp::from_micros(timestamp as u64).map_err(moq_audio::Error::from)?; let data = bytes::Bytes::copy_from_slice(bytemuck::cast_slice(samples)); producer.write(&moq_audio::Frame::new(data, timestamp))?; Ok(()) @@ -1613,6 +1626,13 @@ mod tests { .await; assert!(stamps[0] < cut, "the first play's keyframe: {stamps:?}"); assert!(stamps.windows(2).all(|pair| pair[0] < pair[1]), "{stamps:?}"); + // Stamped with when it showed, though encoded ahead of it. + let shown = publish.catalog.clock().capture(first).unwrap(); + assert!( + stamps[0].as_micros().abs_diff(shown.as_micros()) <= 1, + "the first picture as it showed: {:?}, not {shown:?}", + stamps[0] + ); // Ended, it shows nothing new, yet gives its session back once off // program and unread. @@ -1715,7 +1735,7 @@ mod tests { let container = moq_mux::catalog::hang::Container::try_from(config).unwrap(); let mut track = moq_mux::container::Consumer::new(track, container); // Two and a half loops of a second, each with half a second of sound. - pictures(sources.get("bed").unwrap(), 0, 25).await; + let begun = pictures(sources.get("bed").unwrap(), 0, 25).await; let now = kio::Waiter::noop(); let mut stamps = Vec::new(); let silences = until(|| { @@ -1730,6 +1750,9 @@ mod tests { }) .await; assert_eq!(silences, 2, "each loop's silence is kept: {stamps:?}"); + // Stamped with when it sounded, though written ahead of it. + let sounded = std::time::Duration::from(publish.catalog.clock().capture(begun).unwrap()); + assert_eq!(stamps[0], sounded, "the first sound as it sounded"); }) .await; } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index c3fc923e..df6436f7 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1704,7 +1704,10 @@ impl Driver { async fn open(&self, own: &mut Own<'_, '_>, frame: &moq_video::Frame, config: &VideoConfig) { if let Some(listing) = own.listing.filter(|_| !own.opened) { let video = crate::publish::Video::new(listing, frame.size(), framerate(config), frame.surface.color()); - own.track = self.publish(video.await); + // Pictures are stamped with when they show on `own`'s timeline, + // which may be ahead of when they encode. + let video = video.await.and_then(|mut video| video.anchor(own.origin).map(|()| video)); + own.track = self.publish(video); // Only once it has, so a play cut off while it opens leaves the // next picture to open it. own.opened = true; From 2622772e749a880372fd50cb4fba493faa86da36 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 08:06:05 -0800 Subject: [PATCH 6/7] style(studio): format Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/source.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index df6436f7..cb1dcf63 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -1706,7 +1706,9 @@ impl Driver { let video = crate::publish::Video::new(listing, frame.size(), framerate(config), frame.surface.color()); // Pictures are stamped with when they show on `own`'s timeline, // which may be ahead of when they encode. - let video = video.await.and_then(|mut video| video.anchor(own.origin).map(|()| video)); + let video = video + .await + .and_then(|mut video| video.anchor(own.origin).map(|()| video)); own.track = self.publish(video); // Only once it has, so a play cut off while it opens leaves the // next picture to open it. From 2141f090d6b934f47a6fdbf6e4f0c698c54ef60f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 7 Nov 2026 08:29:48 -0800 Subject: [PATCH 7/7] fix(studio): a stinger cut at its start waits for its clip to begin A media play now begins a bus latency after it is asked, so a stinger cut at zero switched the scenes while its clip's first picture was still on its way. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/compositor.rs | 47 ++++++++++++++++++++++++++++++---- 1 file changed, 42 insertions(+), 5 deletions(-) diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index 88f33402..730a86b5 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -1064,10 +1064,11 @@ fn stinger(feed: Option<&Feed>, cut: Duration, played: &mut Option, now: In if run.ended.is_some_and(|ended| ended <= now) { return Ok(None); } - Ok(Some( - run.begun - .is_some_and(|begun| now.saturating_duration_since(begun) >= cut), - )) + // A play begins a moment after it is asked, once its first picture and + // sound can play together, so even a cut at its start waits for it. + Ok(Some(run.begun.is_some_and(|begun| { + now.checked_duration_since(begun).is_some_and(|into| into >= cut) + }))) } /// The crop that trims what `crop` leaves of a `size` picture evenly from two @@ -1799,7 +1800,7 @@ mod tests { use super::*; use crate::collection::{Blend, MaskMode, Rgba, Scene, Size, Video}; - use crate::source::test::{context, listen, local, start}; + use crate::source::test::{context, listen, local, start, until, until_time}; /// A fade over a second. const FADE: Transition = Transition::Fade { @@ -2857,6 +2858,42 @@ mod tests { .await; } + /// A stinger cut at its clip's very start switches the scenes no sooner + /// than the clip's first picture shows, which is a moment after it plays. + #[tokio::test(start_paused = true)] + async fn a_stinger_cuts_no_sooner_than_its_clip_begins() { + local(async { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("sting.mkv"), crate::media::tests::CLIP).unwrap(); + let (context, _origin) = context(dir.path()); + let mut sources = crate::source::Sources::new(context); + let mut collection = Collection::new(); + let media = Source::Media { + path: "sting.mkv".into(), + looping: false, + restart: false, + }; + collection.sources.insert("sting".into(), media); + collection.transition = Transition::Stinger { + source: "sting".into(), + cut: Duration::ZERO, + }; + let drivers = sources.sync(&collection); + start(&sources, drivers).await; + let feed = sources.get("sting"); + + let mut played = None; + let cut = |played: &mut Option| stinger(feed, Duration::ZERO, played, kio::clock::now()).unwrap(); + assert_eq!(cut(&mut played), Some(false), "it plays the clip"); + let begun = until(|| feed.unwrap().run().begun).await; + assert!(begun > kio::clock::now(), "the clip begins ahead"); + assert_eq!(cut(&mut played), Some(false), "not before the clip begins"); + until_time(begun).await; + assert_eq!(cut(&mut played), Some(true), "as the clip begins"); + }) + .await; + } + /// A stinger's clip covers a canvas of another shape unstretched, trimmed /// evenly from the sides that overflow, after its source's own crop. #[test]