diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 0a435947794..8a6ba8a4b11 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1649,11 +1649,11 @@ 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(), }; + let is_paused = self.flag.load(Ordering::Acquire); if is_paused { if inner.pause_started_at.is_none() { @@ -2471,7 +2471,7 @@ impl OutputPipelineBuilder> { build_ctx.done_tx, None, &path, - shared_pause, + shared_pause.clone(), true, video_start_gate, start_gate, @@ -2497,6 +2497,7 @@ impl OutputPipelineBuilder> { 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, @@ -2568,7 +2569,7 @@ impl OutputPipelineBuilder { build_ctx.done_tx, Some(first_tx), &path, - shared_pause, + shared_pause.clone(), false, None, start_gate, @@ -2587,6 +2588,7 @@ impl OutputPipelineBuilder { 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()), @@ -4172,6 +4174,7 @@ pub struct OutputPipeline { stop_token: Option, done_fut: DoneFut, pause_flag: Arc, + shared_pause: SharedWallClockPause, cancel_token: CancellationToken, video_frame_count: Arc, video_timestamp_span: Arc, @@ -4322,10 +4325,12 @@ impl OutputPipeline { pub fn pause(&self) { self.pause_flag.store(true, atomic::Ordering::Release); + self.shared_pause.check(); } pub fn resume(&self) { self.pause_flag.store(false, atomic::Ordering::Release); + self.shared_pause.check(); } pub fn video_info(&self) -> Option { @@ -4874,6 +4879,25 @@ mod tests { mod shared_pause_state { use super::*; + #[test] + fn check_waiting_on_resume_does_not_start_a_new_pause() { + let flag = Arc::new(AtomicBool::new(true)); + let pause = SharedWallClockPause::new(flag.clone()); + assert!(pause.check().0); + + let resuming = pause.inner.lock().unwrap(); + let waiting = std::thread::spawn({ + let pause = pause.clone(); + move || pause.check() + }); + std::thread::sleep(Duration::from_millis(50)); + flag.store(false, Ordering::Release); + drop(resuming); + + assert!(!waiting.join().unwrap().0); + assert!(pause.inner.lock().unwrap().pause_started_at.is_none()); + } + fn frame_ts(index: u64) -> Duration { Duration::from_nanos(index * 33_333_333 + 400) } @@ -7197,6 +7221,46 @@ mod tests { (timestamps, capture_duration) } + #[tokio::test] + async fn pause_without_video_frames_is_excluded_on_resume() { + let temp_dir = tempfile::tempdir().unwrap(); + let (sender, receiver) = flume::bounded(4); + let sent = Arc::new(std::sync::Mutex::new(Vec::new())); + let pipeline = OutputPipeline::builder(temp_dir.path().join("paused.mp4")) + .with_video::>(ChannelVideoSourceConfig::new( + VideoInfo::from_raw(cap_media_info::RawVideoFormat::Bgra, 16, 16, 30), + receiver, + )) + .build::(sent.clone()) + .await + .unwrap(); + + for expected_frames in 1..=2 { + sender + .send_async(StaticFrame { + timestamp: Timestamp::Instant(Instant::now()), + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(1), async { + while sent.lock().unwrap().len() < expected_frames { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + if expected_frames == 1 { + pipeline.pause(); + tokio::time::sleep(Duration::from_millis(300)).await; + pipeline.resume(); + } + } + pipeline.stop().await.unwrap(); + + let timestamps = sent.lock().unwrap(); + assert!(timestamps[1] < Duration::from_millis(100), "{timestamps:?}"); + } + #[tokio::test] async fn primed_pipeline_discards_video_captured_before_the_start_gate_arms() { let temp_dir = tempfile::tempdir().unwrap();