diff --git a/apps/desktop-gpui/src/main_window.rs b/apps/desktop-gpui/src/main_window.rs index 36774a8806..d93088dbe4 100644 --- a/apps/desktop-gpui/src/main_window.rs +++ b/apps/desktop-gpui/src/main_window.rs @@ -1679,6 +1679,11 @@ impl MainWindow { this.incomplete_recording = incomplete.into_iter().next(); if this.incomplete_recording.is_none() { this.recovery_error = None; + } else if std::env::var("CAP_GPUI_AUTO_RECOVER").is_ok_and(|value| value == "1") { + // The crash harness presses Recover, since unprivileged + // synthetic clicks are dropped (as for every + // `CAP_GPUI_AUTO_*` hook). + this.process_incomplete_recording(true, window, cx); } cx.notify(); window.refresh(); diff --git a/apps/desktop-gpui/src/recording.rs b/apps/desktop-gpui/src/recording.rs index 7c91d53e27..1cf1b8a6ec 100644 --- a/apps/desktop-gpui/src/recording.rs +++ b/apps/desktop-gpui/src/recording.rs @@ -1411,7 +1411,7 @@ pub(crate) fn ensure_finalization_storage(project_path: &std::path::Path) -> any } pub fn recover_instant_recording(project_path: &std::path::Path) -> anyhow::Result { - use cap_project::{InstantRecordingMeta, RecordingMeta, RecordingMetaInner}; + use cap_project::{InstantRecordingMeta, RecordingMeta, RecordingMetaInner, UploadMeta}; let mut meta = RecordingMeta::load_for_project(project_path) .map_err(|error| anyhow!("loading recording metadata: {error}"))?; @@ -1450,6 +1450,9 @@ pub fn recover_instant_recording(project_path: &std::path::Path) -> anyhow::Resu .map(|decoder| decoder.rate()) }) .transpose()?; + if matches!(meta.upload, Some(UploadMeta::SegmentUpload { .. })) { + seal_recovered_instant_tracks(project_path); + } meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::Complete { fps: fps.round() as u32, sample_rate, @@ -1459,6 +1462,46 @@ pub fn recover_instant_recording(project_path: &std::path::Path) -> anyhow::Resu Ok(project_path.to_path_buf()) } +/// Seals the tracks of a recovered Instant recording so the upload queue can +/// send what the crash left: the crash leaves each manifest unfinished, and the +/// upload resume only accepts finished ones. Recovery stands if sealing fails; +/// the recording is saved locally either way, and its upload then fails as it +/// did before sealing existed. +/// +/// Recovery can rescue a final fragment the manifest never listed, so the +/// recovered `output.mp4` may hold more than the sealed tracks. Whenever +/// sealing sets any media aside, or fails, the local recording is kept after +/// its upload verifies, so it is never the only complete copy that gets +/// deleted. +pub(crate) fn seal_recovered_instant_tracks(project_path: &std::path::Path) { + use cap_enc_ffmpeg::segmented_stream::SegmentMediaType; + let mut omitted = false; + for (track, media_type) in [ + ("display", SegmentMediaType::Video), + ("audio", SegmentMediaType::Audio), + ] { + let directory = project_path.join("content").join(track); + if !directory.join("manifest.json").exists() { + continue; + } + match cap_recording::upload_resume::seal_interrupted_track(&directory, media_type) { + Ok(segments) => tracing::info!(track, segments, "Sealed a recovered Instant track"), + Err(error) => { + omitted = true; + tracing::warn!(track, %error, "Could not seal a recovered Instant track") + } + } + omitted |= std::fs::read_dir(&directory).map_or(true, |entries| { + entries + .flatten() + .any(|entry| entry.file_name().to_string_lossy().ends_with(".unsealed")) + }); + } + if omitted && let Err(error) = crate::upload::queue::record_preserve_local(project_path) { + tracing::warn!(%error, "Could not keep a recovered recording after its upload"); + } +} + /// The first segment's display track, which is what /// `handle_recording_finish` hands `create_screenshot` /// (`apps/desktop/src-tauri/src/recording.rs:3415-3429`). @@ -2810,6 +2853,117 @@ mod tests { #[cfg(target_os = "linux")] use std::time::Duration; + fn copy_tree(from: &std::path::Path, to: &std::path::Path) { + std::fs::create_dir_all(to).unwrap(); + for entry in std::fs::read_dir(from).unwrap() { + let entry = entry.unwrap(); + let target = to.join(entry.file_name()); + if entry.file_type().unwrap().is_dir() { + copy_tree(&entry.path(), &target); + } else { + std::fs::copy(entry.path(), target).unwrap(); + } + } + } + + /// A real Instant recording copied mid-capture, which is what a crash + /// leaves, then recovered: Recover saves it locally and seals its track, + /// so the upload resume accepts every segment the crash saved. + #[test] + fn recovering_a_crashed_instant_recording_seals_it_for_upload() { + use cap_enc_ffmpeg::segmented_stream::{ + SegmentMediaType, SegmentedVideoEncoder, SegmentedVideoEncoderConfig, + }; + use cap_project::{ + InstantRecordingMeta, RecordingMeta, RecordingMetaInner, S3UploadMeta, UploadMeta, + VideoUploadInfo, + }; + ffmpeg::init().unwrap(); + let live = tempfile::tempdir().unwrap(); + let display = live.path().join("display"); + let info = cap_media_info::VideoInfo { + pixel_format: cap_media_info::Pixel::NV12, + width: 64, + height: 64, + time_base: ffmpeg::Rational(1, 1_000_000), + frame_rate: ffmpeg::Rational(30, 1), + }; + let mut encoder = SegmentedVideoEncoder::init( + display.clone(), + info, + SegmentedVideoEncoderConfig { + segment_duration: std::time::Duration::from_secs(1), + ..Default::default() + }, + ) + .unwrap(); + for index in 0..300u64 { + let mut frame = ffmpeg::frame::Video::new(ffmpeg::format::Pixel::NV12, 64, 64); + for plane in 0..frame.planes() { + frame.data_mut(plane).fill((index % 200) as u8); + } + encoder + .queue_frame( + frame, + std::time::Duration::from_nanos(index * 1_000_000_000 / 30), + ) + .unwrap(); + } + let root = tempfile::tempdir().unwrap(); + let project = root.path().join("crashed.cap"); + copy_tree(&display, &project.join("content/display")); + encoder.finish().unwrap(); + + let manifest: Value = serde_json::from_slice( + &std::fs::read(project.join("content/display/manifest.json")).unwrap(), + ) + .unwrap(); + assert_eq!(manifest["is_complete"], Value::Bool(false)); + let finished = manifest["segments"] + .as_array() + .unwrap() + .iter() + .filter(|segment| segment["is_complete"] == Value::Bool(true)) + .count(); + assert!(finished >= 1, "the crash saved no finished segment"); + assert!( + cap_recording::upload_resume::collect_segment_events(&project, false).is_err(), + "a crashed recording must not be resumable before Recover" + ); + let video = VideoUploadInfo { + id: "crashedvideo01".into(), + link: "https://cap.test/s/crashedvideo01".into(), + config: S3UploadMeta { + id: "crashedvideo01".into(), + }, + }; + RecordingMeta { + platform: None, + project_path: project.clone(), + pretty_name: "Crashed".into(), + sharing: None, + inner: RecordingMetaInner::Instant(InstantRecordingMeta::InProgress { + recording: true, + }), + upload: Some(UploadMeta::SegmentUpload { + video_id: video.id.clone(), + pre_created_video: video, + recording_dir: project.clone(), + }), + } + .save_for_project() + .unwrap(); + + recover_instant_recording(&project).unwrap(); + assert!(project.join("content/output.mp4").is_file()); + let events = cap_recording::upload_resume::collect_segment_events(&project, false).unwrap(); + let segments = events + .iter() + .filter(|event| !event.is_init && event.media_type == SegmentMediaType::Video) + .count(); + assert_eq!(segments, finished); + } + #[cfg(target_os = "linux")] #[test] fn instant_screen_requested_camera_cannot_bypass_processing_preparation() { diff --git a/apps/desktop-gpui/src/upload.rs b/apps/desktop-gpui/src/upload.rs index d952cb86b0..53878a1977 100644 --- a/apps/desktop-gpui/src/upload.rs +++ b/apps/desktop-gpui/src/upload.rs @@ -2746,6 +2746,58 @@ mod tests { } } + /// The finished segment manifest this app uploads, checked against the + /// files the web server's own parser is tested with + /// (`apps/web/__tests__/unit/desktop-segments-manifest-contract.test.ts`), + /// so neither side can change the format without the other noticing. Run + /// with `CAP_UPDATE_CONTRACT_FIXTURES=1` to rewrite the files after an + /// intended change. + #[test] + fn finished_segment_manifests_match_the_server_contract_fixtures() { + let fixtures = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../web/__tests__/fixtures/desktop-segments"); + for (name, audio) in [("with-audio", 2), ("video-only", 0)] { + let mut manifest = SegmentUploadManifest::default(); + manifest.record(&segment_event(0, 0.0, true, SegmentMediaType::Video)); + for (index, duration) in [(1, 2.0), (2, 2.0), (3, 1.25)] { + manifest.record(&segment_event( + index, + duration, + false, + SegmentMediaType::Video, + )); + } + if audio > 0 { + manifest.record(&segment_event(0, 0.0, true, SegmentMediaType::Audio)); + for index in 1..=audio { + manifest.record(&segment_event(index, 3.0, false, SegmentMediaType::Audio)); + } + } + manifest.is_complete = true; + let uploaded = serde_json::to_value(&manifest).unwrap(); + let path = fixtures.join(format!("finished-manifest-{name}.json")); + if std::env::var_os("CAP_UPDATE_CONTRACT_FIXTURES").is_some() { + std::fs::create_dir_all(&fixtures).unwrap(); + let mut json = Vec::new(); + let formatter = serde_json::ser::PrettyFormatter::with_indent(b"\t"); + let mut serializer = serde_json::Serializer::with_formatter(&mut json, formatter); + serde::Serialize::serialize(&uploaded, &mut serializer).unwrap(); + json.push(b'\n'); + std::fs::write(&path, json).unwrap(); + } + let fixture: serde_json::Value = serde_json::from_slice( + &std::fs::read(&path).unwrap_or_else(|error| panic!("{}: {error}", path.display())), + ) + .unwrap(); + assert_eq!( + fixture, + uploaded, + "{} no longer matches what the app uploads", + path.display() + ); + } + } + #[test] fn segment_manifest_requires_init_and_orders_video_and_audio() { let mut manifest = SegmentUploadManifest::default(); diff --git a/apps/desktop-gpui/src/upload/instant_cleanup.rs b/apps/desktop-gpui/src/upload/instant_cleanup.rs index 738f09c484..1e1fcd71ea 100644 --- a/apps/desktop-gpui/src/upload/instant_cleanup.rs +++ b/apps/desktop-gpui/src/upload/instant_cleanup.rs @@ -11,9 +11,12 @@ const RETRY_MAX_SECS: i64 = 6 * 60 * 60; const INTERRUPTED_GRACE_SECS: i64 = 30 * 60; const SWEEP_INTERVAL: Duration = Duration::from_secs(5 * 60); const MEDIA_SCAN_DEPTH: usize = 3; +const CRASHED_ADOPTION_AGE_SECS: i64 = 10 * 60; +const CRASHED_ERROR: &str = "Cap closed before this recording captured anything."; static JOURNAL_LOCK: Mutex<()> = Mutex::new(()); static SESSION: OnceLock = OnceLock::new(); +static LAUNCHED_AT: OnceLock = OnceLock::new(); #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "kebab-case")] @@ -145,6 +148,10 @@ fn session_id() -> &'static str { SESSION.get_or_init(crate::store::new_uuid_v4) } +fn launched_at() -> i64 { + *LAUNCHED_AT.get_or_init(now) +} + fn journal_path() -> PathBuf { crate::store::app_data_dir().join(JOURNAL_FILE) } @@ -428,7 +435,8 @@ fn meta_claims(meta: &RecordingMeta, video_id: &str) -> bool { /// True when the folder holds recorded media, or when part of it cannot be /// read: media that cannot be checked counts as present. Only a missing folder -/// counts as empty. +/// counts as empty. A segment still named `.m4s.tmp` counts too, since a crash +/// can leave a complete fragment under that name and recovery can rescue it. fn has_recorded_media(directory: &Path, depth: usize) -> bool { let entries = match std::fs::read_dir(directory) { Ok(entries) => entries, @@ -453,7 +461,9 @@ fn has_recorded_media(directory: &Path, depth: usize) -> bool { .and_then(|name| name.to_str()) .unwrap_or_default() .to_ascii_lowercase(); - name.ends_with(".m4s") || (name.ends_with(".mp4") && !name.starts_with("init")) + let segment = + name.ends_with(".m4s") || name.ends_with(".m4s.tmp") || name.ends_with(".m4s.unsealed"); + segment || (name.ends_with(".mp4") && !name.starts_with("init")) }) } @@ -522,10 +532,133 @@ fn locally_protected( false } +/// An Instant recording that crashed after capture began but before it saved +/// any media leaves a server video nothing can fill: the journal stops +/// tracking a recording once it starts, and there is nothing to recover. This +/// journals such a recording as interrupted so the sweep deletes that video, +/// but only when it is not live anywhere (its upload lock is free), its upload +/// state was saved before this launch and over ten minutes ago, it is still in +/// progress, and its folder holds no media. The sweep's own checks then still +/// apply before anything is deleted. +fn adopt_crashed_at(path: &Path, recording_dirs: &[PathBuf], launched_at: i64, now: i64) -> usize { + let mut found = Vec::new(); + for directory in recording_dirs { + let Ok(entries) = std::fs::read_dir(directory) else { + continue; + }; + for entry in entries.flatten() { + if let Some(crashed) = crashed_without_media(&entry.path(), launched_at, now) { + found.push(crashed); + } + } + } + if found.is_empty() { + return 0; + } + let mut adopted = 0; + if let Err(error) = update_journal(path, |entries| { + for crashed in found { + if entries + .iter() + .all(|entry| entry.video_id != crashed.video_id) + { + entries.push(crashed); + adopted += 1; + } + } + }) { + tracing::warn!(%error, "Instant cleanup could not record crashed recordings"); + return 0; + } + adopted +} + +fn crashed_without_media(project: &Path, launched_at: i64, now: i64) -> Option { + let state = queue::read_state(project).ok().flatten()?; + if !matches!( + state.phase, + queue::UploadPhase::Recording + | queue::UploadPhase::Retrying + | queue::UploadPhase::Failed + | queue::UploadPhase::NeedsAuthentication + ) { + return None; + } + let owner_id = state.owner_id()?.to_string(); + let saved = std::fs::symlink_metadata(project.join(queue::STATE_FILE)) + .and_then(|metadata| metadata.modified()) + .ok()? + .duration_since(std::time::UNIX_EPOCH) + .ok()? + .as_secs() as i64; + if saved >= launched_at || now.saturating_sub(saved) < CRASHED_ADOPTION_AGE_SECS { + return None; + } + let meta = RecordingMeta::load_for_project(project).ok()?; + if !matches!( + meta.inner, + RecordingMetaInner::Instant(InstantRecordingMeta::InProgress { .. }) + ) || !meta_claims(&meta, state.video_id()) + { + return None; + } + let _ownership = cap_recording::upload_resume::UploadLock::acquire(project).ok()?; + if has_recorded_media(&project.join("content"), MEDIA_SCAN_DEPTH) { + return None; + } + Some(Entry { + video_id: state.video_id().to_string(), + server_url: state.server_url().to_string(), + owner_id, + session: session_id().to_string(), + stage: Stage::Abandoned, + reason: Some(CleanupReason::Interrupted), + cause: None, + created_at: now, + attempts: 0, + next_attempt_at: now, + project_dir: Some(project.to_path_buf()), + }) +} + +/// After the sweep deletes the server video of an interrupted recording, +/// marks the local copy failed so the library stops showing it in progress. +/// Only a recording that still claims the video, is still in progress and +/// still holds no media is changed, and nothing is deleted. +fn mark_crashed_recording_failed(project: &Path, video_id: &str) { + let Ok(_ownership) = cap_recording::upload_resume::UploadLock::acquire(project) else { + return; + }; + let Ok(mut meta) = RecordingMeta::load_for_project(project) else { + return; + }; + if !matches!( + meta.inner, + RecordingMetaInner::Instant(InstantRecordingMeta::InProgress { .. }) + ) || !meta_claims(&meta, video_id) + || has_recorded_media(&project.join("content"), MEDIA_SCAN_DEPTH) + { + return; + } + meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::Failed { + error: CRASHED_ERROR.to_string(), + }); + if let Err(error) = meta.save_for_project() { + tracing::warn!(%error, video_id, "Instant cleanup could not mark a crashed recording failed"); + return; + } + if let Err(error) = queue::record_cancelled(project) { + tracing::warn!(%error, video_id, "Instant cleanup could not close a crashed upload"); + } +} + trait SweepBackend { fn account_ready(&self, entry: &Entry) -> bool; fn protected(&self, entry: &Entry) -> impl Future + Send; fn delete(&self, entry: &Entry) -> impl Future> + Send; + fn deleted(&self, _entry: &Entry) -> impl Future + Send { + async {} + } } #[derive(Debug, Default, PartialEq)] @@ -571,6 +704,7 @@ async fn sweep(path: &Path, session: &str, now: i64, backend: &impl SweepBackend } match backend.delete(&entry).await { Ok(()) => { + backend.deleted(&entry).await; summary.deleted += 1; remove.push(entry.video_id); } @@ -633,10 +767,39 @@ impl SweepBackend for LiveSweep { )) .await } + + async fn deleted(&self, entry: &Entry) { + if entry.effective_reason() != CleanupReason::Interrupted { + return; + } + let Some(project) = entry.project_dir.clone() else { + return; + }; + let video_id = entry.video_id.clone(); + let _ = + tokio::task::spawn_blocking(move || mark_crashed_recording_failed(&project, &video_id)) + .await; + } } async fn run() { loop { + let adopted = tokio::task::spawn_blocking(|| { + adopt_crashed_at( + &journal_path(), + &crate::library::known_recordings_dirs(), + launched_at(), + now(), + ) + }) + .await + .unwrap_or(0); + if adopted > 0 { + tracing::info!( + adopted, + "Instant cleanup found recordings that crashed before capturing" + ); + } let summary = sweep(&journal_path(), session_id(), now(), &LiveSweep).await; if summary != SweepSummary::default() { tracing::info!(?summary, "Instant cleanup sweep finished"); @@ -650,6 +813,7 @@ pub(crate) fn init(cx: &mut gpui::App) { return; } session_id(); + launched_at(); gpui_tokio::Tokio::spawn(cx, run()).detach(); } @@ -1211,6 +1375,229 @@ mod tests { assert!(!normal); } + /// A project left the way a crash during capture leaves it: metadata in + /// progress, an upload state saved `saved_secs_ago`, and no media. + fn crashed_project( + recordings: &Path, + name: &str, + phase: queue::UploadPhase, + saved_secs_ago: u64, + ) -> PathBuf { + let project = write_project(recordings, name, "video", in_progress()); + queue::record_capture( + &project, + &VideoUploadInfo { + id: "video".into(), + link: "https://cap.test/s/video".into(), + config: S3UploadMeta { id: "video".into() }, + }, + true, + true, + ) + .unwrap(); + let path = project.join(queue::STATE_FILE); + let mut state: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + state["owner_id"] = serde_json::json!("owner"); + state["server_url"] = serde_json::json!("https://cap.test"); + state["phase"] = serde_json::to_value(phase).unwrap(); + std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap(); + std::fs::File::options() + .write(true) + .open(&path) + .unwrap() + .set_modified(std::time::SystemTime::now() - Duration::from_secs(saved_secs_ago)) + .unwrap(); + project + } + + #[test] + fn a_recording_that_crashed_before_capturing_is_adopted_with_its_own_identity() { + let fixture = Fixture::new(); + let recordings = fixture.recordings(); + let project = crashed_project( + &recordings, + "crashed.cap", + queue::UploadPhase::Recording, + 3600, + ); + let now = now(); + let dirs = std::slice::from_ref(&recordings); + assert_eq!(adopt_crashed_at(&fixture.journal(), dirs, now, now), 1); + let entries = fixture.entries(); + assert_eq!(entries.len(), 1); + let entry = &entries[0]; + assert_eq!(entry.video_id, "video"); + assert_eq!(entry.server_url, "https://cap.test"); + assert_eq!(entry.owner_id, "owner"); + assert_eq!(entry.stage, Stage::Abandoned); + assert_eq!(entry.reason, Some(CleanupReason::Interrupted)); + assert_eq!(entry.project_dir.as_deref(), Some(project.as_path())); + assert!(entry.due("another-session", now)); + assert!(!locally_protected(dirs, "video", true, Some(&project))); + assert_eq!(adopt_crashed_at(&fixture.journal(), dirs, now, now), 0); + assert_eq!(fixture.entries().len(), 1); + } + + #[test] + fn a_crashed_recording_is_left_alone_unless_it_is_certainly_abandoned() { + use queue::UploadPhase; + let now = now(); + type Prepare = fn(&Path) -> Option; + let cases: Vec<(&str, UploadPhase, u64, i64, Prepare)> = vec![ + ("saved recently", UploadPhase::Recording, 300, now, |_| None), + ( + "saved after this launch", + UploadPhase::Recording, + 3600, + now - 7200, + |_| None, + ), + ( + "live elsewhere", + UploadPhase::Recording, + 3600, + now, + |project| Some(cap_recording::upload_resume::UploadLock::acquire(project).unwrap()), + ), + ("has media", UploadPhase::Recording, 3600, now, |project| { + std::fs::write(project.join("content/display/segment_001.m4s"), b"media").unwrap(); + None + }), + ( + "has a pending segment", + UploadPhase::Recording, + 3600, + now, + |project| { + std::fs::write( + project.join("content/display/segment_001.m4s.tmp"), + b"media", + ) + .unwrap(); + None + }, + ), + ("finished", UploadPhase::Recording, 3600, now, |project| { + let mut meta = RecordingMeta::load_for_project(project).unwrap(); + meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::Complete { + fps: 30, + sample_rate: None, + }); + meta.save_for_project().unwrap(); + None + }), + ( + "another video", + UploadPhase::Recording, + 3600, + now, + |project| { + write_project( + project.parent().unwrap(), + "case.cap", + "other", + in_progress(), + ); + None + }, + ), + ("no owner", UploadPhase::Recording, 3600, now, |project| { + let path = project.join(queue::STATE_FILE); + let mut state: Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + state["owner_id"] = Value::Null; + std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap(); + std::fs::File::options() + .write(true) + .open(&path) + .unwrap() + .set_modified(std::time::SystemTime::now() - Duration::from_secs(3600)) + .unwrap(); + None + }), + ("uploading", UploadPhase::Pending, 3600, now, |_| None), + ("verified", UploadPhase::Verified, 3600, now, |_| None), + ("cancelled", UploadPhase::Cancelled, 3600, now, |_| None), + ]; + for (case, phase, saved_secs_ago, launched, prepare) in cases { + let fixture = Fixture::new(); + let recordings = fixture.recordings(); + let project = crashed_project(&recordings, "case.cap", phase, saved_secs_ago); + let _held = prepare(&project); + assert_eq!( + adopt_crashed_at( + &fixture.journal(), + std::slice::from_ref(&recordings), + launched, + now + ), + 0, + "{case}" + ); + assert!(fixture.entries().is_empty(), "{case}"); + } + } + + #[test] + fn a_crashed_recording_is_marked_failed_once_its_video_is_deleted() { + let fixture = Fixture::new(); + let recordings = fixture.recordings(); + let project = crashed_project( + &recordings, + "crashed.cap", + queue::UploadPhase::Recording, + 3600, + ); + mark_crashed_recording_failed(&project, "video"); + let meta = RecordingMeta::load_for_project(&project).unwrap(); + assert!(matches!( + meta.inner, + RecordingMetaInner::Instant(InstantRecordingMeta::Failed { ref error }) if error == CRASHED_ERROR + )); + assert_eq!( + queue::read_state(&project).unwrap().unwrap().phase, + queue::UploadPhase::Cancelled + ); + + for (case, prepare) in [ + ( + "has media", + (|project: &Path| { + std::fs::write(project.join("content/display/segment_001.m4s"), b"media") + .unwrap() + }) as fn(&Path), + ), + ("has a pending segment", |project: &Path| { + std::fs::write( + project.join("content/display/segment_001.m4s.tmp"), + b"media", + ) + .unwrap() + }), + ("another video", |project: &Path| { + let name = project.file_name().unwrap().to_str().unwrap(); + write_project(project.parent().unwrap(), name, "other", in_progress()); + }), + ] { + let name = format!("{}.cap", case.replace(' ', "-")); + let project = crashed_project(&recordings, &name, queue::UploadPhase::Recording, 3600); + prepare(&project); + mark_crashed_recording_failed(&project, "video"); + assert!( + matches!( + RecordingMeta::load_for_project(&project).unwrap().inner, + RecordingMetaInner::Instant(InstantRecordingMeta::InProgress { .. }) + ), + "{case}" + ); + assert_eq!( + queue::read_state(&project).unwrap().unwrap().phase, + queue::UploadPhase::Recording, + "{case}" + ); + } + } + #[test] fn an_unreadable_project_that_claims_the_video_protects_it() { let fixture = Fixture::new(); diff --git a/apps/desktop-gpui/src/upload/queue.rs b/apps/desktop-gpui/src/upload/queue.rs index 9d8495c1e4..ca95811122 100644 --- a/apps/desktop-gpui/src/upload/queue.rs +++ b/apps/desktop-gpui/src/upload/queue.rs @@ -9,7 +9,7 @@ use std::time::SystemTime; static STATE_CHANGED: tokio::sync::Notify = tokio::sync::Notify::const_new(); -const STATE_FILE: &str = "instant-upload.json"; +pub(super) const STATE_FILE: &str = "instant-upload.json"; const MAX_STATE_BYTES: u64 = 64 * 1024; const CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(45); const RECONCILE_INTERVAL: Duration = Duration::from_secs(30); @@ -59,6 +59,14 @@ impl UploadState { &self.video_id } + pub(crate) fn server_url(&self) -> &str { + &self.server_url + } + + pub(crate) fn owner_id(&self) -> Option<&str> { + self.owner_id.as_deref() + } + pub(super) fn request_context(&self) -> Result { auth::UploadRequestContext::for_upload(&self.server_url, self.owner_id.as_deref()) .map_err(|error| error.to_string()) @@ -276,6 +284,16 @@ pub(super) fn record_cancelled(project: &Path) -> Result<(), String> { Ok(()) } +/// Keeps the local recording after its upload verifies, whatever the +/// delete-after-upload setting says. +pub(crate) fn record_preserve_local(project: &Path) -> Result<(), String> { + if let Some(mut state) = read_state(project)? { + state.preserve_local = true; + write_state(project, &state)?; + } + Ok(()) +} + pub(crate) fn record_failure(project: &Path, error: &str) -> Result<(), String> { if let Some(mut state) = read_state(project)? { state.fail(error.to_string(), now()); diff --git a/apps/desktop-gpui/src/upload/queue/verify_resume_tests.rs b/apps/desktop-gpui/src/upload/queue/verify_resume_tests.rs index 6b07395236..2030bf8d9f 100644 --- a/apps/desktop-gpui/src/upload/queue/verify_resume_tests.rs +++ b/apps/desktop-gpui/src/upload/queue/verify_resume_tests.rs @@ -175,6 +175,132 @@ fn saved_instant_upload_resumes_after_quit_mid_transfer() { }); } +/// Leaves a saved project the way a crash mid-recording does: both manifests +/// unfinished with a partial next segment on disk, the metadata still in +/// progress, and the upload state still `Recording`. +fn crash_mid_recording(project: &Path) -> Vec<(PathBuf, Vec)> { + let mut partials = Vec::new(); + for track in ["display", "audio"] { + let directory = project.join("content").join(track); + let path = directory.join("manifest.json"); + let mut manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap(); + let next = manifest["segments"].as_array().unwrap().len() + 1; + manifest["is_complete"] = serde_json::json!(false); + manifest["segments"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "path": format!("segment_{next:03}.m4s"), + "index": next, + "duration": 0.0, + "is_complete": false, + "file_size": null, + })); + std::fs::write(&path, serde_json::to_vec(&manifest).unwrap()).unwrap(); + let partial = directory.join(format!("segment_{next:03}.m4s.tmp")); + let bytes = format!("partial {track}").into_bytes(); + std::fs::write(&partial, &bytes).unwrap(); + partials.push((partial, bytes)); + } + let mut meta = RecordingMeta::load_for_project(project).unwrap(); + meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::InProgress { recording: true }); + meta.sharing = None; + meta.save_for_project().unwrap(); + let mut state = read_state(project).unwrap().unwrap(); + state.phase = UploadPhase::Recording; + write_state(project, &state).unwrap(); + partials +} + +/// What Recover leaves for the queue, without the local remux this mock +/// project cannot decode: sealed tracks and saved metadata. +fn recover(project: &Path) { + crate::recording::seal_recovered_instant_tracks(project); + let mut meta = RecordingMeta::load_for_project(project).unwrap(); + meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::Complete { + fps: 30, + sample_rate: Some(48_000), + }); + meta.save_for_project().unwrap(); +} + +/// The app crashes mid-recording and the user recovers the recording on the +/// next launch. The queue then uploads every segment the crash saved under +/// the same video id and verifies it. The unfinished tail is kept on disk, and +/// so is the whole recording, even with delete-after-upload on, because the +/// upload leaves that tail out. +#[test] +fn recovered_instant_recording_uploads_what_the_crash_saved() { + let case = "queue_recovered_instant_upload"; + let _shared_client = crate::auth::SHARED_API_CLIENT_TEST_LOCK.blocking_lock(); + let scratch = Scratch::new(case); + virtual_time(async { + let mock = MockCap::start(Vec::new()).await; + scratch.sign_in(&mock.url); + let store = scratch.0.join("store"); + let mut settings: serde_json::Value = + serde_json::from_slice(&std::fs::read(&store).unwrap()).unwrap(); + settings["general_settings"]["deleteInstantRecordingsAfterUpload"] = + serde_json::json!(true); + std::fs::write(&store, serde_json::to_vec(&settings).unwrap()).unwrap(); + let (project, video, audio) = saved_instant_project(&scratch.0, &mock.url); + let partials = crash_mid_recording(&project); + recover(&project); + + assert!( + read_state(&project).unwrap().unwrap().preserve_local, + "a recording whose sealed tracks leave media out must stay local" + ); + + let manager = Arc::new(Manager::new(LiveBackend, CONFIRMATION_TIMEOUT)); + launch(&manager, &project).await; + settle(&manager).await; + assert_verified(&mock, &project, &video, &audio); + assert_eq!(mock.confirmations().len(), 1); + assert_eq!(mock.deletes(), 0); + for (partial, bytes) in partials { + let aside = partial.with_file_name(format!( + "{}.unsealed", + partial.file_name().unwrap().to_string_lossy() + )); + assert_eq!(std::fs::read(aside).unwrap(), bytes); + } + }); +} + +/// Without sealing, the same recovered recording can never upload: the +/// resume refuses its unfinished manifests and the cloud copy stays empty. +#[test] +fn unsealed_recovered_instant_recording_never_uploads() { + let case = "queue_unsealed_recovered_instant_upload"; + let _shared_client = crate::auth::SHARED_API_CLIENT_TEST_LOCK.blocking_lock(); + let scratch = Scratch::new(case); + virtual_time(async { + let mock = MockCap::start(Vec::new()).await; + scratch.sign_in(&mock.url); + let (project, _, _) = saved_instant_project(&scratch.0, &mock.url); + crash_mid_recording(&project); + let mut meta = RecordingMeta::load_for_project(&project).unwrap(); + meta.inner = RecordingMetaInner::Instant(InstantRecordingMeta::Complete { + fps: 30, + sample_rate: Some(48_000), + }); + meta.save_for_project().unwrap(); + + let manager = Arc::new(Manager::new(LiveBackend, CONFIRMATION_TIMEOUT)); + let ownership = UploadLock::acquire(&project).unwrap(); + let state = read_state(&project).unwrap().unwrap(); + let error = manager + .resume(project.clone(), state, ownership) + .await + .unwrap_err(); + assert!(error.contains("not complete"), "{error}"); + assert!(mock.confirmations().is_empty()); + assert_eq!(mock.stats().media_bytes, 0); + }); +} + /// The app quits while the queue waits for cloud verification. The relaunch /// confirms the already uploaded recording without sending it again. #[test] diff --git a/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-video-only.json b/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-video-only.json new file mode 100644 index 0000000000..703a23d030 --- /dev/null +++ b/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-video-only.json @@ -0,0 +1,21 @@ +{ + "version": 2, + "video_init_uploaded": true, + "audio_init_uploaded": false, + "video_segments": [ + { + "index": 1, + "duration": 2.0 + }, + { + "index": 2, + "duration": 2.0 + }, + { + "index": 3, + "duration": 1.25 + } + ], + "audio_segments": [], + "is_complete": true +} diff --git a/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-with-audio.json b/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-with-audio.json new file mode 100644 index 0000000000..8f95c70a4e --- /dev/null +++ b/apps/web/__tests__/fixtures/desktop-segments/finished-manifest-with-audio.json @@ -0,0 +1,30 @@ +{ + "version": 2, + "video_init_uploaded": true, + "audio_init_uploaded": true, + "video_segments": [ + { + "index": 1, + "duration": 2.0 + }, + { + "index": 2, + "duration": 2.0 + }, + { + "index": 3, + "duration": 1.25 + } + ], + "audio_segments": [ + { + "index": 1, + "duration": 3.0 + }, + { + "index": 2, + "duration": 3.0 + } + ], + "is_complete": true +} diff --git a/apps/web/__tests__/unit/desktop-segments-manifest-contract.test.ts b/apps/web/__tests__/unit/desktop-segments-manifest-contract.test.ts new file mode 100644 index 0000000000..9817ca7d3f --- /dev/null +++ b/apps/web/__tests__/unit/desktop-segments-manifest-contract.test.ts @@ -0,0 +1,38 @@ +import { Video } from "@cap/web-domain"; +import { Schema } from "effect"; +import { describe, expect, it } from "vitest"; +import { readCompletedRecordingManifest } from "@/lib/desktop-recording-verification"; +import videoOnly from "../fixtures/desktop-segments/finished-manifest-video-only.json"; +import withAudio from "../fixtures/desktop-segments/finished-manifest-with-audio.json"; + +// The desktop app's own test writes these files from the manifest it uploads +// (`finished_segment_manifests_match_the_server_contract_fixtures` in +// apps/desktop-gpui/src/upload.rs), so a format change on either side fails +// one of the two tests. +describe("desktop segment manifest contract", () => { + it.each([ + ["with audio", withAudio, true, [1, 2]], + ["video only", videoOnly, false, []], + ])( + "finalization accepts the finished manifest the app uploads (%s)", + (_, fixture, hasAudio, audioSegments) => { + const completed = readCompletedRecordingManifest(JSON.stringify(fixture)); + expect(completed.hasAudio).toBe(hasAudio); + expect(completed.videoSegments).toEqual([1, 2, 3]); + expect(completed.audioSegments).toEqual(audioSegments); + }, + ); + + it.each([ + ["with audio", withAudio], + ["video only", videoOnly], + ])( + "recovery reads the finished manifest the app uploads (%s)", + (_, fixture) => { + const manifest = Schema.decodeUnknownSync(Video.SegmentManifest)(fixture); + expect(manifest.is_complete).toBe(true); + expect(manifest.video_init_uploaded).toBe(true); + expect(manifest.video_segments).toHaveLength(3); + }, + ); +}); diff --git a/crates/recording/src/upload_resume.rs b/crates/recording/src/upload_resume.rs index c8bf118fca..27f4b8a1e1 100644 --- a/crates/recording/src/upload_resume.rs +++ b/crates/recording/src/upload_resume.rs @@ -332,6 +332,132 @@ fn collect_track( Ok(events) } +/// Seals the manifest of a track whose recording was interrupted, so its +/// upload can resume from what was saved. Keeps the run of segments from +/// `segment_001` that every check in `collect_track` accepts, renames the +/// unfinished tail and any other media the sealed manifest does not list to +/// `.unsealed` (never deleting it), and marks the manifest complete. +/// Returns how many segments it kept. A manifest that is already complete is +/// left as it is, so sealing twice changes nothing. +pub fn seal_interrupted_track( + directory: &Path, + media_type: SegmentMediaType, +) -> Result { + metadata(directory, true)?; + let manifest_path = directory.join("manifest.json"); + metadata(&manifest_path, false)?; + let bytes = fs::read(&manifest_path).map_err(|error| error.to_string())?; + let manifest: Manifest = serde_json::from_slice(&bytes) + .map_err(|error| format!("{}: {error}", manifest_path.display()))?; + let (version, kind, stream_kind) = match media_type { + SegmentMediaType::Video => (5, "m4s_segments", ffmpeg::media::Type::Video), + SegmentMediaType::Audio => (2, "m4s_audio_segments", ffmpeg::media::Type::Audio), + }; + if manifest.version != version || manifest.kind != kind || manifest.init_segment != "init.mp4" { + return Err(format!( + "{}: recording manifest is not supported", + manifest_path.display() + )); + } + if manifest.is_complete { + return Ok(manifest.segments.len()); + } + let init = directory.join(&manifest.init_segment); + metadata(&init, false)?; + let input = + ffmpeg::format::input(&init).map_err(|error| format!("{}: {error}", init.display()))?; + if input.streams().best(stream_kind).is_none() { + return Err(format!( + "{}: required initialization stream is missing", + init.display() + )); + } + drop(input); + + let mut raw: serde_json::Value = serde_json::from_slice(&bytes) + .map_err(|error| format!("{}: {error}", manifest_path.display()))?; + let mut entries: Vec<(Segment, serde_json::Value)> = manifest + .segments + .into_iter() + .zip( + raw["segments"] + .as_array() + .cloned() + .ok_or_else(|| format!("{}: segments missing", manifest_path.display()))?, + ) + .collect(); + entries.sort_by_key(|(segment, _)| segment.index); + let mut kept = Vec::new(); + let mut total_duration = 0.0; + for (position, (segment, entry)) in entries.into_iter().enumerate() { + let path = directory.join(&segment.path); + let usable = u64::from(segment.index) == position as u64 + 1 + && segment.path == format!("segment_{:03}.m4s", segment.index) + && segment.is_complete + && segment.duration.is_finite() + && segment.duration > 0.0 + && segment.file_size.is_some_and(|size| size > 0) + && metadata(&path, false).is_ok_and(|actual| Some(actual.len()) == segment.file_size) + && cap_enc_ffmpeg::fragmented_mp4::tail_is_complete(&path).unwrap_or(false); + if !usable { + break; + } + total_duration += segment.duration; + kept.push((segment.path, entry)); + } + if kept.is_empty() { + return Err(format!( + "{}: no complete segments to resume", + manifest_path.display() + )); + } + + let listed: HashSet<&str> = kept.iter().map(|(path, _)| path.as_str()).collect(); + let mut set_aside = Vec::new(); + for entry in fs::read_dir(directory).map_err(|error| error.to_string())? { + let entry = entry.map_err(|error| error.to_string())?; + let name = entry.file_name().to_string_lossy().into_owned(); + let media = name.ends_with(".m4s") || name.ends_with(".m4s.tmp"); + if !media || listed.contains(name.as_str()) { + continue; + } + if entry + .file_type() + .map_err(|error| error.to_string())? + .is_symlink() + { + return Err(format!( + "{}: linked recording input", + entry.path().display() + )); + } + set_aside.push(entry.path()); + } + for path in set_aside { + let target = path.with_file_name(format!( + "{}.unsealed", + path.file_name().unwrap_or_default().to_string_lossy() + )); + if target.exists() { + return Err(format!("{}: already set aside", target.display())); + } + fs::rename(&path, &target).map_err(|error| format!("{}: {error}", path.display()))?; + } + + let count = kept.len(); + raw["segments"] = serde_json::Value::Array(kept.into_iter().map(|(_, entry)| entry).collect()); + raw["total_duration"] = serde_json::json!(total_duration); + raw["is_complete"] = serde_json::Value::Bool(true); + let sealed = serde_json::to_vec(&raw).map_err(|error| error.to_string())?; + let temporary = directory.join(format!(".manifest-{}.sealing", uuid::Uuid::new_v4())); + fs::write(&temporary, sealed).map_err(|error| error.to_string())?; + if let Err(error) = fs::rename(&temporary, &manifest_path) { + let _ = fs::remove_file(&temporary); + return Err(format!("{}: {error}", manifest_path.display())); + } + Ok(count) +} + #[cfg(test)] mod tests { use super::*; @@ -703,6 +829,213 @@ mod tests { assert!(collect_segment_events(project.path(), true).is_err()); } + fn track_manifest(project: &Path, track: &str) -> PathBuf { + project.join("content").join(track).join("manifest.json") + } + + /// Lists copies of `segment_001` as further segments, so a track holds + /// `total` complete segments. + fn extend_track(project: &Path, track: &str, total: u32) { + let directory = project.join("content").join(track); + let path = track_manifest(project, track); + let mut manifest: serde_json::Value = + serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + let first = manifest["segments"][0].clone(); + let bytes = fs::read(directory.join("segment_001.m4s")).unwrap(); + let segments = manifest["segments"].as_array_mut().unwrap(); + segments.truncate(1); + for index in 2..=total { + let name = format!("segment_{index:03}.m4s"); + fs::write(directory.join(&name), &bytes).unwrap(); + let mut entry = first.clone(); + entry["index"] = serde_json::json!(index); + entry["path"] = serde_json::json!(name); + segments.push(entry); + } + fs::write(path, serde_json::to_vec(&manifest).unwrap()).unwrap(); + } + + /// Leaves a finished track the way a crash does: an in-progress manifest + /// whose next segment is unfinished, with that segment's partial bytes on + /// disk. Returns how many complete segments it lists, and the partial bytes. + fn interrupt_track(project: &Path, track: &str) -> (usize, Vec) { + let path = track_manifest(project, track); + let mut manifest: serde_json::Value = + serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + let count = manifest["segments"].as_array().unwrap().len(); + let next = count + 1; + manifest["is_complete"] = serde_json::json!(false); + manifest["total_duration"] = serde_json::Value::Null; + manifest["segments"] + .as_array_mut() + .unwrap() + .push(serde_json::json!({ + "path": format!("segment_{next:03}.m4s"), + "index": next, + "duration": 0.0, + "is_complete": false, + "file_size": null, + })); + fs::write(&path, serde_json::to_vec(&manifest).unwrap()).unwrap(); + let partial = b"partial fragment".to_vec(); + fs::write( + path.with_file_name(format!("segment_{next:03}.m4s.tmp")), + &partial, + ) + .unwrap(); + (count, partial) + } + + fn snapshot(directory: &Path) -> Vec<(String, Vec)> { + let mut files: Vec<_> = fs::read_dir(directory) + .unwrap() + .map(|entry| { + let path = entry.unwrap().path(); + ( + path.file_name().unwrap().to_string_lossy().into_owned(), + fs::read(path).unwrap(), + ) + }) + .collect(); + files.sort(); + files + } + + #[test] + fn an_interrupted_recording_resumes_from_its_saved_segments_once_sealed() { + let project = fixture(); + let mut tracks = Vec::new(); + for (track, media_type) in [ + ("display", SegmentMediaType::Video), + ("audio", SegmentMediaType::Audio), + ] { + extend_track(project.path(), track, 3); + tracks.push((track, media_type, interrupt_track(project.path(), track))); + } + assert!(collect_segment_events(project.path(), true).is_err()); + for (track, media_type, (count, partial)) in &tracks { + let directory = project.path().join("content").join(track); + assert_eq!(seal_interrupted_track(&directory, *media_type), Ok(*count)); + let next = count + 1; + assert_eq!( + &fs::read(directory.join(format!("segment_{next:03}.m4s.tmp.unsealed"))).unwrap(), + partial + ); + } + let events = collect_segment_events(project.path(), true).unwrap(); + for (_, media_type, (count, _)) in &tracks { + let segments = events + .iter() + .filter(|event| !event.is_init && event.media_type == *media_type) + .count(); + assert_eq!(segments, *count); + } + for (track, media_type, (count, _)) in &tracks { + let directory = project.path().join("content").join(track); + let before = snapshot(&directory); + assert_eq!(seal_interrupted_track(&directory, *media_type), Ok(*count)); + assert_eq!(snapshot(&directory), before); + } + } + + #[test] + fn sealing_keeps_only_segments_before_the_first_failed_check() { + for (damage, kept) in [ + ("truncate", 1), + ("incomplete", 1), + ("gap", 1), + ("missing", 1), + ] { + let project = fixture(); + extend_track(project.path(), "display", 4); + interrupt_track(project.path(), "display"); + let directory = project.path().join("content/display"); + match damage { + "truncate" => { + let path = directory.join("segment_002.m4s"); + let mut bytes = fs::read(&path).unwrap(); + bytes.pop(); + fs::write(path, bytes).unwrap(); + } + "incomplete" => edit_manifest(project.path(), |manifest| { + manifest["segments"][1]["is_complete"] = serde_json::json!(false) + }), + "gap" => edit_manifest(project.path(), |manifest| { + manifest["segments"].as_array_mut().unwrap().remove(1); + }), + _ => fs::remove_file(directory.join("segment_002.m4s")).unwrap(), + } + assert_eq!( + seal_interrupted_track(&directory, SegmentMediaType::Video), + Ok(kept), + "{damage}" + ); + for index in 2..=4 { + let original = directory.join(format!("segment_{index:03}.m4s")); + assert!(!original.exists(), "{damage}: {index}"); + let aside = original.with_file_name(format!("segment_{index:03}.m4s.unsealed")); + assert_eq!( + aside.exists(), + damage != "missing" || index != 2, + "{damage}: {index}" + ); + } + let events = collect_segment_events(project.path(), false).unwrap(); + let video = events + .iter() + .filter(|event| !event.is_init && event.media_type == SegmentMediaType::Video) + .count(); + assert_eq!(video, kept, "{damage}"); + } + } + + #[test] + fn sealing_changes_nothing_when_it_cannot_keep_a_segment() { + type Edit = fn(&mut serde_json::Value); + let cases: [(&str, Edit); 4] = [ + ("no complete segment", |manifest| { + manifest["segments"][0]["is_complete"] = serde_json::json!(false) + }), + ("unsupported version", |manifest| { + manifest["version"] = serde_json::json!(4) + }), + ("unsupported kind", |manifest| { + manifest["type"] = serde_json::json!("m4s_audio_segments") + }), + ("escaping init", |manifest| { + manifest["init_segment"] = serde_json::json!("../init.mp4") + }), + ]; + for (case, edit) in cases { + let project = fixture(); + interrupt_track(project.path(), "display"); + edit_manifest(project.path(), edit); + let directory = project.path().join("content/display"); + let before = snapshot(&directory); + assert!( + seal_interrupted_track(&directory, SegmentMediaType::Video).is_err(), + "{case}" + ); + assert_eq!(snapshot(&directory), before, "{case}"); + } + } + + #[cfg(unix)] + #[test] + fn sealing_refuses_linked_media_without_moving_anything() { + let project = fixture(); + extend_track(project.path(), "display", 2); + interrupt_track(project.path(), "display"); + let directory = project.path().join("content/display"); + let outside = project.path().join("outside.m4s"); + fs::write(&outside, b"outside").unwrap(); + std::os::unix::fs::symlink(&outside, directory.join("segment_009.m4s")).unwrap(); + let before = snapshot(&directory); + assert!(seal_interrupted_track(&directory, SegmentMediaType::Video).is_err()); + assert_eq!(snapshot(&directory), before); + assert_eq!(fs::read(outside).unwrap(), b"outside"); + } + #[cfg(unix)] #[test] fn linked_manifest_init_fragment_and_track_are_withheld() {