From 2b47d7b2ea7a210938b94c67e915595ed6812c42 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:46:52 +0900 Subject: [PATCH 01/19] fix: align mixer inputs to their capture timestamps --- crates/recording/src/sources/audio_mixer.rs | 279 +++++++++++++++++--- 1 file changed, 235 insertions(+), 44 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index f4d0b8cce9d..32ff5b7cab1 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -14,7 +14,9 @@ use std::{ use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; -use crate::output_pipeline::{AudioFrame, HealthSender, PipelineHealthEvent, emit_health}; +use crate::output_pipeline::{ + AudioFrame, HealthSender, PipelineHealthEvent, emit_health, trim_audio_frame_front, +}; const DEFAULT_BUFFER_TIMEOUT: Duration = Duration::from_millis(100); const MIN_BUFFER_TIMEOUT_WIRED: Duration = Duration::from_millis(20); @@ -38,6 +40,7 @@ struct MixerSource { buffer_last: Option<(Timestamp, Duration)>, last_input_timestamp: Option, clock_state: SourceClockState, + samples_supplied: usize, } pub struct AudioMixerBuilder { @@ -101,6 +104,7 @@ impl AudioMixerBuilder { buffer_last: None, last_input_timestamp: None, clock_state: SourceClockState::new(clock_label), + samples_supplied: 0, }); } @@ -423,6 +427,11 @@ impl AudioMixer { self._aformat = graph.aformat; self.abuffersink = graph.abuffersink; self._filter_graph = graph.filter_graph; + for source in &mut self.sources { + source.samples_supplied = + (self.samples_out as u128 * source.info.sample_rate as u128 + / Self::INFO.sample_rate as u128) as usize; + } self.max_buffer_timeout = self .sources .iter() @@ -631,48 +640,6 @@ impl AudioMixer { } } - if let Some(start_timestamp) = self.start_timestamp - && let Some(elapsed_since_start) = now - .duration_since(self.timestamps) - .checked_sub(start_timestamp.duration_since(self.timestamps)) - && elapsed_since_start > self.max_buffer_timeout - { - for source in &mut self.sources { - if source.buffer_last.is_none() { - let rate = source.info.rate(); - let buffer_timeout = source.buffer_timeout; - - let mut remaining = elapsed_since_start; - while remaining > buffer_timeout { - let chunk_samples = samples_for_timeout(rate, buffer_timeout); - let frame_duration = duration_from_samples(chunk_samples, rate); - - let mut frame = ffmpeg::frame::Audio::new( - source.info.sample_format, - chunk_samples, - source.info.channel_layout(), - ); - - for i in 0..frame.planes() { - frame.data_mut(i).fill(0); - } - - frame.set_rate(source.info.rate() as u32); - - let timestamp = - start_timestamp + elapsed_since_start.saturating_sub(remaining); - source.buffer_last = Some((timestamp, frame_duration)); - source.buffer.push_front(AudioFrame::new(frame, timestamp)); - - if frame_duration.is_zero() { - break; - } - - remaining = remaining.saturating_sub(frame_duration); - } - } - } - } Ok(()) } @@ -709,12 +676,49 @@ impl AudioMixer { } for (i, source) in self.sources.iter_mut().enumerate() { + let rate = source.info.rate(); for buffer in source.buffer.drain(..) { - let result = self.abuffers[i].source().add(&buffer.inner); + let position = + sample_position(buffer.timestamp, start_timestamp, self.timestamps, rate); + while position > source.samples_supplied as i128 { + let samples = ((position - source.samples_supplied as i128) as usize) + .min(samples_for_timeout(rate, source.buffer_timeout)); + let silence = silence_frame(source.info, samples); + self.abuffers[i] + .source() + .add(&silence) + .with_context(|| format!("Audio mixer source {i} silence input failed"))?; + source.samples_supplied += samples; + } + let overlap = (source.samples_supplied as i128 - position).max(0) as usize; + let frame = if overlap == 0 { + buffer.inner + } else if overlap >= buffer.inner.samples() { + continue; + } else { + trim_audio_frame_front(&buffer.inner, overlap) + .context("Audio mixer input overlap trim failed")? + }; + let samples = frame.samples(); + let result = self.abuffers[i].source().add(&frame); if self.required_stop.is_some() { result .with_context(|| format!("Audio mixer source {i} filter input failed"))?; } + source.samples_supplied += samples; + } + if source.last_input_timestamp.is_none() { + let position = sample_position(now, start_timestamp, self.timestamps, rate) + - samples_for_timeout(rate, source.buffer_timeout) as i128; + while position > source.samples_supplied as i128 { + let samples = ((position - source.samples_supplied as i128) as usize) + .min(samples_for_timeout(rate, source.buffer_timeout)); + let silence = silence_frame(source.info, samples); + self.abuffers[i].source().add(&silence).with_context(|| { + format!("Audio mixer source {i} startup silence failed") + })?; + source.samples_supplied += samples; + } } } @@ -771,6 +775,31 @@ impl AudioMixer { } } +fn sample_position( + timestamp: Timestamp, + start: Timestamp, + timestamps: Timestamps, + rate: i32, +) -> i128 { + let nanos = timestamp.duration_since(timestamps).as_nanos() as i128 + - start.duration_since(timestamps).as_nanos() as i128; + (nanos * rate as i128 + 500_000_000) / 1_000_000_000 +} + +fn silence_frame(info: AudioInfo, samples: usize) -> ffmpeg::frame::Audio { + let mut frame = ffmpeg::frame::Audio::new(info.sample_format, samples, info.channel_layout()); + let silence = if matches!(info.sample_format, cap_media_info::Sample::U8(_)) { + 128 + } else { + 0 + }; + for plane in 0..frame.planes() { + frame.data_mut(plane).fill(silence); + } + frame.set_rate(info.sample_rate); + frame +} + fn detect_format_mismatch(current: &AudioInfo, frame: &ffmpeg::frame::Audio) -> Option { let frame_rate: u32 = frame.rate(); let frame_channels: usize = frame.channels() as usize; @@ -965,6 +994,168 @@ mod test { const ONE_SECOND: Duration = Duration::from_secs(1); const SAMPLES_SECOND: usize = SOURCE_INFO.rate() as usize; + struct PcmProbe { + mixer: AudioMixer, + inputs: Vec>, + output: mpsc::Receiver, + pcm: Vec<[f32; 2]>, + } + + impl PcmProbe { + fn new(sources: usize) -> Self { + let (sender, output) = mpsc::channel(128); + let mut builder = AudioMixerBuilder::new(); + let inputs = (0..sources) + .map(|_| { + let (sender, receiver) = mpsc::channel(32); + builder.add_source(AudioMixer::INFO, receiver); + sender + }) + .collect(); + Self { + mixer: builder.build(sender).unwrap(), + inputs, + output, + pcm: Vec::new(), + } + } + + fn at(&self, millis: u64) -> Timestamp { + Timestamp::Instant(self.mixer.timestamps.instant() + Duration::from_millis(millis)) + } + + fn send(&mut self, source: usize, millis: u64, samples: usize, peak: Option) { + let mut bytes = vec![0; samples * 8]; + if let Some(peak) = peak { + let offset = peak * 8 + source * 4; + bytes[offset..offset + 4].copy_from_slice(&0.5_f32.to_le_bytes()); + } + let timestamp = self.at(millis); + self.inputs[source] + .try_send(AudioFrame::new( + AudioMixer::INFO.wrap_frame(&bytes), + timestamp, + )) + .unwrap(); + } + + fn tick(&mut self, millis: u64) { + assert!( + self.mixer + .tick(self.mixer.timestamps, self.at(millis)) + .unwrap() + ); + while let Ok(Some(frame)) = self.output.try_next() { + for pair in frame.inner.data(0)[..frame.inner.samples() * 8].chunks_exact(8) { + self.pcm.push([ + f32::from_le_bytes(pair[..4].try_into().unwrap()), + f32::from_le_bytes(pair[4..].try_into().unwrap()), + ]); + } + } + } + + fn peaks(&self, channel: usize) -> Vec { + self.pcm + .iter() + .enumerate() + .filter_map(|(index, sample)| (sample[channel].abs() > 0.01).then_some(index)) + .collect() + } + } + + #[test] + fn pcm_arrival_schedules_preserve_sixty_ms_source_offset() { + for early_poll in [false, true] { + let mut probe = PcmProbe::new(2); + probe.send(1, 200, 24_000, Some(9_600)); + if early_poll { + probe.tick(200); + probe.tick(254); + } + probe.send(0, 260, 24_000, Some(9_600)); + probe.send(1, 700, 2_880, None); + probe.tick(300); + assert_eq!(probe.peaks(1), [9_600]); + assert_eq!(probe.peaks(0), [12_480]); + assert_eq!(probe.peaks(0)[0] - probe.peaks(1)[0], 2_880); + } + } + + #[test] + fn pcm_arrival_schedules_align_simultaneous_capture_events() { + for early_poll in [false, true] { + let mut probe = PcmProbe::new(2); + probe.send(1, 200, 24_000, Some(9_600)); + if early_poll { + probe.tick(200); + probe.tick(254); + } + probe.send(0, 260, 24_000, Some(6_720)); + probe.send(1, 700, 2_880, None); + probe.tick(300); + assert_eq!(probe.peaks(0), [9_600]); + assert_eq!(probe.peaks(1), [9_600]); + } + } + + #[test] + fn pcm_continuous_system_and_late_mic_follow_capture_time() { + for early_poll in [false, true] { + let mut probe = PcmProbe::new(2); + probe.send(0, 0, 24_000, Some(14_400)); + if early_poll { + probe.tick(0); + probe.tick(254); + } + probe.send(1, 200, 14_400, Some(4_800)); + probe.tick(300); + assert_eq!(probe.peaks(0), [14_400]); + assert_eq!(probe.peaks(1), [14_400]); + } + } + + #[test] + fn pcm_sparse_system_and_pause_resume_preserve_later_gaps() { + let mut probe = PcmProbe::new(2); + probe.send(1, 0, 48_000, Some(14_400)); + probe.send(0, 0, 960, None); + probe.send(0, 300, 960, Some(0)); + probe.send(0, 600, 960, Some(0)); + probe.send(0, 900, 4_800, Some(0)); + probe.tick(1_000); + assert_eq!(probe.peaks(0), [14_400, 28_800, 43_200]); + assert_eq!(probe.peaks(1), [14_400]); + probe.send(0, 1_500, 4_800, Some(0)); + probe.send(1, 1_500, 4_800, Some(0)); + probe.tick(1_600); + assert_eq!(probe.peaks(0), [14_400, 28_800, 43_200, 72_000]); + assert_eq!(probe.peaks(1), [14_400, 72_000]); + } + + #[test] + fn pcm_single_input_is_unchanged_and_reset_keeps_the_supplied_cursor() { + let mut probe = PcmProbe::new(1); + probe.send(0, 200, 4_800, Some(2_400)); + probe.tick(300); + assert_eq!(probe.pcm.len(), 4_800); + assert_eq!(probe.peaks(0), [2_400]); + for (index, sample) in probe.pcm.iter().enumerate() { + assert_eq!( + *sample, + if index == 2_400 { + [0.5, 0.0] + } else { + [0.0, 0.0] + } + ); + } + probe.mixer.force_reset_source(mixer_source_label(0), 2_000); + probe.send(0, 400, 4_800, Some(0)); + probe.tick(500); + assert_eq!(probe.peaks(0), [2_400, 9_600]); + } + #[tokio::test] async fn mix_sources() { let (tx, mut output_rx) = mpsc::channel(4); From 02b9dd1b27df182de5dfe10b8672543018db565e Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:47:23 +0900 Subject: [PATCH 02/19] fix: retain mixer input gaps below the buffering timeout --- crates/recording/src/sources/audio_mixer.rs | 22 ++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index 32ff5b7cab1..23727c25830 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -553,10 +553,11 @@ impl AudioMixer { 0 }; - let remap = - source - .clock_state - .remap(&self.master_clock, raw_timestamp, frame_duration_ns); + let remap = source.clock_state.remap_preserving_cadence( + &self.master_clock, + raw_timestamp, + frame_duration_ns, + ); if matches!(remap.outcome, SourceClockOutcome::HardReset) { let flushed = source.buffer.len(); @@ -783,7 +784,7 @@ fn sample_position( ) -> i128 { let nanos = timestamp.duration_since(timestamps).as_nanos() as i128 - start.duration_since(timestamps).as_nanos() as i128; - (nanos * rate as i128 + 500_000_000) / 1_000_000_000 + (nanos * rate as i128 + 500_000_000).div_euclid(1_000_000_000) } fn silence_frame(info: AudioInfo, samples: usize) -> ffmpeg::frame::Audio { @@ -1133,6 +1134,17 @@ mod test { assert_eq!(probe.peaks(1), [14_400, 72_000]); } + #[test] + fn pcm_sub_timeout_gap_is_not_smoothed_away() { + let mut probe = PcmProbe::new(2); + probe.send(1, 0, 4_800, None); + probe.send(0, 0, 960, Some(0)); + probe.send(0, 30, 960, Some(0)); + probe.send(0, 50, 2_400, Some(0)); + probe.tick(100); + assert_eq!(probe.peaks(0), [0, 1_440, 2_400]); + } + #[test] fn pcm_single_input_is_unchanged_and_reset_keeps_the_supplied_cursor() { let mut probe = PcmProbe::new(1); From 2a559efa0d424c33934ca72a256146d12b2cb2bc Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:50:35 +0900 Subject: [PATCH 03/19] fix: remove obsolete mixer duration helper --- crates/recording/src/sources/audio_mixer.rs | 8 -------- 1 file changed, 8 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index 23727c25830..e5ad1f79e0d 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -972,14 +972,6 @@ fn samples_for_timeout(rate: i32, timeout: Duration) -> usize { samples.max(1.0) as usize } -fn duration_from_samples(samples: usize, rate: i32) -> Duration { - if rate <= 0 { - return Duration::ZERO; - } - - Duration::from_secs_f64(samples as f64 / rate as f64) -} - #[cfg(test)] mod test { use futures::{SinkExt, StreamExt}; From 89169ff3e029f0f527e1d4555ccfa43209e0dd14 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:55:24 +0900 Subject: [PATCH 04/19] test: drain sparse mixer samples with trailing input --- crates/recording/src/sources/audio_mixer.rs | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index e5ad1f79e0d..cbf7ea4a2d0 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -1116,12 +1116,21 @@ mod test { probe.send(0, 300, 960, Some(0)); probe.send(0, 600, 960, Some(0)); probe.send(0, 900, 4_800, Some(0)); - probe.tick(1_000); - assert_eq!(probe.peaks(0), [14_400, 28_800, 43_200]); + probe.send(0, 1_000, 4_800, None); + probe.send(1, 1_000, 4_800, None); + probe.tick(1_100); + assert_eq!( + probe.peaks(0), + [14_400, 28_800, 43_200], + "{} output samples", + probe.pcm.len() + ); assert_eq!(probe.peaks(1), [14_400]); probe.send(0, 1_500, 4_800, Some(0)); probe.send(1, 1_500, 4_800, Some(0)); - probe.tick(1_600); + probe.send(0, 1_600, 4_800, None); + probe.send(1, 1_600, 4_800, None); + probe.tick(1_700); assert_eq!(probe.peaks(0), [14_400, 28_800, 43_200, 72_000]); assert_eq!(probe.peaks(1), [14_400, 72_000]); } From 51742b6efafc59609a48b66e62651cb193a3f7f0 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:01:31 +0900 Subject: [PATCH 05/19] fix: place sparse mixer gaps only with integer sample positions --- crates/recording/src/sources/audio_mixer.rs | 86 ++++++++++----------- 1 file changed, 39 insertions(+), 47 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index cbf7ea4a2d0..3d2e1034e53 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -508,7 +508,6 @@ impl AudioMixer { let clock_anchor = self.master_clock.timestamps().instant(); for (index, source) in self.sources.iter_mut().enumerate() { let rate = source.info.rate(); - let buffer_timeout = source.buffer_timeout; loop { let AudioFrame { @@ -574,44 +573,6 @@ impl AudioMixer { let timestamp = Timestamp::Instant(clock_anchor + remap.duration()); - if let Some((buffer_last_timestamp, buffer_last_duration)) = source.buffer_last { - let timestamp_elapsed = timestamp.duration_since(self.timestamps); - let buffer_last_elapsed = buffer_last_timestamp.duration_since(self.timestamps); - - if timestamp_elapsed > buffer_last_elapsed { - let elapsed_since_last_frame = - timestamp_elapsed.saturating_sub(buffer_last_elapsed); - - if let Some(gap) = elapsed_since_last_frame - .checked_sub(buffer_last_duration) - .filter(|&diff| diff >= buffer_timeout) - { - debug!(?gap, "Gap between last buffer frame, inserting silence"); - - let silence_samples_needed = (gap.as_secs_f64()) * rate as f64; - let silence_samples_count = silence_samples_needed.ceil() as usize; - - let mut frame = ffmpeg::frame::Audio::new( - source.info.sample_format, - silence_samples_count, - source.info.channel_layout(), - ); - - for i in 0..frame.planes() { - frame.data_mut(i).fill(0); - } - - frame.set_rate(source.info.rate() as u32); - - let silence_duration = - Duration::from_secs_f64(silence_samples_count as f64 / rate as f64); - let timestamp = buffer_last_timestamp + buffer_last_duration; - source.buffer_last = Some((timestamp, silence_duration)); - source.buffer.push_back(AudioFrame::new(frame, timestamp)); - } - } - } - source.buffer_last = Some(( timestamp, Duration::from_secs_f64(frame.samples() as f64 / rate as f64), @@ -1018,12 +979,21 @@ mod test { } fn send(&mut self, source: usize, millis: u64, samples: usize, peak: Option) { + self.send_at(source, self.at(millis), samples, peak); + } + + fn send_at( + &mut self, + source: usize, + timestamp: Timestamp, + samples: usize, + peak: Option, + ) { let mut bytes = vec![0; samples * 8]; if let Some(peak) = peak { let offset = peak * 8 + source * 4; bytes[offset..offset + 4].copy_from_slice(&0.5_f32.to_le_bytes()); } - let timestamp = self.at(millis); self.inputs[source] .try_send(AudioFrame::new( AudioMixer::INFO.wrap_frame(&bytes), @@ -1135,6 +1105,32 @@ mod test { assert_eq!(probe.peaks(1), [14_400, 72_000]); } + #[test] + fn pcm_thousand_sparse_packets_preserve_samples_without_rounding_drift() { + for spacing_ns in [300_000_000, 300_006_944] { + let mut probe = PcmProbe::new(2); + let mut expected = Vec::new(); + for packet in 0..=1_001_u64 { + let timestamp = Timestamp::Instant( + probe.mixer.timestamps.instant() + Duration::from_nanos(packet * spacing_ns), + ); + probe.send_at(0, timestamp, 960, (packet < 1_001).then_some(0)); + probe.send_at(1, timestamp, 960, None); + probe.tick(packet * spacing_ns / 1_000_000 + 20); + if packet < 1_001 { + let fractional_samples = if spacing_ns == 300_000_000 { + 0 + } else { + (packet + 1) / 3 + }; + expected.push((packet * 14_400 + fractional_samples) as usize); + } + } + assert_eq!(probe.peaks(0), expected); + assert!(probe.peaks(1).is_empty()); + } + } + #[test] fn pcm_sub_timeout_gap_is_not_smoothed_away() { let mut probe = PcmProbe::new(2); @@ -1633,12 +1629,12 @@ mod test { let source = &mut mixer.sources[0]; - assert_eq!(source.buffer.len(), 3); + assert_eq!(source.buffer.len(), 2); assert!(source.rx.try_next().is_err()); assert_eq!( source.buffer[1].timestamp.duration_since(mixer.timestamps), - ONE_SECOND / 2 + ONE_SECOND ); assert_eq!(source.buffer[1].samples(), SOURCE_INFO.rate() as usize / 2); } @@ -1714,14 +1710,10 @@ mod test { let source = &mut mixer.sources[0]; - assert_eq!(source.buffer.len(), 2); + assert_eq!(source.buffer.len(), 1); assert!(source.rx.try_next().is_err()); let item = &source.buffer[0]; - assert_eq!(item.timestamp.duration_since(start), ONE_SECOND / 2); - assert_eq!(item.inner.samples(), SOURCE_INFO.rate() as usize / 2); - - let item = &source.buffer[1]; assert_eq!(item.timestamp.duration_since(start), ONE_SECOND); assert_eq!(item.inner.samples(), SOURCE_INFO.rate() as usize / 2); } From 68995c975a35e1623f4e53438b08bd9925dacd84 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:14:25 +0900 Subject: [PATCH 06/19] fix: wait for input watermarks before committing mixed audio --- crates/recording/src/sources/audio_mixer.rs | 251 ++++++++++++++++---- 1 file changed, 204 insertions(+), 47 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index 3d2e1034e53..fb1abec7704 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -23,6 +23,7 @@ const MIN_BUFFER_TIMEOUT_WIRED: Duration = Duration::from_millis(20); const MIN_BUFFER_TIMEOUT_WIRELESS: Duration = Duration::from_millis(90); const MAX_BUFFER_TIMEOUT: Duration = Duration::from_millis(250); const BUFFER_TIMEOUT_HEADROOM: f64 = 2.5; +const STARTUP_GRACE: Duration = Duration::from_secs(1); pub const MAX_BUFFERING_TICKS: u32 = 45; pub const AUDIO_OUTPUT_FRAMES: u32 = cap_timestamp::AUDIO_OUTPUT_FRAMES as u32; @@ -41,6 +42,7 @@ struct MixerSource { last_input_timestamp: Option, clock_state: SourceClockState, samples_supplied: usize, + late_samples_dropped: u64, } pub struct AudioMixerBuilder { @@ -105,6 +107,7 @@ impl AudioMixerBuilder { last_input_timestamp: None, clock_state: SourceClockState::new(clock_label), samples_supplied: 0, + late_samples_dropped: 0, }); } @@ -192,6 +195,15 @@ impl AudioMixerBuilder { std::thread::sleep(Duration::from_millis(5)); } + for source in &mixer.sources { + if source.late_samples_dropped > 0 { + info!( + source = source.clock_state.name(), + late_samples_dropped = source.late_samples_dropped, + "AudioMixer: late input sample count" + ); + } + } Ok(()) } } @@ -573,10 +585,7 @@ impl AudioMixer { let timestamp = Timestamp::Instant(clock_anchor + remap.duration()); - source.buffer_last = Some(( - timestamp, - Duration::from_secs_f64(frame.samples() as f64 / rate as f64), - )); + source.buffer_last = Some((timestamp, Duration::from_nanos(frame_duration_ns))); source.buffer.push_back(AudioFrame::new(frame, timestamp)); } } @@ -585,8 +594,13 @@ impl AudioMixer { self.handle_source_format_change(index, old_info, new_info)?; } - if self.start_timestamp.is_none() { - self.start_timestamp = self + if self.samples_out == 0 + && self + .sources + .iter() + .all(|source| source.samples_supplied == 0) + { + let earliest = self .sources .iter() .filter_map(|s| s.buffer.front()) @@ -597,7 +611,8 @@ impl AudioMixer { }) .map(|v| v.timestamp); - if self.start_timestamp.is_some() { + if let Some(earliest) = earliest { + self.start_timestamp = Some(earliest); self.wall_clock_start = Some(now); } } @@ -637,50 +652,80 @@ impl AudioMixer { self.force_reset_source(source, starvation_ms); } - for (i, source) in self.sources.iter_mut().enumerate() { - let rate = source.info.rate(); - for buffer in source.buffer.drain(..) { - let position = - sample_position(buffer.timestamp, start_timestamp, self.timestamps, rate); - while position > source.samples_supplied as i128 { - let samples = ((position - source.samples_supplied as i128) as usize) - .min(samples_for_timeout(rate, source.buffer_timeout)); - let silence = silence_frame(source.info, samples); - self.abuffers[i] - .source() - .add(&silence) - .with_context(|| format!("Audio mixer source {i} silence input failed"))?; - source.samples_supplied += samples; - } - let overlap = (source.samples_supplied as i128 - position).max(0) as usize; - let frame = if overlap == 0 { - buffer.inner - } else if overlap >= buffer.inner.samples() { - continue; - } else { - trim_audio_frame_front(&buffer.inner, overlap) - .context("Audio mixer input overlap trim failed")? - }; - let samples = frame.samples(); - let result = self.abuffers[i].source().add(&frame); - if self.required_stop.is_some() { - result - .with_context(|| format!("Audio mixer source {i} filter input failed"))?; + let grace_elapsed = now + .duration_since(self.timestamps) + .saturating_sub(start_timestamp.duration_since(self.timestamps)) + >= STARTUP_GRACE; + let mut watermark = None; + let mut waiting_for_input = false; + for source in &self.sources { + if let Some((timestamp, duration)) = source.buffer_last { + let end = timestamp + duration; + if watermark.is_none_or(|previous: Timestamp| { + end.duration_since(self.timestamps) < previous.duration_since(self.timestamps) + }) { + watermark = Some(end); } - source.samples_supplied += samples; + } else if !grace_elapsed { + waiting_for_input = true; } - if source.last_input_timestamp.is_none() { - let position = sample_position(now, start_timestamp, self.timestamps, rate) - - samples_for_timeout(rate, source.buffer_timeout) as i128; - while position > source.samples_supplied as i128 { - let samples = ((position - source.samples_supplied as i128) as usize) - .min(samples_for_timeout(rate, source.buffer_timeout)); - let silence = silence_frame(source.info, samples); - self.abuffers[i].source().add(&silence).with_context(|| { - format!("Audio mixer source {i} startup silence failed") - })?; + } + + if let Some(watermark) = watermark.filter(|_| !waiting_for_input) { + for (i, source) in self.sources.iter_mut().enumerate() { + let rate = source.info.rate(); + let limit = sample_position(watermark, start_timestamp, self.timestamps, rate) + .max(0) as usize; + while let Some(mut buffer) = source.buffer.pop_front() { + let position = + sample_position(buffer.timestamp, start_timestamp, self.timestamps, rate); + let overlap = (source.samples_supplied as i128 - position).max(0) as usize; + if overlap > 0 { + let dropped = overlap.min(buffer.inner.samples()); + if source.late_samples_dropped == 0 { + warn!( + source = source.clock_state.name(), + dropped_samples = dropped, + "AudioMixer: dropping input samples that arrived after their time was committed" + ); + } + source.late_samples_dropped = + source.late_samples_dropped.saturating_add(dropped as u64); + if dropped == buffer.inner.samples() { + continue; + } + buffer.inner = trim_audio_frame_front(&buffer.inner, dropped) + .context("Audio mixer input overlap trim failed")?; + buffer.timestamp = start_timestamp + + samples_duration(source.samples_supplied, source.info.sample_rate); + } + let gap_end = (position.max(0) as usize).min(limit); + supply_silence(source, &mut self.abuffers[i], i, gap_end)?; + if source.samples_supplied >= limit { + source.buffer.push_front(buffer); + break; + } + let samples = buffer.inner.samples().min(limit - source.samples_supplied); + if samples < buffer.inner.samples() { + let remaining = trim_audio_frame_front(&buffer.inner, samples) + .context("Audio mixer watermark split failed")?; + source.buffer.push_front(AudioFrame::new( + remaining, + buffer.timestamp + samples_duration(samples, source.info.sample_rate), + )); + buffer.inner.set_samples(samples); + } + let result = self.abuffers[i].source().add(&buffer.inner); + if self.required_stop.is_some() { + result.with_context(|| { + format!("Audio mixer source {i} filter input failed") + })?; + } source.samples_supplied += samples; } + if source.buffer_last.is_none() { + supply_silence(source, &mut self.abuffers[i], i, limit)?; + } } } @@ -737,6 +782,30 @@ impl AudioMixer { } } +fn samples_duration(samples: usize, rate: u32) -> Duration { + Duration::from_nanos((samples as u128 * 1_000_000_000 / u128::from(rate)) as u64) +} + +fn supply_silence( + source: &mut MixerSource, + input: &mut ffmpeg::filter::Context, + index: usize, + limit: usize, +) -> anyhow::Result<()> { + while source.samples_supplied < limit { + let samples = (limit - source.samples_supplied).min(samples_for_timeout( + source.info.rate(), + source.buffer_timeout, + )); + input + .source() + .add(&silence_frame(source.info, samples)) + .with_context(|| format!("Audio mixer source {index} silence input failed"))?; + source.samples_supplied += samples; + } + Ok(()) +} + fn sample_position( timestamp: Timestamp, start: Timestamp, @@ -953,6 +1022,7 @@ mod test { inputs: Vec>, output: mpsc::Receiver, pcm: Vec<[f32; 2]>, + deliveries: Vec<(usize, u64)>, } impl PcmProbe { @@ -971,6 +1041,7 @@ mod test { inputs, output, pcm: Vec::new(), + deliveries: Vec::new(), } } @@ -1009,6 +1080,7 @@ mod test { .unwrap() ); while let Ok(Some(frame)) = self.output.try_next() { + self.deliveries.push((self.pcm.len(), millis)); for pair in frame.inner.data(0)[..frame.inner.samples() * 8].chunks_exact(8) { self.pcm.push([ f32::from_le_bytes(pair[..4].try_into().unwrap()), @@ -1131,6 +1203,91 @@ mod test { } } + #[test] + fn pcm_delayed_first_delivery_retains_captured_samples() { + for delivery_ms in [400, 800, 1_000] { + let mut probe = PcmProbe::new(2); + for chunk in 0..delivery_ms / 20 { + probe.send(0, chunk * 20, 960, None); + if (chunk + 1) * 20 < delivery_ms { + probe.tick((chunk + 1) * 20); + } + } + probe.send(1, 200, 48_000, Some(2_400)); + probe.send(0, delivery_ms, 57_600 - delivery_ms as usize * 48, None); + probe.tick(delivery_ms); + eprintln!( + "DELAYED delivery_ms={delivery_ms} peaks={:?}", + probe.peaks(1) + ); + assert_eq!(probe.peaks(1), [12_000]); + } + } + + #[test] + fn pcm_first_input_can_precede_the_uncommitted_mixer_origin() { + let mut probe = PcmProbe::new(2); + probe.send(0, 200, 24_000, Some(4_800)); + probe.tick(300); + assert!(probe.pcm.is_empty()); + probe.send(1, 0, 33_600, Some(14_400)); + probe.tick(400); + assert_eq!(probe.peaks(0), [14_400]); + assert_eq!(probe.peaks(1), [14_400]); + } + + #[test] + fn pcm_first_input_after_grace_drops_and_counts_only_committed_samples() { + let mut probe = PcmProbe::new(2); + probe.send(0, 0, 48_000, None); + probe.tick(999); + assert!(probe.pcm.is_empty()); + probe.tick(1_000); + assert_eq!(probe.pcm.len(), 48_000); + assert_eq!(probe.deliveries[0].1, 1_000); + probe.send(1, 200, 48_000, Some(2_400)); + probe.send(0, 1_000, 9_600, Some(0)); + probe.tick(1_200); + assert!(probe.peaks(1).is_empty()); + assert_eq!(probe.mixer.sources[1].late_samples_dropped, 38_400); + assert_eq!(probe.peaks(0), [48_000]); + probe.send(1, 1_200, 960, Some(0)); + probe.send(0, 1_200, 960, None); + probe.tick(1_220); + assert_eq!(probe.peaks(1), [57_600]); + assert_eq!(probe.mixer.sources[1].late_samples_dropped, 38_400); + } + + #[test] + fn pcm_watermark_limits_commitment_to_the_shortest_input() { + let mut probe = PcmProbe::new(2); + probe.send(0, 0, 4_800, Some(4_000)); + probe.send(1, 0, 960, None); + probe.tick(20); + assert_eq!(probe.pcm.len(), 960); + assert_eq!(probe.mixer.sources[0].samples_supplied, 960); + assert_eq!(probe.mixer.sources[1].samples_supplied, 960); + probe.send(1, 20, 3_840, Some(3_040)); + probe.tick(100); + assert_eq!(probe.pcm.len(), 4_800); + assert_eq!(probe.peaks(0), [4_000]); + assert_eq!(probe.peaks(1), [4_000]); + } + + #[test] + fn pcm_continuous_input_watermarks_add_at_most_one_buffer() { + let mut probe = PcmProbe::new(2); + for chunk in 0..10 { + probe.send(0, chunk * 20, 960, None); + probe.tick(chunk * 20 + 20); + assert!(probe.pcm.len() >= chunk as usize * 960); + probe.send(1, chunk * 20, 960, None); + probe.tick(chunk * 20 + 20); + assert_eq!(probe.pcm.len(), (chunk as usize + 1) * 960); + } + assert_eq!(probe.deliveries[0].1, 20); + } + #[test] fn pcm_sub_timeout_gap_is_not_smoothed_away() { let mut probe = PcmProbe::new(2); From 44a90c81cccb64e7f58c465d408512fc23456fa5 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:19:54 +0900 Subject: [PATCH 07/19] fix: keep split mixer buffers on the absolute sample cursor --- crates/recording/src/sources/audio_mixer.rs | 23 ++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index fb1abec7704..c2174063411 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -711,7 +711,11 @@ impl AudioMixer { .context("Audio mixer watermark split failed")?; source.buffer.push_front(AudioFrame::new( remaining, - buffer.timestamp + samples_duration(samples, source.info.sample_rate), + start_timestamp + + samples_duration( + source.samples_supplied + samples, + source.info.sample_rate, + ), )); buffer.inner.set_samples(samples); } @@ -1288,6 +1292,23 @@ mod test { assert_eq!(probe.deliveries[0].1, 20); } + #[test] + fn pcm_watermark_splits_do_not_accumulate_timestamp_rounding() { + let mut probe = PcmProbe::new(2); + probe.send(0, 0, 50_001, Some(49_999)); + for sample in 0..50_001 { + let timestamp = Timestamp::Instant( + probe.mixer.timestamps.instant() + + Duration::from_nanos(sample * 1_000_000_000 / 48_000), + ); + probe.send_at(1, timestamp, 1, None); + probe.tick(sample / 48); + } + assert_eq!(probe.peaks(0), [49_999]); + assert_eq!(probe.mixer.sources[0].late_samples_dropped, 0); + assert_eq!(probe.pcm.len(), 50_001); + } + #[test] fn pcm_sub_timeout_gap_is_not_smoothed_away() { let mut probe = PcmProbe::new(2); From 1da9787b2cc96a36bbfa911b88d284105805e01d Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:46:15 +0900 Subject: [PATCH 08/19] fix: drain captured audio when stopping the mixer --- crates/recording/src/output_pipeline/core.rs | 106 ++++++++++++- crates/recording/src/sources/audio_mixer.rs | 155 +++++++++++++++++-- 2 files changed, 244 insertions(+), 17 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 96057442171..0c35351ae69 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1,5 +1,5 @@ use super::start_gate::{AudioAdmission, RecordingStartGate}; -use crate::sources::audio_mixer::AudioMixer; +use crate::sources::audio_mixer::{AudioMixer, MIXER_STOP_TIMEOUT}; use anyhow::{Context, anyhow}; use cap_media_info::{AudioInfo, VideoInfo}; use cap_timestamp::{MasterClock, SourceClockOutcome, SourceClockState, Timestamp, Timestamps}; @@ -3805,6 +3805,7 @@ impl PreparedAudioSources { audio_anchor: AudioAnchor, ) { let audio_info = self.audio_info; + let mixed_audio = self.erased_audio_sources.len() > 1; let has_wireless_source = self.has_wireless_source; let health_tx = setup_ctx.health_tx().clone(); let master_clock = setup_ctx.master_clock().clone(); @@ -3905,16 +3906,49 @@ impl PreparedAudioSources { None }; + let mixed_sources_stopped = was_cancelled && mixed_audio; + if mixed_sources_stopped { + let stopped = tokio::time::timeout( + Duration::from_secs(1), + stop_audio_sources(&mut self.erased_audio_sources, required_scope.as_ref()), + ) + .await; + if stopped.is_err() { + let error = anyhow!("Mixed audio sources did not stop before the deadline"); + if let Some(scope) = &required_scope { + scope.fail_required(error.to_string()); + } + let _ = audio_error.get_or_insert(error); + } + self.erased_audio_sources.clear(); + } + if was_cancelled && !audio_degraded { let drain_start = std::time::Instant::now(); + let drain_deadline = tokio::time::Instant::now() + MIXER_STOP_TIMEOUT; let max_drain_frames = 500u64; let mut drained = 0u64; let mut skipped = 0u64; let mut degraded_during_drain = false; while drained < max_drain_frames { - match self.audio_rx.try_next() { - Ok(Some(frame)) => match process_audio_frame( + let next = if mixed_audio { + match tokio::time::timeout_at(drain_deadline, self.audio_rx.next()).await { + Ok(frame) => frame, + Err(_) => { + let error = anyhow!("Mixed audio output did not close before the drain deadline"); + if let Some(scope) = &required_scope { + scope.fail_required(error.to_string()); + } + let _ = audio_error.get_or_insert(error); + break; + } + } + } else { + self.audio_rx.try_next().ok().flatten() + }; + match next { + Some(frame) => match process_audio_frame( AudioFrameProcessContext { audio_info: &audio_info, sample_rate, @@ -3965,7 +3999,7 @@ impl PreparedAudioSources { skipped += 1; } }, - Ok(None) | Err(_) => break, + None => break, } } @@ -3986,7 +4020,9 @@ impl PreparedAudioSources { let final_pause_duration = shared_pause.total_pause_duration(); - if let Some(scope) = &required_scope { + if let Some(scope) = &required_scope + && !mixed_sources_stopped + { stop_audio_sources(&mut self.erased_audio_sources, Some(scope)).await; } @@ -4104,7 +4140,7 @@ impl PreparedAudioSources { let _ = gap_summary_slot.set(gap_tracker.gap_summary()); } - if required_scope.is_none() { + if required_scope.is_none() && !mixed_sources_stopped { stop_audio_sources(&mut self.erased_audio_sources, None).await; } @@ -9235,6 +9271,64 @@ mod tests { } } + struct StopFrameSource { + output: mpsc::Sender, + final_frame: Option, + } + + impl AudioSource for StopFrameSource { + type Config = Option; + + fn setup( + final_frame: Self::Config, + output: mpsc::Sender, + _: &mut SetupCtx, + ) -> impl Future> + Send + 'static { + std::future::ready(Ok(Self { + output, + final_frame, + })) + } + + fn audio_info(&self) -> AudioInfo { + AudioMixer::INFO + } + + async fn stop(&mut self) -> anyhow::Result<()> { + if let Some(frame) = self.final_frame.take() { + self.output.send(frame).await?; + } + Ok(()) + } + } + + #[tokio::test] + async fn mixed_pipeline_stop_receives_audio_delivered_during_source_stop() { + let temp = tempfile::tempdir().unwrap(); + let timestamps = Timestamps::now(); + let sent = Arc::new(std::sync::Mutex::new(Vec::new())); + let pipeline = OutputPipeline::builder(temp.path().join("mixed-stop.ogg")) + .with_timestamps(timestamps) + .with_audio_source::(None) + .with_audio_source::(Some(AudioFrame::new( + AudioMixer::INFO.empty_frame(4_800), + Timestamp::Instant(timestamps.instant()), + ))) + .build::(sent.clone()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(3), pipeline.stop()) + .await + .unwrap() + .unwrap(); + let frames = sent.lock().unwrap(); + assert_eq!( + frames.iter().map(|frame| frame.samples).sum::(), + 4_800 + ); + assert_eq!(frames[0].timestamp, Duration::ZERO); + } + struct AudioTimelineHarness { info: AudioInfo, timestamps: Timestamps, diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index c2174063411..19717f263a0 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -24,6 +24,7 @@ const MIN_BUFFER_TIMEOUT_WIRELESS: Duration = Duration::from_millis(90); const MAX_BUFFER_TIMEOUT: Duration = Duration::from_millis(250); const BUFFER_TIMEOUT_HEADROOM: f64 = 2.5; const STARTUP_GRACE: Duration = Duration::from_secs(1); +pub(crate) const MIXER_STOP_TIMEOUT: Duration = Duration::from_secs(2); pub const MAX_BUFFERING_TICKS: u32 = 45; pub const AUDIO_OUTPUT_FRAMES: u32 = cap_timestamp::AUDIO_OUTPUT_FRAMES as u32; @@ -43,6 +44,7 @@ struct MixerSource { clock_state: SourceClockState, samples_supplied: usize, late_samples_dropped: u64, + ended: bool, } pub struct AudioMixerBuilder { @@ -108,6 +110,7 @@ impl AudioMixerBuilder { clock_state: SourceClockState::new(clock_label), samples_supplied: 0, late_samples_dropped: 0, + ended: false, }); } @@ -146,6 +149,8 @@ impl AudioMixerBuilder { buffering: MixerBufferingTracker::new(), health_tx: self.health_tx, required_stop: self.required_stop, + drain_deadline: None, + finishing: false, }) } @@ -177,16 +182,11 @@ impl AudioMixerBuilder { .is_some_and(CancellationToken::is_cancelled) { info!("Mixer stop flag triggered"); + mixer.finish()?; break; } - #[cfg(target_os = "macos")] - let now = Timestamp::MachAbsoluteTime(cap_timestamp::MachAbsoluteTimestamp::now()); - #[cfg(windows)] - let now = - Timestamp::PerformanceCounter(cap_timestamp::PerformanceCounterTimestamp::now()); - #[cfg(not(any(target_os = "macos", windows)))] - let now = Timestamp::Instant(Instant::now()); + let now = mixer_now(); if !mixer.tick(start, now)? { info!("Mixer output closed"); @@ -227,6 +227,8 @@ pub struct AudioMixer { buffering: MixerBufferingTracker, health_tx: Option, required_stop: Option, + drain_deadline: Option, + finishing: bool, } #[derive(Debug, Clone)] @@ -522,12 +524,19 @@ impl AudioMixer { let rate = source.info.rate(); loop { + if self + .drain_deadline + .is_some_and(|deadline| Instant::now() >= deadline) + { + source.rx.close(); + } let AudioFrame { inner: frame, timestamp: raw_timestamp, } = match source.rx.try_next() { Ok(Some(frame)) => frame, Ok(None) => { + source.ended = true; anyhow::ensure!( self.required_stop .as_ref() @@ -624,6 +633,30 @@ impl AudioMixer { self.tick_with_output(now, |sink, frame| sink.sink().frame(frame)) } + fn finish(&mut self) -> anyhow::Result<()> { + if let Some(stop) = &self.required_stop { + stop.cancel(); + } + self.drain_deadline = Some(Instant::now() + STARTUP_GRACE); + loop { + self.buffer_sources(mixer_now())?; + if self.sources.iter().all(|source| source.ended) { + break; + } + std::thread::sleep(Duration::from_millis(1)); + } + self.finishing = true; + self.drain_deadline = Some(Instant::now() + STARTUP_GRACE); + if !self.tick(self.timestamps, mixer_now())? { + return Ok(()); + } + for input in &mut self.abuffers { + input.source().flush().context("Audio mixer flush failed")?; + } + self.tick(self.timestamps, mixer_now())?; + Ok(()) + } + fn tick_with_output( &mut self, now: Timestamp, @@ -639,7 +672,11 @@ impl AudioMixer { }; let wall_now = Instant::now(); - let laggard = self.detect_laggard_source(now); + let laggard = if self.finishing { + None + } else { + self.detect_laggard_source(now) + }; let action = self .buffering .observe(wall_now, laggard, self.health_tx.as_ref()); @@ -662,11 +699,17 @@ impl AudioMixer { if let Some((timestamp, duration)) = source.buffer_last { let end = timestamp + duration; if watermark.is_none_or(|previous: Timestamp| { - end.duration_since(self.timestamps) < previous.duration_since(self.timestamps) + let end = end.duration_since(self.timestamps); + let previous = previous.duration_since(self.timestamps); + if self.finishing { + end > previous + } else { + end < previous + } }) { watermark = Some(end); } - } else if !grace_elapsed { + } else if !grace_elapsed && !self.finishing { waiting_for_input = true; } } @@ -727,7 +770,7 @@ impl AudioMixer { } source.samples_supplied += samples; } - if source.buffer_last.is_none() { + if source.buffer_last.is_none() || self.finishing { supply_silence(source, &mut self.abuffers[i], i, limit)?; } } @@ -765,6 +808,11 @@ impl AudioMixer { match self.output.try_send(frame) { Ok(()) => break, Err(err) if err.is_full() => { + anyhow::ensure!( + self.drain_deadline + .is_none_or(|deadline| Instant::now() < deadline), + "Audio mixer output did not drain before the stop deadline" + ); frame = err.into_inner(); std::thread::sleep(Duration::from_millis(1)); } @@ -786,6 +834,21 @@ impl AudioMixer { } } +fn mixer_now() -> Timestamp { + #[cfg(target_os = "macos")] + { + Timestamp::MachAbsoluteTime(cap_timestamp::MachAbsoluteTimestamp::now()) + } + #[cfg(windows)] + { + Timestamp::PerformanceCounter(cap_timestamp::PerformanceCounterTimestamp::now()) + } + #[cfg(not(any(target_os = "macos", windows)))] + { + Timestamp::Instant(Instant::now()) + } +} + fn samples_duration(samples: usize, rate: u32) -> Duration { Duration::from_nanos((samples as u128 * 1_000_000_000 / u128::from(rate)) as u64) } @@ -1083,6 +1146,10 @@ mod test { .tick(self.mixer.timestamps, self.at(millis)) .unwrap() ); + self.collect(millis); + } + + fn collect(&mut self, millis: u64) { while let Ok(Some(frame)) = self.output.try_next() { self.deliveries.push((self.pcm.len(), millis)); for pair in frame.inner.data(0)[..frame.inner.samples() * 8].chunks_exact(8) { @@ -1320,6 +1387,72 @@ mod test { assert_eq!(probe.peaks(0), [0, 1_440, 2_400]); } + #[test] + fn pcm_stop_before_startup_grace_preserves_captured_audio() { + let mut probe = PcmProbe::new(2); + probe.send(1, 0, 4_800, Some(960)); + probe.tick(100); + assert!(probe.pcm.is_empty()); + probe.inputs.clear(); + probe.mixer.finish().unwrap(); + probe.collect(100); + assert_eq!(probe.peaks(1), [960]); + assert_eq!(probe.pcm.len(), 4_800); + } + + #[test] + fn pcm_stop_drains_the_longest_input_and_pads_closed_sources() { + for millis in [100, 1_500] { + let mut probe = PcmProbe::new(2); + probe.send(0, 0, 960, None); + probe.send(1, 0, 4_800, Some(4_799)); + probe.tick(millis); + probe.inputs.clear(); + probe.mixer.finish().unwrap(); + probe.collect(millis); + assert_eq!(probe.peaks(1), [4_799]); + assert_eq!(probe.pcm.len(), 4_800); + } + } + + #[test] + fn pcm_stop_waits_for_queued_capture_before_closing_inputs() { + let mut probe = PcmProbe::new(2); + probe.send(1, 0, 960, None); + let mut input = probe.inputs.pop().unwrap(); + probe.inputs.clear(); + let timestamp = probe.at(20); + let worker = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(20)); + let mut bytes = vec![0; 3_840 * 8]; + bytes[3_000 * 8 + 4..3_000 * 8 + 8].copy_from_slice(&0.5_f32.to_le_bytes()); + input + .try_send(AudioFrame::new( + AudioMixer::INFO.wrap_frame(&bytes), + timestamp, + )) + .unwrap(); + }); + probe.mixer.finish().unwrap(); + worker.join().unwrap(); + probe.collect(100); + assert_eq!(probe.peaks(1), [3_960]); + assert_eq!(probe.pcm.len(), 4_800); + } + + #[test] + fn pcm_stop_bounds_a_producer_that_never_closes() { + let mut probe = PcmProbe::new(2); + probe.send(1, 0, 4_800, Some(960)); + let start = Instant::now(); + probe.mixer.finish().unwrap(); + assert!(start.elapsed() < MIXER_STOP_TIMEOUT); + probe.collect(100); + assert_eq!(probe.peaks(1), [960]); + assert_eq!(probe.pcm.len(), 4_800); + assert!(probe.inputs.iter_mut().all(|input| input.is_closed())); + } + #[test] fn pcm_single_input_is_unchanged_and_reset_keeps_the_supplied_cursor() { let mut probe = PcmProbe::new(1); From c2ad9f17c94f19d4b41b0671df593f89dc167f4f Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 19:21:08 +0900 Subject: [PATCH 09/19] fix: reconcile delayed mixed audio with capture-time pauses --- crates/recording/src/output_pipeline/core.rs | 529 ++++++++++++++++++- crates/recording/src/sources/audio_mixer.rs | 10 + 2 files changed, 518 insertions(+), 21 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 0c35351ae69..c58abd4772a 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -2038,7 +2038,23 @@ impl SharedPauseState { } } +#[derive(Clone, Copy)] +struct CapturePause { + start: Instant, + end: Option, +} + +impl CapturePause { + fn duration_before(self, instant: Instant) -> Duration { + self.end + .unwrap_or(instant) + .min(instant) + .saturating_duration_since(self.start) + } +} + struct SharedWallClockPauseInner { + capture_pauses: Vec, pause_started_at: Option, total_pause_duration: Duration, } @@ -2054,12 +2070,91 @@ impl SharedWallClockPause { Self { flag, inner: Arc::new(std::sync::Mutex::new(SharedWallClockPauseInner { + capture_pauses: Vec::new(), pause_started_at: None, total_pause_duration: Duration::ZERO, })), } } + fn set_paused_at(&self, paused: bool, instant: Instant) { + let mut inner = self.inner.lock().unwrap_or_else(|error| error.into_inner()); + let active = inner + .capture_pauses + .last_mut() + .filter(|pause| pause.end.is_none()); + match (paused, active) { + (true, None) => inner.capture_pauses.push(CapturePause { + start: instant, + end: None, + }), + (false, Some(pause)) => pause.end = Some(instant.max(pause.start)), + _ => {} + } + self.flag.store(paused, Ordering::Release); + } + + fn captured_pause_duration(&self, instant: Instant) -> Duration { + let inner = self.inner.lock().unwrap_or_else(|error| error.into_inner()); + inner + .capture_pauses + .iter() + .map(|pause| pause.duration_before(instant)) + .sum() + } + + fn unpaused_offset(&self, timestamps: Timestamps, offset: Duration) -> Duration { + offset.saturating_sub(self.captured_pause_duration(timestamps.instant() + offset)) + } + + fn captured_audio_ranges( + &self, + frame: &AudioFrame, + timestamps: Timestamps, + rate: u32, + ) -> Option> { + let inner = self.inner.lock().unwrap_or_else(|error| error.into_inner()); + if inner.capture_pauses.is_empty() { + return None; + } + let start = timestamps.instant() + frame.timestamp.duration_since(timestamps); + let samples = frame.inner.samples(); + let index = |instant: Instant| { + (instant.saturating_duration_since(start).as_nanos() * u128::from(rate)) + .div_ceil(1_000_000_000) + .min(samples as u128) as usize + }; + let mut ranges = Vec::new(); + let mut cursor = 0; + for pause in &inner.capture_pauses { + let begin = index(pause.start); + if cursor < begin { + ranges.push((cursor, begin)); + } + cursor = cursor.max(pause.end.map(index).unwrap_or(samples)); + } + if cursor < samples { + ranges.push((cursor, samples)); + } + Some( + ranges + .into_iter() + .map(|(begin, end)| { + let instant = start + + Duration::from_nanos( + (begin as u128 * 1_000_000_000 / u128::from(rate)) as u64, + ); + let paused = inner + .capture_pauses + .iter() + .map(|pause| pause.duration_before(instant)) + .sum(); + (begin, end, paused) + }) + .collect(), + ) + } + pub fn check(&self) -> (bool, Duration) { let is_paused = self.flag.load(Ordering::Acquire); let mut inner = match self.inner.lock() { @@ -2884,7 +2979,7 @@ impl OutputPipelineBuilder> { build_ctx.done_tx, None, &path, - shared_pause, + shared_pause.clone(), true, video_start_gate, start_gate, @@ -2910,7 +3005,7 @@ impl OutputPipelineBuilder> { video_info: Some(video_info), stop_token: Some(build_ctx.stop_token.clone().drop_guard()), done_fut: build_ctx.done_rx, - pause_flag: build_ctx.pause_flag, + shared_pause, cancel_token: build_ctx.stop_token, video_frame_count, video_timestamp_span, @@ -2983,7 +3078,7 @@ impl OutputPipelineBuilder { build_ctx.done_tx, Some(first_tx), &path, - shared_pause, + shared_pause.clone(), false, None, start_gate, @@ -3002,7 +3097,7 @@ impl OutputPipelineBuilder { stop_token: Some(build_ctx.stop_token.clone().drop_guard()), video_info: None, done_fut: build_ctx.done_rx, - pause_flag: build_ctx.pause_flag, + shared_pause, cancel_token: build_ctx.stop_token, video_frame_count: Arc::new(AtomicU64::new(0)), video_timestamp_span: Arc::new(VideoTimestampSpan::default()), @@ -3848,6 +3943,7 @@ impl PreparedAudioSources { muxer: &muxer, health_tx: &health_tx, shared_pause: &shared_pause, + capture_time_pause: mixed_audio, video_start_gate: video_start_gate.as_ref(), start_gate: &start_gate.gate, allow_audio_degradation, @@ -3895,13 +3991,14 @@ impl PreparedAudioSources { { scope.fail_required(format!("Required audio processing failed: {error:#}")); } + let cancelled_at = Instant::now(); + let pause_at_stop = if mixed_audio { + shared_pause.captured_pause_duration(cancelled_at) + } else { + shared_pause.total_pause_duration() + }; let cancellation_target_elapsed = if was_cancelled && !audio_degraded { - Some( - timestamps - .instant() - .elapsed() - .saturating_sub(shared_pause.total_pause_duration()), - ) + Some(cancelled_at.saturating_duration_since(timestamps.instant()).saturating_sub(pause_at_stop)) } else { None }; @@ -3956,6 +4053,7 @@ impl PreparedAudioSources { muxer: &muxer, health_tx: &health_tx, shared_pause: &shared_pause, + capture_time_pause: mixed_audio, video_start_gate: video_start_gate.as_ref(), start_gate: &start_gate.gate, allow_audio_degradation, @@ -4018,7 +4116,7 @@ impl PreparedAudioSources { } } - let final_pause_duration = shared_pause.total_pause_duration(); + let final_pause_duration = if mixed_audio { pause_at_stop } else { shared_pause.total_pause_duration() }; if let Some(scope) = &required_scope && !mixed_sources_stopped @@ -4055,7 +4153,14 @@ impl PreparedAudioSources { // before comparing. let track_target = gap_tracker .track_start_offset() - .map(|offset| target_elapsed.saturating_sub(offset)) + .map(|offset| { + let offset = if mixed_audio { + shared_pause.unpaused_offset(timestamps, offset) + } else { + offset + }; + target_elapsed.saturating_sub(offset) + }) .unwrap_or(target_elapsed); let tail_padding = audio_tail_padding_duration(audio_elapsed, track_target); let start_samples = timestamp_generator.total_samples; @@ -4182,6 +4287,7 @@ struct AudioFrameProcessContext<'a, TMutex: AudioMuxer> { muxer: &'a Arc>, health_tx: &'a HealthSender, shared_pause: &'a SharedWallClockPause, + capture_time_pause: bool, video_start_gate: Option<&'a VideoStartGate>, start_gate: &'a Option, allow_audio_degradation: bool, @@ -4230,10 +4336,53 @@ fn audio_epoch(timestamps: Timestamps, start_gate: Option<&RecordingStartGate>) async fn process_audio_frame( ctx: AudioFrameProcessContext<'_, TMutex>, - state: AudioFrameProcessState<'_>, + mut state: AudioFrameProcessState<'_>, + frame: AudioFrame, +) -> anyhow::Result { + if !ctx.capture_time_pause { + return process_audio_frame_part(&ctx, &mut state, frame, None).await; + } + let Some(mut ranges) = + ctx.shared_pause + .captured_audio_ranges(&frame, ctx.timestamps, ctx.sample_rate) + else { + return process_audio_frame_part(&ctx, &mut state, frame, Some(Duration::ZERO)).await; + }; + if ranges.len() == 1 && ranges[0].0 == 0 && ranges[0].1 == frame.inner.samples() { + let (_, _, paused) = ranges.pop().unwrap(); + return process_audio_frame_part(&ctx, &mut state, frame, Some(paused)).await; + } + if ranges.is_empty() { + *state.dropped_during_pause += 1; + return Ok(AudioFrameOutcome::DroppedPaused); + } + let mut outcome = AudioFrameOutcome::DropFrame; + for (begin, end, paused) in ranges { + let mut inner = trim_audio_frame_front(&frame.inner, begin) + .context("Could not split mixed audio at a capture-time pause boundary")?; + inner.set_samples(end - begin); + let offset = Duration::from_nanos( + (begin as u128 * 1_000_000_000 / u128::from(ctx.sample_rate)) as u64, + ); + let part = AudioFrame::new(inner, frame.timestamp + offset); + match process_audio_frame_part(&ctx, &mut state, part, Some(paused)).await? { + AudioFrameOutcome::AudioDegraded => return Ok(AudioFrameOutcome::AudioDegraded), + AudioFrameOutcome::Sent => outcome = AudioFrameOutcome::Sent, + _ => {} + } + } + Ok(outcome) +} + +async fn process_audio_frame_part( + ctx: &AudioFrameProcessContext<'_, TMutex>, + state: &mut AudioFrameProcessState<'_>, mut frame: AudioFrame, + captured_pause_duration: Option, ) -> anyhow::Result { - let (is_paused, total_pause_duration) = ctx.shared_pause.check(); + let (is_paused, total_pause_duration) = captured_pause_duration + .map(|duration| (false, duration)) + .unwrap_or_else(|| ctx.shared_pause.check()); if is_paused { *state.dropped_during_pause += 1; @@ -4380,11 +4529,19 @@ async fn process_audio_frame( state.gap_tracker.mark_started(frame.timestamp, observed_at); let sample_based_before = state.timestamp_generator.next_timestamp(0); + let timeline_pause_duration = if ctx.capture_time_pause && !total_pause_duration.is_zero() { + let first = state.gap_tracker.first_frame_ts.unwrap(); + total_pause_duration.saturating_sub(ctx.shared_pause.captured_pause_duration( + ctx.timestamps.instant() + first.duration_since(ctx.timestamps), + )) + } else { + total_pause_duration + }; if let Some(overlap_duration) = state.gap_tracker.detect_overlap( frame.timestamp, sample_based_before, - total_pause_duration, + timeline_pause_duration, observed_at, ) { let trim_samples = duration_to_sample_count(overlap_duration, ctx.sample_rate) as usize; @@ -4425,7 +4582,7 @@ async fn process_audio_frame( if let Some(gap_duration) = state.gap_tracker.detect_gap( frame.timestamp, sample_based_before, - total_pause_duration, + timeline_pause_duration, observed_at, ) { let start_samples = state.timestamp_generator.total_samples; @@ -4653,7 +4810,7 @@ pub struct OutputPipeline { video_info: Option, stop_token: Option, done_fut: DoneFut, - pause_flag: Arc, + shared_pause: SharedWallClockPause, cancel_token: CancellationToken, video_frame_count: Arc, video_timestamp_span: Arc, @@ -4809,11 +4966,11 @@ impl OutputPipeline { } pub fn pause(&self) { - self.pause_flag.store(true, atomic::Ordering::Release); + self.shared_pause.set_paused_at(true, Instant::now()); } pub fn resume(&self) { - self.pause_flag.store(false, atomic::Ordering::Release); + self.shared_pause.set_paused_at(false, Instant::now()); } pub fn video_info(&self) -> Option { @@ -9233,6 +9390,7 @@ mod tests { struct RecordingAudioMuxer { sent: Arc>>, + pcm: Option>>>, } impl Muxer for RecordingAudioMuxer { @@ -9249,7 +9407,10 @@ mod tests { where Self: Sized, { - Ok(Self { sent: config }) + Ok(Self { + sent: config, + pcm: None, + }) } fn finish(&mut self, _timestamp: Duration) -> anyhow::Result> { @@ -9263,6 +9424,19 @@ mod tests { frame: AudioFrame, timestamp: Duration, ) -> anyhow::Result<()> { + if let Some(pcm) = &self.pcm { + let mut pcm = pcm.lock().unwrap(); + pcm.extend( + frame.inner.data(0)[..frame.inner.samples() * 8] + .chunks_exact(8) + .map(|pair| { + [ + f32::from_le_bytes(pair[..4].try_into().unwrap()), + f32::from_le_bytes(pair[4..].try_into().unwrap()), + ] + }), + ); + } self.sent.lock().unwrap().push(SentAudioFrame { samples: frame.inner.samples(), timestamp, @@ -9335,6 +9509,8 @@ mod tests { master_clock: Arc, muxer: Arc>, sent: Arc>>, + pcm: Arc>>, + capture_time_pause: bool, health_tx: HealthSender, shared_pause: SharedWallClockPause, timestamp_generator: AudioTimestampGenerator, @@ -9355,7 +9531,11 @@ mod tests { let timestamps = Timestamps::now(); let master_clock = MasterClock::new(timestamps, TEST_SAMPLE_RATE); let sent = Arc::new(std::sync::Mutex::new(Vec::new())); - let muxer = Arc::new(Mutex::new(RecordingAudioMuxer { sent: sent.clone() })); + let pcm = Arc::new(std::sync::Mutex::new(Vec::new())); + let muxer = Arc::new(Mutex::new(RecordingAudioMuxer { + sent: sent.clone(), + pcm: None, + })); let (health_tx, _) = new_health_channel(); let shared_pause = SharedWallClockPause::new(Arc::new(AtomicBool::new(false))); @@ -9365,6 +9545,8 @@ mod tests { master_clock: master_clock.clone(), muxer, sent, + pcm, + capture_time_pause: false, health_tx, shared_pause, timestamp_generator: AudioTimestampGenerator::from_master_clock(master_clock), @@ -9416,7 +9598,14 @@ mod tests { let timestamp = Timestamp::Instant(self.timestamps.instant() + capture_offset); let observed_at = self.timestamps.instant() + wall_offset; let frame = AudioFrame::new(make_test_frame(&self.info, samples, 0.0), timestamp); + self.process_frame_at(frame, observed_at).await + } + async fn process_frame_at( + &mut self, + frame: AudioFrame, + observed_at: Instant, + ) -> AudioFrameOutcome { process_audio_frame( AudioFrameProcessContext { audio_info: &self.info, @@ -9425,6 +9614,7 @@ mod tests { muxer: &self.muxer, health_tx: &self.health_tx, shared_pause: &self.shared_pause, + capture_time_pause: self.capture_time_pause, video_start_gate: None, start_gate: &self.start_gate, allow_audio_degradation: true, @@ -9449,6 +9639,34 @@ mod tests { .unwrap() } + fn mixed() -> Self { + let mut harness = Self::new(); + harness.info = AudioMixer::INFO; + harness.capture_time_pause = true; + harness.muxer = Arc::new(Mutex::new(RecordingAudioMuxer { + sent: harness.sent.clone(), + pcm: Some(harness.pcm.clone()), + })); + harness + } + + fn pause_at(&self, millis: u64, paused: bool) { + self.shared_pause.set_paused_at( + paused, + self.timestamps.instant() + Duration::from_millis(millis), + ); + } + + fn peaks(&self, channel: usize) -> Vec { + self.pcm + .lock() + .unwrap() + .iter() + .enumerate() + .filter_map(|(index, sample)| (sample[channel].abs() > 0.01).then_some(index)) + .collect() + } + fn committed_audio(&mut self) -> Duration { self.timestamp_generator.next_timestamp(0) } @@ -9466,6 +9684,275 @@ mod tests { } } + fn mixed_pause_frame( + timestamps: Timestamps, + start_ms: u64, + samples: usize, + peaks: &[(usize, usize)], + ) -> AudioFrame { + let mut bytes = vec![0; samples * 8]; + for &(sample, channel) in peaks { + let offset = sample * 8 + channel * 4; + bytes[offset..offset + 4].copy_from_slice(&0.5_f32.to_le_bytes()); + } + AudioFrame::new( + AudioMixer::INFO.wrap_frame(&bytes), + Timestamp::Instant(timestamps.instant() + Duration::from_millis(start_ms)), + ) + } + + #[tokio::test] + async fn mixed_pause_splits_capture_boundaries_independent_of_delivery() { + for late_delivery in [false, true] { + let mut harness = AudioTimelineHarness::mixed(); + let first = mixed_pause_frame(harness.timestamps, 0, 4_800, &[(960, 0), (960, 1)]); + if !late_delivery { + harness + .process_frame_at( + first, + harness.timestamps.instant() + Duration::from_millis(100), + ) + .await; + } + harness.pause_at(100, true); + harness.pause_at(300, false); + if late_delivery { + let first = + mixed_pause_frame(harness.timestamps, 0, 4_800, &[(960, 0), (960, 1)]); + harness + .process_frame_at( + first, + harness.timestamps.instant() + Duration::from_millis(600), + ) + .await; + } + let crossing = mixed_pause_frame( + harness.timestamps, + 100, + 19_200, + &[(4_800, 0), (4_800, 1), (10_560, 0), (10_560, 1)], + ); + harness + .process_frame_at( + crossing, + harness.timestamps.instant() + Duration::from_millis(600), + ) + .await; + assert_eq!(harness.peaks(0), [960, 5_760]); + assert_eq!(harness.peaks(1), [960, 5_760]); + assert_eq!(harness.timestamp_generator.total_samples, 14_400); + assert_eq!(harness.overlap_dropped_frames(), 0); + assert_eq!(harness.total_silence(), Duration::ZERO); + } + } + + #[tokio::test] + async fn mixed_pause_live_inputs_match_backlogged_delivery() { + for late_delivery in [false, true] { + let mut harness = AudioTimelineHarness::mixed(); + let mut builder = AudioMixer::builder() + .with_timestamps(harness.timestamps) + .with_master_clock(harness.master_clock.clone()); + let (output, mut received) = mpsc::channel(64); + let mut inputs = Vec::new(); + for _ in 0..2 { + let (input, receiver) = mpsc::channel(32); + builder.add_source(AudioMixer::INFO, receiver); + inputs.push(input); + } + let mut mixer = builder.build(output).unwrap(); + for chunk in 0..25 { + let millis = chunk * 20; + if millis == 100 { + harness.pause_at(100, true); + } + if millis == 300 { + harness.pause_at(300, false); + } + for (channel, input) in inputs.iter_mut().enumerate() { + let peaks = if [20, 200, 320].contains(&millis) { + vec![(0, channel)] + } else { + Vec::new() + }; + input + .try_send(mixed_pause_frame(harness.timestamps, millis, 960, &peaks)) + .unwrap(); + } + if !late_delivery { + mixer + .tick_for_capture_pause( + harness.timestamps.instant() + Duration::from_millis(millis + 20), + ) + .unwrap(); + while let Ok(Some(frame)) = received.try_next() { + harness + .process_frame_at( + frame, + harness.timestamps.instant() + + Duration::from_millis(millis + 20), + ) + .await; + } + } + } + drop(inputs); + mixer.finish_for_capture_pause().unwrap(); + while let Ok(Some(frame)) = received.try_next() { + harness + .process_frame_at( + frame, + harness.timestamps.instant() + Duration::from_millis(800), + ) + .await; + } + assert_eq!(harness.peaks(0), [960, 5_760]); + assert_eq!(harness.peaks(1), [960, 5_760]); + assert_eq!(harness.timestamp_generator.total_samples, 14_400); + assert_eq!(harness.overlap_dropped_frames(), 0); + assert_eq!(harness.total_silence(), Duration::ZERO); + } + } + + #[tokio::test] + async fn mixed_pause_keeps_watermark_held_capture_with_absent_input() { + for stop_while_paused in [false, true] { + let mut harness = AudioTimelineHarness::mixed(); + let mut builder = AudioMixer::builder() + .with_timestamps(harness.timestamps) + .with_master_clock(harness.master_clock.clone()); + let (system, system_rx) = mpsc::channel(8); + let (mut mic, mic_rx) = mpsc::channel(8); + builder.add_source(AudioMixer::INFO, system_rx); + builder.add_source(AudioMixer::INFO, mic_rx); + let (output, mut received) = mpsc::channel(32); + let mut mixer = builder.build(output).unwrap(); + mic.try_send(mixed_pause_frame(harness.timestamps, 0, 4_800, &[(960, 1)])) + .unwrap(); + mixer + .tick_for_capture_pause( + harness.timestamps.instant() + Duration::from_millis(100), + ) + .unwrap(); + assert!(received.try_next().is_err()); + harness.pause_at(100, true); + if stop_while_paused { + drop(system); + drop(mic); + mixer.finish_for_capture_pause().unwrap(); + } else { + mixer + .tick_for_capture_pause( + harness.timestamps.instant() + Duration::from_millis(1_000), + ) + .unwrap(); + } + while let Ok(Some(frame)) = received.try_next() { + harness + .process_frame_at( + frame, + harness.timestamps.instant() + Duration::from_millis(1_000), + ) + .await; + } + assert_eq!(harness.peaks(1), [960]); + if stop_while_paused { + assert_eq!(harness.timestamp_generator.total_samples, 4_800); + } + assert_eq!(harness.dropped_during_pause, 0); + } + } + + #[tokio::test] + async fn mixed_pause_drops_held_paused_capture_and_keeps_resumed_samples() { + let mut harness = AudioTimelineHarness::mixed(); + harness.pause_at(100, true); + harness.pause_at(300, false); + let frame = mixed_pause_frame( + harness.timestamps, + 0, + 24_000, + &[(960, 0), (9_600, 0), (19_200, 0)], + ); + harness + .process_frame_at( + frame, + harness.timestamps.instant() + Duration::from_millis(800), + ) + .await; + assert_eq!(harness.peaks(0), [960, 9_600]); + assert_eq!(harness.timestamp_generator.total_samples, 14_400); + assert_eq!(harness.total_silence(), Duration::ZERO); + assert_eq!(harness.overlap_dropped_frames(), 0); + } + + #[tokio::test] + async fn mixed_pause_before_first_sample_and_repeated_cycles_keep_the_timeline() { + let mut harness = AudioTimelineHarness::mixed(); + harness.pause_at(20, true); + harness.pause_at(100, false); + harness.pause_at(200, true); + harness.pause_at(300, false); + harness.pause_at(350, true); + harness.pause_at(400, false); + let frame = mixed_pause_frame( + harness.timestamps, + 100, + 19_200, + &[(0, 0), (4_800, 0), (9_600, 0), (12_000, 0), (14_400, 0)], + ); + harness + .process_frame_at( + frame, + harness.timestamps.instant() + Duration::from_millis(800), + ) + .await; + let first = harness + .shared_pause + .unpaused_offset(harness.timestamps, Duration::from_millis(100)); + let stopped = harness + .shared_pause + .unpaused_offset(harness.timestamps, Duration::from_millis(500)); + assert_eq!(first, Duration::from_millis(20)); + assert_eq!(stopped.saturating_sub(first), Duration::from_millis(250)); + assert_eq!(harness.peaks(0), [0, 4_800, 7_200]); + assert_eq!(harness.timestamp_generator.total_samples, 12_000); + assert_eq!(harness.total_silence(), Duration::ZERO); + assert_eq!(harness.overlap_dropped_frames(), 0); + } + + #[tokio::test] + async fn mixed_pipeline_stop_while_paused_drains_pre_pause_capture() { + let temp = tempfile::tempdir().unwrap(); + let timestamps = Timestamps::now(); + let sent = Arc::new(std::sync::Mutex::new(Vec::new())); + let pipeline = OutputPipeline::builder(temp.path().join("mixed-paused-stop.ogg")) + .with_timestamps(timestamps) + .with_audio_source::(None) + .with_audio_source::(Some(mixed_pause_frame( + timestamps, + 0, + 4_800, + &[(960, 1)], + ))) + .build::(sent.clone()) + .await + .unwrap(); + pipeline + .shared_pause + .set_paused_at(true, timestamps.instant() + Duration::from_millis(100)); + tokio::time::timeout(Duration::from_secs(3), pipeline.stop()) + .await + .unwrap() + .unwrap(); + let frames = sent.lock().unwrap(); + assert_eq!( + frames.iter().map(|frame| frame.samples).sum::(), + 4_800 + ); + assert_eq!(frames[0].timestamp, Duration::ZERO); + } + #[tokio::test(flavor = "current_thread")] async fn primed_track_holds_audio_until_the_start_gate_arms() { let gate = RecordingStartGate::new(); diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index 19717f263a0..e031ab817b6 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -633,6 +633,16 @@ impl AudioMixer { self.tick_with_output(now, |sink, frame| sink.sink().frame(frame)) } + #[cfg(test)] + pub(crate) fn tick_for_capture_pause(&mut self, now: Instant) -> anyhow::Result { + self.tick(self.timestamps, Timestamp::Instant(now)) + } + + #[cfg(test)] + pub(crate) fn finish_for_capture_pause(&mut self) -> anyhow::Result<()> { + self.finish() + } + fn finish(&mut self) -> anyhow::Result<()> { if let Some(stop) = &self.required_stop { stop.cancel(); From fe088a42b00e005a8b557f9656c32b737aff435b Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:09:41 +0900 Subject: [PATCH 10/19] fix: absorb mixer input timestamp jitter within 2 ms Mixer inputs were placed at the exact sample of every capture timestamp. Receipt-time stamps (Linux) jitter by milliseconds, so each late callback became inserted silence followed by dropped input. Packets within 2 ms of the source's sample cursor now continue it untouched; between 2 and 10 ms the cursor moves at most one sample per 256 (clock drift); from 10 ms a packet starts at its capture time as before. --- crates/recording/src/sources/audio_mixer.rs | 94 ++++++++++++++++++++- 1 file changed, 92 insertions(+), 2 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index e031ab817b6..e22c8d5a1f6 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -732,7 +732,12 @@ impl AudioMixer { while let Some(mut buffer) = source.buffer.pop_front() { let position = sample_position(buffer.timestamp, start_timestamp, self.timestamps, rate); - let overlap = (source.samples_supplied as i128 - position).max(0) as usize; + let correction = placement_correction( + position - source.samples_supplied as i128, + source.info.sample_rate, + buffer.inner.samples(), + ); + let overlap = (-correction).max(0) as usize; if overlap > 0 { let dropped = overlap.min(buffer.inner.samples()); if source.late_samples_dropped == 0 { @@ -752,7 +757,7 @@ impl AudioMixer { buffer.timestamp = start_timestamp + samples_duration(source.samples_supplied, source.info.sample_rate); } - let gap_end = (position.max(0) as usize).min(limit); + let gap_end = (source.samples_supplied + correction.max(0) as usize).min(limit); supply_silence(source, &mut self.abuffers[i], i, gap_end)?; if source.samples_supplied >= limit { source.buffer.push_front(buffer); @@ -859,6 +864,35 @@ fn mixer_now() -> Timestamp { } } +/// Input timestamps jitter around their sample clock (Linux stamps samples +/// with their receipt time) and drift from the host clock by a few parts per +/// million. Within this tolerance a packet continues the source's samples +/// untouched. +const PLACEMENT_JITTER_TOLERANCE: Duration = Duration::from_millis(2); +/// A packet this far from where the source's samples have reached starts at +/// its capture time: a real gap is filled with silence, input that arrived +/// after its time was committed is dropped. +const PLACEMENT_REALIGN_THRESHOLD: Duration = Duration::from_millis(10); + +/// Samples to add (silence, positive) or drop (negative) before a packet +/// of `packet` samples whose capture time lies `drift` samples after the +/// source's supplied samples. Between the tolerance and the realign +/// threshold the source moves by at most one sample per 256 (a 0.4% rate +/// change), so clock drift never causes an audible jump. +fn placement_correction(drift: i128, rate: u32, packet: usize) -> i128 { + let samples = + |duration: Duration| duration.as_nanos() as i128 * i128::from(rate) / 1_000_000_000; + let tolerance = samples(PLACEMENT_JITTER_TOLERANCE); + if drift.abs() >= samples(PLACEMENT_REALIGN_THRESHOLD) { + drift + } else if drift.abs() > tolerance { + let step = (packet as i128 / 256).max(1); + drift.signum() * (drift.abs() - tolerance).min(step) + } else { + 0 + } +} + fn samples_duration(samples: usize, rate: u32) -> Duration { Duration::from_nanos((samples as u128 * 1_000_000_000 / u128::from(rate)) as u64) } @@ -1463,6 +1497,62 @@ mod test { assert!(probe.inputs.iter_mut().all(|input| input.is_closed())); } + #[test] + fn pcm_timestamp_jitter_within_tolerance_keeps_samples_contiguous() { + let mut probe = PcmProbe::new(1); + let base = probe.mixer.timestamps.instant() + Duration::from_millis(100); + let jitter_us = [ + 0_i64, 1_500, -800, 1_900, 300, -1_900, 1_200, 0, 700, -1_500, + ]; + for (packet, jitter) in jitter_us.into_iter().enumerate() { + let micros = 20_000 * packet as i64 + jitter; + let timestamp = base + Duration::from_micros(micros.max(0) as u64); + probe.send_at(0, Timestamp::Instant(timestamp), 960, Some(0)); + probe.tick(20 * packet as u64 + 125); + } + assert_eq!( + probe.peaks(0), + (0..jitter_us.len()) + .map(|packet| packet * 960) + .collect::>() + ); + assert_eq!(probe.mixer.sources[0].late_samples_dropped, 0); + } + + #[test] + fn pcm_clock_drift_is_corrected_one_sample_at_a_time() { + let mut probe = PcmProbe::new(1); + let base = probe.mixer.timestamps.instant() + Duration::from_millis(100); + // 2,000 ppm: each 960-sample packet's capture time runs 1.92 + // samples ahead of its sample count. + let spacing = Duration::from_nanos(20_040_000); + let packets = 200; + for packet in 0..packets { + probe.send_at( + 0, + Timestamp::Instant(base + spacing * packet as u32), + 960, + Some(0), + ); + probe.tick(20 * packet as u64 + 125); + } + let peaks = probe.peaks(0); + assert_eq!(peaks.len(), packets); + for pair in peaks.windows(2) { + assert!( + (960..=963).contains(&(pair[1] - pair[0])), + "a drift correction jumped: {pair:?}" + ); + } + let last_position = + (spacing.as_nanos() * (packets as u128 - 1) * 48_000 / 1_000_000_000) as usize; + assert!( + last_position.abs_diff(*peaks.last().unwrap()) <= 96 + 2, + "drift not followed: {} vs {last_position}", + peaks.last().unwrap() + ); + } + #[test] fn pcm_single_input_is_unchanged_and_reset_keeps_the_supplied_cursor() { let mut probe = PcmProbe::new(1); From a591d9bcd4c76e305b3d4eabb862c45bbf1878c4 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:09:41 +0900 Subject: [PATCH 11/19] fix: keep mixed audio when sources stop slowly A mixed audio source that took over 1 s to stop, or a mixer that had not closed 2 s later, failed the recording. Bluetooth inputs can take that long to tear down. Both now record a cleanup error and keep the audio drained so far, as a failed source stop already did. --- crates/recording/src/output_pipeline/core.rs | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index c58abd4772a..f7858bfe277 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -4010,12 +4010,16 @@ impl PreparedAudioSources { stop_audio_sources(&mut self.erased_audio_sources, required_scope.as_ref()), ) .await; + // A slow device teardown (Bluetooth inputs can take + // seconds) costs at most the audio after this point: + // the mixer closes its inputs at its own drain deadline. if stopped.is_err() { - let error = anyhow!("Mixed audio sources did not stop before the deadline"); + warn!("Mixed audio sources did not stop within 1 s; draining what the mixer has"); if let Some(scope) = &required_scope { - scope.fail_required(error.to_string()); + scope.record_cleanup_error( + "Mixed audio sources did not stop before the deadline".to_string(), + ); } - let _ = audio_error.get_or_insert(error); } self.erased_audio_sources.clear(); } @@ -4033,11 +4037,12 @@ impl PreparedAudioSources { match tokio::time::timeout_at(drain_deadline, self.audio_rx.next()).await { Ok(frame) => frame, Err(_) => { - let error = anyhow!("Mixed audio output did not close before the drain deadline"); + warn!(drained, "Mixed audio output did not close before the drain deadline; keeping the audio drained so far"); if let Some(scope) = &required_scope { - scope.fail_required(error.to_string()); + scope.record_cleanup_error( + "Mixed audio output did not close before the drain deadline".to_string(), + ); } - let _ = audio_error.get_or_insert(error); break; } } From 3e2e2128517066fe3f253752b6f17972d62fec62 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:20:51 +0900 Subject: [PATCH 12/19] fix: drop undelivered mixed audio at the stop deadline instead of failing An encoder that stopped draining the mixer's output past the stop deadline returned an error from finish(), which failed the recording. The mixer now stops delivering and keeps the audio already sent. --- crates/recording/src/sources/audio_mixer.rs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index e22c8d5a1f6..e6d0669ba0a 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -823,11 +823,17 @@ impl AudioMixer { match self.output.try_send(frame) { Ok(()) => break, Err(err) if err.is_full() => { - anyhow::ensure!( - self.drain_deadline - .is_none_or(|deadline| Instant::now() < deadline), - "Audio mixer output did not drain before the stop deadline" - ); + // At stop, a consumer that stopped draining costs the + // audio not yet delivered, never the recording. + if self + .drain_deadline + .is_some_and(|deadline| Instant::now() >= deadline) + { + warn!( + "Audio mixer output did not drain before the stop deadline; dropping the rest" + ); + return Ok(false); + } frame = err.into_inner(); std::thread::sleep(Duration::from_millis(1)); } From 4ac1650dae734e15b1a3485f6e860982e5103bda Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:24:52 +0900 Subject: [PATCH 13/19] fix: stop mixed audio sources together and clip the drain at Stop Sources were stopped one after another, so a microphone kept capturing while system audio tore down, and that audio after Stop reached the file through the mixer drain. All sources now stop together, the drain keeps only audio captured before Stop, and the source-stop and drain timeouts no longer record cleanup errors (Instant treats any cleanup error as a failed recording); the mixer closes its inputs at its own deadline. --- crates/recording/src/output_pipeline/core.rs | 101 ++++++++++++++----- 1 file changed, 78 insertions(+), 23 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index f7858bfe277..28dff3f2440 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -3861,6 +3861,45 @@ fn spawn_video_encoder, TVideo: V started } +async fn stop_audio_sources_together( + sources: &mut [ErasedAudioSource], + required_scope: Option<&PipelineBuildScope>, +) { + let results = futures::future::join_all( + sources + .iter_mut() + .map(|source| (source.stop_fn)(source.inner.as_mut())), + ) + .await; + for (source, result) in sources.iter_mut().zip(results) { + let cleanup_scope = source.cleanup_scope.take(); + if let Err(error) = result + && let Some(scope) = cleanup_scope.as_ref().or(required_scope) + { + scope.record_cleanup_error(error.to_string()); + } + } +} + +/// Mixed audio drained at stop: keeps what was captured before `stop`, also +/// when it arrives late, and drops what sources captured after it. +fn clip_to_capture_time( + mut frame: AudioFrame, + timestamps: Timestamps, + stop: Instant, +) -> Option { + let start = timestamps.instant() + frame.timestamp.duration_since(timestamps); + let kept = stop.saturating_duration_since(start).as_nanos() * u128::from(frame.inner.rate()) + / 1_000_000_000; + if kept == 0 { + return None; + } + if (kept as usize) < frame.inner.samples() { + frame.inner.set_samples(kept as usize); + } + Some(frame) +} + async fn stop_audio_sources( sources: &mut [ErasedAudioSource], required_scope: Option<&PipelineBuildScope>, @@ -4005,22 +4044,15 @@ impl PreparedAudioSources { let mixed_sources_stopped = was_cancelled && mixed_audio; if mixed_sources_stopped { - let stopped = tokio::time::timeout( - Duration::from_secs(1), - stop_audio_sources(&mut self.erased_audio_sources, required_scope.as_ref()), + // The mixer finishes once every input has ended (or at its + // own drain deadline), so its sources stop first, together: + // a source still capturing while another tears down would + // feed audio from after Stop into the drain. + stop_audio_sources_together( + &mut self.erased_audio_sources, + required_scope.as_ref(), ) .await; - // A slow device teardown (Bluetooth inputs can take - // seconds) costs at most the audio after this point: - // the mixer closes its inputs at its own drain deadline. - if stopped.is_err() { - warn!("Mixed audio sources did not stop within 1 s; draining what the mixer has"); - if let Some(scope) = &required_scope { - scope.record_cleanup_error( - "Mixed audio sources did not stop before the deadline".to_string(), - ); - } - } self.erased_audio_sources.clear(); } @@ -4038,14 +4070,10 @@ impl PreparedAudioSources { Ok(frame) => frame, Err(_) => { warn!(drained, "Mixed audio output did not close before the drain deadline; keeping the audio drained so far"); - if let Some(scope) = &required_scope { - scope.record_cleanup_error( - "Mixed audio output did not close before the drain deadline".to_string(), - ); - } break; } } + .and_then(|frame| clip_to_capture_time(frame, timestamps, cancelled_at)) } else { self.audio_rx.try_next().ok().flatten() }; @@ -6655,6 +6683,28 @@ mod tests { } } + mod clip_to_capture_time { + use super::super::*; + + fn frame_at(timestamps: Timestamps, millis: u64) -> AudioFrame { + AudioFrame::new( + AudioMixer::INFO.wrap_frame(&vec![0; 960 * 8]), + Timestamp::Instant(timestamps.instant() + Duration::from_millis(millis)), + ) + } + + #[test] + fn keeps_only_samples_captured_before_stop() { + let timestamps = Timestamps::now(); + let stop = timestamps.instant() + Duration::from_millis(110); + let clipped = clip_to_capture_time(frame_at(timestamps, 100), timestamps, stop); + assert_eq!(clipped.unwrap().inner.samples(), 480); + let whole = clip_to_capture_time(frame_at(timestamps, 50), timestamps, stop); + assert_eq!(whole.unwrap().inner.samples(), 960); + assert!(clip_to_capture_time(frame_at(timestamps, 110), timestamps, stop).is_none()); + } + } + mod audio_tail_padding { use super::*; @@ -9496,16 +9546,18 @@ mod tests { .build::(sent.clone()) .await .unwrap(); + // The 100 ms frame was captured entirely before Stop. + tokio::time::sleep_until((timestamps.instant() + Duration::from_millis(150)).into()) + .await; tokio::time::timeout(Duration::from_secs(3), pipeline.stop()) .await .unwrap() .unwrap(); let frames = sent.lock().unwrap(); - assert_eq!( - frames.iter().map(|frame| frame.samples).sum::(), - 4_800 - ); + // The delivered frame is the only audio, so the track starts with + // it and is then padded with silence out to Stop. assert_eq!(frames[0].timestamp, Duration::ZERO); + assert!(frames.iter().map(|frame| frame.samples).sum::() >= 4_800); } struct AudioTimelineHarness { @@ -9946,6 +9998,9 @@ mod tests { pipeline .shared_pause .set_paused_at(true, timestamps.instant() + Duration::from_millis(100)); + // The 100 ms frame was captured entirely before Stop. + tokio::time::sleep_until((timestamps.instant() + Duration::from_millis(150)).into()) + .await; tokio::time::timeout(Duration::from_secs(3), pipeline.stop()) .await .unwrap() From ca7a2455f218c8a961287f6002c2e73e9b1ef106 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:26:49 +0900 Subject: [PATCH 14/19] fix: keep mixer input captured before the clock origin at its capture time The source clock remap clamps a packet captured before the master clock's origin to the origin. The first real-capture microphone packet started 3.6 ms early, so it was stamped at the origin with all 480 samples and overlapped the next packet by 174 samples; the mixer then moved the microphone back over two seconds. The mixer now drops only the samples captured before the origin. --- crates/recording/src/sources/audio_mixer.rs | 46 ++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/crates/recording/src/sources/audio_mixer.rs b/crates/recording/src/sources/audio_mixer.rs index e6d0669ba0a..498c73aeb7f 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -592,7 +592,28 @@ impl AudioMixer { source.last_input_timestamp = None; } - let timestamp = Timestamp::Instant(clock_anchor + remap.duration()); + // The remap clamps a packet captured before the master clock's + // origin to the origin. Drop only the samples that precede it, + // so the rest of the packet keeps its capture time instead of + // overlapping the next packet. + let (frame, master_ns, frame_duration_ns) = if remap.master_ns == 0 + && remap.raw_ns < 0 + && rate > 0 + { + let before_ns = u128::from(remap.raw_ns.unsigned_abs()); + let before = (before_ns * rate as u128).div_ceil(1_000_000_000) as usize; + let Some(kept) = trim_audio_frame_front(&frame, before) else { + continue; + }; + let first_ns = (before as u128 * 1_000_000_000 / rate as u128) + .saturating_sub(before_ns) as u64; + let kept_ns = (kept.samples() as u128 * 1_000_000_000 / rate as u128) as u64; + (kept, first_ns, kept_ns) + } else { + (frame, remap.master_ns, frame_duration_ns) + }; + + let timestamp = Timestamp::Instant(clock_anchor + Duration::from_nanos(master_ns)); source.buffer_last = Some((timestamp, Duration::from_nanos(frame_duration_ns))); source.buffer.push_back(AudioFrame::new(frame, timestamp)); @@ -1503,6 +1524,29 @@ mod test { assert!(probe.inputs.iter_mut().all(|input| input.is_closed())); } + #[test] + fn pcm_input_captured_before_the_clock_origin_keeps_its_capture_time() { + let mut probe = PcmProbe::new(1); + let origin = probe.mixer.timestamps.instant(); + // 480 samples from 4 ms before the origin: 192 precede it, and the + // marker at sample 300 was captured 2.25 ms after it. + probe.send_at( + 0, + Timestamp::Instant(origin.checked_sub(Duration::from_millis(4)).unwrap()), + 480, + Some(300), + ); + probe.send_at( + 0, + Timestamp::Instant(origin + Duration::from_millis(6)), + 480, + Some(0), + ); + probe.tick(100); + assert_eq!(probe.peaks(0), [108, 288]); + assert_eq!(probe.mixer.sources[0].late_samples_dropped, 0); + } + #[test] fn pcm_timestamp_jitter_within_tolerance_keeps_samples_contiguous() { let mut probe = PcmProbe::new(1); From 7c4c17ab430cb998322d8de424fba542575477c5 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:26:49 +0900 Subject: [PATCH 15/19] test: give Instant system audio the microphone's gated offset With mixer inputs aligned, system audio sits on the microphone's timeline (system minus mic 0 ms) and shares its gated offset, replacing the -60 ms and -6.7 ms mixer-mode entries. --- crates/verify/expectations.toml | 25 ++++++++----------------- 1 file changed, 8 insertions(+), 17 deletions(-) diff --git a/crates/verify/expectations.toml b/crates/verify/expectations.toml index 09db7f24934..3a498a886ac 100644 --- a/crates/verify/expectations.toml +++ b/crates/verify/expectations.toml @@ -300,14 +300,6 @@ bound = { count = 1, missing = 2, lost = 2 } reason = "NEW: gated loopback-only system audio sometimes loses markers (2 of 11 lost in 1 of 2 fresh recordings, with the track 958 ms early). Bound: at most 2 markers lost, none unexpected" link = "" -[[xfail]] -case = "export/instant/output/mp4/native" -class = "sync-system" -flaky = true -bound = { count = 1, offset-ms = [-2.7, 28.3], mic-offset-ms = [5, 34], system-minus-mic-ms = [-7.7, -5.7] } -reason = "Instant mixer input timing and the gate phase produce a -6.7 ms system-minus-mic mode, within 1 ms. Only this inter-source signature with the microphone inside its unchanged 5 to 34 ms bound is excused. Coordinator decision 09:20, 2026-10-16." -link = "https://github.com/CapSoftware/Cap/tree/fix/instant-mixer-input-alignment" - [[xfail]] case = "export/instant/output/mp4/native" class = "sync-mic" @@ -315,6 +307,14 @@ bound = { count = 1, offset-ms = [5, 34] } reason = "NEW: see export/instant-gated-*: mic audio up to one video frame late after the start gate (14.2 to 15.3 ms measured)" link = "" + +[[xfail]] +case = "export/instant/output/mp4/native" +class = "sync-system" +bound = { count = 1, offset-ms = [5, 34], mic-offset-ms = [5, 34], system-minus-mic-ms = [-1, 1] } +reason = "NEW: the mixer places system audio on the microphone's timeline (system minus mic 0 ms), so it shares the gated mic offset of the sync-mic entry above (13.3 to 15.4 ms measured)" +link = "" + [[xfail]] case = "*/basic/transition/*60*/avf" class = "stale" @@ -662,12 +662,3 @@ platforms = ["macos"] bound = { count = 1 } reason = "NEW (bugs B1 and C, see loss-time): nothing recoverable (no frames)" link = "https://github.com/CapSoftware/Cap/pull/2643" - -[[xfail]] -case = "export/instant/output/mp4/native" -class = "sync-system" -platforms = ["macos"] -flaky = true -bound = { count = 1, offset-ms = [-56, -25], mic-offset-ms = [5, 34], system-minus-mic-ms = [-61, -59] } -reason = "Instant mixer input timing produces a -60 ms system-minus-mic mode, within 1 ms. Only this inter-source signature with the microphone inside its unchanged 5 to 34 ms bound is excused. Coordinator decision 09:20, 2026-10-16." -link = "https://github.com/CapSoftware/Cap/tree/fix/instant-mixer-input-alignment" From a841eed984bb458440fdfbe267de99d8cae9b36a Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:33:36 +0900 Subject: [PATCH 16/19] test: pin the Instant system audio entry to the microphone timeline The report test pinned the two removed mixer-mode entries; it now checks the single strict entry accepts system audio within 1 ms of the mic and rejects both old modes. --- crates/verify/src/report.rs | 47 +++++++++++++------------------------ 1 file changed, 16 insertions(+), 31 deletions(-) diff --git a/crates/verify/src/report.rs b/crates/verify/src/report.rs index 9ecabe828dc..c35b36d83dc 100644 --- a/crates/verify/src/report.rs +++ b/crates/verify/src/report.rs @@ -646,14 +646,17 @@ mod tests { ); } #[test] - fn instant_system_offset_expectation_requires_its_exact_signature() { + fn instant_system_audio_expectation_requires_the_microphone_timeline() { let entries = cap_verify_expect::parse(include_str!("../expectations.toml")).unwrap(); let entries: Vec<_> = entries .iter() - .filter(|entry| entry.class == "sync-system" && entry.flaky) + .filter(|entry| { + entry.class == "sync-system" && entry.case == "export/instant/output/mp4/native" + }) .collect(); - assert_eq!(entries.len(), 2); - let metrics = |offset, mic| { + assert_eq!(entries.len(), 1); + assert!(!entries[0].flaky); + let metrics = |offset: f64, mic: f64| { BTreeMap::from([ ("count".into(), cap_verify_expect::Range::at(1.0)), ("offset-ms".into(), cap_verify_expect::Range::at(offset)), @@ -665,38 +668,20 @@ mod tests { ]) }; let accepts = |metrics: &BTreeMap| { - entries - .iter() - .any(|entry| cap_verify_expect::within(&entry.bound, Some(metrics)).is_ok()) + cap_verify_expect::within(&entries[0].bound, Some(metrics)).is_ok() }; - for (offset, mic) in [(-46.1, 13.9), (7.8, 14.5), (-47.1, 13.9), (8.8, 14.5)] { + for (offset, mic) in [(14.0, 14.0), (13.3, 13.3), (15.4, 15.0)] { assert!(accepts(&metrics(offset, mic))); } - for (offset, mic) in [ - (-47.2, 13.9), - (-45.0, 13.9), - (6.7, 14.5), - (8.9, 14.5), - (-25.0, 35.0), - (-56.0, 4.0), - (0.0, 14.5), - ] { + // The -60 ms and -6.7 ms mixer modes, and system audio off the mic. + for (offset, mic) in [(-46.1, 13.9), (7.8, 14.5), (16.0, 14.5), (14.0, 35.0)] { assert!(!accepts(&metrics(offset, mic))); } - for required in ["mic-offset-ms", "system-minus-mic-ms"] { - let mut absent = metrics(-46.1, 13.9); - absent.remove(required); - assert!(!accepts(&absent)); - } - for entry in &entries { - assert!(entry.matches("export/instant/output/mp4/native", "macos")); + let mut absent = metrics(14.0, 14.0); + absent.remove("system-minus-mic-ms"); + assert!(!accepts(&absent)); + for platform in ["macos", "windows", "linux"] { + assert!(entries[0].matches("export/instant/output/mp4/native", platform)); } - assert_eq!( - entries - .iter() - .filter(|entry| entry.matches("export/instant/output/mp4/native", "linux")) - .count(), - 1 - ); } } From 95171b1a7b841ccfa95ba99634b89449437aaf8c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:39:26 +0900 Subject: [PATCH 17/19] fix: bound the mixed source stop with the mixer's stop deadline A source whose device teardown hangs held the pipeline stop forever. The stop now waits at most MIXER_STOP_TIMEOUT, logs, and drains what the mixer has; it records no cleanup error, which Instant treats as failure. --- crates/recording/src/output_pipeline/core.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 28dff3f2440..2c3f9202991 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -4048,11 +4048,21 @@ impl PreparedAudioSources { // own drain deadline), so its sources stop first, together: // a source still capturing while another tears down would // feed audio from after Stop into the drain. - stop_audio_sources_together( - &mut self.erased_audio_sources, - required_scope.as_ref(), + // A device whose teardown hangs must not hang the + // recording: past the deadline the drain keeps what the + // mixer has, and dropping the sources ends their capture. + if tokio::time::timeout( + MIXER_STOP_TIMEOUT, + stop_audio_sources_together( + &mut self.erased_audio_sources, + required_scope.as_ref(), + ), ) - .await; + .await + .is_err() + { + warn!("Mixed audio sources did not stop within the stop deadline"); + } self.erased_audio_sources.clear(); } From c9f875e0f64c1a6b0f08a7569579fa7d0bd565f2 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:16:03 +0900 Subject: [PATCH 18/19] fix: count every pause for video and drop hung sources without rerunning their stop Pause and Resume now advance the wall-clock pause total at their own instants, so video removes the same pause time as mixed audio even when a static screen sends no frame through check() during the pause. A mixed source that misses the stop deadline is dropped without its cleanup scope, so the stop that hung is not spawned again as scope cleanup. --- crates/recording/src/output_pipeline/core.rs | 34 ++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 2c3f9202991..55537921bde 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -2091,6 +2091,17 @@ impl SharedWallClockPause { (false, Some(pause)) => pause.end = Some(instant.max(pause.start)), _ => {} } + // The wall-clock total `check` reports advances here too, with the + // same instants: a static screen sends no video frame through + // `check` across Pause and Resume, and video must still remove + // exactly the pause mixed audio removes. + if paused { + inner.pause_started_at.get_or_insert(instant); + } else if let Some(started) = inner.pause_started_at.take() { + inner.total_pause_duration = inner + .total_pause_duration + .saturating_add(instant.saturating_duration_since(started)); + } self.flag.store(paused, Ordering::Release); } @@ -4062,6 +4073,11 @@ impl PreparedAudioSources { .is_err() { warn!("Mixed audio sources did not stop within the stop deadline"); + // Dropping a source with its cleanup scope would run + // the stop that just hung again, as scope cleanup. + for source in &mut self.erased_audio_sources { + source.cleanup_scope = None; + } } self.erased_audio_sources.clear(); } @@ -6260,6 +6276,24 @@ mod tests { mod shared_pause_state { use super::*; + #[test] + fn wall_clock_pause_counts_pauses_no_frame_observed() { + let pause = SharedWallClockPause::new(Arc::new(AtomicBool::new(false))); + let start = Instant::now(); + pause.set_paused_at(true, start); + pause.set_paused_at(false, start + Duration::from_millis(300)); + assert_eq!(pause.total_pause_duration(), Duration::from_millis(300)); + assert_eq!( + pause.captured_pause_duration(start + Duration::from_secs(1)), + Duration::from_millis(300) + ); + assert_eq!(pause.check(), (false, Duration::from_millis(300))); + pause.set_paused_at(true, start + Duration::from_secs(1)); + assert!(pause.check().0); + pause.set_paused_at(false, start + Duration::from_millis(1_200)); + assert_eq!(pause.check(), (false, Duration::from_millis(500))); + } + fn frame_ts(index: u64) -> Duration { Duration::from_nanos(index * 33_333_333 + 400) } From 5569b930775db45abcf2b265761145f636ca3e79 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 02:03:11 +0900 Subject: [PATCH 19/19] fix: read the pause flag under the pause lock check() read the flag before taking the lock set_paused_at holds while it records a pause, so a frame holding a stale flag could end a pause that had just started and lose its time from the video total. --- crates/recording/src/output_pipeline/core.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 55537921bde..dee920e3d43 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -2167,11 +2167,13 @@ impl SharedWallClockPause { } pub fn check(&self) -> (bool, Duration) { - let is_paused = self.flag.load(Ordering::Acquire); let mut inner = match self.inner.lock() { Ok(guard) => guard, Err(poisoned) => poisoned.into_inner(), }; + // Read under the lock `set_paused_at` holds while it updates both, + // so a stale flag cannot end a pause it just started. + let is_paused = self.flag.load(Ordering::Acquire); if is_paused { if inner.pause_started_at.is_none() {