fix(recording): exclude pauses that arrive with no media

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<TVideo: VideoSource> OutputPipelineBuilder<HasVideo<TVideo>> {
             build_ctx.done_tx,
             None,
             &path,
-            shared_pause,
+            shared_pause.clone(),
             true,
             video_start_gate,
             start_gate,
@@ -2497,6 +2497,7 @@ impl<TVideo: VideoSource> OutputPipelineBuilder<HasVideo<TVideo>> {
             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<NoVideo> {
             build_ctx.done_tx,
             Some(first_tx),
             &path,
-            shared_pause,
+            shared_pause.clone(),
             false,
             None,
             start_gate,
@@ -2587,6 +2588,7 @@ impl OutputPipelineBuilder<NoVideo> {
             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<DropGuard>,
     done_fut: DoneFut,
     pause_flag: Arc<AtomicBool>,
+    shared_pause: SharedWallClockPause,
     cancel_token: CancellationToken,
     video_frame_count: Arc<AtomicU64>,
     video_timestamp_span: Arc<VideoTimestampSpan>,
@@ -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<VideoInfo> {
@@ -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::<ChannelVideoSource<StaticFrame>>(ChannelVideoSourceConfig::new(
+                    VideoInfo::from_raw(cap_media_info::RawVideoFormat::Bgra, 16, 16, 30),
+                    receiver,
+                ))
+                .build::<ObservedMuxer>(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();