Encoder error handling improvements (#1305)

* more clippy stuff + handle encoder retries for enc-avfoundation

* don't save pr caches

* mostly fix windows

* audio input NotReadyForMore

* fix macos

* hard error on write_trailer

* fix lock error

* beter enc-avfoundation error handling and windows fixes

* macos

* use double result for muxer partial failure

* cleanup

* better error reporting

Author: Brendonovich

.github/workflows/ci.yml

@@ -103,6 +103,7 @@ jobs:
         uses: swatinem/rust-cache@v2
         with:
           shared-key: ${{ matrix.settings.target }}
+          save-if: ${{ github.ref == 'refs/heads/main' }}
 
       - name: Create .env file in root
         run: |
@@ -169,6 +170,7 @@ jobs:
         uses: swatinem/rust-cache@v2
         with:
           shared-key: ${{ matrix.settings.target }}
+          save-if: ${{ github.ref == 'refs/heads/main' }}
 
       - uses: ./.github/actions/setup-js
 

.github/workflows/publish.yml

@@ -173,6 +173,7 @@ jobs:
         uses: swatinem/rust-cache@v2
         with:
           shared-key: ${{ matrix.settings.target }}
+          save-if: ${{ github.ref == 'refs/heads/main' }}
 
       - uses: ./.github/actions/setup-js
 

Cargo.lock

@@ -64,6 +64,7 @@ percent-encoding = "2.3.1"
 [workspace.lints.rust]
 deprecated = "allow"
 unexpected_cfgs = "allow"
+unused_must_use = "deny"
 
 [workspace.lints.clippy]
 dbg_macro = "deny"

Cargo.toml

@@ -196,15 +196,15 @@ pub async fn focus_window(window_id: WindowId) -> Result<(), String> {
                 if GetWindowPlacement(hwnd, &mut wp).is_ok() {
                     // Restore using the previous placement to avoid resizing
                     wp.showCmd = SW_RESTORE.0 as u32;
-                    SetWindowPlacement(hwnd, &wp);
+                    let _ = SetWindowPlacement(hwnd, &wp);
                 } else {
                     // Fallback to simple restore if placement fails
-                    ShowWindow(hwnd, SW_RESTORE);
+                    let _ = ShowWindow(hwnd, SW_RESTORE);
                 }
             }
 
             // Always try to bring to foreground
-            SetForegroundWindow(hwnd);
+            let _ = SetForegroundWindow(hwnd);
         }
     }
 

apps/desktop/src-tauri/src/target_select_overlay.rs

@@ -230,6 +230,7 @@ const WARMUP_GUARD_SAMPLES: u32 = 3;
 const WARMUP_SPIKE_RATIO: f64 = 50.0;
 #[cfg(not(target_os = "macos"))]
 const FALLBACK_WIRED_LATENCY_SECS: f64 = 0.03;
+#[cfg(target_os = "macos")]
 const WIRELESS_FALLBACK_LATENCY_SECS: f64 = 0.20;
 const WIRELESS_MIN_LATENCY_SECS: f64 = 0.12;
 
@@ -615,6 +616,7 @@ mod macos {
 }
 
 #[cfg(test)]
+#[allow(clippy::unchecked_duration_subtraction)]
 mod tests {
     use super::*;
     use std::time::Instant;

crates/audio/src/latency.rs

@@ -10,7 +10,7 @@ use std::{
     ptr::{self, null, null_mut},
     time::{Duration, Instant},
 };
-use tracing::{trace, warn};
+use tracing::*;
 use windows::{
     Win32::{
         Foundation::*,

crates/camera-directshow/src/lib.rs

@@ -6,7 +6,7 @@ use std::{
     ffi::{OsStr, OsString},
     fmt::{Debug, Display},
     ops::Deref,
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use windows::Win32::Media::{DirectShow::*, KernelStreaming::*, MediaFoundation::*};

crates/camera-windows/src/lib.rs

@@ -2,7 +2,7 @@ use cap_media_info::{AudioInfo, VideoInfo};
 use cidre::{cm::SampleTimingInfo, objc::Obj, *};
 use ffmpeg::frame;
 use std::{ops::Sub, path::PathBuf, time::Duration};
-use tracing::{debug, error, info, trace};
+use tracing::*;
 
 // before pausing at all, subtract 0.
 // on pause, record last frame time.
@@ -48,23 +48,23 @@ pub enum InitError {
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum QueueVideoFrameError {
+pub enum QueueFrameError {
     #[error("AppendError/{0}")]
     AppendError(arc::R<ns::Exception>),
     #[error("Failed")]
     Failed,
+    #[error("Construct/{0}")]
+    Construct(cidre::os::Error),
+    #[error("NotReadyForMore")]
+    NotReadyForMore,
 }
 
 #[derive(thiserror::Error, Debug)]
-pub enum QueueAudioFrameError {
-    #[error("No audio input")]
-    NoAudioInput,
-    #[error("Not ready")]
-    NotReady,
-    #[error("Setup/{0}")]
-    Setup(cidre::os::Error),
-    #[error("AppendError/{0}")]
-    AppendError(&'static cidre::ns::Exception),
+pub enum FinishError {
+    #[error("NotWriting")]
+    NotWriting,
+    #[error("NoFrames")]
+    NoFrames,
     #[error("Failed")]
     Failed,
 }
@@ -219,11 +219,15 @@ impl MP4Encoder {
         &mut self,
         frame: arc::R<cm::SampleBuf>,
         timestamp: Duration,
-    ) -> Result<(), QueueVideoFrameError> {
-        if self.is_paused || !self.video_input.is_ready_for_more_media_data() {
+    ) -> Result<(), QueueFrameError> {
+        if self.is_paused {
             return Ok(());
         };
 
+        if !self.video_input.is_ready_for_more_media_data() {
+            return Err(QueueFrameError::NotReadyForMore);
+        }
+
         if !self.is_writing {
             self.is_writing = true;
             self.asset_writer
@@ -270,10 +274,7 @@ impl MP4Encoder {
         timing.pts = cm::Time::new(pts_duration.as_millis() as i64, 1_000);
         let frame = frame.copy_with_new_timing(&[timing]).unwrap();
 
-        self.video_input
-            .append_sample_buf(&frame)
-            .map_err(|e| QueueVideoFrameError::AppendError(e.retained()))
-            .and_then(|v| v.then_some(()).ok_or(QueueVideoFrameError::Failed))?;
+        append_sample_buf(&mut self.video_input, &self.asset_writer, &frame)?;
 
         self.video_frames_appended += 1;
         self.last_timestamp = Some(timestamp);
@@ -285,26 +286,26 @@ impl MP4Encoder {
     /// in the timebase of 1 / sample rate
     pub fn queue_audio_frame(
         &mut self,
-        frame: frame::Audio,
+        frame: &frame::Audio,
         timestamp: Duration,
-    ) -> Result<(), QueueAudioFrameError> {
+    ) -> Result<(), QueueFrameError> {
         if self.is_paused || !self.is_writing {
             return Ok(());
         }
 
+        let Some(audio_input) = &mut self.audio_input else {
+            return Err(QueueFrameError::Failed);
+        };
+
         if let Some(pause_timestamp) = self.pause_timestamp
             && let Some(gap) = timestamp.checked_sub(pause_timestamp)
         {
             self.timestamp_offset += gap;
             self.pause_timestamp = None;
         }
 
-        let Some(audio_input) = &mut self.audio_input else {
-            return Err(QueueAudioFrameError::NoAudioInput);
-        };
-
         if !audio_input.is_ready_for_more_media_data() {
-            return Ok(());
+            return Err(QueueFrameError::NotReadyForMore);
         }
 
         let audio_desc = cat::audio::StreamBasicDesc::common_f32(
@@ -316,11 +317,11 @@ impl MP4Encoder {
         let total_data = frame.samples() * frame.channels() as usize * frame.format().bytes();
 
         let mut block_buf =
-            cm::BlockBuf::with_mem_block(total_data, None).map_err(QueueAudioFrameError::Setup)?;
+            cm::BlockBuf::with_mem_block(total_data, None).map_err(QueueFrameError::Construct)?;
 
         let block_buf_slice = block_buf
             .as_mut_slice()
-            .map_err(QueueAudioFrameError::Setup)?;
+            .map_err(QueueFrameError::Construct)?;
 
         if frame.is_planar() {
             let mut offset = 0;
@@ -335,7 +336,7 @@ impl MP4Encoder {
         }
 
         let format_desc =
-            cm::AudioFormatDesc::with_asbd(&audio_desc).map_err(QueueAudioFrameError::Setup)?;
+            cm::AudioFormatDesc::with_asbd(&audio_desc).map_err(QueueFrameError::Construct)?;
 
         let mut pts_duration = timestamp
             .checked_sub(self.timestamp_offset)
@@ -344,7 +345,7 @@ impl MP4Encoder {
         if let Some(last_pts) = self.last_audio_pts
             && pts_duration <= last_pts
         {
-            let frame_duration = Self::audio_frame_duration(&frame);
+            let frame_duration = Self::audio_frame_duration(frame);
             let adjusted_pts = last_pts + frame_duration;
 
             trace!(
@@ -383,12 +384,9 @@ impl MP4Encoder {
             }],
             &[],
         )
-        .map_err(QueueAudioFrameError::Setup)?;
+        .map_err(QueueFrameError::Construct)?;
 
-        audio_input
-            .append_sample_buf(&buffer)
-            .map_err(QueueAudioFrameError::AppendError)
-            .and_then(|v| v.then_some(()).ok_or(QueueAudioFrameError::Failed))?;
+        append_sample_buf(audio_input, &self.asset_writer, &buffer)?;
 
         self.audio_frames_appended += 1;
         self.last_timestamp = Some(timestamp);
@@ -449,13 +447,14 @@ impl MP4Encoder {
         self.is_paused = false;
     }
 
-    pub fn finish(&mut self, timestamp: Option<Duration>) {
+    pub fn finish(&mut self, timestamp: Option<Duration>) -> Result<(), FinishError> {
         if !self.is_writing {
-            return;
+            return Err(FinishError::NotWriting);
         }
 
         let Some(mut most_recent_frame) = self.most_recent_frame.take() else {
-            return;
+            warn!("Encoder attempted to finish with no frame");
+            return Err(FinishError::NoFrames);
         };
 
         // We extend the video to the provided timestamp if possible
@@ -489,16 +488,17 @@ impl MP4Encoder {
         debug!("Appended {} video frames", self.video_frames_appended);
         debug!("Appended {} audio frames", self.audio_frames_appended);
 
-        // debug!("First video timestamp: {:?}", self.first_timestamp);
-        // debug!("Last video timestamp: {:?}", self.last_pts);
+        wait_for_writer_finished(&self.asset_writer).map_err(|_| FinishError::Failed)?;
 
         info!("Finished writing");
+
+        Ok(())
     }
 }
 
 impl Drop for MP4Encoder {
     fn drop(&mut self) {
-        self.finish(None);
+        let _ = self.finish(None);
     }
 }
 
@@ -621,3 +621,35 @@ impl SampleBufExt for cm::SampleBuf {
         }
     }
 }
+
+fn append_sample_buf(
+    input: &mut av::AssetWriterInput,
+    writer: &av::AssetWriter,
+    frame: &cm::SampleBuf,
+) -> Result<(), QueueFrameError> {
+    match input.append_sample_buf(frame) {
+        Ok(true) => {}
+        Ok(false) => {
+            if writer.status() == av::asset::writer::Status::Failed {
+                return Err(QueueFrameError::Failed);
+            }
+            if writer.status() == av::asset::writer::Status::Writing {
+                return Err(QueueFrameError::NotReadyForMore);
+            }
+        }
+        Err(e) => return Err(QueueFrameError::AppendError(e.retained())),
+    }
+
+    Ok(())
+}
+
+fn wait_for_writer_finished(writer: &av::AssetWriter) -> Result<(), ()> {
+    use av::asset::writer::Status;
+    loop {
+        match writer.status() {
+            Status::Completed | Status::Cancelled => return Ok(()),
+            Status::Failed | Status::Unknown => return Err(()),
+            Status::Writing => std::thread::sleep(Duration::from_millis(2)),
+        }
+    }
+}

crates/enc-avfoundation/src/mp4.rs

@@ -104,8 +104,8 @@ impl AACEncoder {
         self.base.send_frame(frame, timestamp, output)
     }
 
-    pub fn finish(&mut self, output: &mut format::context::Output) -> Result<(), ffmpeg::Error> {
-        self.base.finish(output)
+    pub fn flush(&mut self, output: &mut format::context::Output) -> Result<(), ffmpeg::Error> {
+        self.base.flush(output)
     }
 }
 
@@ -114,7 +114,7 @@ impl AudioEncoder for AACEncoder {
         let _ = self.send_frame(frame, Duration::MAX, output);
     }
 
-    fn finish(&mut self, output: &mut format::context::Output) {
-        let _ = self.finish(output);
+    fn flush(&mut self, output: &mut format::context::Output) -> Result<(), ffmpeg::Error> {
+        self.flush(output)
     }
 }