From 8a03f9e952c5a4894597f7ae5efa6bda4387e204 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sat, 17 Oct 2026 13:55:00 +0900 Subject: [PATCH 1/5] fix(recording): Windows encoders repeat a stalled frame only after three intervals, behind real time --- crates/recording/src/output_pipeline/core.rs | 6 + .../src/output_pipeline/frame_pacing.rs | 191 +++++++++++++ crates/recording/src/output_pipeline/mod.rs | 2 + .../output_pipeline/oop_fragmented_m4s_win.rs | 65 ++--- crates/recording/src/output_pipeline/win.rs | 268 +++++------------- .../src/output_pipeline/win_fragmented_m4s.rs | 92 ++---- 6 files changed, 329 insertions(+), 295 deletions(-) create mode 100644 crates/recording/src/output_pipeline/frame_pacing.rs diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index e28e3fc97fa..38c3650b7d7 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1991,6 +1991,12 @@ pub struct SharedPauseState { } impl SharedPauseState { + /// The flag that is set while the recording is paused. + #[cfg(windows)] + pub(crate) fn flag(&self) -> Arc { + self.flag.clone() + } + pub fn new(flag: Arc) -> Self { Self { flag, diff --git a/crates/recording/src/output_pipeline/frame_pacing.rs b/crates/recording/src/output_pipeline/frame_pacing.rs new file mode 100644 index 00000000000..1d0de4906fe --- /dev/null +++ b/crates/recording/src/output_pipeline/frame_pacing.rs @@ -0,0 +1,191 @@ +//! Repeats the last frame while a Windows video source is stalled, so the +//! encoder keeps a steady track, without taking the timestamp of a frame that +//! is only late. +//! +//! The Windows encoder loops used to repeat the last frame as soon as one +//! frame interval passed without a new one, stamped one interval after it: +//! the slot a slightly late real frame is stamped for. The real frame then +//! landed on the same timestamp and the muxer moved it by one tick, so a +//! camera with ordinary arrival jitter recorded duplicated frames followed by +//! frames shown for 11 µs (16 of 262 in real-capture run 39000141528). + +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + mpsc::{Receiver, RecvTimeoutError}, + }, + time::{Duration, Instant}, +}; + +/// Frame intervals of silence before the last frame is repeated. A repeat is +/// stamped one interval after the previous frame and sent once a frame for +/// that slot would be two intervals late. +const STALL_INTERVALS: u32 = 3; + +pub(crate) enum Paced { + /// A frame from the source and its timestamp. + Frame(F, Duration), + /// Encode the last frame again at this timestamp. + Repeat(Duration), + End, +} + +pub(crate) struct FramePacer { + interval: Duration, + last: Option, +} + +struct LastFrame { + timestamp: Duration, + arrived: Instant, + repeats: u32, +} + +impl FramePacer { + pub(crate) fn new(frame_rate: u32) -> Self { + Self { + interval: Duration::from_secs_f64(1.0 / f64::from(frame_rate.max(1))), + last: None, + } + } + + /// Blocks until a frame arrives, the source has been silent long enough + /// to repeat the last one, or the stream ends. While `paused`, nothing is + /// repeated and the next frame starts afresh. + pub(crate) fn next( + &mut self, + rx: &Receiver>, + paused: &AtomicBool, + ) -> Paced { + loop { + let received = match &self.last { + None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected), + Some(last) => { + let due = last.arrived + self.interval * (STALL_INTERVALS + last.repeats); + rx.recv_timeout(due.saturating_duration_since(Instant::now())) + } + }; + match received { + Ok(Some((frame, timestamp))) => { + self.last = Some(LastFrame { + timestamp, + arrived: Instant::now(), + repeats: 0, + }); + return Paced::Frame(frame, timestamp); + } + Ok(None) | Err(RecvTimeoutError::Disconnected) => return Paced::End, + Err(RecvTimeoutError::Timeout) => { + if paused.load(Ordering::Acquire) { + self.last = None; + continue; + } + let Some(last) = self.last.as_mut() else { + continue; + }; + last.repeats += 1; + return Paced::Repeat(last.timestamp + self.interval * last.repeats); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc::sync_channel; + + const FPS: u32 = 50; + const INTERVAL: Duration = Duration::from_millis(20); + + fn at(n: u32) -> Duration { + INTERVAL * n + } + + #[test] + fn jittered_frames_pass_through_without_repeats() { + let (tx, rx) = sync_channel(64); + let sender = std::thread::spawn(move || { + for n in 0..20u32 { + // Frames arrive up to 15 ms (three quarters of an interval) late. + std::thread::sleep(if n % 2 == 0 { + INTERVAL + INTERVAL * 3 / 4 + } else { + INTERVAL / 4 + }); + tx.send(Some((n, at(n)))).unwrap(); + } + tx.send(None).unwrap(); + }); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + let mut seen = Vec::new(); + loop { + match pacer.next(&rx, &paused) { + Paced::Frame(n, ts) => seen.push((n, ts)), + Paced::Repeat(ts) => panic!("repeat at {ts:?} for a frame that was only late"), + Paced::End => break, + } + } + sender.join().unwrap(); + assert_eq!(seen, (0..20).map(|n| (n, at(n))).collect::>()); + } + + #[test] + fn a_stall_repeats_the_last_frame_on_its_grid_behind_real_time() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((7u32, at(10)))).unwrap(); + let start = Instant::now(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(7, ts) if ts == at(10))); + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(first, at(11)); + assert!(start.elapsed() >= INTERVAL * STALL_INTERVALS); + let Paced::Repeat(second) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(second, at(12)); + assert!(start.elapsed() >= INTERVAL * (STALL_INTERVALS + 1)); + // The source resumes with a frame captured after the stall. + tx.send(Some((8, at(16)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(8, ts) if ts == at(16))); + } + + #[test] + fn nothing_repeats_while_paused() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((1u32, at(0)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + paused.store(true, Ordering::Release); + let sender = std::thread::spawn(move || { + std::thread::sleep(INTERVAL * 8); + tx.send(Some((2, at(40)))).unwrap(); + tx.send(None).unwrap(); + }); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(2, ts) if ts == at(40))); + assert!(matches!(pacer.next(&rx, &paused), Paced::End)); + sender.join().unwrap(); + } + + #[test] + fn the_stream_ends_on_none_or_disconnect() { + let paused = AtomicBool::new(false); + let (tx, rx) = sync_channel::>(1); + tx.send(None).unwrap(); + assert!(matches!( + FramePacer::new(FPS).next(&rx, &paused), + Paced::End + )); + drop(tx); + assert!(matches!( + FramePacer::new(FPS).next(&rx, &paused), + Paced::End + )); + } +} diff --git a/crates/recording/src/output_pipeline/mod.rs b/crates/recording/src/output_pipeline/mod.rs index e8f069a70d2..d5223a33987 100644 --- a/crates/recording/src/output_pipeline/mod.rs +++ b/crates/recording/src/output_pipeline/mod.rs @@ -1,6 +1,8 @@ mod async_camera; mod core; pub mod ffmpeg; +#[cfg(any(windows, test))] +mod frame_pacing; #[cfg(target_os = "macos")] mod macos_fragmented_m4s; #[cfg(target_os = "macos")] diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs index 4c67f24eb36..d90b19cf7c9 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s_win.rs @@ -2,6 +2,7 @@ use super::core::{ BlockingThreadFinish, DiskSpaceMonitor, HealthSender, PipelineHealthEvent, SharedHealthSender, resolve_oop_thread_finish, wait_for_blocking_thread_finish, }; +use super::frame_pacing::{FramePacer, Paced}; use super::oop_muxer::{ MuxerSubprocessConfig, MuxerSubprocessError, RespawningMuxerSubprocess, VideoStreamInit, resolve_muxer_binary, @@ -258,6 +259,7 @@ impl WindowsOOPFragmentedM4SMuxer { let base_path = self.base_path.clone(); let video_config = self.video_config; + let pause_flag = self.pause.flag(); let segment_duration = self.segment_duration; let preset = self.preset; let bpp = self.bpp; @@ -363,7 +365,7 @@ impl WindowsOOPFragmentedM4SMuxer { return Err(err); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_ffmpeg_frame: Option = None; let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; @@ -460,27 +462,30 @@ impl WindowsOOPFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); - let (ffmpeg_frame, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => match frame.as_ffmpeg() { - Ok(f) => { - last_ffmpeg_frame = Some(reference_video_frame(&f)); - last_timestamp = Some(ts); - (Some(f), ts) - } - Err(e) => { - warn!("Windows OOP failed to convert D3D11 frame: {e:?}"); - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => (None, Duration::ZERO), + let (ffmpeg_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => { + last_timestamp = Some(ts); + match frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(reference_video_frame(&f)); + (Some(f), ts) + } + Err(e) => { + warn!("Windows OOP failed to convert D3D11 frame: {e:?}"); + duplicated_frames += 1; + (last_ffmpeg_frame.as_ref().map(reference_video_frame), ts) } } - }, - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + } + Paced::Repeat(ts) => { + let Some(f) = &last_ffmpeg_frame else { + continue; + }; + last_timestamp = Some(ts); + duplicated_frames += 1; + (Some(reference_video_frame(f)), ts) + } + Paced::End => { let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); loop { @@ -516,17 +521,6 @@ impl WindowsOOPFragmentedM4SMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => continue, - } - } }; let convert_elapsed_ms = convert_start.elapsed().as_millis(); @@ -543,15 +537,6 @@ impl WindowsOOPFragmentedM4SMuxer { } let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - last_timestamp = Some(ts); - } - } - Ok(None) | Err(_) => break, - } continue; }; diff --git a/crates/recording/src/output_pipeline/win.rs b/crates/recording/src/output_pipeline/win.rs index f291851c3b4..7ecb8a0e4d5 100644 --- a/crates/recording/src/output_pipeline/win.rs +++ b/crates/recording/src/output_pipeline/win.rs @@ -1,6 +1,9 @@ use crate::{ AudioFrame, AudioMuxer, Muxer, TaskPool, VideoFrame, VideoMuxer, - output_pipeline::{HealthSender, PipelineHealthEvent, emit_health}, + output_pipeline::{ + HealthSender, PipelineHealthEvent, emit_health, + frame_pacing::{FramePacer, Paced}, + }, screen_capture, }; use anyhow::{Context, anyhow}; @@ -337,70 +340,43 @@ impl Muxer for WindowsMuxer { match encoder { either::Left((mut encoder, mut muxer)) => { trace!("Running native encoder with frame pacing"); - let frame_interval = Duration::from_secs_f64(1.0 / config.frame_rate as f64); + let mut pacer = FramePacer::new(config.frame_rate); let mut last_texture: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count: u64 = 0; let mut frames_reused: u64 = 0; let result = encoder.run( Arc::new(AtomicBool::default()), - || { - loop { - match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_texture = Some(frame.texture().clone()); - last_timestamp = Some(timestamp); - } - Ok(None) => { - trace!("End of stream signal received"); - return Ok(None); - } - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } else if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - frames_reused += 1; - if frames_reused.is_multiple_of(30) { - debug!( - frames_reused = frames_reused, - frame_count = frame_count, - "Frame pacing: reusing frames due to slow capture" - ); - } - } - } - Err(RecvTimeoutError::Disconnected) => { - trace!("Channel disconnected"); - return Ok(None); - } + || loop { + let (texture, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + let texture = frame.texture().clone(); + last_texture = Some(texture.clone()); + (texture, timestamp) } - - if let (Some(texture), Some(ts)) = (&last_texture, last_timestamp) { - let normalized_ts = normalize_timestamp(ts, &mut first_timestamp); - frame_count += 1; - let frame_time = duration_to_timespan(normalized_ts); - return Ok(Some((texture.clone(), frame_time))); - } else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - let texture = frame.texture().clone(); - last_texture = Some(texture.clone()); - last_timestamp = Some(timestamp); - let normalized_ts = - normalize_timestamp(timestamp, &mut first_timestamp); - frame_count = 1; - let frame_time = duration_to_timespan(normalized_ts); - return Ok(Some((texture, frame_time))); - } - Ok(None) | Err(_) => return Ok(None), + Paced::Repeat(timestamp) => { + let Some(texture) = &last_texture else { + continue; + }; + frames_reused += 1; + if frames_reused.is_multiple_of(30) { + debug!( + frames_reused = frames_reused, + frame_count = frame_count, + "Frame pacing: reusing frames due to slow capture" + ); } + (texture.clone(), timestamp) } - } + Paced::End => { + trace!("End of stream signal received"); + return Ok(None); + } + }; + let normalized_ts = normalize_timestamp(timestamp, &mut first_timestamp); + frame_count += 1; + return Ok(Some((texture, duration_to_timespan(normalized_ts)))); }, |output_sample| { let Ok(mut output) = output.lock() else { @@ -438,63 +414,33 @@ impl Muxer for WindowsMuxer { } either::Right(mut encoder) => { trace!("Running FFmpeg encoder with frame pacing"); - let frame_interval = Duration::from_secs_f64(1.0 / config.frame_rate as f64); + let mut pacer = FramePacer::new(config.frame_rate); let mut last_ffmpeg_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; loop { - let (ffmpeg_frame, ts) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_timestamp = Some(timestamp); - match frame.as_ffmpeg() { - Ok(f) => { - last_ffmpeg_frame = Some(reference_video_frame(&f)); - (Some(f), timestamp) - } - Err(e) => { - warn!("Failed to convert frame: {e:?}"); - ( - last_ffmpeg_frame - .as_ref() - .map(reference_video_frame), - timestamp, - ) - } - } - } - Ok(None) => break, - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; + let (ffmpeg_frame, ts) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => match frame.as_ffmpeg() { + Ok(f) => { + last_ffmpeg_frame = Some(reference_video_frame(&f)); + (Some(f), timestamp) } - if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); + Err(e) => { + warn!("Failed to convert frame: {e:?}"); ( - last_ffmpeg_frame - .as_ref() - .map(reference_video_frame), - new_ts, + last_ffmpeg_frame.as_ref().map(reference_video_frame), + timestamp, ) - } else { - continue; } - } - Err(RecvTimeoutError::Disconnected) => break, + }, + Paced::Repeat(timestamp) => ( + last_ffmpeg_frame.as_ref().map(reference_video_frame), + timestamp, + ), + Paced::End => break, }; let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_timestamp = Some(timestamp); - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - } - } - Ok(None) | Err(_) => break, - } continue; }; @@ -927,75 +873,38 @@ impl Muxer for WindowsCameraMuxer { frame_rate ); - let frame_interval = Duration::from_secs_f64(1.0 / frame_rate as f64); + let mut pacer = FramePacer::new(frame_rate); let mut last_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count = 0u64; let mut camera_buffers = CameraBuffers::new(); let result = encoder.run( Arc::new(AtomicBool::default()), - || { - loop { - match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame); - last_timestamp = Some(timestamp); - } - Ok(None) => { - trace!("End of camera stream signal received"); - return Ok(None); - } - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } else if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - } - } - Err(RecvTimeoutError::Disconnected) => { - trace!("Camera channel disconnected"); - return Ok(None); - } + || loop { + let timestamp = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + last_frame = Some(frame); + timestamp } - - if let (Some(frame), Some(ts)) = (&last_frame, last_timestamp) { - let normalized_ts = - normalize_camera_timestamp(ts, &mut first_timestamp); - frame_count += 1; - if frame_count.is_multiple_of(30) { - debug!( - "Windows camera encoder: processed {} frames", - frame_count - ); - } - let texture = - upload_mf_buffer_to_texture(&d3d_device, frame, &mut camera_buffers)?; - return Ok(Some((texture, duration_to_timespan(normalized_ts)))); - } else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - let normalized_ts = normalize_camera_timestamp( - timestamp, - &mut first_timestamp, - ); - frame_count = 1; - let texture = upload_mf_buffer_to_texture( - &d3d_device, - &frame, - &mut camera_buffers, - )?; - return Ok(Some((texture, duration_to_timespan(normalized_ts)))); - } - Ok(None) | Err(_) => return Ok(None), - } + Paced::Repeat(timestamp) => timestamp, + Paced::End => { + trace!("End of camera stream signal received"); + return Ok(None); } + }; + let Some(frame) = &last_frame else { + continue; + }; + let normalized_ts = + normalize_camera_timestamp(timestamp, &mut first_timestamp); + frame_count += 1; + if frame_count.is_multiple_of(30) { + debug!("Windows camera encoder: processed {} frames", frame_count); } + let texture = + upload_mf_buffer_to_texture(&d3d_device, frame, &mut camera_buffers)?; + return Ok(Some((texture, duration_to_timespan(normalized_ts)))); }, |output_sample| { let mut output = match output.lock() { @@ -1047,44 +956,21 @@ impl Muxer for WindowsCameraMuxer { frame_rate ); - let frame_interval = Duration::from_secs_f64(1.0 / frame_rate as f64); + let mut pacer = FramePacer::new(frame_rate); let mut last_frame: Option = None; let mut first_timestamp: Option = None; - let mut last_timestamp: Option = None; let mut frame_count = 0u64; loop { - let (frame_to_encode, ts) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - (Some(frame), timestamp) + let ts = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, timestamp) => { + last_frame = Some(frame); + timestamp } - Ok(None) => break, - Err(RecvTimeoutError::Timeout) => { - if pause_flag.load(Ordering::Acquire) { - last_timestamp = None; - continue; - } - if let Some(last_ts) = last_timestamp { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - (last_frame.clone(), new_ts) - } else { - continue; - } - } - Err(RecvTimeoutError::Disconnected) => break, + Paced::Repeat(timestamp) => timestamp, + Paced::End => break, }; - - let Some(frame) = frame_to_encode else { - match video_rx.recv() { - Ok(Some((frame, timestamp))) => { - last_frame = Some(frame.clone()); - last_timestamp = Some(timestamp); - } - Ok(None) | Err(_) => break, - } + let Some(frame) = last_frame.clone() else { continue; }; diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 303c82ec61a..67278a91340 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -2,6 +2,7 @@ use super::core::{ BlockingThreadFinish, DiskSpaceMonitor, HealthSender, PipelineHealthEvent, SharedHealthSender, combine_finish_errors, wait_for_blocking_thread_finish, }; +use super::frame_pacing::{FramePacer, Paced}; use super::win::reference_video_frame; use crate::{ AudioFrame, AudioMuxer, Muxer, SharedPauseState, TaskPool, VideoMuxer, @@ -359,6 +360,7 @@ impl WindowsFragmentedM4SMuxer { let video_config = self.video_config; let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); + let pause_flag = self.pause.flag(); let encoder_handle = std::thread::Builder::new() .name("win-m4s-segment-encoder".to_string()) .spawn(move || { @@ -368,9 +370,8 @@ impl WindowsFragmentedM4SMuxer { return Err(anyhow!("Failed to send ready signal - receiver dropped")); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_ffmpeg_frame: Option = None; - let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; let mut slow_convert_count = 0u32; @@ -444,27 +445,26 @@ impl WindowsFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); - let (ffmpeg_frame, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => match frame.as_ffmpeg() { + let (ffmpeg_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => match frame.as_ffmpeg() { Ok(f) => { last_ffmpeg_frame = Some(reference_video_frame(&f)); - last_timestamp = Some(ts); (Some(f), ts) } Err(e) => { warn!("Failed to convert D3D11 frame: {e:?}"); - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => (None, Duration::ZERO), - } + duplicated_frames += 1; + (last_ffmpeg_frame.as_ref().map(reference_video_frame), ts) } }, - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + Paced::Repeat(ts) => { + let Some(f) = &last_ffmpeg_frame else { + continue; + }; + duplicated_frames += 1; + (Some(reference_video_frame(f)), ts) + } + Paced::End => { let mut drained = 0u64; let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); @@ -503,31 +503,11 @@ impl WindowsFragmentedM4SMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_ffmpeg_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(reference_video_frame(f)), new_ts) - } - _ => continue, - } - } }; let convert_elapsed_ms = convert_start.elapsed().as_millis(); let Some(ffmpeg_frame) = ffmpeg_frame else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - if let Ok(f) = frame.as_ffmpeg() { - last_ffmpeg_frame = Some(f); - last_timestamp = Some(ts); - } - } - Ok(None) | Err(_) => break, - } continue; }; @@ -828,6 +808,7 @@ impl WindowsFragmentedM4SCameraMuxer { let video_config = self.video_config; let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); + let pause_flag = self.pause.flag(); let encoder_handle = std::thread::Builder::new() .name("win-m4s-camera-segment-encoder".to_string()) .spawn(move || { @@ -839,9 +820,8 @@ impl WindowsFragmentedM4SCameraMuxer { )); } - let frame_interval = Duration::from_secs_f64(1.0 / video_config.fps() as f64); + let mut pacer = FramePacer::new(video_config.fps()); let mut last_frame: Option = None; - let mut last_timestamp: Option = None; let mut first_timestamp: Option = None; let mut slow_convert_count = 0u32; @@ -940,13 +920,19 @@ impl WindowsFragmentedM4SCameraMuxer { break; } - let (frame_to_encode, timestamp) = match video_rx.recv_timeout(frame_interval) { - Ok(Some((frame, ts))) => { + let (camera_frame, timestamp) = match pacer.next(&video_rx, &pause_flag) { + Paced::Frame(frame, ts) => { last_frame = Some(frame.clone()); - last_timestamp = Some(ts); - (Some(frame), ts) + (frame, ts) } - Ok(None) | Err(RecvTimeoutError::Disconnected) => { + Paced::Repeat(ts) => { + let Some(frame) = &last_frame else { + continue; + }; + duplicated_frames += 1; + (frame.clone(), ts) + } + Paced::End => { let mut drained = 0u64; let drain_start = std::time::Instant::now(); let drain_timeout = Duration::from_millis(500); @@ -987,28 +973,6 @@ impl WindowsFragmentedM4SCameraMuxer { } break; } - Err(RecvTimeoutError::Timeout) => { - match (&last_frame, last_timestamp) { - (Some(f), Some(last_ts)) => { - let new_ts = last_ts.saturating_add(frame_interval); - last_timestamp = Some(new_ts); - duplicated_frames += 1; - (Some(f.clone()), new_ts) - } - _ => continue, - } - } - }; - - let Some(camera_frame) = frame_to_encode else { - match video_rx.recv() { - Ok(Some((frame, ts))) => { - last_frame = Some(frame); - last_timestamp = Some(ts); - } - Ok(None) | Err(_) => break, - } - continue; }; let normalized_ts = normalize_timestamp(timestamp, &mut first_timestamp); From 4e199f4c7c03c7bba359f6be57e410f4aaf693a8 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sun, 18 Oct 2026 07:40:32 +0900 Subject: [PATCH 2/5] style(recording): drop imports the pacer replaced --- crates/recording/src/output_pipeline/win.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/recording/src/output_pipeline/win.rs b/crates/recording/src/output_pipeline/win.rs index 7ecb8a0e4d5..cf9003cf2d8 100644 --- a/crates/recording/src/output_pipeline/win.rs +++ b/crates/recording/src/output_pipeline/win.rs @@ -15,8 +15,8 @@ use std::{ path::PathBuf, sync::{ Arc, Mutex, - atomic::{AtomicBool, Ordering}, - mpsc::{RecvTimeoutError, SyncSender, TrySendError, sync_channel}, + atomic::AtomicBool, + mpsc::{SyncSender, TrySendError, sync_channel}, }, time::Duration, }; From 099d1f960368625353f4572662657879a930a68c Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sun, 18 Oct 2026 07:42:24 +0900 Subject: [PATCH 3/5] test(recording): frame pacing jitter test with margins a loaded runner cannot oversleep --- crates/recording/src/output_pipeline/frame_pacing.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/crates/recording/src/output_pipeline/frame_pacing.rs b/crates/recording/src/output_pipeline/frame_pacing.rs index 1d0de4906fe..d82db607965 100644 --- a/crates/recording/src/output_pipeline/frame_pacing.rs +++ b/crates/recording/src/output_pipeline/frame_pacing.rs @@ -96,8 +96,10 @@ mod tests { use super::*; use std::sync::mpsc::sync_channel; - const FPS: u32 = 50; - const INTERVAL: Duration = Duration::from_millis(20); + // Long intervals so a loaded runner's oversleep (tens of ms) never + // reaches the three-interval stall in the jitter test. + const FPS: u32 = 5; + const INTERVAL: Duration = Duration::from_millis(200); fn at(n: u32) -> Duration { INTERVAL * n @@ -107,8 +109,8 @@ mod tests { fn jittered_frames_pass_through_without_repeats() { let (tx, rx) = sync_channel(64); let sender = std::thread::spawn(move || { - for n in 0..20u32 { - // Frames arrive up to 15 ms (three quarters of an interval) late. + for n in 0..10u32 { + // Frames arrive up to three quarters of an interval late. std::thread::sleep(if n % 2 == 0 { INTERVAL + INTERVAL * 3 / 4 } else { @@ -129,7 +131,7 @@ mod tests { } } sender.join().unwrap(); - assert_eq!(seen, (0..20).map(|n| (n, at(n))).collect::>()); + assert_eq!(seen, (0..10).map(|n| (n, at(n))).collect::>()); } #[test] From 99c663d6a4bf4a3bae7f08f70aecf29fd5e2ca85 Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:35:07 +0900 Subject: [PATCH 4/5] fix(recording): Windows frame pacer keeps the last frame through a pause and resumes repeats after it --- crates/recording/src/output_pipeline/core.rs | 1 - .../src/output_pipeline/frame_pacing.rs | 71 ++++++++++++++++--- 2 files changed, 60 insertions(+), 12 deletions(-) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 38c3650b7d7..9045431f630 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -1991,7 +1991,6 @@ pub struct SharedPauseState { } impl SharedPauseState { - /// The flag that is set while the recording is paused. #[cfg(windows)] pub(crate) fn flag(&self) -> Arc { self.flag.clone() diff --git a/crates/recording/src/output_pipeline/frame_pacing.rs b/crates/recording/src/output_pipeline/frame_pacing.rs index d82db607965..67c43605b13 100644 --- a/crates/recording/src/output_pipeline/frame_pacing.rs +++ b/crates/recording/src/output_pipeline/frame_pacing.rs @@ -23,9 +23,7 @@ use std::{ const STALL_INTERVALS: u32 = 3; pub(crate) enum Paced { - /// A frame from the source and its timestamp. Frame(F, Duration), - /// Encode the last frame again at this timestamp. Repeat(Duration), End, } @@ -33,6 +31,7 @@ pub(crate) enum Paced { pub(crate) struct FramePacer { interval: Duration, last: Option, + paused_since: Option, } struct LastFrame { @@ -46,23 +45,44 @@ impl FramePacer { Self { interval: Duration::from_secs_f64(1.0 / f64::from(frame_rate.max(1))), last: None, + paused_since: None, } } /// Blocks until a frame arrives, the source has been silent long enough - /// to repeat the last one, or the stream ends. While `paused`, nothing is - /// repeated and the next frame starts afresh. + /// to repeat the last one, or the stream ends. Nothing repeats while + /// `paused`. Frame timestamps exclude paused time, so on resume the last + /// frame counts as having arrived that much later, and repeats continue + /// on its grid two intervals behind real time: a source that sends + /// nothing until the screen changes still gets the resumed span. Waits + /// are cut to half an interval so a pause is timed to within that. pub(crate) fn next( &mut self, rx: &Receiver>, paused: &AtomicBool, ) -> Paced { + let slice = self.interval / 2; loop { - let received = match &self.last { + let is_paused = paused.load(Ordering::Acquire); + match (is_paused, self.paused_since) { + (true, None) => self.paused_since = Some(Instant::now()), + (false, Some(since)) => { + if let Some(last) = self.last.as_mut() { + last.arrived += since.elapsed(); + } + self.paused_since = None; + } + _ => {} + } + let due = self + .last + .as_ref() + .map(|last| last.arrived + self.interval * (STALL_INTERVALS + last.repeats)); + let received = match due { None => rx.recv().map_err(|_| RecvTimeoutError::Disconnected), - Some(last) => { - let due = last.arrived + self.interval * (STALL_INTERVALS + last.repeats); - rx.recv_timeout(due.saturating_duration_since(Instant::now())) + Some(_) if is_paused => rx.recv_timeout(slice), + Some(due) => { + rx.recv_timeout(due.saturating_duration_since(Instant::now()).min(slice)) } }; match received { @@ -76,8 +96,7 @@ impl FramePacer { } Ok(None) | Err(RecvTimeoutError::Disconnected) => return Paced::End, Err(RecvTimeoutError::Timeout) => { - if paused.load(Ordering::Acquire) { - self.last = None; + if is_paused || due.is_some_and(|due| Instant::now() < due) { continue; } let Some(last) = self.last.as_mut() else { @@ -94,7 +113,7 @@ impl FramePacer { #[cfg(test)] mod tests { use super::*; - use std::sync::mpsc::sync_channel; + use std::sync::{Arc, mpsc::sync_channel}; // Long intervals so a loaded runner's oversleep (tens of ms) never // reaches the three-interval stall in the jitter test. @@ -175,6 +194,36 @@ mod tests { sender.join().unwrap(); } + #[test] + fn repeats_resume_after_a_pause_on_the_last_frames_grid() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = Arc::new(AtomicBool::new(false)); + tx.send(Some((1u32, at(5)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + let start = Instant::now(); + paused.store(true, Ordering::Release); + let resumer = { + let paused = paused.clone(); + std::thread::spawn(move || { + std::thread::sleep(INTERVAL * 6); + paused.store(false, Ordering::Release); + }) + }; + // A static screen: nothing arrives after the resume either. + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat after the resume"); + }; + assert_eq!(first, at(6)); + assert!(start.elapsed() >= INTERVAL * (6 + STALL_INTERVALS)); + let Paced::Repeat(second) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(second, at(7)); + resumer.join().unwrap(); + drop(tx); + } + #[test] fn the_stream_ends_on_none_or_disconnect() { let paused = AtomicBool::new(false); From 232e4d29e8c9df6a49f5a1370d5d6cfc39b3a17d Mon Sep 17 00:00:00 2001 From: Richie McIlroy <33632126+richiemcilroy@users.noreply.github.com> Date: Sun, 18 Oct 2026 09:49:12 +0900 Subject: [PATCH 5/5] fix(recording): a frame that arrives during a pause ends the pacer's pause adjustment --- .../src/output_pipeline/frame_pacing.rs | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/crates/recording/src/output_pipeline/frame_pacing.rs b/crates/recording/src/output_pipeline/frame_pacing.rs index 67c43605b13..3dd9719bf85 100644 --- a/crates/recording/src/output_pipeline/frame_pacing.rs +++ b/crates/recording/src/output_pipeline/frame_pacing.rs @@ -87,6 +87,9 @@ impl FramePacer { }; match received { Ok(Some((frame, timestamp))) => { + // The pause so far happened before this frame; if it + // continues, the next call times it from now. + self.paused_since = None; self.last = Some(LastFrame { timestamp, arrived: Instant::now(), @@ -224,6 +227,28 @@ mod tests { drop(tx); } + #[test] + fn a_frame_that_arrives_while_paused_ends_the_pause_adjustment() { + let (tx, rx) = sync_channel(4); + let mut pacer = FramePacer::new(FPS); + let paused = AtomicBool::new(false); + tx.send(Some((1u32, at(0)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(1, _))); + paused.store(true, Ordering::Release); + tx.send(Some((2, at(1)))).unwrap(); + assert!(matches!(pacer.next(&rx, &paused), Paced::Frame(2, _))); + let arrived = Instant::now(); + paused.store(false, Ordering::Release); + std::thread::sleep(INTERVAL * 6); + // Frame 2 has been up for six intervals: its first repeat is overdue. + let Paced::Repeat(first) = pacer.next(&rx, &paused) else { + panic!("expected a repeat"); + }; + assert_eq!(first, at(2)); + assert!(arrived.elapsed() < INTERVAL * 8); + drop(tx); + } + #[test] fn the_stream_ends_on_none_or_disconnect() { let paused = AtomicBool::new(false);