Merge pull request #1716 from CapSoftware/hls-recording-reliability

segmented MP4 upload pipeline with live HLS playback and server-side muxing

Author: richiemcilroy

Cargo.lock

@@ -94,6 +94,12 @@ unnecessary_lazy_evaluations = "deny"
 needless_range_loop = "deny"
 manual_clamp = "deny"
 
+[profile.dev]
+debug = 1
+
+[profile.dev.package."*"]
+debug = 0
+
 # Optimize for smaller binary size
 [profile.release]
 panic = "unwind"

Cargo.toml

@@ -265,6 +265,34 @@ pub struct Organization {
     pub owner_id: String,
 }
 
+pub async fn signal_recording_complete(
+    app: &AppHandle,
+    video_id: &str,
+) -> Result<(), AuthedApiError> {
+    let resp = app
+        .authed_api_request("/api/upload/recording-complete", |client, url| {
+            client
+                .post(url)
+                .header("Content-Type", "application/json")
+                .json(&serde_json::json!({
+                    "videoId": video_id,
+                }))
+        })
+        .await
+        .map_err(|err| format!("api/signal_recording_complete/request: {err}"))?;
+
+    if !resp.status().is_success() {
+        let status = resp.status().as_u16();
+        let error_body = resp
+            .text()
+            .await
+            .unwrap_or_else(|_| "<no response body>".to_string());
+        return Err(format!("api/signal_recording_complete/{status}: {error_body}").into());
+    }
+
+    Ok(())
+}
+
 pub async fn fetch_organizations(app: &AppHandle) -> Result<Vec<Organization>, AuthedApiError> {
     let resp = app
         .authed_api_request("/api/desktop/organizations", |client, url| client.get(url))

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

@@ -266,7 +266,7 @@ impl CameraPreviewManager {
                 })
                 .ok();
 
-            let _ = rt.block_on(tokio::time::timeout(Duration::from_millis(250), drop_rx));
+            wait_for_shutdown_signal(&rt, drop_rx, Duration::from_millis(250));
 
             shutdown_complete_tx.send(()).ok();
             info!("DONE");
@@ -276,6 +276,12 @@ impl CameraPreviewManager {
     }
 }
 
+fn wait_for_shutdown_signal(runtime: &Runtime, receiver: oneshot::Receiver<()>, timeout: Duration) {
+    runtime.block_on(async move {
+        let _ = tokio::time::timeout(timeout, receiver).await;
+    });
+}
+
 // Internal events for the persistent camera renderer architecture.
 //
 // The camera preview uses a persistent WGPU renderer that stays alive across
@@ -1025,7 +1031,9 @@ async fn resize_window(
 
 #[cfg(test)]
 mod tests {
-    use super::preferred_alpha_mode;
+    use super::{preferred_alpha_mode, wait_for_shutdown_signal};
+    use std::thread;
+    use tokio::{runtime::Runtime, sync::oneshot, time::Duration};
     use wgpu::CompositeAlphaMode;
 
     #[test]
@@ -1066,6 +1074,19 @@ mod tests {
         );
         assert_eq!(preferred_alpha_mode(&[]), CompositeAlphaMode::Opaque);
     }
+
+    #[test]
+    fn wait_for_shutdown_signal_can_run_from_plain_thread() {
+        let runtime = Runtime::new().unwrap();
+        let (tx, rx) = oneshot::channel();
+
+        let handle = thread::spawn(move || {
+            wait_for_shutdown_signal(&runtime, rx, Duration::from_millis(100));
+        });
+
+        tx.send(()).ok();
+        handle.join().unwrap();
+    }
 }
 
 fn render_solid_frame(color: [u8; 4], width: u32, height: u32) -> (Vec<u8>, u32) {

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

@@ -0,0 +1,122 @@
+use tokio::task::JoinHandle;
+
+pub(crate) fn run_while_active<T, FExit, F>(is_exiting: FExit, operation: F) -> Option<T>
+where
+    FExit: Fn() -> bool,
+    F: FnOnce() -> T,
+{
+    if is_exiting() {
+        None
+    } else {
+        Some(operation())
+    }
+}
+
+pub(crate) fn collect_device_inventory<TCamera, TMicrophone, FExit, FCamera, FMicrophone>(
+    is_exiting: FExit,
+    camera_permitted: bool,
+    microphone_permitted: bool,
+    list_cameras: FCamera,
+    list_microphones: FMicrophone,
+) -> Option<(Vec<TCamera>, Vec<TMicrophone>)>
+where
+    FExit: Fn() -> bool,
+    FCamera: FnOnce() -> Vec<TCamera>,
+    FMicrophone: FnOnce() -> Vec<TMicrophone>,
+{
+    if is_exiting() {
+        return None;
+    }
+
+    let cameras = if camera_permitted {
+        if is_exiting() {
+            return None;
+        }
+
+        list_cameras()
+    } else {
+        Vec::new()
+    };
+
+    if is_exiting() {
+        return None;
+    }
+
+    let microphones = if microphone_permitted {
+        if is_exiting() {
+            return None;
+        }
+
+        list_microphones()
+    } else {
+        Vec::new()
+    };
+
+    if is_exiting() {
+        return None;
+    }
+
+    Some((cameras, microphones))
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum AppExitAction {
+    #[cfg(target_os = "macos")]
+    Process(i32),
+    #[cfg(not(target_os = "macos"))]
+    Runtime(i32),
+}
+
+pub(crate) fn app_exit_action(exit_code: i32) -> AppExitAction {
+    #[cfg(target_os = "macos")]
+    {
+        AppExitAction::Process(exit_code)
+    }
+
+    #[cfg(not(target_os = "macos"))]
+    {
+        AppExitAction::Runtime(exit_code)
+    }
+}
+
+pub(crate) fn read_target_under_cursor<TDisplay, TWindow, FExit, FDisplay, FWindow>(
+    is_exiting: FExit,
+    display: FDisplay,
+    window: FWindow,
+) -> Option<(Option<TDisplay>, Option<TWindow>)>
+where
+    FExit: Fn() -> bool,
+    FDisplay: FnOnce() -> Option<TDisplay>,
+    FWindow: FnOnce() -> Option<TWindow>,
+{
+    if is_exiting() {
+        return None;
+    }
+
+    let display = display();
+
+    if is_exiting() {
+        return None;
+    }
+
+    let window = window();
+
+    if is_exiting() {
+        return None;
+    }
+
+    Some((display, window))
+}
+
+pub(crate) fn abort_join_handles<T>(
+    tasks: impl IntoIterator<Item = JoinHandle<T>>,
+    task: Option<JoinHandle<T>>,
+) {
+    for task in tasks {
+        task.abort();
+    }
+
+    if let Some(task) = task {
+        task.abort();
+    }
+}