perf(cli): stream eligible MP4 export audio

apps/cli/src/export.rs

@@ -3,11 +3,11 @@ use std::{
     path::{Path, PathBuf},
     sync::{
         Arc, Mutex,
-        atomic::{AtomicU32, Ordering},
+        atomic::{AtomicBool, AtomicU32, Ordering},
     },
 };
 
-use cap_export::{ExporterBase, make_cursor_only_project};
+use cap_export::{ExporterBase, ExporterBuilder, Mp4ExporterBase, make_cursor_only_project};
 use cap_project::{RecordingMeta, RecordingMetaInner, XY};
 use clap::{Args, ValueEnum};
 use serde::{Deserialize, Serialize};
@@ -121,6 +121,24 @@ pub enum CliExportSettings {
 }
 
 impl CliExportSettings {
+    async fn prepare(self, builder: ExporterBuilder) -> Result<PreparedExport, String> {
+        match self {
+            Self::Mp4(settings) => builder
+                .build_for_mp4(Arc::new(AtomicBool::new(false)))
+                .await
+                .map(|base| PreparedExport::Mp4(settings, base)),
+            Self::Gif(settings) => builder
+                .build()
+                .await
+                .map(|base| PreparedExport::Gif(settings, base)),
+            Self::Mov(settings) => builder
+                .build()
+                .await
+                .map(|base| PreparedExport::Mov(settings, base)),
+        }
+        .map_err(|error| format!("Exporter build error: {error}"))
+    }
+
     fn fps(&self) -> u32 {
         match self {
             Self::Mp4(settings) => settings.fps,
@@ -152,6 +170,33 @@ impl CliExportSettings {
     }
 }
 
+enum PreparedExport {
+    Mp4(cap_export::mp4::Mp4ExportSettings, Mp4ExporterBase),
+    Gif(cap_export::gif::GifExportSettings, ExporterBase),
+    Mov(cap_export::mov::MovExportSettings, ExporterBase),
+}
+
+impl PreparedExport {
+    fn total_frames(&self, fps: u32) -> u32 {
+        match self {
+            Self::Mp4(_, base) => base.total_frames(fps),
+            Self::Gif(_, base) | Self::Mov(_, base) => base.total_frames(fps),
+        }
+    }
+
+    async fn export(
+        self,
+        on_progress: impl FnMut(u32) -> bool + Send + 'static,
+    ) -> Result<PathBuf, String> {
+        match self {
+            Self::Mp4(settings, base) => settings.export_prepared(base, on_progress).await,
+            Self::Gif(settings, base) => settings.export(base, on_progress).await,
+            Self::Mov(settings, base) => settings.export(base, on_progress).await,
+        }
+        .map_err(|error| format!("Exporter error: {error}"))
+    }
+}
+
 fn default_fps(format: ExportFormat) -> u32 {
     match format {
         ExportFormat::Mp4 | ExportFormat::Mov => 60,
@@ -530,12 +575,8 @@ impl Export {
             builder = builder.with_config(make_cursor_only_project(meta.project_config()));
         }
 
-        let exporter_base = builder
-            .build()
-            .await
-            .map_err(|v| format!("Exporter build error: {v}"))?;
-
-        let total_frames = exporter_base.total_frames(settings.fps());
+        let exporter = settings.prepare(builder).await?;
+        let total_frames = exporter.total_frames(settings.fps());
 
         if progress_json {
             emit_export_message(
@@ -568,12 +609,7 @@ impl Export {
             true
         };
 
-        let output_path = match settings {
-            CliExportSettings::Mp4(settings) => settings.export(exporter_base, on_progress).await,
-            CliExportSettings::Gif(settings) => settings.export(exporter_base, on_progress).await,
-            CliExportSettings::Mov(settings) => settings.export(exporter_base, on_progress).await,
-        }
-        .map_err(|v| format!("Exporter error: {v}"))?;
+        let output_path = exporter.export(on_progress).await?;
 
         // Defense in depth: an export that renders no frames writes an empty (~few hundred byte) file
         // but otherwise "succeeds". An agent must never silently get/upload that, so fail loudly and
@@ -874,26 +910,22 @@ pub async fn export_project_default(project_path: PathBuf) -> Result<PathBuf, St
         return prepare_instant_output(project_path).await;
     }
 
-    let exporter_base = ExporterBase::builder(project_path)
-        .with_force_ffmpeg_decoder(settings.force_ffmpeg_decoder())
-        .build()
-        .await
-        .map_err(|v| format!("Exporter build error: {v}"))?;
+    let exporter = settings
+        .prepare(
+            ExporterBase::builder(project_path)
+                .with_force_ffmpeg_decoder(settings.force_ffmpeg_decoder()),
+        )
+        .await?;
 
-    let total_frames = exporter_base.total_frames(settings.fps());
+    let total_frames = exporter.total_frames(settings.fps());
     let rendered = Arc::new(AtomicU32::new(0));
     let progress_rendered = Arc::clone(&rendered);
     let on_progress = move |frame_index: u32| {
         progress_rendered.store((frame_index + 1).min(total_frames), Ordering::Relaxed);
         true
     };
 
-    let output_path = match settings {
-        CliExportSettings::Mp4(settings) => settings.export(exporter_base, on_progress).await,
-        CliExportSettings::Gif(settings) => settings.export(exporter_base, on_progress).await,
-        CliExportSettings::Mov(settings) => settings.export(exporter_base, on_progress).await,
-    }
-    .map_err(|v| format!("Exporter error: {v}"))?;
+    let output_path = exporter.export(on_progress).await?;
 
     // Same 0-frame guard as Export::run_inner: a recording with missing media renders an empty,
     // unplayable file that otherwise "succeeds", and `cap upload --export` would sign + upload it and
@@ -1058,6 +1090,153 @@ mod tests {
         assert!(!project.path().join(".recovery.lock").exists());
     }
 
+    fn playable_studio_project() -> tempfile::TempDir {
+        let project = playable_instant_project();
+        std::fs::rename(
+            project.path().join("content/output.mp4"),
+            project.path().join("display.mp4"),
+        )
+        .unwrap();
+        let audio = std::process::Command::new("ffmpeg")
+            .args([
+                "-nostdin",
+                "-v",
+                "error",
+                "-f",
+                "lavfi",
+                "-i",
+                "sine=frequency=440:sample_rate=48000:duration=2",
+                "-ac",
+                "2",
+                "-c:a",
+                "aac",
+            ])
+            .arg(project.path().join("mic.m4a"))
+            .output()
+            .unwrap();
+        assert!(audio.status.success(), "{audio:?}");
+        std::fs::write(
+            project.path().join("recording-meta.json"),
+            serde_json::to_vec(&serde_json::json!({
+                "pretty_name": "CLI streaming test",
+                "segments": [{
+                    "display": {"path": "display.mp4", "fps": 30, "start_time": 0.0},
+                    "mic": {"path": "mic.m4a", "start_time": 0.0}
+                }]
+            }))
+            .unwrap(),
+        )
+        .unwrap();
+        cap_project::ProjectConfiguration {
+            timeline: Some(
+                serde_json::from_value(serde_json::json!({
+                    "segments": [{
+                        "recordingSegment": 0, "start": 0.0, "end": 1.0, "timescale": 1.0
+                    }],
+                    "zoomSegments": []
+                }))
+                .unwrap(),
+            ),
+            ..Default::default()
+        }
+        .write(project.path())
+        .unwrap();
+        project
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ffmpeg and a wgpu adapter"]
+    async fn studio_mp4_streams_and_existing_destinations_keep_eager_preparation() {
+        let project = playable_studio_project();
+        let output = project.path().join("render.mp4");
+        let settings = settings_from_flags(&ExportFlags {
+            fps: Some(30),
+            resolution: Some("64x64".into()),
+            force_ffmpeg_decoder: true,
+            ..Default::default()
+        })
+        .unwrap();
+        let builder = || {
+            ExporterBase::builder(project.path().to_path_buf())
+                .with_force_ffmpeg_decoder(true)
+                .with_output_path(output.clone())
+        };
+        let prepared = settings.prepare(builder()).await.unwrap();
+        assert!(matches!(&prepared, PreparedExport::Mp4(_, base) if base.uses_streaming_audio()));
+        assert_eq!(prepared.total_frames(30), 30);
+        assert_eq!(prepared.export(|_| true).await.unwrap(), output);
+        let original = std::fs::read(&output).unwrap();
+        let prepared = settings.prepare(builder()).await.unwrap();
+        assert!(matches!(&prepared, PreparedExport::Mp4(_, base) if !base.uses_streaming_audio()));
+        drop(prepared);
+        assert_eq!(std::fs::read(&output).unwrap(), original);
+
+        for format in [ExportFormat::Gif, ExportFormat::Mov] {
+            let settings = settings_from_flags(&ExportFlags {
+                format: Some(format),
+                ..Default::default()
+            })
+            .unwrap();
+            let prepared = settings.prepare(builder()).await.unwrap();
+            assert!(matches!(
+                prepared,
+                PreparedExport::Gif(_, _) | PreparedExport::Mov(_, _)
+            ));
+        }
+    }
+
+    #[tokio::test]
+    #[ignore = "requires ffmpeg, ffprobe and a wgpu adapter"]
+    async fn studio_mp4_rejects_damaged_audio_after_the_selected_video() {
+        let project = playable_studio_project();
+        let audio = project.path().join("mic.m4a");
+        let packets = std::process::Command::new("ffprobe")
+            .args([
+                "-v",
+                "error",
+                "-show_packets",
+                "-show_entries",
+                "packet=pts_time,pos,size",
+                "-of",
+                "json",
+            ])
+            .arg(&audio)
+            .output()
+            .unwrap();
+        assert!(packets.status.success());
+        let packets: serde_json::Value = serde_json::from_slice(&packets.stdout).unwrap();
+        let packet = packets["packets"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|packet| packet["pts_time"].as_str().unwrap().parse::<f64>().unwrap() > 1.5)
+            .unwrap();
+        let position: usize = packet["pos"].as_str().unwrap().parse().unwrap();
+        let size: usize = packet["size"].as_str().unwrap().parse().unwrap();
+        let mut bytes = std::fs::read(&audio).unwrap();
+        bytes[position..position + size].fill(0xff);
+        std::fs::write(audio, bytes).unwrap();
+
+        let output = project.path().join("render.mp4");
+        let prepared = settings_from_flags(&ExportFlags {
+            fps: Some(30),
+            resolution: Some("64x64".into()),
+            ..Default::default()
+        })
+        .unwrap()
+        .prepare(
+            ExporterBase::builder(project.path().to_path_buf())
+                .with_force_ffmpeg_decoder(true)
+                .with_output_path(output.clone()),
+        )
+        .await
+        .unwrap();
+        let error = prepared.export(|_| true).await.unwrap_err();
+        assert!(error.contains("mic.m4a"), "{error}");
+        assert!(error.contains("send-packet"), "{error}");
+        assert!(!output.exists());
+    }
+
     #[tokio::test]
     async fn instant_export_needing_repair_cannot_rewrite_an_owned_upload() {
         let project = playable_instant_project();

crates/export/examples/export-path-comparison.rs

@@ -0,0 +1,93 @@
+use cap_export::{
+    ExporterBase,
+    mp4::{ExportCompression, Mp4ExportSettings},
+};
+use cap_project::XY;
+use std::{
+    path::PathBuf,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, AtomicU64, Ordering},
+    },
+    time::Instant,
+};
+
+fn main() {
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .thread_stack_size(16 * 1024 * 1024)
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(run());
+}
+
+async fn run() {
+    let mut args = std::env::args_os().skip(1);
+    let mode = args.next().expect("eager or streaming");
+    assert!(mode == "eager" || mode == "streaming");
+    let root = PathBuf::from(args.next().expect("project path"));
+    let output = PathBuf::from(args.next().expect("unused output path"));
+    assert!(!output.exists());
+    let first_progress = Arc::new(AtomicU64::new(u64::MAX));
+    let recorded = first_progress.clone();
+    let start = Instant::now();
+    let on_progress = move |_| {
+        let _ = recorded.compare_exchange(
+            u64::MAX,
+            start.elapsed().as_micros() as u64,
+            Ordering::Relaxed,
+            Ordering::Relaxed,
+        );
+        true
+    };
+    let builder = ExporterBase::builder(root)
+        .with_output_path(output.clone())
+        .with_force_ffmpeg_decoder(true);
+    let settings = Mp4ExportSettings {
+        fps: 15,
+        resolution_base: XY::new(640, 360),
+        compression: ExportCompression::Web,
+        custom_bpp: None,
+        force_ffmpeg_decoder: true,
+        optimize_filesize: false,
+    };
+    let prepared_seconds;
+    let streaming;
+    if mode == "streaming" {
+        let base = builder
+            .build_for_mp4(Arc::new(AtomicBool::new(false)))
+            .await
+            .expect("prepare streaming");
+        prepared_seconds = start.elapsed().as_secs_f64();
+        streaming = base.uses_streaming_audio();
+        settings
+            .export_prepared(base, on_progress)
+            .await
+            .expect("export streaming");
+    } else {
+        let base = builder.build().await.expect("prepare eager");
+        prepared_seconds = start.elapsed().as_secs_f64();
+        streaming = false;
+        settings
+            .export(base, on_progress)
+            .await
+            .expect("export eager");
+    }
+    let elapsed = start.elapsed().as_secs_f64();
+    let maxrss_kib = std::fs::read_to_string("/proc/self/status")
+        .ok()
+        .and_then(|status| {
+            status.lines().find_map(|line| {
+                line.strip_prefix("VmHWM:")?
+                    .split_whitespace()
+                    .next()?
+                    .parse::<u64>()
+                    .ok()
+            })
+        });
+    println!(
+        "EXPORT_PATH {}",
+        serde_json::json!({"mode":mode.to_string_lossy(), "streaming_audio":streaming,"prepare_seconds":prepared_seconds,"total_seconds":elapsed,"first_progress_seconds":first_progress.load(Ordering::Relaxed) as f64 / 1_000_000.0,"maxrss_kib":maxrss_kib,"output":output})
+    );
+}