diff --git a/.github/workflows/sync-tests.yml b/.github/workflows/sync-tests.yml index e7e71598a42..2e7cd3cb247 100644 --- a/.github/workflows/sync-tests.yml +++ b/.github/workflows/sync-tests.yml @@ -115,7 +115,27 @@ jobs: run: | cargo test --locked -p cap-timestamp -p cap-enc-ffmpeg cargo test --locked -p cap-recording --lib - cargo test --locked -p cap-rendering + # --nocapture so a WARP-adapter notch skip prints instead of + # looking identical to a pass in the CI log. + cargo test --locked -p cap-rendering -- --nocapture + + # Real encoders + DASH muxer + remux/validation over full instant-mode + # scenarios: pause/resume excision, stall-recovery bursts with + # same-microsecond timestamps, segment assembly and A/V alignment. + - name: Instant mode scenario harness + shell: bash + run: | + cargo test --locked -p cap-recording --test instant_mode_scenarios + + # The hardware harnesses (real screen/mic recordings) only run on + # developer machines, but they must keep compiling: nothing else + # builds the full test target set, and hardware_instant_recording + # once rotted into breaking every non-macOS test build via ungated + # macOS-only imports. + - name: Compile recording test harnesses + shell: bash + run: | + cargo check --locked -p cap-recording --tests # The AVFoundation encoder (studio camera/display on macOS) has its own # pts handling; its fps-matrix duration tests guard against re-timing @@ -185,7 +205,12 @@ jobs: print("| Case | Result | Detail |") print("| --- | --- | --- |") for case in report.get("cases", []): - verdict = "PASS" if case["pass"] else "FAIL" + if not case["pass"]: + verdict = "FAIL" + elif "skipped:" in case["detail"]: + verdict = "SKIP" + else: + verdict = "PASS" detail = case["detail"].replace("|", "\\|") print(f"| {case['name']} | {verdict} | {detail} |") PYEOF diff --git a/apps/desktop/src-tauri/src/telemetry.rs b/apps/desktop/src-tauri/src/telemetry.rs index 7d1139361bd..10d1d908667 100644 --- a/apps/desktop/src-tauri/src/telemetry.rs +++ b/apps/desktop/src-tauri/src/telemetry.rs @@ -118,7 +118,13 @@ pub enum AnalyticsEvent { fn truncate_reason(mut s: String) -> String { const MAX_LEN: usize = 240; if s.len() > MAX_LEN { - s.truncate(MAX_LEN); + // Reasons carry NSError debug strings whose localized text is + // multi-byte; String::truncate panics off a char boundary. + let end = (0..=MAX_LEN) + .rev() + .find(|&i| s.is_char_boundary(i)) + .unwrap_or(0); + s.truncate(end); s.push('…'); } s diff --git a/crates/enc-avfoundation/src/mp4.rs b/crates/enc-avfoundation/src/mp4.rs index bc3cea69af4..8e6d742ec43 100644 --- a/crates/enc-avfoundation/src/mp4.rs +++ b/crates/enc-avfoundation/src/mp4.rs @@ -80,7 +80,11 @@ pub enum QueueFrameError { AppendError(arc::R), #[error("Failed")] Failed, - #[error("WriterFailed/{0}")] + // Debug-format the NSError: Display is only the localized description + // ("The operation could not be completed"), which hides the code and the + // NSUnderlyingError (e.g. -11800/-16364 InvalidTimestamp) needed to + // diagnose field reports. + #[error("WriterFailed/{0:?}")] WriterFailed(arc::R), #[error("Finished")] Finished, @@ -509,6 +513,16 @@ impl MP4Encoder { { self.timestamp_offset += gap; self.pause_timestamp = None; + // A frame held across the pause may carry a deferred offset + // snapshotted before the gap existed; applying it verbatim on + // append would overwrite the gap-adjusted offset and re-insert + // the pause into every later video and audio timestamp. Shift it + // by the gap so apply-on-append stays correct. + if let Some(pending) = self.pending_video_frame.as_mut() + && let Some(deferred) = pending.deferred_offset + { + pending.deferred_offset = Some(deferred + gap); + } } if !self.instant_mode @@ -525,9 +539,21 @@ impl MP4Encoder { } } - let mut pts_duration = timestamp - .checked_sub(self.timestamp_offset) - .unwrap_or(Duration::ZERO); + // The writer only sees whole microseconds (write_pending_frame builds + // SampleTimingInfo on a 1MHz timescale via as_micros), while remapped + // capture timestamps carry nanosecond precision. During stall-recovery + // bursts two frames can land inside the same microsecond: they pass a + // nanosecond-space monotonicity check but collapse into duplicate + // writer PTS, which AVAssetWriter reports asynchronously a few frames + // later as -11800/-16364 (InvalidTimestamp), killing the recording. + // Truncate first so the tie correction below operates in the same + // units the writer sees. + let mut pts_duration = Duration::from_micros( + timestamp + .checked_sub(self.timestamp_offset) + .unwrap_or(Duration::ZERO) + .as_micros() as u64, + ); let mut deferred_offset: Option = None; @@ -606,6 +632,13 @@ impl MP4Encoder { { self.timestamp_offset += gap; self.pause_timestamp = None; + // Same as the video path: keep a held frame's deferred offset in + // step with the consumed pause gap. + if let Some(pending) = self.pending_video_frame.as_mut() + && let Some(deferred) = pending.deferred_offset + { + pending.deferred_offset = Some(deferred + gap); + } } if !self.session_started { @@ -770,6 +803,7 @@ impl MP4Encoder { Ok(()) => {} Err(QueueFrameError::WriterFailed(err)) => { error!( + error = ?err, video_frames = self.video_frames_appended, audio_frames = self.audio_frames_appended, audio_pts_value = pts_value, @@ -831,6 +865,7 @@ impl MP4Encoder { } Err(QueueFrameError::WriterFailed(err)) => { error!( + error = ?err, video_frames = self.video_frames_appended, audio_frames = self.audio_frames_appended, pts_us = pending.pts.as_micros() as i64, @@ -880,8 +915,23 @@ impl MP4Encoder { return; }; - self.flush_pending_video(); - + // Deliberately keep the pending frame instead of flushing it here. + // Flushing wrote it with the full nominal duration, and the first + // post-resume frame ties against its pts and gets bumped +1us — + // landing inside the flushed sample's extent. Overlapping extents are + // the sporadic AVAssetWriter failure shape reproduced in the + // overlapping-extents tests. Held until resume, the pending frame is + // written with the real (clamped) forward gap and extents stay + // disjoint. The trade: when the last pre-pause sample was video, the + // first post-resume frame maps to exactly the held frame's pts, ties, + // and bumps +1us — the final pre-pause frame keeps a 1us extent and + // is effectively never displayed. Total timeline length is preserved + // (the 1us comes out of the resume frame's slot), which the + // container-duration assertions pin. + // finish_start still flushes it with nominal duration when the + // recording stops while paused. Holding it retains one capture-pool + // pixel buffer for the pause duration; upstream drops paused frames + // before they reach us, so the pool never contends on it. self.pause_timestamp = Some(timestamp); self.is_paused = true; } @@ -1283,7 +1333,9 @@ mod tests { } fn test_output_path(name: &str) -> PathBuf { - let path = std::env::temp_dir().join(format!("cap_test_{name}.mp4")); + // Namespaced per process: two concurrent runs of this binary sharing + // a fixed path fail AVAssetWriter init with "Cannot Save" mid-suite. + let path = std::env::temp_dir().join(format!("cap_test_{name}_{}.mp4", std::process::id())); let _ = std::fs::remove_file(&path); path } @@ -1325,6 +1377,27 @@ mod tests { create_pixel_buffer_pool_with_format(width, height, cidre::cv::PixelFormat::_420V) } + // Mirrors the production encoder-thread retry loop: paravirtualized CI + // runners have no hardware VideoToolbox, so the writer input reports + // NotReadyForMore often enough that single-shot queue calls drop frames + // and count-based assertions flake. + fn queue_video_frame_with_retry( + encoder: &mut MP4Encoder, + frame: arc::R, + timestamp: Duration, + ) -> Result { + for _ in 0..1000 { + match encoder.queue_video_frame(frame.clone(), timestamp) { + Ok(()) => return Ok(true), + Err(QueueFrameError::NotReadyForMore) => { + std::thread::sleep(Duration::from_micros(200)); + } + Err(e) => return Err(e), + } + } + Ok(false) + } + fn create_test_video_frame( pool: &cidre::cv::PixelBufPool, pts_us: i64, @@ -3683,6 +3756,304 @@ mod tests { let _ = std::fs::remove_file(&output); } + #[test] + fn regression_same_microsecond_pts_pair_is_bumped_apart() { + let output = test_output_path("same_us_pts_bump"); + let video = valid_video_config(); + + let mut encoder = MP4Encoder::init(output.clone(), video, None, None).unwrap(); + let pool = create_pixel_buffer_pool(1920, 1080); + + let base = Duration::from_micros(33_333); + let first = base + Duration::from_nanos(200); + let second = base + Duration::from_nanos(800); + + let frame_a = create_test_video_frame(&pool, 33_333, 33_333); + let frame_b = create_test_video_frame(&pool, 33_333, 33_333); + assert!(queue_video_frame_with_retry(&mut encoder, frame_a, first).unwrap()); + assert!(queue_video_frame_with_retry(&mut encoder, frame_b, second).unwrap()); + + assert_eq!( + encoder.last_video_pts, + Some(Duration::from_micros(33_333)), + "first frame must be written at the truncated microsecond" + ); + assert_eq!( + encoder.pending_video_frame.as_ref().map(|p| p.pts), + Some(Duration::from_micros(33_334)), + "second frame in the same microsecond must be bumped one whole microsecond" + ); + + let _ = encoder.finish(Some(Duration::from_micros(66_666))); + let _ = std::fs::remove_file(&output); + } + + #[test] + fn regression_same_microsecond_pts_bursts_survive_writer() { + // Field failure from the 0.5.8 reports (studio + camera on macOS): + // remapped capture timestamps carry nanosecond precision, and a + // stall-recovery burst can put two frames inside the same + // microsecond. The writer quantizes PTS to whole microseconds, so + // without entry quantization the pair reaches AVAssetWriter as + // duplicate timestamps and the writer dies asynchronously with + // -11800/-16364 (InvalidTimestamp) a few frames later. + let output = test_output_path("same_us_pts_bursts"); + let video = valid_video_config(); + + let mut encoder = MP4Encoder::init(output.clone(), video, None, None).unwrap(); + let pool = create_pixel_buffer_pool(1920, 1080); + + let mut errors = Vec::new(); + let mut appended = 0u64; + let mut attempted = 0u64; + + 'frames: for i in 0..360u64 { + let base_us = i * 33_333; + let mut timestamps = vec![Duration::from_micros(base_us) + Duration::from_nanos(200)]; + if i % 30 == 10 { + timestamps.push(Duration::from_micros(base_us) + Duration::from_nanos(800)); + } + + for ts in timestamps { + let frame = create_test_video_frame(&pool, base_us as i64, 33_333); + attempted += 1; + match queue_video_frame_with_retry(&mut encoder, frame, ts) { + Ok(true) => appended += 1, + Ok(false) => {} + Err(e) => { + errors.push(format!("{e:?} at frame {i}")); + break 'frames; + } + } + } + } + + assert!( + errors.is_empty(), + "Same-microsecond PTS bursts must not fail the writer: {errors:?}" + ); + assert!( + appended >= attempted * 9 / 10, + "expected most frames to queue, got {appended}/{attempted}" + ); + + let finish = encoder.finish(Some(Duration::from_secs(13))); + assert!(finish.is_ok(), "Finish failed: {finish:?}"); + + let _ = std::fs::remove_file(&output); + } + + #[test] + fn regression_pause_resume_keeps_sample_extents_disjoint() { + let output = test_output_path("pause_resume_extents"); + let video = valid_video_config(); + + let mut encoder = MP4Encoder::init(output.clone(), video, None, None).unwrap(); + let pool = create_pixel_buffer_pool(1920, 1080); + + let mut errors = Vec::new(); + let mut queue = |encoder: &mut MP4Encoder, ts: Duration, label: &str| { + let frame = create_test_video_frame(&pool, ts.as_micros() as i64, 33_333); + match queue_video_frame_with_retry(encoder, frame, ts) { + Ok(_) => {} + Err(e) => errors.push(format!("{e:?} at {label}")), + } + }; + + for i in 0..60u64 { + queue( + &mut encoder, + Duration::from_micros(i * 33_333) + Duration::from_nanos(400), + "pre-pause", + ); + } + + let pre_pause_last = Duration::from_micros(59 * 33_333) + Duration::from_nanos(400); + encoder.pause(); + assert!( + encoder.pending_video_frame.is_some(), + "pause must hold the pending frame instead of flushing it with nominal duration" + ); + encoder.resume(); + + // Upstream excises the pause from the timeline, so the first resumed + // frame can tie the last pre-pause frame within the same microsecond. + queue( + &mut encoder, + pre_pause_last + Duration::from_nanos(200), + "resume-tie", + ); + + for i in 61..120u64 { + queue( + &mut encoder, + Duration::from_micros(i * 33_333) + Duration::from_nanos(400), + "post-resume", + ); + } + + assert!( + errors.is_empty(), + "pause/resume with a same-microsecond resume tie must not fail the writer: {errors:?}" + ); + + let finish = encoder.finish(Some(Duration::from_micros(120 * 33_333))); + assert!(finish.is_ok(), "Finish failed: {finish:?}"); + + let duration = container_duration_secs(&output); + assert!( + (3.8..=4.2).contains(&duration), + "held-frame pause must not distort the muxed timeline: 120 frames at 30fps \ + should span ~4.0s, container reports {duration:.3}s" + ); + + let _ = std::fs::remove_file(&output); + } + + #[test] + fn regression_stop_while_paused_flushes_pending_frame() { + let output = test_output_path("stop_while_paused"); + let video = valid_video_config(); + + let mut encoder = MP4Encoder::init(output.clone(), video, None, None).unwrap(); + let pool = create_pixel_buffer_pool(1920, 1080); + + for i in 0..30u64 { + let ts = Duration::from_micros(i * 33_333); + let frame = create_test_video_frame(&pool, (i * 33_333) as i64, 33_333); + let queued = queue_video_frame_with_retry(&mut encoder, frame, ts).unwrap(); + assert!(queued, "frame {i} exhausted the writer-ready retry budget"); + } + + encoder.pause(); + assert!(encoder.pending_video_frame.is_some()); + let appended_before_finish = encoder.video_frames_appended; + assert_eq!( + appended_before_finish, 29, + "all queued frames but the held one should be appended before finish" + ); + + // The finish-time flush appends without a readiness check; wait for + // the input to drain so the held frame cannot be dropped on a slow + // runner. + for _ in 0..1000 { + if encoder.video_input.is_ready_for_more_media_data() { + break; + } + std::thread::sleep(Duration::from_micros(200)); + } + + let finish = encoder.finish(Some(Duration::from_secs(1))); + assert!( + finish.is_ok(), + "stopping while paused must flush the held frame and finalize: {finish:?}" + ); + assert!(encoder.pending_video_frame.is_none()); + assert_eq!( + encoder.video_frames_appended, + appended_before_finish + 1, + "the held frame must reach the writer during finish" + ); + + let _ = std::fs::remove_file(&output); + } + + #[test] + fn regression_deferred_offset_shifts_with_pause_gap_across_hold() { + // Covers the deferred-offset gap shift: a frame that was tie-bumped + // (so it carries deferred_offset = Some) held across a SECOND pause + // must have that snapshot shifted by the consumed gap. Applied + // verbatim on append, the stale snapshot overwrites the gap-adjusted + // timestamp_offset and every later timestamp jumps forward by the + // pause length. + let output = test_output_path("deferred_offset_pause_gap"); + let video = valid_video_config(); + + let mut encoder = MP4Encoder::init(output.clone(), video, None, None).unwrap(); + let pool = create_pixel_buffer_pool(1920, 1080); + + let step = Duration::from_micros(33_333); + let mut errors: Vec = Vec::new(); + let mut queue = |encoder: &mut MP4Encoder, ts: Duration, label: &str| { + let frame = create_test_video_frame(&pool, ts.as_micros() as i64, 33_333); + match queue_video_frame_with_retry(encoder, frame, ts) { + Ok(_) => {} + Err(e) => errors.push(format!("{e:?} at {label}")), + } + }; + + for i in 0..30u64 { + queue( + &mut encoder, + step * i as u32 + Duration::from_nanos(400), + "pre", + ); + } + let t29 = step * 29 + Duration::from_nanos(400); + + encoder.pause(); + encoder.resume(); + + // Pause excision maps this frame back onto t29's microsecond: it + // tie-bumps and becomes the held frame with deferred_offset = Some. + let gap1 = Duration::from_secs(2) + Duration::from_nanos(100); + let resume1 = t29 + gap1; + queue(&mut encoder, resume1, "resume-tie"); + assert!( + encoder + .pending_video_frame + .as_ref() + .is_some_and(|p| p.deferred_offset.is_some()), + "the tie-bumped resume frame must carry a deferred offset for this test to bite" + ); + + // Second pause with the deferred-carrying frame still held. + encoder.pause(); + encoder.resume(); + + let gap2 = Duration::from_secs(1) + step; + let resume2 = resume1 + gap2; + queue(&mut encoder, resume2, "second-resume"); + + for k in 1..=10u64 { + queue(&mut encoder, resume2 + step * k as u32, "post"); + } + + assert!(errors.is_empty(), "no queue call may fail: {errors:?}"); + + // Both pause gaps must be excised from the mapping. A stale deferred + // snapshot (missing gap2) would leave timestamp_offset ~1s short and + // push every later pts forward by that much. + let expected_offset = gap1 + gap2; + let offset_error = encoder.timestamp_offset.abs_diff(expected_offset); + assert!( + offset_error < Duration::from_micros(5), + "timestamp_offset must track both consumed gaps: expected ~{expected_offset:?}, \ + got {:?}", + encoder.timestamp_offset + ); + + let last_pts = encoder.last_video_pts.expect("frames were written"); + assert!( + last_pts < step * 41, + "written pts must continue at frame cadence after the held-frame pauses, \ + got {last_pts:?} (a value ~1s larger means the stale deferred offset \ + re-inserted the second pause)" + ); + + let finish = encoder.finish(Some(resume2 + step * 11)); + assert!(finish.is_ok(), "Finish failed: {finish:?}"); + + let duration = container_duration_secs(&output); + assert!( + (1.2..=1.7).contains(&duration), + "42 frames at 30fps must span ~1.4s regardless of pauses, container reports \ + {duration:.3}s" + ); + + let _ = std::fs::remove_file(&output); + } + #[test] fn regression_wired_mic_timestamp_gap_is_preserved() { let output = test_output_path("wired_mic_timestamp_gap"); diff --git a/crates/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index a176d2c689a..2937c419115 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -1344,4 +1344,172 @@ mod tests { assert!(crate::remux::probe_video_can_decode(&output_path).unwrap_or(false)); } + + #[test] + fn stall_recovery_burst_with_same_microsecond_timestamps_survives() { + // Replays the 0.5.8 field-failure timeline shape end to end: normal + // cadence with nanosecond-fraction timestamps, a multi-second system + // stall, then a recovery burst of backlogged frames landing hundreds + // of nanoseconds apart (same microsecond, same 90kHz tick), plus an + // exact duplicate and a backwards blip. The instant-mode encoder must + // accept every frame, keep encoded PTS strictly monotonic, and the + // production remux + decode of the segments must succeed. + ffmpeg::init().ok(); + + let temp = tempfile::tempdir().unwrap(); + let base_path = temp.path().to_path_buf(); + + let mut encoder = SegmentedVideoEncoder::init( + base_path.clone(), + test_video_info(), + SegmentedVideoEncoderConfig { + segment_duration: Duration::from_millis(500), + ..Default::default() + }, + ) + .unwrap(); + + let frame_ns = 33_333_333u64; + let mut timestamps: Vec = Vec::new(); + for i in 0..60u64 { + timestamps.push(Duration::from_nanos(i * frame_ns + 400)); + } + let stall_end = 60 * frame_ns + 2_000_000_000; + for i in 0..12u64 { + timestamps.push(Duration::from_nanos(stall_end + i * 300)); + } + timestamps.push(Duration::from_nanos(stall_end + 11 * 300)); + timestamps.push(Duration::from_nanos(stall_end.saturating_sub(5_000_000))); + for i in 1..=60u64 { + timestamps.push(Duration::from_nanos(stall_end + i * frame_ns)); + } + + for (i, &ts) in timestamps.iter().enumerate() { + let frame = create_test_frame(320, 240); + encoder + .queue_frame(frame, ts) + .unwrap_or_else(|e| panic!("frame {i} at {ts:?} rejected: {e}")); + } + + encoder.finish().unwrap(); + + let mut segment_paths: Vec = std::fs::read_dir(&base_path) + .unwrap() + .filter_map(|e| e.ok().map(|e| e.path())) + .filter(|p| p.extension().is_some_and(|ext| ext == "m4s")) + .collect(); + segment_paths.sort(); + assert!( + segment_paths.len() >= 3, + "expected multiple media segments, got {segment_paths:?}" + ); + + let concat_path = base_path.join("concat_test.mp4"); + let mut concatenated = std::fs::read(base_path.join(INIT_SEGMENT_NAME)).unwrap(); + for segment in &segment_paths { + concatenated.extend(std::fs::read(segment).unwrap()); + } + std::fs::write(&concat_path, concatenated).unwrap(); + + let mut input = format::input(&concat_path).unwrap(); + let stream_index = input + .streams() + .best(ffmpeg::media::Type::Video) + .unwrap() + .index(); + + let mut pts_ticks: Vec = input + .packets() + .filter_map(|(stream, packet)| { + (stream.index() == stream_index) + .then_some(packet.pts()) + .flatten() + }) + .collect(); + pts_ticks.sort_unstable(); + + assert_eq!( + pts_ticks.len(), + timestamps.len(), + "every queued frame must be encoded (ties bumped, never dropped)" + ); + for pair in pts_ticks.windows(2) { + assert!( + pair[1] > pair[0], + "encoded pts must be strictly monotonic, found {} then {} (duplicate PTS is the \ + -16364 failure class)", + pair[0], + pair[1] + ); + } + + let remuxed_path = temp.path().join("stall-burst-output.mp4"); + crate::remux::concatenate_m4s_segments_with_init( + &base_path.join(INIT_SEGMENT_NAME), + &segment_paths, + &remuxed_path, + ) + .unwrap(); + assert!(crate::remux::probe_video_can_decode(&remuxed_path).unwrap_or(false)); + } + + #[test] + fn default_config_cuts_segments_from_encoder_keyframe_cadence() { + // Production always runs segment_duration == the encoder GOP + // (DEFAULT_KEYFRAME_INTERVAL_SECS), so segment cuts depend on the + // encoder emitting keyframes at its configured cadence — no caller + // forces I-frames. Pin that contract: if the GOP options regress + // (g/keyint_min or the default interval), segments stop cutting and + // this fails. Test helpers that force I-frames at shorter cadences + // cannot catch that. + ffmpeg::init().ok(); + + let temp = tempfile::tempdir().unwrap(); + let base_path = temp.path().to_path_buf(); + + let mut encoder = SegmentedVideoEncoder::init( + base_path.clone(), + test_video_info(), + SegmentedVideoEncoderConfig::default(), + ) + .unwrap(); + + // 6.6s at 30fps with untouched frame kinds. + for i in 0..200u64 { + let frame = create_test_frame(320, 240); + encoder + .queue_frame(frame, Duration::from_nanos(i * 33_333_333)) + .unwrap(); + } + encoder.finish().unwrap(); + + let mut segment_paths: Vec = std::fs::read_dir(&base_path) + .unwrap() + .filter_map(|e| e.ok().map(|e| e.path())) + .filter(|p| p.extension().is_some_and(|ext| ext == "m4s")) + .collect(); + segment_paths.sort(); + assert!( + (2..=5).contains(&segment_paths.len()), + "6.6s at the default 2s segment/GOP cadence must cut ~3 media segments \ + from the encoder's own keyframes, got {}: {segment_paths:?}", + segment_paths.len() + ); + + let remuxed_path = temp.path().join("default-cadence-output.mp4"); + crate::remux::concatenate_m4s_segments_with_init( + &base_path.join(INIT_SEGMENT_NAME), + &segment_paths, + &remuxed_path, + ) + .unwrap(); + let duration = crate::remux::get_media_duration(&remuxed_path) + .expect("assembled duration readable") + .as_secs_f64(); + assert!( + (6.0..=7.2).contains(&duration), + "assembled output must carry the full 6.6s of content, got {duration:.2}s" + ); + assert!(crate::remux::probe_video_can_decode(&remuxed_path).unwrap_or(false)); + } } diff --git a/crates/enc-ffmpeg/src/remux.rs b/crates/enc-ffmpeg/src/remux.rs index d9554ba8913..bf17ad6e2e0 100644 --- a/crates/enc-ffmpeg/src/remux.rs +++ b/crates/enc-ffmpeg/src/remux.rs @@ -1048,7 +1048,7 @@ mod tests { // cadence; deterministic pseudo-jitter stands in for QPC noise. let timestamps: Vec = (0..120) .map(|i| { - let jitter_us = ((i * 7919) % 7000) as u64; // 0..7ms + let jitter_us = (i * 7919) % 7000; // 0..7ms Duration::from_nanos(i * 1_000_000_000 / 30 + jitter_us * 1_000) }) .collect(); diff --git a/crates/mediafoundation-ffmpeg/src/h264.rs b/crates/mediafoundation-ffmpeg/src/h264.rs index 55e8a5c0afc..f1077860e9d 100644 --- a/crates/mediafoundation-ffmpeg/src/h264.rs +++ b/crates/mediafoundation-ffmpeg/src/h264.rs @@ -32,6 +32,8 @@ pub struct H264StreamMuxer { time_base: ffmpeg::Rational, is_finished: bool, frame_count: u64, + last_written_pts: Option, + consecutive_pts_bumps: u64, } impl H264StreamMuxer { @@ -83,6 +85,8 @@ impl H264StreamMuxer { time_base, is_finished: false, frame_count: 0, + last_written_pts: None, + consecutive_pts_bumps: 0, }) } @@ -102,6 +106,43 @@ impl H264StreamMuxer { output.stream(self.stream_index).unwrap().time_base(), ); + // MediaFoundation stamps samples in 100ns ticks, but this stream's + // time base is ~333x coarser (1/(fps*1000)): two strictly increasing + // sample times can land on the same output tick, and the mov muxer + // rejects the duplicate pts/dts — the same unit-mismatch class as the + // AVFoundation -16364 failures. A tie carries no time: advance one + // tick in the writer-visible unit and let real timestamps take over, + // like normalize_input_pts in cap-enc-ffmpeg. pts==dts here (MF + // encoders are configured without B-frames). + if let Some(pts) = packet.pts() { + let pts = match self.last_written_pts { + Some(last) if pts <= last => { + // Bumps are expected in short runs (re-quantization + // ties); a long run means the source clock is stuck and + // the muxed timeline is compressing, which must be + // visible in logs rather than silent. + self.consecutive_pts_bumps += 1; + if self.consecutive_pts_bumps == 30 + || self.consecutive_pts_bumps.is_multiple_of(300) + { + warn!( + consecutive_bumps = self.consecutive_pts_bumps, + "MF sample times are not advancing; muxer is tie-bumping \ + every frame (stuck source clock?)" + ); + } + last + 1 + } + _ => { + self.consecutive_pts_bumps = 0; + pts + } + }; + self.last_written_pts = Some(pts); + packet.set_pts(Some(pts)); + packet.set_dts(Some(pts)); + } + packet.write_interleaved(output)?; Ok(()) diff --git a/crates/recording/src/output_pipeline/core.rs b/crates/recording/src/output_pipeline/core.rs index 902e2226cd1..91355fbf657 100644 --- a/crates/recording/src/output_pipeline/core.rs +++ b/crates/recording/src/output_pipeline/core.rs @@ -3424,6 +3424,127 @@ pub trait VideoMuxer: Muxer { mod tests { use super::*; + mod shared_pause_state { + use super::*; + + fn frame_ts(index: u64) -> Duration { + Duration::from_nanos(index * 33_333_333 + 400) + } + + #[test] + fn pause_resume_produces_strictly_forward_timeline() { + let flag = Arc::new(AtomicBool::new(false)); + let pause = SharedPauseState::new(flag.clone()); + + let mut sent: Vec = Vec::new(); + for i in 0..10 { + sent.push(pause.adjust(frame_ts(i)).unwrap().unwrap()); + } + + flag.store(true, Ordering::Release); + for i in 10..40 { + assert_eq!( + pause.adjust(frame_ts(i)).unwrap(), + None, + "paused frames must be swallowed" + ); + } + flag.store(false, Ordering::Release); + + for i in 40..60 { + sent.push(pause.adjust(frame_ts(i)).unwrap().unwrap()); + } + + for pair in sent.windows(2) { + assert!( + pair[1] > pair[0], + "adjusted timeline must be strictly forward, found {:?} then {:?}", + pair[0], + pair[1] + ); + } + + let resume_step = sent[10].saturating_sub(sent[9]); + assert_eq!( + resume_step, + frame_ts(10).saturating_sub(frame_ts(9)), + "the pause span must be excised: the first resumed frame continues one \ + normal frame step after the last sent frame" + ); + } + + #[test] + fn pause_with_no_frames_is_a_noop() { + let flag = Arc::new(AtomicBool::new(false)); + let pause = SharedPauseState::new(flag.clone()); + + assert_eq!( + pause.adjust(frame_ts(0)).unwrap(), + Some(frame_ts(0)), + "no pause yet, passthrough" + ); + + flag.store(true, Ordering::Release); + flag.store(false, Ordering::Release); + + assert_eq!( + pause.adjust(frame_ts(1)).unwrap(), + Some(frame_ts(1)), + "a pause window with no swallowed frames must not shift the timeline" + ); + } + + #[test] + fn repeated_pause_cycles_accumulate_offsets() { + let flag = Arc::new(AtomicBool::new(false)); + let pause = SharedPauseState::new(flag.clone()); + + let _ = pause.adjust(Duration::from_secs(1)).unwrap(); + + flag.store(true, Ordering::Release); + assert_eq!(pause.adjust(Duration::from_secs(2)).unwrap(), None); + flag.store(false, Ordering::Release); + let after_first = pause.adjust(Duration::from_secs(5)).unwrap().unwrap(); + assert_eq!(after_first, Duration::from_secs(2)); + + flag.store(true, Ordering::Release); + assert_eq!(pause.adjust(Duration::from_secs(6)).unwrap(), None); + flag.store(false, Ordering::Release); + let after_second = pause.adjust(Duration::from_secs(10)).unwrap().unwrap(); + assert_eq!( + after_second, + Duration::from_secs(3), + "both pause spans must stay excised: 10s - (5s-2s) - (10s-6s) = 3s" + ); + } + + #[test] + fn resume_with_backwards_timestamp_does_not_panic_or_stall() { + let flag = Arc::new(AtomicBool::new(false)); + let pause = SharedPauseState::new(flag.clone()); + + let _ = pause.adjust(Duration::from_secs(2)).unwrap(); + + flag.store(true, Ordering::Release); + assert_eq!(pause.adjust(Duration::from_secs(3)).unwrap(), None); + flag.store(false, Ordering::Release); + + let adjusted = pause.adjust(Duration::from_secs(1)).unwrap(); + assert_eq!( + adjusted, + Some(Duration::from_secs(1)), + "a backwards resume timestamp is treated as zero pause delta" + ); + + let next = pause.adjust(Duration::from_secs(4)).unwrap(); + assert_eq!( + next, + Some(Duration::from_secs(4)), + "the timeline keeps flowing after the anomaly" + ); + } + } + mod audio_timestamp_generator { use super::*; diff --git a/crates/recording/src/output_pipeline/macos.rs b/crates/recording/src/output_pipeline/macos.rs index 8991b9c7284..ea764ef3457 100644 --- a/crates/recording/src/output_pipeline/macos.rs +++ b/crates/recording/src/output_pipeline/macos.rs @@ -1,7 +1,8 @@ use crate::{ output_pipeline::{ - AudioFrame, AudioMuxer, BlockingThreadFinish, HealthSender, Muxer, PipelineHealthEvent, - TaskPool, VideoFrame, VideoMuxer, emit_health, wait_for_blocking_thread_finish, + AudioFrame, AudioMuxer, BlockingThreadFinish, DiskSpaceMonitor, DiskSpacePollResult, + HealthSender, Muxer, PipelineHealthEvent, SharedHealthSender, TaskPool, VideoFrame, + VideoMuxer, emit_health, wait_for_blocking_thread_finish, }, sources::screen_capture, }; @@ -28,9 +29,7 @@ const DEFAULT_MP4_MUXER_BUFFER_SIZE_INSTANT: usize = 240; const DEFAULT_MP4_AUDIO_FINISH_TIMEOUT: Duration = Duration::from_secs(2); const DEFAULT_MP4_AUDIO_FINISH_TIMEOUT_INSTANT: Duration = Duration::from_secs(8); -const DISK_SPACE_MIN_START_MB: u64 = 500; -const DISK_SPACE_CRITICAL_MB: u64 = 200; -const DISK_SPACE_CHECK_INTERVAL: Duration = Duration::from_secs(10); +const DISK_SPACE_MIN_START_BYTES: u64 = 500 * 1024 * 1024; fn boost_encoder_thread_qos() { let result = set_current_thread_qos(MacOsQosClass::UserInitiated); @@ -39,15 +38,29 @@ fn boost_encoder_thread_qos() { } } -fn get_available_disk_space_mb(path: &std::path::Path) -> Option { - use std::ffi::CString; - let c_path = CString::new(path.parent().unwrap_or(path).to_str()?).ok()?; - let mut stat: libc::statvfs = unsafe { std::mem::zeroed() }; - let result = unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) }; - if result != 0 { - return None; +// Refuse to start any AVAssetWriter recording (studio, instant, camera) +// without headroom: a writer that dies on a failed async write mid-recording +// loses its moov, so the clean refusal up front is strictly better. +fn check_disk_space_to_start(output_path: &std::path::Path) -> anyhow::Result<()> { + match cap_utils::disk_space::free_bytes_for_path(output_path) { + Ok(available) => { + info!( + available_mb = available / (1024 * 1024), + "Disk space check before recording start" + ); + if available < DISK_SPACE_MIN_START_BYTES { + return Err(anyhow!( + "Insufficient disk space to start recording: {}MB available, {}MB required", + available / (1024 * 1024), + DISK_SPACE_MIN_START_BYTES / (1024 * 1024) + )); + } + } + Err(err) => { + debug!(error = %err, "Disk space preflight probe failed; starting anyway"); + } } - Some((stat.f_bavail as u64).saturating_mul(stat.f_frsize) / (1024 * 1024)) + Ok(()) } fn get_mp4_muxer_buffer_size(instant_mode: bool) -> usize { @@ -273,6 +286,7 @@ pub struct AVFoundationMp4Muxer { audio_channel_pressure: Option, was_paused: bool, fatal_error: SharedFatalError, + health_tx: SharedHealthSender, } #[derive(Default)] @@ -297,18 +311,7 @@ impl Muxer for AVFoundationMp4Muxer { let video_config = video_config.ok_or_else(|| anyhow!("Invariant: No video source provided"))?; - if config.instant_mode - && let Some(available_mb) = get_available_disk_space_mb(&output_path) - { - info!(available_mb, "Disk space check before recording start"); - if available_mb < DISK_SPACE_MIN_START_MB { - return Err(anyhow!( - "Insufficient disk space to start recording: {}MB available, {}MB required", - available_mb, - DISK_SPACE_MIN_START_MB - )); - } - } + check_disk_space_to_start(&output_path)?; let buffer_size = get_mp4_muxer_buffer_size(config.instant_mode); debug!( @@ -356,6 +359,8 @@ impl Muxer for AVFoundationMp4Muxer { let fatal_error = Arc::new(Mutex::new(None)); let video_fatal_error = fatal_error.clone(); let disk_check_path = output_path.clone(); + let health_tx = SharedHealthSender::new(); + let video_health_tx = health_tx.clone(); let is_instant = config.instant_mode; let (channel_pressure, channel_depth) = if is_instant { @@ -386,7 +391,7 @@ impl Muxer for AVFoundationMp4Muxer { } let mut encoder_busy_count = 0u64; - let mut last_disk_check = std::time::Instant::now(); + let mut disk_monitor = DiskSpaceMonitor::new(); while let Ok(Some(msg)) = video_rx.recv() { if let Some(ref depth) = channel_depth { @@ -396,17 +401,23 @@ impl Muxer for AVFoundationMp4Muxer { break; } - if is_instant && last_disk_check.elapsed() >= DISK_SPACE_CHECK_INTERVAL { - last_disk_check = std::time::Instant::now(); - if let Some(available_mb) = get_available_disk_space_mb(&disk_check_path) - && available_mb < DISK_SPACE_CRITICAL_MB - { - let message = format!( - "Disk space critically low ({available_mb}MB), stopping recording to preserve output" - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } + // All modes, not just instant: if the disk fills, the + // AVAssetWriter dies asynchronously mid-write and the + // non-fragmented output loses its moov (unrecoverable). + // Stopping while the writer is alive lets finish() write + // the moov and preserves the recording up to this point. + // DiskSpaceMonitor carries the platform-wide thresholds + // (warn 200MB / stop 50MB) and emits DiskSpaceLow / + // DiskSpaceExhausted so the user sees the real cause. + if let DiskSpacePollResult::Exhausted { bytes_remaining } = + disk_monitor.poll(&disk_check_path, &video_health_tx) + { + let message = format!( + "Disk space exhausted ({}MB left), stopping recording to preserve output", + bytes_remaining / (1024 * 1024) + ); + set_fatal_error(&video_fatal_error, message.clone()); + return Err(anyhow!(message)); } match msg { @@ -447,7 +458,7 @@ impl Muxer for AVFoundationMp4Muxer { let total = video_count_thread .load(std::sync::atomic::Ordering::Relaxed); let message = format!( - "Failed to encode video frame: WriterFailed/{err} \ + "Failed to encode video frame: WriterFailed/{err:?} \ (frame #{total}, ts={timestamp:?})" ); set_fatal_error(&video_fatal_error, message.clone()); @@ -576,7 +587,7 @@ impl Muxer for AVFoundationMp4Muxer { let total = audio_count_thread .load(std::sync::atomic::Ordering::Relaxed); let message = format!( - "Failed to encode audio frame: WriterFailed/{err} \ + "Failed to encode audio frame: WriterFailed/{err:?} \ (frame #{total}, ts={timestamp:?})" ); set_fatal_error(&audio_fatal_error, message.clone()); @@ -650,6 +661,7 @@ impl Muxer for AVFoundationMp4Muxer { audio_channel_pressure, was_paused: false, fatal_error, + health_tx, }) } @@ -667,6 +679,7 @@ impl Muxer for AVFoundationMp4Muxer { } fn set_health_sender(&mut self, tx: HealthSender) { + self.health_tx.set(tx.clone()); self.frame_drops.health_tx = Some(tx); } @@ -920,6 +933,7 @@ pub struct AVFoundationCameraMuxer { audio_channel_pressure: Option, was_paused: bool, fatal_error: SharedFatalError, + health_tx: SharedHealthSender, } #[derive(Default)] @@ -943,6 +957,8 @@ impl Muxer for AVFoundationCameraMuxer { let video_config = video_config.ok_or_else(|| anyhow!("Invariant: No video source provided"))?; + check_disk_space_to_start(&output_path)?; + let is_instant = config.instant_mode; let buffer_size = get_mp4_muxer_buffer_size(is_instant); debug!( @@ -982,6 +998,9 @@ impl Muxer for AVFoundationCameraMuxer { let encoder_clone = encoder.clone(); let fatal_error = Arc::new(Mutex::new(None)); let video_fatal_error = fatal_error.clone(); + let disk_check_path = output_path.clone(); + let health_tx = SharedHealthSender::new(); + let video_health_tx = health_tx.clone(); let encoder_handle = std::thread::Builder::new() .name("mp4-camera-encoder".to_string()) @@ -994,12 +1013,29 @@ impl Muxer for AVFoundationCameraMuxer { let mut total_frames = 0u64; let mut encoder_busy_count = 0u64; + let mut disk_monitor = DiskSpaceMonitor::new(); while let Ok(Some(msg)) = video_rx.recv() { if fatal_error_message(&video_fatal_error).is_some() { break; } + // Same rationale as the screen writer above: stop while + // the AVAssetWriter is still alive so the camera file + // keeps its moov instead of dying on a failed async write + // (finish() runs because the thread exits cleanly with an + // error rather than timing out). + if let DiskSpacePollResult::Exhausted { bytes_remaining } = + disk_monitor.poll(&disk_check_path, &video_health_tx) + { + let message = format!( + "Disk space exhausted ({}MB left), stopping camera recording to preserve output", + bytes_remaining / (1024 * 1024) + ); + set_fatal_error(&video_fatal_error, message.clone()); + return Err(anyhow!(message)); + } + match msg { CameraFrameMessage::Frame(sample_buf, timestamp) => { let mut retry_count = 0; @@ -1037,7 +1073,7 @@ impl Muxer for AVFoundationCameraMuxer { } Err(QueueFrameError::WriterFailed(err)) => { let message = format!( - "Failed to encode camera frame: WriterFailed/{err}" + "Failed to encode camera frame: WriterFailed/{err:?}" ); set_fatal_error(&video_fatal_error, message.clone()); return Err(anyhow!(message)); @@ -1161,7 +1197,7 @@ impl Muxer for AVFoundationCameraMuxer { } Err(QueueFrameError::WriterFailed(err)) => { let message = format!( - "Failed to encode camera audio frame: WriterFailed/{err} \ + "Failed to encode camera audio frame: WriterFailed/{err:?} \ (frame #{total_frames}, ts={timestamp:?})" ); set_fatal_error(&audio_fatal_error, message.clone()); @@ -1227,6 +1263,7 @@ impl Muxer for AVFoundationCameraMuxer { audio_channel_pressure, was_paused: false, fatal_error, + health_tx, }) } @@ -1243,6 +1280,11 @@ impl Muxer for AVFoundationCameraMuxer { } } + fn set_health_sender(&mut self, tx: HealthSender) { + self.health_tx.set(tx.clone()); + self.frame_drops.health_tx = Some(tx); + } + fn finish(&mut self, timestamp: Duration) -> anyhow::Result> { let mut finish_error: Option = None; @@ -1258,14 +1300,37 @@ impl Muxer for AVFoundationCameraMuxer { let mut can_finish_encoder = true; - if let Some(handle) = state.encoder_handle.take() - && let Err(e) = - wait_for_worker(handle, Duration::from_secs(5), "Camera MP4 encoder thread") - { - warn!("{e:#}"); - can_finish_encoder = false; - if finish_error.is_none() { - finish_error = Some(e); + if let Some(handle) = state.encoder_handle.take() { + match wait_for_blocking_thread_finish( + handle, + Duration::from_secs(5), + "Camera MP4 encoder thread", + ) { + BlockingThreadFinish::Clean => {} + // The thread exited with an error: the encoder mutex is + // free, so fall through to encoder.finish() below. When + // the writer is still alive (disk-exhaustion stop, mutex + // poison) finalizing is what preserves the moov — + // skipping it here used to leave camera.mp4 headerless + // on every encoder thread error. When the writer itself + // died (WriterFailed) finish_writing cannot salvage the + // file, but the attempt is harmless and surfaces the + // writer's NSError. + BlockingThreadFinish::Failed(error) => { + warn!("{error:#}"); + if finish_error.is_none() { + finish_error = Some(error); + } + } + // The thread is still alive and may hold the encoder + // mutex; locking it for finish() could block forever. + BlockingThreadFinish::TimedOut(error) => { + warn!("{error:#}"); + can_finish_encoder = false; + if finish_error.is_none() { + finish_error = Some(error); + } + } } } diff --git a/crates/recording/tests/hardware_instant_recording.rs b/crates/recording/tests/hardware_instant_recording.rs index f99c758a58c..7d2fda02b34 100644 --- a/crates/recording/tests/hardware_instant_recording.rs +++ b/crates/recording/tests/hardware_instant_recording.rs @@ -1,3 +1,5 @@ +#![cfg(target_os = "macos")] + use cap_enc_ffmpeg::remux::{ concatenate_m4s_segments_with_init, get_media_duration, merge_video_audio, probe_m4s_can_decode_with_init, probe_media_valid, probe_video_can_decode, @@ -82,8 +84,16 @@ async fn instant_record_with_real_mic_and_screen() { let temp = TempDir::new().unwrap(); let recording_dir = temp.path().join("test_recording.cap"); - let recording_seconds = 15; - eprintln!("Starting {recording_seconds}s instant recording..."); + let record_before_pause = Duration::from_secs(6); + let pause_duration = Duration::from_secs(5); + let record_after_resume = Duration::from_secs(6); + let expected_content_secs = (record_before_pause + record_after_resume).as_secs_f64(); + eprintln!( + "Starting instant recording: {}s, pause {}s, {}s (expecting ~{expected_content_secs}s of content)...", + record_before_pause.as_secs(), + pause_duration.as_secs(), + record_after_resume.as_secs(), + ); let mut builder = instant_recording::Actor::builder( recording_dir.clone(), @@ -102,7 +112,25 @@ async fn instant_record_with_real_mic_and_screen() { let segment_rx = actor_handle.take_segment_rx(); - tokio::time::sleep(Duration::from_secs(recording_seconds)).await; + tokio::time::sleep(record_before_pause).await; + + eprintln!("Pausing for {}s...", pause_duration.as_secs()); + actor_handle + .pause() + .await + .expect("Failed to pause recording"); + assert!( + actor_handle.is_paused().await.expect("is_paused failed"), + "actor should report paused" + ); + tokio::time::sleep(pause_duration).await; + + eprintln!("Resuming..."); + actor_handle + .resume() + .await + .expect("Failed to resume recording"); + tokio::time::sleep(record_after_resume).await; eprintln!("Stopping recording..."); let completed = actor_handle.stop().await.expect("Failed to stop recording"); @@ -307,14 +335,20 @@ async fn instant_record_with_real_mic_and_screen() { "Should be able to read assembled video duration" ); let video_dur_secs = video_duration.unwrap().as_secs_f64(); - eprintln!(" Video duration: {video_dur_secs:.2}s (expected ~{recording_seconds}s)"); + eprintln!( + " Video duration: {video_dur_secs:.2}s (expected ~{expected_content_secs}s of content, \ + pause excised)" + ); assert!( - video_dur_secs > (recording_seconds as f64) * 0.5, - "Video duration ({video_dur_secs:.2}s) should be at least 50% of recording time ({recording_seconds}s)" + video_dur_secs > expected_content_secs * 0.7, + "Video duration ({video_dur_secs:.2}s) should be at least 70% of the recorded content \ + time ({expected_content_secs}s)" ); assert!( - video_dur_secs < (recording_seconds as f64) * 2.0, - "Video duration ({video_dur_secs:.2}s) should be less than 2x recording time ({recording_seconds}s)" + video_dur_secs < expected_content_secs * 1.3, + "Video duration ({video_dur_secs:.2}s) should be under 130% of the recorded content \ + time ({expected_content_secs}s) — a value near wall time means the pause leaked \ + into the timeline" ); let input_ctx = @@ -361,10 +395,20 @@ async fn instant_record_with_real_mic_and_screen() { "Should be able to read assembled audio duration" ); let audio_dur_secs = audio_duration.unwrap().as_secs_f64(); - eprintln!(" Audio duration: {audio_dur_secs:.2}s (expected ~{recording_seconds}s)"); + eprintln!( + " Audio duration: {audio_dur_secs:.2}s (expected ~{expected_content_secs}s of \ + content, pause excised)" + ); + assert!( + audio_dur_secs > expected_content_secs * 0.7, + "Audio duration ({audio_dur_secs:.2}s) should be at least 70% of the recorded \ + content time" + ); assert!( - audio_dur_secs > (recording_seconds as f64) * 0.5, - "Audio duration ({audio_dur_secs:.2}s) should be at least 50% of recording time" + audio_dur_secs < expected_content_secs * 1.3, + "Audio duration ({audio_dur_secs:.2}s) should be under 130% of the recorded \ + content time — a value near wall time means the pause leaked into the audio \ + timeline" ); let av_drift = (video_dur_secs - audio_dur_secs).abs(); diff --git a/crates/recording/tests/hardware_studio_recording.rs b/crates/recording/tests/hardware_studio_recording.rs new file mode 100644 index 00000000000..78d60a4d489 --- /dev/null +++ b/crates/recording/tests/hardware_studio_recording.rs @@ -0,0 +1,174 @@ +#![cfg(target_os = "macos")] + +//! Real-hardware validation of the studio-mode NON-fragmented pipeline: the +//! AVFoundation MP4 writer path that produced the 0.5.8 field failures +//! (-11800/-16364 InvalidTimestamp). Requires Screen Recording permission on +//! the terminal running the test, exactly like `hardware_instant_recording`. +//! +//! Records the primary display through the real studio actor with +//! `fragmented(false)` (the shape a studio recording takes when a camera is +//! active), pauses and resumes mid-recording (which finalizes segment-0 and +//! opens segment-1), then verifies every segment's display.mp4 is a plain +//! finalized MP4 with the expected content duration. + +use cap_enc_ffmpeg::remux::{get_media_duration, probe_media_valid, probe_video_can_decode}; +use cap_recording::sources::screen_capture::ScreenCaptureTarget; +use cap_recording::{SendableShareableContent, studio_recording}; +use std::{path::PathBuf, time::Duration}; +use tempfile::TempDir; + +fn init() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive(tracing::Level::INFO.into()), + ) + .with_test_writer() + .try_init() + .ok(); + ffmpeg::init().expect("failed to initialize ffmpeg"); +} + +#[tokio::test] +async fn studio_nonfragmented_record_pause_resume_with_real_screen() { + init(); + + let primary = scap_targets::Display::primary(); + let display_id = primary.id(); + eprintln!( + "Using primary display: {:?}", + primary.name().unwrap_or_default(), + ); + + let shareable_content: SendableShareableContent = cidre::sc::ShareableContent::current() + .await + .expect( + "Failed to get SCShareableContent. \ + Grant Screen Recording permission to your terminal in \ + System Settings > Privacy & Security > Screen Recording", + ) + .into(); + + let temp = TempDir::new().unwrap(); + let recording_dir = temp.path().join("test_studio_recording.cap"); + + let record_before_pause = Duration::from_secs(6); + let pause_duration = Duration::from_secs(3); + let record_after_resume = Duration::from_secs(6); + let segment_expected_secs = [ + record_before_pause.as_secs_f64(), + record_after_resume.as_secs_f64(), + ]; + + eprintln!( + "Starting studio (non-fragmented) recording: {}s, pause {}s, {}s...", + record_before_pause.as_secs(), + pause_duration.as_secs(), + record_after_resume.as_secs(), + ); + + let actor_handle = studio_recording::Actor::builder( + recording_dir.clone(), + ScreenCaptureTarget::Display { id: display_id }, + ) + .with_fragmented(false) + .with_max_fps(30) + .with_keyboard_capture(false) + .build(Some(shareable_content)) + .await + .expect("Failed to spawn studio recording actor"); + + tokio::time::sleep(record_before_pause).await; + + eprintln!( + "Pausing for {}s (finalizes segment-0)...", + pause_duration.as_secs() + ); + actor_handle.pause().await.expect("Failed to pause"); + assert!( + actor_handle.is_paused().await.expect("is_paused failed"), + "actor should report paused" + ); + tokio::time::sleep(pause_duration).await; + + eprintln!("Resuming (opens segment-1)..."); + actor_handle.resume().await.expect("Failed to resume"); + tokio::time::sleep(record_after_resume).await; + + eprintln!("Stopping recording..."); + let completed = actor_handle.stop().await.expect("Failed to stop recording"); + eprintln!("Recording stopped at {}", completed.project_path.display()); + + let segments_dir = recording_dir.join("content").join("segments"); + let mut segment_dirs: Vec = std::fs::read_dir(&segments_dir) + .expect("segments dir should exist") + .filter_map(|e| { + let path = e.ok()?.path(); + path.is_dir().then_some(path) + }) + .collect(); + segment_dirs.sort(); + + assert_eq!( + segment_dirs.len(), + 2, + "pause/resume must produce exactly two segments, got {segment_dirs:?}" + ); + + let mut total_duration = 0.0f64; + for (i, segment_dir) in segment_dirs.iter().enumerate() { + let display_path = segment_dir.join("display.mp4"); + assert!( + display_path.is_file(), + "segment {i} display.mp4 must be a plain finalized MP4 file \ + (non-fragmented studio path), missing at {}", + display_path.display() + ); + + assert!( + probe_media_valid(&display_path), + "segment {i} display.mp4 must be a valid container" + ); + assert!( + probe_video_can_decode(&display_path).unwrap_or(false), + "segment {i} display.mp4 must be decodable" + ); + + let duration = get_media_duration(&display_path) + .expect("segment display duration should be readable") + .as_secs_f64(); + let expected = segment_expected_secs[i]; + eprintln!(" Segment {i}: {duration:.2}s (expected ~{expected:.0}s)"); + assert!( + duration > expected * 0.6, + "segment {i} duration ({duration:.2}s) should be at least 60% of its recording \ + window ({expected:.0}s)" + ); + assert!( + duration < expected * 1.4, + "segment {i} duration ({duration:.2}s) should be under 140% of its recording \ + window ({expected:.0}s) — a larger value means paused time leaked in" + ); + total_duration += duration; + } + + let expected_content = segment_expected_secs.iter().sum::(); + eprintln!( + " Total content: {total_duration:.2}s (expected ~{expected_content:.0}s, \ + pause excised across segments)" + ); + assert!( + (total_duration - expected_content).abs() < expected_content * 0.4, + "total recorded content ({total_duration:.2}s) should be within 40% of \ + {expected_content:.0}s" + ); + + let meta_path = recording_dir.join("recording-meta.json"); + assert!( + meta_path.exists(), + "recording meta should be persisted at {}", + meta_path.display() + ); + + eprintln!("\n=== ALL CHECKS PASSED ==="); +} diff --git a/crates/recording/tests/instant_mode_scenarios.rs b/crates/recording/tests/instant_mode_scenarios.rs index 7f4f38a9333..7c927b1f96e 100644 --- a/crates/recording/tests/instant_mode_scenarios.rs +++ b/crates/recording/tests/instant_mode_scenarios.rs @@ -9,11 +9,17 @@ use cap_enc_ffmpeg::{ }, }; use cap_media_info::{AudioInfo, VideoInfo}; -use cap_recording::{RecordingHealth, output_validation::validate_instant_recording}; +use cap_recording::{ + RecordingHealth, SharedPauseState, output_validation::validate_instant_recording, +}; use std::{ collections::{HashMap, HashSet}, path::{Path, PathBuf}, - sync::mpsc, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + mpsc, + }, time::Duration, }; use tempfile::TempDir; @@ -139,11 +145,23 @@ fn encode_video_segments( .unwrap(); encoder.set_segment_callback(tx); + // The DASH muxer can only cut segments at keyframes and the encoder's + // GOP is fixed at DEFAULT_KEYFRAME_INTERVAL_SECS (2s), so sub-GOP + // segment durations are only reachable when the source marks I-frames + // at the segment cadence. libx264 honors keyint_min strictly while + // hardware encoders emit extra IDRs, so without this the segment counts + // differ per platform. + let seg_ms = segment_duration.as_millis() as u64; let total_frames = recording_duration_ms / frame_interval_ms; for i in 0..total_frames { - let frame = make_video_frame_patterned(info.width, info.height, i as u32); - let ts = Duration::from_millis(i * frame_interval_ms); - encoder.queue_frame(frame, ts).unwrap(); + let ts_ms = i * frame_interval_ms; + let mut frame = make_video_frame_patterned(info.width, info.height, i as u32); + if seg_ms > 0 && ts_ms % seg_ms < frame_interval_ms { + frame.set_kind(ffmpeg::picture::Type::I); + } + encoder + .queue_frame(frame, Duration::from_millis(ts_ms)) + .unwrap(); } encoder.finish().unwrap(); @@ -1311,11 +1329,18 @@ fn video_audio_duration_alignment() { None, ); - let video_manifest = read_manifest(&video.manifest_path); - let audio_manifest = read_manifest(&audio.manifest_path); + // Compare assembled media durations, not manifest bookkeeping: the DASH + // muxer only opens a new segment file at a keyframe, so a tail that ends + // between keyframes is appended into the previous segment file and the + // manifest's estimated total under-reports it. The assembled output is + // what users get and must carry the full content on both tracks. + let video_mp4 = temp.path().join("video.mp4"); + concatenate_m4s_segments_with_init(&video.init_path, &video.segment_paths, &video_mp4).unwrap(); + let audio_m4a = temp.path().join("audio.m4a"); + concatenate_m4s_segments_with_init(&audio.init_path, &audio.segment_paths, &audio_m4a).unwrap(); - let video_duration = video_manifest["total_duration"].as_f64().unwrap(); - let audio_duration = audio_manifest["total_duration"].as_f64().unwrap(); + let video_duration = get_media_duration(&video_mp4).unwrap().as_secs_f64(); + let audio_duration = get_media_duration(&audio_m4a).unwrap().as_secs_f64(); let diff = (video_duration - audio_duration).abs(); assert!( @@ -2115,10 +2140,21 @@ fn segment_callback_receives_events_during_encoding() { .unwrap(); encoder.set_segment_callback(tx); - for i in 0..15 { + // Segment cuts require keyframes; mark them at the segment cadence so + // sub-GOP segment durations behave the same on every encoder. + let queue_with_cadence = |encoder: &mut SegmentedVideoEncoder, i: u64| { + let ts_ms = i * 33; + let mut frame = make_video_frame(320, 240); + if ts_ms % 200 < 33 { + frame.set_kind(ffmpeg::picture::Type::I); + } encoder - .queue_frame(make_video_frame(320, 240), Duration::from_millis(i * 33)) + .queue_frame(frame, Duration::from_millis(ts_ms)) .unwrap(); + }; + + for i in 0..15 { + queue_with_cadence(&mut encoder, i); } let mid_events: Vec = rx.try_iter().collect(); @@ -2129,9 +2165,7 @@ fn segment_callback_receives_events_during_encoding() { ); for i in 15..45 { - encoder - .queue_frame(make_video_frame(320, 240), Duration::from_millis(i * 33)) - .unwrap(); + queue_with_cadence(&mut encoder, i); } encoder.finish().unwrap(); @@ -2336,6 +2370,267 @@ fn output_file_has_correct_codec() { ); } +#[test] +fn pause_resume_full_pipeline_excises_pause_and_stays_uploadable() { + common::init(); + + let temp = TempDir::new().unwrap(); + let content_dir = temp.path().join("content"); + std::fs::create_dir_all(&content_dir).unwrap(); + + let video_dir = content_dir.join("display"); + let audio_dir = content_dir.join("audio"); + + let mut video_encoder = SegmentedVideoEncoder::init( + video_dir.clone(), + default_video_info(), + SegmentedVideoEncoderConfig { + segment_duration: Duration::from_millis(500), + ..Default::default() + }, + ) + .unwrap(); + let mut audio_encoder = DashAudioSegmentEncoder::init( + audio_dir.clone(), + default_audio_info(), + DashAudioSegmentEncoderConfig { + segment_duration: Duration::from_millis(500), + }, + ) + .unwrap(); + + // The real instant-mode pause path: both muxers swallow frames while the + // shared flag is set and excise the pause span via SharedPauseState, + // exactly like MacOSFragmentedM4SMuxer and DashSegmentedAudioMuxer. + // 3s of capture with the recording paused over [1s, 2s). + let pause_flag = Arc::new(AtomicBool::new(false)); + let video_pause = SharedPauseState::new(pause_flag.clone()); + let audio_pause = SharedPauseState::new(pause_flag.clone()); + let pause_window = Duration::from_secs(1)..Duration::from_secs(2); + + let mut video_sent = 0u64; + for i in 0..90u64 { + let ts = Duration::from_nanos(i * 33_333_333 + 400); + pause_flag.store(pause_window.contains(&ts), Ordering::Release); + if let Some(adjusted) = video_pause.adjust(ts).unwrap() { + video_encoder + .queue_frame(make_video_frame(320, 240), adjusted) + .unwrap(); + video_sent += 1; + } + } + assert!( + (55..=65).contains(&video_sent), + "one third of the video frames should be swallowed by the pause, sent {video_sent}" + ); + + let mut sample_offset = 0u64; + for i in 0..140u64 { + let ts = Duration::from_nanos(i * 1024 * 1_000_000_000 / 48_000); + pause_flag.store(pause_window.contains(&ts), Ordering::Release); + if let Some(adjusted) = audio_pause.adjust(ts).unwrap() { + audio_encoder + .queue_frame(default_audio_frame(1024, sample_offset), adjusted) + .unwrap(); + sample_offset += 1024; + } + } + + video_encoder.finish().unwrap(); + audio_encoder.finish().unwrap(); + + let video_manifest = read_manifest(&video_dir.join("manifest.json")); + let audio_manifest = read_manifest(&audio_dir.join("manifest.json")); + assert!(video_manifest["is_complete"].as_bool().unwrap()); + assert!(audio_manifest["is_complete"].as_bool().unwrap()); + + let video_segs: Vec = video_manifest["segments"] + .as_array() + .unwrap() + .iter() + .filter(|s| s["is_complete"].as_bool().unwrap_or(false)) + .filter_map(|s| { + let p = video_dir.join(s["path"].as_str()?); + p.exists().then_some(p) + }) + .collect(); + let audio_segs: Vec = audio_manifest["segments"] + .as_array() + .unwrap() + .iter() + .filter(|s| s["is_complete"].as_bool().unwrap_or(false)) + .filter_map(|s| { + let p = audio_dir.join(s["path"].as_str()?); + p.exists().then_some(p) + }) + .collect(); + + let video_mp4 = temp.path().join("video.mp4"); + concatenate_m4s_segments_with_init(&video_dir.join("init.mp4"), &video_segs, &video_mp4) + .unwrap(); + let audio_m4a = temp.path().join("audio.m4a"); + concatenate_m4s_segments_with_init(&audio_dir.join("init.mp4"), &audio_segs, &audio_m4a) + .unwrap(); + + let video_dur = get_media_duration(&video_mp4).unwrap().as_secs_f64(); + let audio_dur = get_media_duration(&audio_m4a).unwrap().as_secs_f64(); + + assert!( + (1.6..=2.4).contains(&video_dur), + "3s capture with a 1s pause must produce ~2s of video, got {video_dur:.2}s \ + (a value near 3s means the pause leaked into the timeline)" + ); + assert!( + (1.6..=2.4).contains(&audio_dur), + "3s capture with a 1s pause must produce ~2s of audio, got {audio_dur:.2}s" + ); + assert!( + (video_dur - audio_dur).abs() < 0.5, + "video ({video_dur:.2}s) and audio ({audio_dur:.2}s) must excise the pause identically" + ); + + let merged = content_dir.join("output.mp4"); + merge_video_audio(&video_mp4, &audio_m4a, &merged).unwrap(); + + assert_valid_playable_mp4(&merged); + assert_has_video_stream(&merged); + assert_has_audio_stream(&merged); + + let validation = validate_instant_recording(&merged, Duration::from_secs(2)); + assert!( + validation.health.is_uploadable(), + "paused-and-resumed instant recording must stay uploadable, got {:?}", + validation.health + ); +} + +#[test] +fn stall_recovery_burst_full_pipeline_stays_uploadable() { + common::init(); + + let temp = TempDir::new().unwrap(); + let content_dir = temp.path().join("content"); + std::fs::create_dir_all(&content_dir).unwrap(); + + let video_dir = content_dir.join("display"); + let audio_dir = content_dir.join("audio"); + + let mut video_encoder = SegmentedVideoEncoder::init( + video_dir.clone(), + default_video_info(), + SegmentedVideoEncoderConfig { + segment_duration: Duration::from_millis(500), + ..Default::default() + }, + ) + .unwrap(); + let mut audio_encoder = DashAudioSegmentEncoder::init( + audio_dir.clone(), + default_audio_info(), + DashAudioSegmentEncoderConfig { + segment_duration: Duration::from_millis(500), + }, + ) + .unwrap(); + + // The 0.5.8 field-failure shape at the pipeline level: video delivers + // normally, stalls for 1.5s, then flushes a burst of backlogged frames + // landing nanoseconds apart (same microsecond), while audio keeps + // flowing through the stall. + let frame_ns = 33_333_333u64; + let mut video_timestamps: Vec = Vec::new(); + for i in 0..30u64 { + video_timestamps.push(Duration::from_nanos(i * frame_ns + 400)); + } + let stall_end = 30 * frame_ns + 1_500_000_000; + for i in 0..10u64 { + video_timestamps.push(Duration::from_nanos(stall_end + i * 300)); + } + for i in 1..=45u64 { + video_timestamps.push(Duration::from_nanos(stall_end + i * frame_ns)); + } + + for (i, &ts) in video_timestamps.iter().enumerate() { + video_encoder + .queue_frame(make_video_frame(320, 240), ts) + .unwrap_or_else(|e| panic!("video frame {i} at {ts:?} rejected: {e}")); + } + + let total_capture = Duration::from_nanos(stall_end + 45 * frame_ns); + let mut sample_offset = 0u64; + let mut audio_ts = Duration::ZERO; + while audio_ts < total_capture { + audio_encoder + .queue_frame(default_audio_frame(1024, sample_offset), audio_ts) + .unwrap(); + sample_offset += 1024; + audio_ts = Duration::from_nanos(sample_offset * 1_000_000_000 / 48_000); + } + + video_encoder.finish().unwrap(); + audio_encoder.finish().unwrap(); + + let video_manifest = read_manifest(&video_dir.join("manifest.json")); + let audio_manifest = read_manifest(&audio_dir.join("manifest.json")); + assert!(video_manifest["is_complete"].as_bool().unwrap()); + assert!(audio_manifest["is_complete"].as_bool().unwrap()); + + let video_segs: Vec = video_manifest["segments"] + .as_array() + .unwrap() + .iter() + .filter(|s| s["is_complete"].as_bool().unwrap_or(false)) + .filter_map(|s| { + let p = video_dir.join(s["path"].as_str()?); + p.exists().then_some(p) + }) + .collect(); + let audio_segs: Vec = audio_manifest["segments"] + .as_array() + .unwrap() + .iter() + .filter(|s| s["is_complete"].as_bool().unwrap_or(false)) + .filter_map(|s| { + let p = audio_dir.join(s["path"].as_str()?); + p.exists().then_some(p) + }) + .collect(); + + let video_mp4 = temp.path().join("video.mp4"); + concatenate_m4s_segments_with_init(&video_dir.join("init.mp4"), &video_segs, &video_mp4) + .unwrap(); + let audio_m4a = temp.path().join("audio.m4a"); + concatenate_m4s_segments_with_init(&audio_dir.join("init.mp4"), &audio_segs, &audio_m4a) + .unwrap(); + + let expected_secs = total_capture.as_secs_f64(); + let video_dur = get_media_duration(&video_mp4).unwrap().as_secs_f64(); + let audio_dur = get_media_duration(&audio_m4a).unwrap().as_secs_f64(); + assert!( + (video_dur - expected_secs).abs() < 0.5, + "the stall must stay in the video timeline (expected ~{expected_secs:.2}s, got \ + {video_dur:.2}s); collapsing it desyncs video from audio" + ); + assert!( + (video_dur - audio_dur).abs() < 1.0, + "video ({video_dur:.2}s) and audio ({audio_dur:.2}s) must stay aligned across the stall" + ); + + let merged = content_dir.join("output.mp4"); + merge_video_audio(&video_mp4, &audio_m4a, &merged).unwrap(); + + assert_valid_playable_mp4(&merged); + assert_has_video_stream(&merged); + assert_has_audio_stream(&merged); + + let validation = validate_instant_recording(&merged, total_capture); + assert!( + validation.health.is_uploadable(), + "stall-recovery burst recording must stay uploadable, got {:?}", + validation.health + ); +} + #[test] fn merged_output_preserves_both_stream_durations() { common::init(); diff --git a/crates/recording/tests/sync_matrix.rs b/crates/recording/tests/sync_matrix.rs index 8ef2ac4c7a9..174b42a29f4 100644 --- a/crates/recording/tests/sync_matrix.rs +++ b/crates/recording/tests/sync_matrix.rs @@ -34,6 +34,10 @@ const CONTENT_SECS: f64 = 4.0; /// pts). Covers warmup anchoring, emission jitter and encoder rounding, /// plus scheduler noise on shared CI runners. const ABS_TOLERANCE_SECS: f64 = 0.25; +/// The drift tracker deliberately re-pins pts toward the wall clock by up to +/// this much; heavy over-delivery cases get it as designed headroom on top of +/// the base tolerance. Keep in step with the tracker's re-pin cap. +const DRIFT_REPIN_CAP_SECS: f64 = 0.1; /// Tolerance for the relative structure (pts deltas vs sent deltas), which is /// what actually determines sync drift. The bug class this guards against /// produces errors of a second or more. @@ -256,23 +260,55 @@ async fn run_video_case(case: VideoCase) -> Result { let timestamps = Timestamps::now(); let sent = case.sent.clone(); + // Set once the pipeline consumer exists. The emitter starts before the + // pipeline build so the builder can own the channel receiver; frames + // scheduled during the build window back up in the bounded channel and + // their lateness is structural, not a runner stall. Only lateness on + // frames due after this instant means the runner actually stalled. + let built_at: std::sync::Arc> = + std::sync::Arc::new(std::sync::OnceLock::new()); let emit = { let sent = sent.clone(); let base = timestamps.instant(); let (width, height, content) = (case.width, case.height, case.content); let mut rng = Rng(case.rng_seed); + let built_at = built_at.clone(); + // Over-delivery cases (>240fps, rates production never produces) + // exist to verify drop handling, not consumer throughput: a weak + // runner saturated by the firehose proves nothing, so send blocking + // still counts as falling behind there and earns a loud skip. + let overload_case = case.delivered_fps > 240; tokio::spawn(async move { + let mut max_late = 0.0f64; + let mut last_send_end: Option = None; for (i, &ts) in sent.iter().enumerate() { - tokio::time::sleep_until((base + Duration::from_secs_f64(ts)).into()).await; + let due = base + Duration::from_secs_f64(ts); + tokio::time::sleep_until(due.into()).await; + if built_at.get().is_some_and(|built| due >= *built) { + // At production-representative rates, lateness counts + // only time since the channel was last free: a send_async + // blocked on pipeline backpressure delays the next frame + // too, and that is the pipeline's fault, not a runner + // stall — a consumer-side regression must fail the case, + // not convert it into a skip. + let anchor = if overload_case { + due + } else { + last_send_end.map_or(due, |s| s.max(due)) + }; + max_late = max_late.max(anchor.elapsed().as_secs_f64()); + } let frame = FFmpegVideoFrame { inner: make_video_frame(width, height, i as u64, content, &mut rng), - timestamp: Timestamp::Instant(base + Duration::from_secs_f64(ts)), + timestamp: Timestamp::Instant(due), }; if tx.send_async(frame).await.is_err() { break; } + last_send_end = Some(std::time::Instant::now()); } // Sender drops here, ending the stream. + max_late }) }; @@ -291,8 +327,9 @@ async fn run_video_case(case: VideoCase) -> Result { builder.build::(()).await } .map_err(|e| format!("pipeline build: {e}"))?; + let _ = built_at.set(std::time::Instant::now()); - emit.await.map_err(|e| format!("emit join: {e}"))?; + let max_emit_late = emit.await.map_err(|e| format!("emit join: {e}"))?; // The verification below assumes frames were emitted in real time; when a // saturated runner (or a software encoder drowning in worst-case content) // stalls emission for seconds, pts-vs-wall comparisons are meaningless. @@ -309,6 +346,19 @@ async fn run_video_case(case: VideoCase) -> Result { "skipped: runner fell {emit_lag:.1}s behind real-time emission" )); } + // A stall that later catches up is invisible to the end-of-emission lag + // above, but it still contaminates the checks: frames stamped with their + // scheduled capture time arrive late, and the pipeline's wall-clock + // coupling (drift re-pinning) legitimately moves the muxed pts by about + // the stall size — a 0.3s scheduler stall mid-case reads as a 0.3s "pts + // error" on an otherwise healthy pipeline. Real timestamp bugs reproduce + // on healthy runners; a stalled runner proves nothing either way. + if max_emit_late > 0.1 { + return Ok(format!( + "skipped: runner stalled {max_emit_late:.2}s mid-emission; \ + pts-vs-wall checks are environment-contaminated" + )); + } // Read back the muxed pts. let playable = if fragmented { @@ -318,21 +368,128 @@ async fn run_video_case(case: VideoCase) -> Result { }; let pts = read_video_pts(&playable)?; + // At low frame rates the fixed tolerance is only a frame or two of + // budget, so scheduler jitter on shared runners trips it; express the + // floor in frames as well. The bug class this guards produces errors of + // a second or more either way. When the source over-delivers several + // times faster than the configured rate, the drift tracker deliberately + // re-pins pts toward the wall clock (a designed 0.1s cap) while the + // runner schedules hundreds of timed emissions per second, so the + // headroom on top of that designed deviation has to be wider. + let over_delivery = f64::from(case.delivered_fps) / f64::from(case.fps.max(1)); + let base_rel_tolerance = if over_delivery > 4.0 { + REL_TOLERANCE_SECS + DRIFT_REPIN_CAP_SECS + } else { + REL_TOLERANCE_SECS + }; + let rel_tolerance = base_rel_tolerance.max(2.5 / f64::from(case.fps)); + if pts.len() != sent.len() { - return Err(format!( - "frame count mismatch: sent {} frames, container has {}", + // Beyond any real capture device's rate, losslessness is not a + // pipeline guarantee: the muxer's stall budget drops frames rather + // than block capture (production behavior), and a shared runner + // cannot real-time-encode several hundred fps of worst-case content. + // Timestamp correctness is still enforced below on every frame that + // was muxed; extra frames or heavy loss always fail. + let overload_case = case.delivered_fps > 240; + let coverage = pts.len() as f64 / sent.len() as f64; + if !overload_case || coverage < 0.9 || pts.len() > sent.len() { + return Err(format!( + "frame count mismatch: sent {} frames, container has {} \ + (missing sent indices: {})", + sent.len(), + pts.len(), + unmatched_sent_indices(&sent, &pts, 1.0 / f64::from(case.delivered_fps.max(1))) + )); + } + + let sent_origin = sent[0]; + let pts_origin = pts[0]; + let mut max_rel: f64 = 0.0; + let mut j = 0usize; + for (i, &p) in pts.iter().enumerate() { + let rel_p = p - pts_origin; + while j + 1 < sent.len() + && ((sent[j + 1] - sent_origin) - rel_p).abs() + <= ((sent[j] - sent_origin) - rel_p).abs() + { + j += 1; + } + let rel = (rel_p - (sent[j] - sent_origin)).abs(); + max_rel = max_rel.max(rel); + if rel > rel_tolerance { + return Err(format!( + "muxed frame {i}: no sent timestamp within {rel_tolerance:.3}s \ + (pts {p:.3}s, nearest sent {:.3}s, err {rel:.3}s)", + sent[j] + )); + } + } + // Burst collapse piles muxed frames onto instants the sent timeline + // never had; nearest-matching alone scores that as zero error. Every + // generator dedups its timeline at exactly this threshold + // (period * 0.25), so consecutive sent pairs are never tight and + // sent_tight is identically zero today — the operative bound is the + // 5% absolute slack for boundary effects. The sent term stays as a + // scaling guard in case a future generator legitimately emits + // tighter cadences than its dedup spacing. + let tight = 0.25 / f64::from(case.delivered_fps.max(1)); + let tight_rate = |xs: &[f64]| { + if xs.len() < 2 { + return 0.0; + } + let tight_pairs = xs.windows(2).filter(|w| w[1] - w[0] < tight).count(); + tight_pairs as f64 / (xs.len() - 1) as f64 + }; + let muxed_tight = tight_rate(&pts); + let sent_tight = tight_rate(&sent); + if muxed_tight > sent_tight * 1.5 + 0.05 { + return Err(format!( + "muxed pts cluster far beyond the sent timeline (tight-pair rate \ + {:.1}% vs sent {:.1}% at <{tight:.6}s) — burst collapse under overload", + muxed_tight * 100.0, + sent_tight * 100.0 + )); + } + + // Drops shorten coverage but must not shrink the recorded span + // beyond the dropped tail/head, and must never stretch it. + if let Some((first, last)) = finished.video_timestamp_span { + let span = (last - first).as_secs_f64(); + let expected = sent.last().unwrap() - sent[0]; + if span > expected + 0.25 || span < expected - 0.5 { + return Err(format!( + "video_timestamp_span {span:.3}s does not match sent span \ + {expected:.3}s under overload" + )); + } + } else { + return Err("video_timestamp_span missing".to_string()); + } + + // Gap preservation still holds under drops: dropping frames can only + // widen a container gap, so a collapsed gap is a real timestamp bug. + let max_sent_gap = sent.windows(2).map(|w| w[1] - w[0]).fold(0.0, f64::max); + if max_sent_gap > 1.0 { + let max_pts_gap = pts.windows(2).map(|w| w[1] - w[0]).fold(0.0, f64::max); + if max_pts_gap < max_sent_gap * 0.9 { + return Err(format!( + "{max_sent_gap:.2}s capture gap collapsed to {max_pts_gap:.3}s \ + in the container under overload" + )); + } + } + + return Ok(format!( + "{} of {} frames muxed under {}fps overload (drops allowed), max rel err {max_rel:.3}s", + pts.len(), sent.len(), - pts.len() + case.delivered_fps )); } let mut max_abs: f64 = 0.0; let mut max_rel: f64 = 0.0; - // At low frame rates the fixed tolerance is only a frame or two of - // budget, so scheduler jitter on shared runners trips it; express the - // floor in frames as well. The bug class this guards produces errors of - // a second or more either way. - let rel_tolerance = REL_TOLERANCE_SECS.max(2.5 / f64::from(case.fps)); // The muxed timeline's origin is the first DELIVERED frame: the pipeline // zeroes each track at its first frame and the recorder persists the // track's start_time for cross-track alignment. A random case whose @@ -1076,6 +1233,57 @@ fn read_audio_stats(path: &Path) -> Result<(f64, u16, f64), String> { Ok((samples as f64 / f64::from(rate), channels, rms)) } +/// On a frame-count mismatch, name WHICH sent frames never reached the +/// container: a leading run ("0-4") means a startup stall/race, a spread +/// ("7, 23, 41") means mid-stream drops. Greedy monotone matcher — pts and +/// sent are both origin-normalized and sorted, a pts within 0.6 periods of +/// the sent slot consumes it. +fn unmatched_sent_indices(sent: &[f64], pts: &[f64], period: f64) -> String { + let Some(&sent0) = sent.first() else { + return "none".to_string(); + }; + let pts0 = pts.first().copied().unwrap_or(0.0); + let window = period * 0.6; + let mut missing: Vec = Vec::new(); + let mut i = 0usize; + for (k, &s) in sent.iter().enumerate() { + let rel_s = s - sent0; + while i < pts.len() && (pts[i] - pts0) < rel_s - window { + i += 1; + } + if i < pts.len() && ((pts[i] - pts0) - rel_s).abs() <= window { + i += 1; + } else { + missing.push(k); + } + } + if missing.is_empty() { + return "none (pts shifted rather than missing)".to_string(); + } + let mut runs: Vec = Vec::new(); + let mut start = missing[0]; + let mut prev = missing[0]; + for &m in &missing[1..] { + if m == prev + 1 { + prev = m; + continue; + } + runs.push(if start == prev { + format!("{start}") + } else { + format!("{start}-{prev}") + }); + start = m; + prev = m; + } + runs.push(if start == prev { + format!("{start}") + } else { + format!("{start}-{prev}") + }); + runs.join(", ") +} + fn record(results: &mut Vec, name: String, outcome: Result) { eprintln!( "{name}: {}", @@ -1180,6 +1388,12 @@ fn random_audio_case(rng: &mut Rng) -> AudioCase { /// would pay that stall, overflow the muxer's bounded channel, and drop its /// startup frames (observed as the 15fps/fragmented case losing 3-22 of 60 /// frames depending on load). +/// +/// The warm-up must run PAST the first segment cut, not just the first +/// accepted frame: VideoToolbox defers parts of session bring-up until real +/// packets flow, and the DASH muxer's first segment write has its own +/// first-use cost. A 3-frame warm-up stopped before either happened and the +/// first real case still stalled 1-3s on fast machines running newer macOS. async fn warm_up_video_encoder() { let Ok(temp) = tempfile::tempdir() else { return; @@ -1200,10 +1414,12 @@ async fn warm_up_video_encoder() { else { return; }; - for i in 0..3u64 { + // 2.2s of timestamps crosses the 2s segment boundary; frames are pushed + // as fast as the encoder accepts them (no real-time pacing needed). + for i in 0..66u64 { let frame = FFmpegVideoFrame { inner: make_video_frame(160, 120, i, Content::Flat, &mut rng), - timestamp: Timestamp::Instant(base + Duration::from_millis(i * 30)), + timestamp: Timestamp::Instant(base + Duration::from_millis(i * 33)), }; if tx.send_async(frame).await.is_err() { break; @@ -1214,8 +1430,49 @@ async fn warm_up_video_encoder() { let _ = pipeline.stop().await; } +/// One retry for a failed video case. A cold system pays one-time costs +/// (VideoToolbox service bring-up, first DASH segment write) INSIDE the +/// pipeline where no emitter-side guard can see them; the muxer's stall +/// budget then drops frames mid-case exactly as production would, and the +/// case fails on count with a contiguous missing run. That never repeats on +/// a warm system, while a real timestamp/drop regression reproduces +/// immediately — so a retried pass is labeled loudly instead of hidden. +async fn run_video_case_with_cold_retry(case: VideoCase) -> Result { + match run_video_case(case.clone()).await { + Ok(detail) => Ok(detail), + // Only the cold-start signatures earn a retry: a contiguous + // missing-frame run fails the count check, and a stop timeout is the + // same stall surfacing at teardown. Correctness failures (pts error, + // tight-pair clustering, duration drift) get no second chance — a + // retry there would let a 50%-reproducible regression pass most runs. + Err(first_error) + if first_error.contains("frame count mismatch") + || first_error.contains("Pipeline stop timed out") => + { + eprintln!("case failed cold ({first_error}); retrying once on a warm pipeline"); + run_video_case(case) + .await + .map(|detail| { + format!("passed on retry after cold-start failure ({first_error}); {detail}") + }) + .map_err(|second_error| { + format!("failed twice: {second_error} (first: {first_error})") + }) + } + Err(first_error) => Err(first_error), + } +} + #[tokio::test(flavor = "multi_thread")] async fn synthetic_device_matrix_preserves_sync() { + // Silent without RUST_LOG; with it, pipeline drop/stall warnings become + // visible so a failing case can be diagnosed instead of re-guessed. + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .try_init() + .ok(); + let mut results: Vec = Vec::new(); warm_up_video_encoder().await; @@ -1258,7 +1515,8 @@ async fn synthetic_device_matrix_preserves_sync() { scenario.name(), if fragmented { "fragmented" } else { "mp4" } ); - let outcome = run_video_case(VideoCase::curated(fps, scenario, fragmented)).await; + let outcome = + run_video_case_with_cold_retry(VideoCase::curated(fps, scenario, fragmented)).await; record(&mut results, name, outcome); } @@ -1296,7 +1554,7 @@ async fn synthetic_device_matrix_preserves_sync() { scenario.name(), if fragmented { "fragmented" } else { "mp4" } ); - let outcome = run_video_case(VideoCase::mismatch( + let outcome = run_video_case_with_cold_retry(VideoCase::mismatch( nominal, delivered, scenario, fragmented, )) .await; @@ -1388,7 +1646,7 @@ async fn synthetic_device_matrix_preserves_sync() { ); // Run both legs concurrently, as a real recording does. let (video_outcome, audio_outcome) = - tokio::join!(run_video_case(video), run_audio_case(audio)); + tokio::join!(run_video_case_with_cold_retry(video), run_audio_case(audio)); let outcome = match (video_outcome, audio_outcome) { (Ok(v), Ok(a)) => Ok(format!("video: {v}; audio: {a}")), (Err(e), _) => Err(format!("video leg: {e}")), @@ -1424,4 +1682,20 @@ async fn synthetic_device_matrix_preserves_sync() { .collect::>() .join("\n") ); + + // Environment skips are a pressure valve, not a pass: if half the matrix + // skipped, the run proves nothing and must be loud about it. Retried + // passes spent one of their two shots on a cold failure, so they count + // toward the same degradation budget — a runner that needs the retry + // everywhere proves as little as one that skips everywhere. + let degraded = results + .iter() + .filter(|r| r.detail.contains("skipped:") || r.detail.contains("passed on retry")) + .count(); + assert!( + degraded * 2 <= results.len(), + "{degraded} of {} matrix cases skipped or passed only on retry — runner too \ + degraded for this run to verify anything", + results.len() + ); } diff --git a/crates/rendering/src/layers/notch.rs b/crates/rendering/src/layers/notch.rs index 2c7a1915d43..a20e413b873 100644 --- a/crates/rendering/src/layers/notch.rs +++ b/crates/rendering/src/layers/notch.rs @@ -164,7 +164,7 @@ mod tests { /// A notch spanning x 64..192 and y 0..48 of the output. const BOUNDS: [f32; 4] = [64.0, 0.0, 192.0, 48.0]; - fn device() -> Option<(wgpu::Device, wgpu::Queue)> { + fn device() -> Option<(wgpu::Device, wgpu::Queue, wgpu::AdapterInfo)> { let instance = crate::create_wgpu_instance_sync(); let adapter = pollster::block_on(instance.request_adapter(&wgpu::RequestAdapterOptions { power_preference: wgpu::PowerPreference::LowPower, @@ -173,7 +173,10 @@ mod tests { })) .ok()?; - pollster::block_on(adapter.request_device(&wgpu::DeviceDescriptor::default())).ok() + let info = adapter.get_info(); + let (device, queue) = + pollster::block_on(adapter.request_device(&wgpu::DeviceDescriptor::default())).ok()?; + Some((device, queue, info)) } fn uniforms() -> NotchUniforms { @@ -200,7 +203,14 @@ mod tests { /// Renders the notch over a white frame and returns the RGBA pixels. fn render_with_uniforms(uniforms: NotchUniforms) -> Option> { - let (device, queue) = device()?; + let (pixels, _) = render_with_uniforms_and_adapter(uniforms)?; + Some(pixels) + } + + fn render_with_uniforms_and_adapter( + uniforms: NotchUniforms, + ) -> Option<(Vec, wgpu::AdapterInfo)> { + let (device, queue, adapter_info) = device()?; let mut layer = NotchLayer::new(&device, Arc::new(CompositeVideoFramePipeline::new(&device))); @@ -277,11 +287,37 @@ mod tests { let pixels = readback.slice(..).get_mapped_range().to_vec(); readback.unmap(); - Some(pixels) + Some((pixels, adapter_info)) } - fn render() -> Option> { - render_with_uniforms(uniforms()) + /// Renders on the available adapter, but returns None (skip) when a + /// known-broken software rasterizer produced output that fails the basic + /// sanity of "the clear executed and the layer drew something". Hosted CI + /// GPU stacks break underneath us (the windows-2022 WARP adapter stopped + /// compositing correctly with a runner image update, with no repo + /// change). The escape hatch is limited by NAME to the WARP family: on + /// real hardware and on Ubuntu's lavapipe (DeviceType::Cpu but renders + /// correctly) the shape assertions run at full strength, so a real + /// regression that draws nothing still fails on at least two CI legs + /// instead of skipping everywhere. + fn render_or_skip_broken_software_adapter() -> Option> { + let (pixels, adapter_info) = render_with_uniforms_and_adapter(uniforms())?; + + let is_known_broken_adapter = adapter_info.name.contains("Basic Render Driver") + || adapter_info.name.to_lowercase().contains("warp"); + let cleared_to_white = is_white(pixel(&pixels, 2, OUTPUT - 2)); + let drew_anything = (0..OUTPUT).any(|y| black_run(&pixels, y) > 0); + + if is_known_broken_adapter && !(cleared_to_white && drew_anything) { + eprintln!( + "software adapter '{}' cannot composite this pass (cleared={cleared_to_white}, \ + drew={drew_anything}), skipping", + adapter_info.name + ); + return None; + } + + Some(pixels) } fn pixel(pixels: &[u8], x: u32, y: u32) -> [u8; 4] { @@ -306,8 +342,8 @@ mod tests { #[test] fn draws_an_opaque_notch_that_flares_at_the_top() { - let Some(pixels) = render() else { - eprintln!("no wgpu adapter available, skipping"); + let Some(pixels) = render_or_skip_broken_software_adapter() else { + eprintln!("no usable wgpu adapter available, skipping"); return; }; @@ -348,7 +384,7 @@ mod tests { #[test] fn source_crop_preserves_the_uncropped_shape() { - let Some(full) = render() else { + let Some(full) = render_or_skip_broken_software_adapter() else { return; }; let mut cropped_uniforms = uniforms(); @@ -376,7 +412,7 @@ mod tests { #[test] fn draws_nothing_when_there_is_no_notch() { - let Some((device, queue)) = device() else { + let Some((device, queue, _)) = device() else { return; }; @@ -395,7 +431,7 @@ mod tests { /// Zoom scales the texture; it must not re-rasterize per frame. #[test] fn reuses_the_texture_while_the_unzoomed_size_holds() { - let Some((device, queue)) = device() else { + let Some((device, queue, _)) = device() else { return; };