diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 96057442171..dee920e3d43 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}; @@ -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,18 +2070,110 @@ 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)), + _ => {} + } + // 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); + } + + 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() { 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() { @@ -2884,7 +2992,7 @@ impl OutputPipelineBuilder> { build_ctx.done_tx, None, &path, - shared_pause, + shared_pause.clone(), true, video_start_gate, start_gate, @@ -2910,7 +3018,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 +3091,7 @@ impl OutputPipelineBuilder { build_ctx.done_tx, Some(first_tx), &path, - shared_pause, + shared_pause.clone(), false, None, start_gate, @@ -3002,7 +3110,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()), @@ -3766,6 +3874,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>, @@ -3805,6 +3952,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(); @@ -3847,6 +3995,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, @@ -3894,27 +4043,70 @@ 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 }; + let mixed_sources_stopped = was_cancelled && mixed_audio; + if mixed_sources_stopped { + // 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. + // 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 + .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(); + } + 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(_) => { + warn!(drained, "Mixed audio output did not close before the drain deadline; keeping the audio drained so far"); + break; + } + } + .and_then(|frame| clip_to_capture_time(frame, timestamps, cancelled_at)) + } else { + self.audio_rx.try_next().ok().flatten() + }; + match next { + Some(frame) => match process_audio_frame( AudioFrameProcessContext { audio_info: &audio_info, sample_rate, @@ -3922,6 +4114,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, @@ -3965,7 +4158,7 @@ impl PreparedAudioSources { skipped += 1; } }, - Ok(None) | Err(_) => break, + None => break, } } @@ -3984,9 +4177,11 @@ 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 { + if let Some(scope) = &required_scope + && !mixed_sources_stopped + { stop_audio_sources(&mut self.erased_audio_sources, Some(scope)).await; } @@ -4019,7 +4214,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; @@ -4104,7 +4306,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; } @@ -4146,6 +4348,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, @@ -4194,10 +4397,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; @@ -4344,11 +4590,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; @@ -4389,7 +4643,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; @@ -4617,7 +4871,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, @@ -4773,11 +5027,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 { @@ -6024,6 +6278,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) } @@ -6457,6 +6729,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::*; @@ -9197,6 +9491,7 @@ mod tests { struct RecordingAudioMuxer { sent: Arc>>, + pcm: Option>>>, } impl Muxer for RecordingAudioMuxer { @@ -9213,7 +9508,10 @@ mod tests { where Self: Sized, { - Ok(Self { sent: config }) + Ok(Self { + sent: config, + pcm: None, + }) } fn finish(&mut self, _timestamp: Duration) -> anyhow::Result> { @@ -9227,6 +9525,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, @@ -9235,12 +9546,74 @@ 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(); + // 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(); + // 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 { info: AudioInfo, timestamps: Timestamps, master_clock: Arc, muxer: Arc>, sent: Arc>>, + pcm: Arc>>, + capture_time_pause: bool, health_tx: HealthSender, shared_pause: SharedWallClockPause, timestamp_generator: AudioTimestampGenerator, @@ -9261,7 +9634,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))); @@ -9271,6 +9648,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), @@ -9322,7 +9701,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, @@ -9331,6 +9717,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, @@ -9355,6 +9742,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) } @@ -9372,6 +9787,278 @@ 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)); + // 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 + ); + 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 f4d0b8cce9d..498c73aeb7f 100644 --- a/crates/recording/src/sources/audio_mixer.rs +++ b/crates/recording/src/sources/audio_mixer.rs @@ -14,13 +14,17 @@ 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); 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; @@ -38,6 +42,9 @@ struct MixerSource { buffer_last: Option<(Timestamp, Duration)>, last_input_timestamp: Option, clock_state: SourceClockState, + samples_supplied: usize, + late_samples_dropped: u64, + ended: bool, } pub struct AudioMixerBuilder { @@ -101,6 +108,9 @@ impl AudioMixerBuilder { buffer_last: None, last_input_timestamp: None, clock_state: SourceClockState::new(clock_label), + samples_supplied: 0, + late_samples_dropped: 0, + ended: false, }); } @@ -139,6 +149,8 @@ impl AudioMixerBuilder { buffering: MixerBufferingTracker::new(), health_tx: self.health_tx, required_stop: self.required_stop, + drain_deadline: None, + finishing: false, }) } @@ -170,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"); @@ -188,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(()) } } @@ -211,6 +227,8 @@ pub struct AudioMixer { buffering: MixerBufferingTracker, health_tx: Option, required_stop: Option, + drain_deadline: Option, + finishing: bool, } #[derive(Debug, Clone)] @@ -423,6 +441,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() @@ -499,15 +522,21 @@ 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 { + 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() @@ -544,10 +573,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(); @@ -562,50 +592,30 @@ impl AudioMixer { source.last_input_timestamp = None; } - 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); - } + // 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) + }; - frame.set_rate(source.info.rate() as u32); + let timestamp = Timestamp::Instant(clock_anchor + Duration::from_nanos(master_ns)); - 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), - )); + source.buffer_last = Some((timestamp, Duration::from_nanos(frame_duration_ns))); source.buffer.push_back(AudioFrame::new(frame, timestamp)); } } @@ -614,8 +624,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()) @@ -626,60 +641,53 @@ 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); } } - 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); - } + Ok(()) + } - frame.set_rate(source.info.rate() as u32); + fn tick(&mut self, _start: Timestamps, now: Timestamp) -> anyhow::Result { + self.tick_with_output(now, |sink, frame| sink.sink().frame(frame)) + } - 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)); + #[cfg(test)] + pub(crate) fn tick_for_capture_pause(&mut self, now: Instant) -> anyhow::Result { + self.tick(self.timestamps, Timestamp::Instant(now)) + } - if frame_duration.is_zero() { - break; - } + #[cfg(test)] + pub(crate) fn finish_for_capture_pause(&mut self) -> anyhow::Result<()> { + self.finish() + } - remaining = remaining.saturating_sub(frame_duration); - } - } + 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(&mut self, _start: Timestamps, now: Timestamp) -> anyhow::Result { - self.tick_with_output(now, |sink, frame| sink.sink().frame(frame)) - } - fn tick_with_output( &mut self, now: Timestamp, @@ -695,7 +703,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()); @@ -708,12 +720,94 @@ impl AudioMixer { self.force_reset_source(source, starvation_ms); } - for (i, source) in self.sources.iter_mut().enumerate() { - for buffer in source.buffer.drain(..) { - 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"))?; + 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| { + 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 && !self.finishing { + waiting_for_input = true; + } + } + + 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 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 { + 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 = (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); + 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, + start_timestamp + + samples_duration( + source.samples_supplied + 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() || self.finishing { + supply_silence(source, &mut self.abuffers[i], i, limit)?; } } } @@ -750,6 +844,17 @@ impl AudioMixer { match self.output.try_send(frame) { Ok(()) => break, Err(err) if err.is_full() => { + // 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)); } @@ -771,6 +876,99 @@ 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()) + } +} + +/// 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) +} + +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, + 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).div_euclid(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; @@ -942,14 +1140,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}; @@ -965,6 +1155,477 @@ 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]>, + deliveries: Vec<(usize, u64)>, + } + + 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(), + deliveries: 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) { + 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()); + } + 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() + ); + 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) { + 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.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.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]); + } + + #[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_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_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); + 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_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_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); + 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); + 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); @@ -1429,12 +2090,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); } @@ -1510,14 +2171,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); } 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" 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 - ); } }