diff --git a/apps/desktop-gpui/docs/upload-health-failed-linux.png b/apps/desktop-gpui/docs/upload-health-failed-linux.png new file mode 100644 index 00000000000..7aafc5c0eea Binary files /dev/null and b/apps/desktop-gpui/docs/upload-health-failed-linux.png differ diff --git a/apps/desktop-gpui/docs/upload-health-fast-linux.png b/apps/desktop-gpui/docs/upload-health-fast-linux.png new file mode 100644 index 00000000000..63c59a58875 Binary files /dev/null and b/apps/desktop-gpui/docs/upload-health-fast-linux.png differ diff --git a/apps/desktop-gpui/docs/upload-health-slow-linux.png b/apps/desktop-gpui/docs/upload-health-slow-linux.png new file mode 100644 index 00000000000..6d96bc91e86 Binary files /dev/null and b/apps/desktop-gpui/docs/upload-health-slow-linux.png differ diff --git a/apps/desktop-gpui/src/auth.rs b/apps/desktop-gpui/src/auth.rs index 0ddbed2aacb..761adcb723d 100644 --- a/apps/desktop-gpui/src/auth.rs +++ b/apps/desktop-gpui/src/auth.rs @@ -345,19 +345,7 @@ pub async fn authed_request_with( path: &str, body: Option, ) -> Result { - let (server, token) = if let Ok(context) = UPLOAD_REQUEST.try_with(Clone::clone) { - let token = context.credentials()?; - (context.server_url, token) - } else { - let auth = store::auth_snapshot(); - ( - server_url(), - auth.token.ok_or(AuthApiError::InvalidAuthentication)?, - ) - }; - let url = format!("{server}{path}"); - let mut request = client.request(method, url); - request = apply_desktop_headers(request).bearer_auth(token); + let mut request = authed_request_builder(client, method, path)?; if let Some(body) = body { request = request .header("Content-Type", "application/json") @@ -376,6 +364,25 @@ pub async fn authed_request_with( Ok(response) } +pub(crate) fn authed_request_builder( + client: &reqwest::Client, + method: reqwest::Method, + path: &str, +) -> Result { + let (server, token) = if let Ok(context) = UPLOAD_REQUEST.try_with(Clone::clone) { + let token = context.credentials()?; + (context.server_url, token) + } else { + let auth = store::auth_snapshot(); + ( + server_url(), + auth.token.ok_or(AuthApiError::InvalidAuthentication)?, + ) + }; + let url = format!("{server}{path}"); + Ok(apply_desktop_headers(client.request(method, url)).bearer_auth(token)) +} + pub fn process_auth_data(params: AuthParams) -> Result<(), String> { let (secret, user_id) = match params { AuthParams::ApiKey { api_key, user_id } => (json!({ "api_key": api_key }), user_id), diff --git a/apps/desktop-gpui/src/main.rs b/apps/desktop-gpui/src/main.rs index 9e886003506..fd598f1682d 100644 --- a/apps/desktop-gpui/src/main.rs +++ b/apps/desktop-gpui/src/main.rs @@ -48,6 +48,7 @@ mod main_window; mod media_access; mod menus; mod mode_select_window; +mod network_health; mod onboarding_audio; mod onboarding_doodles; mod onboarding_window; @@ -331,6 +332,7 @@ fn main() { platform::follow_system_reduce_motion(cx); let session = RecordingSession::init(cx); + crate::network_health::init(cx); crate::sign_in::SignIn::init(cx); crate::feeds::Feeds::init(cx); crate::media_access::MediaAccess::init(cx); diff --git a/apps/desktop-gpui/src/main_window.rs b/apps/desktop-gpui/src/main_window.rs index a318d54a0f7..995ae905237 100644 --- a/apps/desktop-gpui/src/main_window.rs +++ b/apps/desktop-gpui/src/main_window.rs @@ -3465,6 +3465,26 @@ impl MainWindow { /// 8px from the window edges (`mx-2`). fn render_header_actions(&self, cx: &mut Context) -> impl IntoElement { let theme = self.theme; + let upload_health = crate::network_health::indicator().map(|health| { + let color = if health.failed { + theme.red_11 + } else if health.limited { + theme.amber_11 + } else { + theme.gray_11 + }; + div() + .id("upload-health") + .tab_index(0) + .flex_shrink_0() + .text_size(px(9.)) + .text_color(color) + .cursor_pointer() + .child(health.label) + .tooltip(move |_, cx| ui::Tooltip::new(&theme, health.tooltip.clone()).view(cx)) + .tooltip_show_delay(ui::TOOLTIP_SHOW_DELAY) + .on_click(cx.listener(|_, _, _, cx| crate::network_health::request(cx, true))) + }); // `IconButton::header`: a 20px hit box with no fill, `text-gray-11` // going to `text-gray-12` on hover. @@ -3509,6 +3529,7 @@ impl MainWindow { .gap(px(4.)) .mx(px(8.)) .min_w_0() + .children(upload_health) .child( icon_button("help", "icons/circle-help.svg", 16.).on_click(cx.listener( |_, _, _window, cx| { diff --git a/apps/desktop-gpui/src/network_health.rs b/apps/desktop-gpui/src/network_health.rs new file mode 100644 index 00000000000..1a31eab705a --- /dev/null +++ b/apps/desktop-gpui/src/network_health.rs @@ -0,0 +1,469 @@ +mod policy; + +use std::{ + sync::{Mutex, MutexGuard, OnceLock}, + time::{Duration, SystemTime}, +}; + +use futures_util::future::{AbortHandle, Abortable}; +use gpui::App; +use policy::{Admission, Measurement, Sample, Scope}; +use serde::Deserialize; + +use crate::{auth, session::RecordingSession, store}; + +const AUTOMATIC_PROBE_INTERVAL: Duration = Duration::from_secs(60 * 60); + +#[derive(Clone, Copy, Default)] +enum Status { + #[default] + Unknown, + Checking, + Ready, + Failed, + Unsupported, +} + +struct Running { + ticket: u64, + abort: AbortHandle, + done: flume::Receiver<()>, +} + +#[derive(Default)] +struct Controller { + admission: Admission, + scope: Option, + measurement: Option, + status: Status, + running: Option, + attempted_at: Option, + applied_width: Option, +} + +fn controller() -> MutexGuard<'static, Controller> { + static STATE: OnceLock> = OnceLock::new(); + STATE + .get_or_init(|| Mutex::new(Controller::default())) + .lock() + .unwrap_or_else(|error| error.into_inner()) +} + +impl Controller { + fn admit_recording(&mut self) { + self.applied_width = None; + self.status = Status::Unknown; + self.admission.admit_recording(); + if let Some(running) = &self.running { + running.abort.abort(); + } + } +} + +fn current_scope() -> Option { + store::auth_snapshot().token?; + Some(Scope { + server: auth::server_url(), + owner: store::store_section("auth") + .get("user_id")? + .as_str()? + .to_string(), + organization: store::store_section(store::RECORDING_SETTINGS) + .get("organizationId") + .and_then(serde_json::Value::as_str) + .filter(|id| !id.is_empty()) + .map(str::to_string), + }) +} + +pub(crate) fn invalidate() { + let mut state = controller(); + state.admission.invalidate(); + if let Some(running) = &state.running { + running.abort.abort(); + } + state.measurement = None; + state.attempted_at = None; + state.status = Status::Unknown; +} + +pub(crate) fn recording_admitted() { + controller().admit_recording(); +} + +pub(crate) fn recording_released() { + let mut state = controller(); + state.admission.release_recording(); + state.applied_width = None; +} + +#[cfg(test)] +pub(crate) fn probe_admitted_for_test() -> bool { + controller().admission.begin_probe().is_some() +} + +pub(crate) async fn wait_for_probe() { + let done = controller() + .running + .as_ref() + .map(|running| running.done.clone()); + if let Some(done) = done { + let _ = done.recv_async().await; + } +} + +pub(crate) fn max_width(configured: u32, organization: Option<&str>) -> u32 { + let Some(scope) = current_scope() else { + return configured; + }; + let mut state = controller(); + let width = state + .measurement + .as_ref() + .and_then(|measurement| measurement.max_width_for(&scope, organization, SystemTime::now())) + .map_or(configured, |cap| configured.min(cap)); + state.applied_width = (width < configured).then_some(width); + width +} + +pub(crate) struct Indicator { + pub label: String, + pub tooltip: String, + pub failed: bool, + pub limited: bool, +} + +pub(crate) fn indicator() -> Option { + let scope = current_scope()?; + let state = controller(); + let measurement = state + .measurement + .as_ref() + .filter(|measurement| measurement.fresh_for(&scope, SystemTime::now())); + let (label, tooltip, failed, limited) = if let Some(width) = state.applied_width { + (format!("Instant limited: {}p", width * 9 / 16), + "This recording uses a lower resolution because repeated upload samples were slow. Your saved setting is unchanged; retest after recording.".into(), false, true) + } else if state.scope.as_ref() == Some(&scope) && matches!(state.status, Status::Checking) { + ( + "Checking upload".into(), + "Checking upload availability and transfer speed.".into(), + false, + false, + ) + } else if let Some(measurement) = measurement { + let speed = measurement + .mbps + .map_or_else(|| "Upload OK".to_string(), |rate| format!("{rate:.1} Mb/s")); + let configured = if store::auth_snapshot().is_upgraded() { + store::GeneralSettings::load().instant_mode_max_resolution + } else { + cap_recording::FREE_INSTANT_MODE_MAX_RESOLUTION + }; + let cap = measurement.max_width.filter(|cap| *cap < configured); + match cap { + Some(width) => ( + format!("{speed} · {}p", width * 9 / 16), + format!("Repeated slow uploads: Instant recordings are limited to {}p. Your saved resolution is unchanged. Click to retest while idle.", width * 9 / 16), + false, true, + ), + None => (speed, "Upload destination is writable. Click to retest while idle. Tests are paused during recording.".into(), false, false), + } + } else { + match if state.scope.as_ref() == Some(&scope) { state.status } else { Status::Unknown } { + Status::Checking => ("Checking upload".into(), "Checking upload availability and transfer speed.".into(), false, false), + Status::Failed => ("Upload check failed".into(), "The upload check failed. Check your connection and storage settings, retry while idle, or contact support. Recording is still available.".into(), true, false), + Status::Unsupported => ("Upload check unavailable".into(), "This server does not support upload diagnostics. Your recording resolution is unchanged.".into(), false, false), + _ => ("Check upload".into(), "Click to check the upload destination and speed while idle.".into(), false, false), + } + }; + Some(Indicator { + label, + tooltip, + failed, + limited, + }) +} + +struct Completion(flume::Sender<()>); + +impl Drop for Completion { + fn drop(&mut self) { + let _ = self.0.send(()); + } +} + +pub(crate) fn request(cx: &mut App, manual: bool) { + let scope = current_scope(); + let launch = { + let mut state = controller(); + if state.scope != scope { + state.admission.invalidate(); + if let Some(running) = &state.running { + running.abort.abort(); + } + state.scope = scope.clone(); + state.measurement = None; + state.attempted_at = None; + state.status = Status::Unknown; + } + let Some(scope) = scope else { + return; + }; + let interval = if manual { + Duration::from_secs(60) + } else { + AUTOMATIC_PROBE_INTERVAL + }; + if state.running.is_some() + || RecordingSession::recording_in_flight(cx) + || state + .attempted_at + .is_some_and(|at| at.elapsed().is_ok_and(|age| age < interval)) + { + return; + } + let Some(ticket) = state.admission.begin_probe() else { + return; + }; + let (abort, registration) = AbortHandle::new_pair(); + let (done, receiver) = flume::bounded(1); + state.running = Some(Running { + ticket, + abort, + done: receiver, + }); + state.status = Status::Checking; + state.attempted_at = Some(SystemTime::now()); + (scope, ticket, registration, Completion(done)) + }; + let (scope, ticket, registration, completion) = launch; + let expected_scope = scope.clone(); + let task = gpui_tokio::Tokio::spawn(cx, async move { + let _completion = completion; + Abortable::new(probe(scope), registration).await + }); + cx.spawn(async move |cx| { + let result = task.await; + cx.update(|cx| { + let mut state = controller(); + if !state + .running + .as_ref() + .is_some_and(|running| running.ticket == ticket) + { + return; + } + state.running = None; + if !state.admission.accepts(ticket) || current_scope().as_ref() != Some(&expected_scope) + { + state.status = Status::Unknown; + if state.measurement.is_none() { + state.attempted_at = None; + } + } else { + match result { + Ok(Ok(Ok(measurement))) => { + state.measurement = Some(measurement); + state.status = Status::Ready; + } + Ok(Ok(Err(ProbeError::Unsupported))) => { + state.measurement = None; + state.status = Status::Unsupported; + } + _ => { + state.measurement = None; + state.status = Status::Failed; + } + } + } + drop(state); + cx.refresh_windows(); + }); + }) + .detach(); + cx.refresh_windows(); +} + +pub(crate) fn init(cx: &mut App) { + cx.spawn(async move |cx| { + loop { + cx.update(|cx| request(cx, false)); + cx.background_executor() + .timer(Duration::from_secs(15)) + .await; + } + }) + .detach(); +} + +#[derive(Debug)] +enum ProbeError { + Failed, + Unsupported, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ProbeResponse { + bytes: usize, + body_read_ms: f64, + organization_id: String, +} + +async fn probe(scope: Scope) -> Result { + let client = auth::api_client().map_err(|_| ProbeError::Failed)?; + let context = auth::UploadRequestContext::for_upload(&scope.server, Some(&scope.owner)) + .map_err(|_| ProbeError::Failed)?; + context + .run(async { + let warm = auth::authed_request_with( + &client, + reqwest::Method::GET, + "/api/desktop/upload-health", + None, + ) + .await + .map_err(|_| ProbeError::Failed)?; + if warm.status() == reqwest::StatusCode::NOT_FOUND { + return Err(ProbeError::Unsupported); + } + if !warm.status().is_success() { + return Err(ProbeError::Failed); + } + warm.bytes().await.map_err(|_| ProbeError::Failed)?; + let mut payload = vec![0_u8; 2 * 1024 * 1024]; + let mut random = 0x3ab1_5fc9_u32; + for byte in &mut payload { + random ^= random << 13; + random ^= random >> 17; + random ^= random << 5; + *byte = random as u8; + } + let payload = bytes::Bytes::from(payload); + let mut samples = [Sample { + bytes: 0, + body_read_ms: 0., + }; 2]; + let mut organization: Option = None; + for (index, sample) in samples.iter_mut().enumerate() { + if current_scope().as_ref() != Some(&scope) { + return Err(ProbeError::Failed); + } + let path = if index == 0 { + "/api/desktop/upload-health?checkStorage=true" + } else { + "/api/desktop/upload-health" + }; + let mut request = + auth::authed_request_builder(&client, reqwest::Method::POST, path) + .map_err(|_| ProbeError::Failed)?; + if let Some(id) = organization.as_ref().or(scope.organization.as_ref()) { + request = request.query(&[("orgId", id)]); + } + let response = request + .header("Content-Type", "application/octet-stream") + .timeout(Duration::from_secs(30)) + .body(payload.clone()) + .send() + .await + .map_err(|_| ProbeError::Failed)?; + if !response.status().is_success() { + return Err(ProbeError::Failed); + } + let response = response + .json::() + .await + .map_err(|_| ProbeError::Failed)?; + if response.bytes != payload.len() + || response.organization_id.is_empty() + || organization + .as_ref() + .is_some_and(|id| *id != response.organization_id) + { + return Err(ProbeError::Failed); + } + organization = Some(response.organization_id); + *sample = Sample { + bytes: response.bytes, + body_read_ms: response.body_read_ms, + }; + } + Ok(Measurement::from_samples( + scope.clone(), + organization.ok_or(ProbeError::Failed)?, + samples, + SystemTime::now(), + )) + }) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + struct ActiveTransfer(Arc); + + impl Drop for ActiveTransfer { + fn drop(&mut self) { + self.0.store(false, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn recording_waits_for_an_in_flight_transfer_to_be_dropped() { + let active = Arc::new(AtomicBool::new(false)); + let mut state = Controller::default(); + let ticket = state.admission.begin_probe().unwrap(); + let (abort, registration) = AbortHandle::new_pair(); + let (done, receiver) = flume::bounded(1); + let (started, ready) = flume::bounded(1); + state.running = Some(Running { + ticket, + abort, + done: receiver.clone(), + }); + let completion = Completion(done); + let transfer_active = active.clone(); + let task = tokio::spawn(async move { + let _completion = completion; + Abortable::new( + async move { + transfer_active.store(true, Ordering::SeqCst); + let _transfer = ActiveTransfer(transfer_active); + started.send(()).unwrap(); + std::future::pending::<()>().await; + }, + registration, + ) + .await + }); + ready.recv_async().await.unwrap(); + assert!(active.load(Ordering::SeqCst)); + state.admit_recording(); + tokio::time::timeout(Duration::from_secs(1), receiver.recv_async()) + .await + .unwrap() + .unwrap(); + assert!(!active.load(Ordering::SeqCst)); + assert!(!state.admission.accepts(ticket)); + assert_eq!(state.admission.begin_probe(), None); + assert!(task.await.unwrap().is_err()); + } + + #[test] + fn dropping_an_unpolled_probe_still_releases_recording_startup() { + let (done, receiver) = flume::bounded(1); + let completion = Completion(done); + let task = async move { + let _completion = completion; + std::future::pending::<()>().await; + }; + drop(task); + assert_eq!(receiver.try_recv(), Ok(())); + } +} diff --git a/apps/desktop-gpui/src/network_health/policy.rs b/apps/desktop-gpui/src/network_health/policy.rs new file mode 100644 index 00000000000..59dc4a3b1d9 --- /dev/null +++ b/apps/desktop-gpui/src/network_health/policy.rs @@ -0,0 +1,256 @@ +use std::time::{Duration, SystemTime}; + +pub const MEASUREMENT_TTL: Duration = Duration::from_secs(15 * 60); + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Scope { + pub server: String, + pub owner: String, + pub organization: Option, +} + +#[derive(Clone, Copy, Debug)] +pub struct Sample { + pub bytes: usize, + pub body_read_ms: f64, +} + +impl Sample { + fn mbps(self) -> Option { + (self.bytes >= 1024 * 1024 + && self.bytes <= 8 * 1024 * 1024 + && self.body_read_ms.is_finite() + && self.body_read_ms > 0.) + .then(|| self.bytes as f64 * 8. / (self.body_read_ms * 1000.)) + } +} + +#[derive(Clone, Debug)] +pub struct Measurement { + pub scope: Scope, + pub organization: String, + pub checked_at: SystemTime, + pub mbps: Option, + pub max_width: Option, +} + +impl Measurement { + pub fn from_samples( + scope: Scope, + organization: String, + samples: [Sample; 2], + now: SystemTime, + ) -> Self { + let rates = samples[0].mbps().zip(samples[1].mbps()); + let mbps = rates.map(|(first, second)| first.max(second)); + let max_width = rates.and_then(|(first, second)| { + if samples.iter().any(|sample| sample.body_read_ms < 500.) + || first.max(second) > first.min(second) * 2. + { + return None; + } + match first.max(second) { + rate if rate < 5. => Some(1280), + rate if rate < 10. => Some(1920), + rate if rate < 20. => Some(2560), + _ => None, + } + }); + Self { + scope, + organization, + checked_at: now, + mbps, + max_width, + } + } + + pub fn fresh_for(&self, scope: &Scope, now: SystemTime) -> bool { + self.scope == *scope + && now + .duration_since(self.checked_at) + .is_ok_and(|age| age < MEASUREMENT_TTL) + } + + pub fn max_width_for( + &self, + scope: &Scope, + organization: Option<&str>, + now: SystemTime, + ) -> Option { + (self.fresh_for(scope, now) && organization == Some(self.organization.as_str())) + .then_some(self.max_width) + .flatten() + } +} + +#[derive(Default)] +pub struct Admission { + recording: bool, + generation: u64, +} + +impl Admission { + pub fn begin_probe(&mut self) -> Option { + if self.recording { + return None; + } + self.invalidate(); + Some(self.generation) + } + + pub fn accepts(&self, generation: u64) -> bool { + !self.recording && self.generation == generation + } + + pub fn admit_recording(&mut self) { + self.recording = true; + self.invalidate(); + } + + pub fn release_recording(&mut self) { + self.recording = false; + } + + pub fn invalidate(&mut self) { + self.generation = self.generation.wrapping_add(1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scope() -> Scope { + Scope { + server: "https://cap.example".into(), + owner: "owner".into(), + organization: None, + } + } + + fn measure(first_mbps: f64, second_mbps: f64) -> Measurement { + let sample = |mbps| Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: 2. * 1024. * 1024. * 8. / (mbps * 1000.), + }; + Measurement::from_samples( + scope(), + "org-1".into(), + [sample(first_mbps), sample(second_mbps)], + SystemTime::now(), + ) + } + + #[test] + fn only_repeatably_slow_samples_lower_resolution() { + assert_eq!(measure(3., 4.).max_width, Some(1280)); + assert_eq!(measure(6., 7.).max_width, Some(1920)); + assert_eq!(measure(12., 16.).max_width, Some(2560)); + assert_eq!(measure(20., 20.).max_width, None); + } + + #[test] + fn a_fast_or_inconsistent_sample_preserves_quality() { + assert_eq!(measure(3., 100.).max_width, None); + assert_eq!(measure(3., 9.).max_width, None); + assert_eq!(measure(100., 100.).max_width, None); + } + + #[test] + fn invalid_or_buffered_samples_never_apply_a_cap() { + for duration in [0., -1., f64::NAN, f64::INFINITY] { + let sample = Sample { + bytes: 2 * 1024 * 1024, + body_read_ms: duration, + }; + let result = + Measurement::from_samples(scope(), "org-1".into(), [sample; 2], SystemTime::now()); + assert_eq!(result.max_width, None); + assert_eq!(result.mbps, None); + } + let sample = Sample { + bytes: 100, + body_read_ms: 1000., + }; + assert_eq!( + Measurement::from_samples(scope(), "org-1".into(), [sample; 2], SystemTime::now()) + .max_width, + None + ); + } + + #[test] + fn threshold_boundaries_use_the_less_restrictive_tier() { + assert_eq!(measure(5., 5.).max_width, Some(1920)); + assert_eq!(measure(10., 10.).max_width, Some(2560)); + assert_eq!(measure(20., 20.).max_width, None); + } + + #[test] + fn stale_account_and_server_measurements_are_ignored() { + let result = measure(3., 3.); + assert!(result.fresh_for(&scope(), result.checked_at)); + assert!(!result.fresh_for(&scope(), result.checked_at + MEASUREMENT_TTL)); + let mut other = scope(); + other.owner = "other".into(); + assert!(!result.fresh_for(&other, result.checked_at)); + other = scope(); + other.server = "https://other.example".into(); + assert!(!result.fresh_for(&other, result.checked_at)); + } + + #[test] + fn recording_admission_rejects_in_flight_and_new_probe_results() { + let mut admission = Admission::default(); + let ticket = admission.begin_probe().unwrap(); + assert!(admission.accepts(ticket)); + admission.admit_recording(); + assert!(!admission.accepts(ticket)); + assert_eq!(admission.begin_probe(), None); + admission.release_recording(); + assert!(!admission.accepts(ticket)); + let next = admission.begin_probe().unwrap(); + assert!(admission.accepts(next)); + assert!(!admission.accepts(ticket)); + } + + #[test] + fn storage_selection_and_actual_recording_destination_bound_the_cap() { + let result = measure(3., 3.); + assert_eq!( + result.max_width_for(&scope(), Some("org-1"), result.checked_at), + Some(1280) + ); + assert_eq!( + result.max_width_for(&scope(), Some("org-2"), result.checked_at), + None + ); + assert_eq!( + result.max_width_for(&scope(), None, result.checked_at), + None + ); + let mut other = scope(); + other.organization = Some("different-selection".into()); + assert!(!result.fresh_for(&other, result.checked_at)); + } + + #[test] + fn invalidation_rejects_late_results_without_entering_recording() { + let mut admission = Admission::default(); + let ticket = admission.begin_probe().unwrap(); + admission.invalidate(); + assert!(!admission.accepts(ticket)); + assert!(admission.begin_probe().is_some()); + } + + #[test] + fn moving_the_wall_clock_backwards_preserves_the_users_resolution() { + let result = measure(3., 3.); + let earlier = result + .checked_at + .checked_sub(Duration::from_secs(1)) + .unwrap(); + assert!(!result.fresh_for(&scope(), earlier)); + } +} diff --git a/apps/desktop-gpui/src/recording.rs b/apps/desktop-gpui/src/recording.rs index b9bb2cf8b9c..f3d0586db4a 100644 --- a/apps/desktop-gpui/src/recording.rs +++ b/apps/desktop-gpui/src/recording.rs @@ -1994,6 +1994,7 @@ async fn start_internal( config: StartConfig, #[cfg(target_os = "linux")] attempt: Option, ) -> anyhow::Result { + crate::network_health::wait_for_probe().await; #[cfg(target_os = "linux")] validate_linux_camera_request( config.mode, @@ -2020,24 +2021,24 @@ async fn start_internal( .filter(|organization| !organization.is_empty()) .map(str::to_string); - let pre_created_video = if config.mode == RecordingMode::Instant { - Some( - crate::upload::prepare_instant_upload( - matches!(config.target, ScreenCaptureTarget::CameraOnly), - project_name, - organization_id, - ) - .await - .map_err(anyhow::Error::msg)?, + let (pre_created_video, upload_organization_id) = if config.mode == RecordingMode::Instant { + let (video, organization) = crate::upload::prepare_instant_upload( + matches!(config.target, ScreenCaptureTarget::CameraOnly), + project_name, + organization_id, ) + .await + .map_err(anyhow::Error::msg)?; + (Some(video), organization) } else { - None + (None, None) }; let result = start_attempt_with_upload( config, project_dir, pre_created_video.clone(), + upload_organization_id, #[cfg(target_os = "linux")] attempt.clone(), ) @@ -2070,6 +2071,7 @@ async fn start_attempt_with_upload( config: StartConfig, project_dir: PathBuf, pre_created_video: Option, + upload_organization_id: Option, #[cfg(target_os = "linux")] attempt: Option, ) -> anyhow::Result { #[cfg(target_os = "linux")] @@ -2225,6 +2227,11 @@ async fn start_attempt_with_upload( } else { cap_recording::FREE_INSTANT_MODE_MAX_RESOLUTION }; + let instant_max_resolution = if config.mode == RecordingMode::Instant { + crate::network_health::max_width(instant_max_resolution, upload_organization_id.as_deref()) + } else { + instant_max_resolution + }; #[cfg(target_os = "macos")] let excluded_windows = { diff --git a/apps/desktop-gpui/src/session.rs b/apps/desktop-gpui/src/session.rs index ad48a204a55..758af25e8aa 100644 --- a/apps/desktop-gpui/src/session.rs +++ b/apps/desktop-gpui/src/session.rs @@ -462,8 +462,8 @@ struct SessionGlobal(Entity); impl Global for SessionGlobal {} impl RecordingSession { - pub fn init(cx: &mut App) -> Entity { - let session = cx.new(|_| Self { + fn new() -> Self { + Self { phase: Phase::Idle, active: None, error: None, @@ -502,7 +502,11 @@ impl RecordingSession { clean_control: CleanControlState::default(), #[cfg(target_os = "linux")] show_controls_after_pause: false, - }); + } + } + + pub fn init(cx: &mut App) -> Entity { + let session = cx.new(|_| Self::new()); cx.set_global(SessionGlobal(session.clone())); session } @@ -675,6 +679,7 @@ impl RecordingSession { self.clean_control.invalidate(); self.clean_control.uncertain = false; } + crate::network_health::recording_admitted(); self.phase = Phase::Starting; self.error = None; self.storage_warning = false; @@ -831,6 +836,8 @@ impl RecordingSession { } else { false }; + #[cfg(not(target_os = "linux"))] + let startup_joined = true; this.update(cx, |this, cx| { if this.recording_generation != generation || this.phase != Phase::Starting { if let Ok(Ok(active)) = result { @@ -848,39 +855,11 @@ impl RecordingSession { } Ok(Err(error)) => { tracing::error!("recording failed to start: {error:#}"); - this.error = Some(format!("{error:#}")); - this.abort_start_cue(); - this.countdown_remaining = None; - #[cfg(target_os = "linux")] - { - this.phase = if startup_joined { - Phase::Idle - } else { - Phase::Starting - }; - } - #[cfg(not(target_os = "linux"))] - { - this.phase = Phase::Idle; - } + this.fail_start(format!("{error:#}"), startup_joined); } Err(join_error) => { tracing::error!("recording start task died: {join_error}"); - this.error = Some("Recording task failed.".into()); - this.abort_start_cue(); - this.countdown_remaining = None; - #[cfg(target_os = "linux")] - { - this.phase = if startup_joined { - Phase::Idle - } else { - Phase::Starting - }; - } - #[cfg(not(target_os = "linux"))] - { - this.phase = Phase::Idle; - } + this.fail_start("Recording task failed.".into(), startup_joined); } } #[cfg(target_os = "linux")] @@ -905,6 +884,20 @@ impl RecordingSession { .detach(); } + fn fail_start(&mut self, error: String, cleanup_joined: bool) { + self.error = Some(error); + self.abort_start_cue(); + self.countdown_remaining = None; + self.phase = if cleanup_joined { + Phase::Idle + } else { + Phase::Starting + }; + if cleanup_joined { + crate::network_health::recording_released(); + } + } + fn activate_primed(&mut self, cx: &mut Context) { if let Some(active) = self.primed.take() { self.activate(active, cx); @@ -1700,6 +1693,7 @@ impl RecordingSession { { self.failure_monitor = None; } + crate::network_health::recording_admitted(); self.phase = Phase::Starting; self.started_at = None; cx.notify(); @@ -1950,6 +1944,7 @@ impl RecordingSession { self.storage_monitor = None; self.countdown_remaining = None; self.phase = Phase::Idle; + crate::network_health::recording_released(); self.mic_muted = false; self.storage_warning = false; self.stopped_elapsed = None; @@ -2653,4 +2648,27 @@ mod studio_editor_presentation_tests { state.closed(2); assert_eq!(state, StudioEditorPresentation::Dismissed); } + #[test] + fn failed_starts_release_upload_checks_only_after_capture_cleanup() { + let mut session = RecordingSession::new(); + for error in [ + "Missing device", + "Upload setup failed", + "Recording task failed.", + ] { + crate::network_health::recording_admitted(); + session.phase = Phase::Starting; + assert!(!crate::network_health::probe_admitted_for_test()); + session.fail_start(error.into(), true); + assert_eq!(session.phase, Phase::Idle); + assert_eq!(session.error.as_deref(), Some(error)); + assert!(crate::network_health::probe_admitted_for_test()); + } + crate::network_health::recording_admitted(); + session.fail_start("Cleanup unconfirmed".into(), false); + assert_eq!(session.phase, Phase::Starting); + assert!(!crate::network_health::probe_admitted_for_test()); + session.fail_start("Cleanup joined".into(), true); + assert!(crate::network_health::probe_admitted_for_test()); + } } diff --git a/apps/desktop-gpui/src/settings_pages.rs b/apps/desktop-gpui/src/settings_pages.rs index c0d5d6bb421..7437f1e8c1b 100644 --- a/apps/desktop-gpui/src/settings_pages.rs +++ b/apps/desktop-gpui/src/settings_pages.rs @@ -20,12 +20,12 @@ use std::{ time::Duration, }; +use cap_cli_install as cli_install; use gpui::{ AppContext as _, Context, Entity, FontWeight, Hsla, InteractiveElement, IntoElement, MouseButton, ParentElement, SharedString, StatefulInteractiveElement, Styled, Window, div, prelude::FluentBuilder, px, rgb, svg, }; -use cap_cli_install as cli_install; use serde::Deserialize; use serde_json::{Map, Value, json}; @@ -4543,6 +4543,7 @@ impl SettingsWindow { Ok((200, _)) => { this.s3_fetch(window, cx); this.info_dialog("S3 configuration saved successfully", window, cx); + crate::network_health::invalidate(); } _ => this.pages.s3.error = Some("Failed to save S3 config".to_string()), } @@ -4579,6 +4580,7 @@ impl SettingsWindow { Ok((200, _)) => { this.s3_fetch(window, cx); this.info_dialog("S3 configuration deleted successfully", window, cx); + crate::network_health::invalidate(); } _ => this.pages.s3.error = Some("Failed to delete S3 config".to_string()), } @@ -4775,6 +4777,7 @@ impl SettingsWindow { this.pages.storage = Some(parsed); } if connected { + crate::network_health::invalidate(); this.pages.gdrive.waiting = false; this.gdrive_fetch_s3_presence(window, cx); } @@ -4892,7 +4895,10 @@ impl SettingsWindow { |this, result, window, cx| { this.pages.gdrive.set_active_pending = false; match result { - Ok((200, _)) => this.gdrive_refresh(window, cx), + Ok((200, _)) => { + crate::network_health::invalidate(); + this.gdrive_refresh(window, cx); + } _ => { this.pages.gdrive.error = Some("Failed to update active storage provider".to_string()) @@ -4934,6 +4940,7 @@ impl SettingsWindow { Ok((200, _)) => { this.gdrive_refresh(window, cx); this.info_dialog("Google Drive disconnected", window, cx); + crate::network_health::invalidate(); } _ => { this.pages.gdrive.error = diff --git a/apps/desktop-gpui/src/settings_window.rs b/apps/desktop-gpui/src/settings_window.rs index cd7dd5b65d4..2cb02ee2f42 100644 --- a/apps/desktop-gpui/src/settings_window.rs +++ b/apps/desktop-gpui/src/settings_window.rs @@ -3209,6 +3209,22 @@ impl SettingsWindow { ); let mut rows = vec![resolution.into_any_element()]; + if let Some(health) = crate::network_health::indicator() { + rows.push( + div() + .p(px(12.)) + .text_size(px(12.)) + .text_color(Hsla::from(if health.failed { + theme.red_11 + } else if health.limited { + theme.amber_11 + } else { + theme.gray_11 + })) + .child(health.label) + .into_any_element(), + ); + } if let Some(InstantQualityNotice::SaveFailed) = self.instant_quality_notice { rows.push( div() diff --git a/apps/desktop-gpui/src/snapshot.rs b/apps/desktop-gpui/src/snapshot.rs index ea53a80bfc3..40747f1ba03 100644 --- a/apps/desktop-gpui/src/snapshot.rs +++ b/apps/desktop-gpui/src/snapshot.rs @@ -38,15 +38,29 @@ pub fn arm_from_env(cx: &mut App) { } else if target == "main" { Some(cx.global::().main.into()) } else { - cx.global::() - .onboarding - .map(Into::into) - }; + cx.global::() + .onboarding + .map(Into::into) + }; let Some(handle) = handle else { tracing::warn!("snapshot: window handle missing"); return None; }; - match handle.update(cx, |_, window, _| crate::platform::window_number(window)) { + match handle.update(cx, |_, window, _| { + #[cfg(target_os = "linux")] + { + use raw_window_handle::{HasWindowHandle, RawWindowHandle}; + match HasWindowHandle::window_handle(window).ok()?.as_raw() { + RawWindowHandle::Xlib(handle) => isize::try_from(handle.window).ok(), + RawWindowHandle::Xcb(handle) => { + isize::try_from(handle.window.get()).ok() + } + _ => None, + } + } + #[cfg(not(target_os = "linux"))] + crate::platform::window_number(window) + }) { Ok(number) => number, Err(error) => { tracing::warn!("snapshot: onboarding update failed: {error}"); diff --git a/apps/desktop-gpui/src/upload.rs b/apps/desktop-gpui/src/upload.rs index d8f11085955..c56a92b6970 100644 --- a/apps/desktop-gpui/src/upload.rs +++ b/apps/desktop-gpui/src/upload.rs @@ -42,6 +42,14 @@ struct SignedUploadTarget { headers: HashMap, } +#[derive(Deserialize)] +struct CreatedVideo { + #[serde(flatten)] + config: S3UploadMeta, + #[serde(default, rename = "organizationId")] + organization_id: Option, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum UploadResult { Success(String), @@ -233,7 +241,7 @@ pub async fn prepare_instant_upload( camera_only: bool, project_name: String, organization_id: Option, -) -> Result { +) -> Result<(VideoUploadInfo, Option), String> { if store_auth_missing() { return Err("Please sign in to use instant recording".to_string()); } @@ -260,11 +268,14 @@ pub async fn prepare_instant_upload( error => format!("Could not create the shareable link: {error}"), })?; - Ok(VideoUploadInfo { - id: config.id.clone(), - link: format!("{}/s/{}", auth::server_url(), config.id), - config, - }) + Ok(( + VideoUploadInfo { + id: config.config.id.clone(), + link: format!("{}/s/{}", auth::server_url(), config.config.id), + config: config.config, + }, + config.organization_id, + )) } pub(crate) fn start_instant_upload( @@ -1421,6 +1432,7 @@ async fn create_or_get_video( "desktopMP4", ) .await + .map(|created| created.config) } async fn create_or_get_video_with_mode( @@ -1430,7 +1442,7 @@ async fn create_or_get_video_with_mode( meta: Option<&VideoMeta>, organization_id: Option, recording_mode: &str, -) -> Result { +) -> Result { let mut path = format!( "/api/desktop/video/create?recordingMode={}", urlencoding(recording_mode) @@ -1472,7 +1484,7 @@ async fn create_or_get_video_with_mode( .text() .await .map_err(|error| AuthApiError::Other(format!("Failed to read response body: {error}")))?; - serde_json::from_str::(&text) + serde_json::from_str::(&text) .map_err(|_| AuthApiError::Other("Invalid video creation response".into())) } @@ -2505,6 +2517,19 @@ fn urlencoding(value: &str) -> String { mod tests { use super::*; + #[test] + fn created_video_accepts_legacy_and_organization_bound_responses() { + for organization in [None, Some("org-recording")] { + let mut response = json!({ "id": "video-recording" }); + if let Some(id) = organization { + response["organizationId"] = json!(id); + } + let created: CreatedVideo = serde_json::from_value(response).unwrap(); + assert_eq!(created.config.id, "video-recording"); + assert_eq!(created.organization_id.as_deref(), organization); + } + } + #[tokio::test] async fn presigned_transfer_errors_never_expose_secret_urls() { let client = reqwest::Client::new(); diff --git a/apps/desktop-gpui/upload-health.md b/apps/desktop-gpui/upload-health.md new file mode 100644 index 00000000000..2778f1d8fc0 --- /dev/null +++ b/apps/desktop-gpui/upload-health.md @@ -0,0 +1,105 @@ +# Upload diagnostics + +The signed-in GPUI app checks its upload path at startup and at most once an hour +while idle. The header indicator can request a manual check, with a one-minute +cooldown. Signing out or changing the server invalidates the account-bound result; +saving/deleting S3 settings or connecting, disconnecting, or switching Google Drive +also invalidates it. Changing the selected recording organization invalidates the +result. Checks do not block recording when diagnostics are unavailable. + +## Measurement and resolution + +The client reuses one HTTP client, warms the authenticated connection with a GET, +then sends two 2 MiB binary samples. The server measures the interval between the +first and last nonempty body chunks. Authentication, cold start, storage access, +and object cleanup are outside that interval. A proxy that buffers an entire +sample can produce an inconclusive result; that result preserves the configured +resolution. + +A cap requires two valid samples, both lasting at least 500 ms, whose rates differ +by no more than a factor of two. The faster sample selects the maximum width: + +| Transfer rate | Maximum width | +| --- | --- | +| Below 5 Mbps | 1280 (720p) | +| 5 to below 10 Mbps | 1920 (1080p) | +| 10 to below 20 Mbps | 2560 (1440p) | +| At least 20 Mbps, inconsistent or inconclusive | Configured resolution | + +The effective width is the minimum of this cap and the existing plan/user limit. +Saved settings and Studio recordings are unchanged. A measurement expires after +15 minutes, including time spent asleep; a backwards wall-clock change invalidates +it. The header and Quality settings show an active cap. Once recording starts, +the applied-cap indicator stays visible for that recording even if the measurement +expires. + +Recording admission cancels the active probe and invalidates its generation. +Engine startup waits for the probe future to be dropped before continuing; late +results cannot update the measurement or start another probe during recording. +Failed startup releases admission only after capture cleanup has joined. A cap +also requires the organization returned when creating the recording to match the +probe destination; missing organization information from an older server preserves +the configured resolution. + +## Storage check and bounds + +Only the first sample checks storage. The endpoint uses the recording organization +selection order: a valid saved organization, a valid default, then the earliest +owned or joined organization. Both samples must resolve to the same organization. +After validating current membership, it resolves the same writable access used for recordings +(organization/personal S3, Google Drive, or default storage). It writes at most +1 KiB of synthetic data under `/.cap-upload-health/` and attempts +cleanup even when writing fails. Write failures make the check unavailable; +cleanup failures are logged and do not turn a successful upload into a failure. +The check verifies storage write access, not screen capture or encoder correctness. +Drive probes use a null video ID and the configured root folder, without creating +recording folders or warning files. Ordinary recording uploads retain their existing +folder layout and video association. + +Cookie-authenticated POSTs require same-origin Fetch Metadata, or an exact configured +Origin when Fetch Metadata is absent. Validated API keys remain usable by native +clients without browser headers; malformed authorization cannot bypass this guard. + +Body input is streamed with an 8 MiB limit; the reader retains only the small +storage sample. Storage writes time out after ten seconds, cleanup after three, +and the request after 25 seconds. A check transfers 4 MiB in total. The process +limiter allows four POSTs per authenticated user per minute and retains at most +1024 rate-limit entries. For a distributed deployment, configure the existing +Vercel Firewall integration's `rl_desktop_upload_health` rule with the desired +per-user limit; the process limit alone is not a distributed quota. + +## Verification + +Native Linux UI captures use an isolated profile and synthetic API responses; +they demonstrate rendering, not measured network performance or a real recording: + +- [Slow upload with a visible 720p cap](docs/upload-health-slow-linux.png) +- [Fast upload without a cap](docs/upload-health-fast-linux.png) +- [Failed upload check](docs/upload-health-failed-linux.png) + +```sh +bun run --cwd apps/web test __tests__/unit/upload-health.test.ts __tests__/unit/upload-health-route.test.ts __tests__/unit/desktop-video-create.test.ts __tests__/unit/google-drive-storage.test.ts +./scripts/prepare-gpui-dependency.sh +# Run scripts/setup.js with the same target/native dependencies as desktop CI. +cd apps/desktop-gpui +cargo check --locked -p cap-desktop-gpui +cargo test --locked -p cap-desktop-gpui --bin cap-gpui network_health +cargo test --locked -p cap-desktop-gpui --bin cap-gpui failed_starts_release_upload_checks_only_after_capture_cleanup +cargo test --locked -p cap-desktop-gpui --bin cap-gpui created_video_accepts_legacy_and_organization_bound_responses +``` + +Manual desktop checks before acceptance: + +1. On macOS and Windows, sign in and capture the header and Quality settings with + a successful probe and an applied cap. Confirm the app layout still fits. +2. Compare a fast connection with added RTT against a genuinely throttled upload. + A single slow sample or an inconclusive/buffered sample must not lower quality. +3. Start recording during the first POST, during storage processing, and between + samples. Confirm no later sample starts and recording proceeds after cancellation. +4. Change account, server and selected storage; the old result must not be applied. +5. Make the selected storage unwritable while keeping the app API reachable. The + indicator must report the failed check; merely reaching the app is insufficient. +6. Deny deletion after allowing writes. The upload check stays successful and + cleanup is logged. Operators can remove residual objects under the probe prefix. +7. Check an older self-hosted server without this endpoint: diagnostics are + unavailable and the configured recording resolution is preserved. diff --git a/apps/web/__tests__/unit/desktop-video-create.test.ts b/apps/web/__tests__/unit/desktop-video-create.test.ts index 795c0aaeb93..07688d33575 100644 --- a/apps/web/__tests__/unit/desktop-video-create.test.ts +++ b/apps/web/__tests__/unit/desktop-video-create.test.ts @@ -500,6 +500,7 @@ describe("GET /create", () => { const body = await response.json(); expect(body.id).toBeTruthy(); + expect(body.organizationId).toBe(orgValues?.id); }); it("does not provision when the user already belongs to an org", async () => { @@ -633,6 +634,7 @@ describe("GET /create", () => { ); expect(response.status).toBe(200); + expect((await response.json()).organizationId).toBe("org-1"); expect(mockDb.transaction).toHaveBeenCalledTimes(1); expect(insertedValues(schema.videos)).toMatchObject({ @@ -660,6 +662,7 @@ describe("GET /create", () => { const response = await app.request("https://cap.test/create?orgId=org-2"); expect(response.status).toBe(200); + expect((await response.json()).organizationId).toBe("org-2"); expect(mockDb.transaction).toHaveBeenCalledTimes(1); expect(insertedValues(schema.videos)).toMatchObject({ diff --git a/apps/web/__tests__/unit/google-drive-storage.test.ts b/apps/web/__tests__/unit/google-drive-storage.test.ts index b426c9b57af..a4f13cb01c2 100644 --- a/apps/web/__tests__/unit/google-drive-storage.test.ts +++ b/apps/web/__tests__/unit/google-drive-storage.test.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { Organisation, Storage, User, Video } from "@cap/web-domain"; import { Cause, Effect, Exit, Layer, Option } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { checkUploadStorage } from "@/lib/upload-health"; vi.mock("@cap/env", () => ({ serverEnv: () => ({ @@ -1845,6 +1846,70 @@ const getStorageAccess = (harness: RepoHarness) => { ); }; +describe("Google Drive upload-health objects", () => { + it.each(["video", "userVideo"] as const)( + "probes %s storage without video IDs or recording folders", + async (layout) => { + const harness = makeRepoHarness(); + vi.spyOn(harness.repo, "getGoogleDriveConfig").mockReturnValue( + Effect.succeed(makeConfig(layout)), + ); + const reserve = harness.repo.reserveObject; + vi.spyOn(harness.repo, "reserveObject").mockImplementation((input) => { + if (input.videoId && input.videoId.length > 15) + return Effect.die(new Error("videoId exceeds database width")); + return reserve(input); + }); + let uploaded = false; + const fetchMock = vi.fn( + async (input: string | URL | Request, init?: RequestInit) => { + const url = String(input); + if (url.includes("/files/generateIds")) + return jsonResponse({ ids: ["probe-file"] }); + if (url.includes("uploadType=resumable") && init?.method === "POST") { + expect(requestBody(init)).toMatchObject({ + parents: ["root-folder"], + appProperties: { + capObjectKey: `${videoOwnerId}/.cap-upload-health/probe`, + }, + }); + return response(200, { Location: "https://upload.test/probe" }); + } + if (url === "https://upload.test/probe" && init?.method === "PUT") { + uploaded = true; + return response(200); + } + if (url.includes("/files/probe-file") && init?.method === "DELETE") + return response(204); + throw new Error( + `Unexpected Drive probe request: ${init?.method} ${url}`, + ); + }, + ); + vi.stubGlobal("fetch", fetchMock); + const access = await getStorageAccess(harness); + await Effect.runPromise( + checkUploadStorage({ + userId: videoOwnerId, + bytes: new Uint8Array([1, 2, 3]), + uuid: () => "probe", + put: (key, bytes) => access.putProbeObject(key, bytes), + remove: (key) => access.deleteObject(key), + onCleanupFailure: () => { + throw new Error("Cleanup failed"); + }, + }), + ); + expect(uploaded).toBe(true); + expect(harness.reservations).toHaveLength(1); + expect(harness.reservations[0]?.videoId).toBeNull(); + expect(harness.reservations[0]?.metadata?.videoId).toBeUndefined(); + expect(harness.objects.size).toBe(0); + expect(fetchMock).toHaveBeenCalledTimes(4); + }, + ); +}); + const getStorageObjectExit = async (harness: RepoHarness) => { const access = await getStorageAccess(harness); return Effect.runPromiseExit(access.getObject(resultKey)); diff --git a/apps/web/__tests__/unit/upload-health-route.test.ts b/apps/web/__tests__/unit/upload-health-route.test.ts new file mode 100644 index 00000000000..77c8f95f010 --- /dev/null +++ b/apps/web/__tests__/unit/upload-health-route.test.ts @@ -0,0 +1,333 @@ +import { HttpApiBuilder, HttpServer } from "@effect/platform"; +import { Context, Effect, Layer } from "effect"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const state = vi.hoisted(() => ({ + writes: [] as string[], + deletes: [] as string[], + storageAvailable: true, + cleanupAvailable: true, + organizationAllowed: true, + cookieValid: false, + apiKeyValid: true, + rateChecks: 0, + selectedOrganization: "", + rateLimited: false, +})); + +vi.mock("@cap/web-backend", async () => { + const { Context } = await import("effect"); + return { Storage: Context.GenericTag("UploadHealthTestStorage") }; +}); + +vi.mock("@/lib/rate-limit", () => ({ + isRateLimited: async () => { + state.rateChecks++; + return state.rateLimited; + }, + RATE_LIMIT_IDS: { DESKTOP_UPLOAD_HEALTH: "rl_desktop_upload_health" }, +})); + +vi.mock("@/lib/desktop-recording-organization", async (importOriginal) => { + const { Organisation } = await import("@cap/web-domain"); + const original = + await importOriginal< + typeof import("@/lib/desktop-recording-organization") + >(); + return { + ...original, + getDesktopRecordingOrganization: async ( + _user: string, + requested?: string, + ) => + state.organizationAllowed + ? original.selectDesktopRecordingOrganization( + [{ id: "org-default" }, { id: "org-selected" }].map(({ id }) => ({ + id: Organisation.OrganisationId.make(id), + })), + requested ? Organisation.OrganisationId.make(requested) : undefined, + Organisation.OrganisationId.make("org-default"), + ) + : undefined, + }; +}); + +vi.mock("@cap/env", () => ({ + serverEnv: () => ({ WEB_URL: "https://cap.test" }), +})); +vi.mock("@cap/database/auth/auth-options", () => ({ + getServerSession: async () => + state.cookieValid ? { user: { id: "test-user" } } : null, +})); + +vi.mock("@/lib/server", async () => { + const storage = Context.GenericTag("UploadHealthTestStorage"); + const { HttpAuthMiddlewareLive } = await import( + "../../../../packages/web-backend/src/Auth" + ); + const { Database } = await import( + "../../../../packages/web-backend/src/Database" + ); + const user = { + id: "test-user", + email: "dev@example.test", + activeOrganizationId: "org-dashboard", + image: null, + }; + const database = { + use: (run: (db: unknown) => Promise) => + Effect.promise(() => + run({ + select: () => ({ + from: () => ({ + leftJoin: () => ({ + where: async () => (state.apiKeyValid ? [{ users: user }] : []), + }), + where: async () => [user], + }), + }), + }), + ), + }; + const authentication = HttpAuthMiddlewareLive.pipe( + Layer.provide( + Layer.succeed( + Database, + database as unknown as Context.Tag.Service, + ), + ), + ); + + const access = { + putProbeObject: (key: string) => + Effect.suspend(() => { + state.writes.push(key); + return state.storageAvailable + ? Effect.void + : Effect.fail("write failed"); + }), + deleteObject: (key: string) => + Effect.suspend(() => { + state.deletes.push(key); + return state.cleanupAvailable + ? Effect.void + : Effect.fail("delete failed"); + }), + }; + return { + apiToHandler: (live: Parameters[0]) => + HttpApiBuilder.toWebHandler( + live.pipe( + Layer.provide(authentication), + Layer.provide( + Layer.succeed(storage, { + getWritableAccessForUser: ( + _user: string, + organization: string, + ) => { + state.selectedOrganization = organization; + return Effect.succeed({ access }); + }, + }), + ), + Layer.provide(HttpServer.layerContext), + ), + ).handler, + }; +}); + +import { GET, POST } from "@/app/api/desktop/upload-health/route"; + +function request( + checkStorage = true, + body = new Uint8Array(1024), + token = "12345678-1234-1234-1234-123456789abc", +) { + return new Request( + `http://localhost/api/desktop/upload-health${checkStorage ? "?checkStorage=true" : ""}`, + { + method: "POST", + headers: { authorization: `Bearer ${token}` }, + body, + }, + ); +} + +let clock = 0; +beforeEach(() => { + clock += 60_000; + vi.spyOn(Date, "now").mockReturnValue(clock); + state.writes = []; + state.deletes = []; + state.storageAvailable = true; + state.cleanupAvailable = true; + state.organizationAllowed = true; + state.rateLimited = false; + state.cookieValid = false; + state.apiKeyValid = true; + state.rateChecks = 0; + state.selectedOrganization = ""; +}); + +describe("authenticated upload-health API", () => { + it("rejects unauthenticated probes without touching storage", async () => { + const response = await POST(request(true, new Uint8Array(1), "wrong")); + expect(response.status).toBe(401); + expect(state.writes).toEqual([]); + }); + + it("warms the connection without writing or deleting an object", async () => { + const response = await GET( + new Request("http://localhost/api/desktop/upload-health", { + headers: { + authorization: "Bearer 12345678-1234-1234-1234-123456789abc", + }, + }), + ); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ ok: true }); + expect(state.writes).toEqual([]); + }); + + it("checks the selected storage under the user's prefix and returns body timing", async () => { + const response = await POST(request()); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + bytes: 1024, + bodyReadMs: 0, + organizationId: "org-default", + }); + expect(state.writes[0]).toMatch(/^test-user\/\.cap-upload-health\//); + expect(state.deletes).toEqual(state.writes); + }); + + it("measures subsequent samples without another storage write", async () => { + const response = await POST(request(false)); + expect(response.status).toBe(200); + expect(state.writes).toEqual([]); + }); + + it("reports no available organization without touching storage", async () => { + state.organizationAllowed = false; + const response = await POST(request()); + expect(response.status).toBe(503); + expect(state.writes).toEqual([]); + }); + + it("returns unavailable when the chosen storage cannot accept a write", async () => { + state.storageAvailable = false; + const response = await POST(request()); + expect(response.status).toBe(503); + }); + + it("rejects empty and oversized streamed bodies without a storage write", async () => { + expect((await POST(request(true, new Uint8Array()))).status).toBe(400); + expect( + (await POST(request(true, new Uint8Array(8 * 1024 * 1024 + 1)))).status, + ).toBe(413); + expect(state.writes).toEqual([]); + }); + + it("rejects an oversized Content-Length before reading or writing", async () => { + const probe = request(); + probe.headers.set("content-length", String(8 * 1024 * 1024 + 1)); + expect((await POST(probe)).status).toBe(413); + expect(state.writes).toEqual([]); + }); + + it("keeps upload availability successful when object cleanup fails", async () => { + state.cleanupAvailable = false; + vi.spyOn(console, "warn").mockImplementation(() => {}); + expect((await POST(request())).status).toBe(200); + expect(console.warn).toHaveBeenCalledOnce(); + }); + + it("rate-limits repeated authenticated probes", async () => { + for (let index = 0; index < 4; index++) { + expect((await POST(request(false))).status).toBe(200); + } + expect((await POST(request())).status).toBe(429); + expect(state.writes).toEqual([]); + }); + + it("rejects a distributed rate-limit decision before writing", async () => { + state.rateLimited = true; + expect((await POST(request())).status).toBe(429); + expect(state.writes).toEqual([]); + }); +}); + +describe("upload probe credential boundary", () => { + const cookieProbe = (headers: Record = {}) => { + state.cookieValid = true; + return new Request( + "https://cap.test/api/desktop/upload-health?checkStorage=true", + { + method: "POST", + headers: { + cookie: "next-auth.session-token=valid", + "content-type": "text/plain", + ...headers, + }, + body: "probe", + }, + ); + }; + it.each>([ + {}, + { origin: "https://attacker.test" }, + { origin: "null" }, + { "sec-fetch-site": "cross-site", origin: "https://attacker.test" }, + { "sec-fetch-site": "same-site", origin: "https://sub.cap.test" }, + { "sec-fetch-site": "none", origin: "https://cap.test" }, + { authorization: "Bearer invalid", origin: "https://attacker.test" }, + { authorization: "Bearer", origin: "https://attacker.test" }, + ])( + "rejects ambient cookie POSTs before rate limiting/storage: %j", + async (headers) => { + expect((await POST(cookieProbe(headers))).status).toBe(403); + expect(state.rateChecks).toBe(0); + expect(state.writes).toEqual([]); + expect(state.deletes).toEqual([]); + }, + ); + it.each>([ + { "sec-fetch-site": "same-origin" }, + { origin: "https://cap.test" }, + ])("allows legitimate same-origin cookies: %j", async (headers) => { + expect((await POST(cookieProbe(headers))).status).toBe(200); + expect(state.writes).toHaveLength(1); + }); + it("does not fall back to a cookie for an invalid API key", async () => { + state.apiKeyValid = false; + expect( + ( + await POST( + cookieProbe({ + authorization: "Bearer 00000000-0000-0000-0000-000000000000", + origin: "https://attacker.test", + }), + ) + ).status, + ).toBe(401); + expect(state.writes).toEqual([]); + }); + it("checks the saved desktop organization instead of the dashboard organization", async () => { + const probe = request(); + const url = new URL(probe.url); + url.searchParams.set("orgId", "org-selected"); + const response = await POST(new Request(url, probe)); + expect(response.status).toBe(200); + expect(state.selectedOrganization).toBe("org-selected"); + expect(await response.json()).toMatchObject({ + organizationId: "org-selected", + }); + }); + it("uses the default organization for a stale desktop selection", async () => { + const probe = request(); + const url = new URL(probe.url); + url.searchParams.set("orgId", "departed-org"); + expect((await POST(new Request(url, probe))).status).toBe(200); + expect(state.selectedOrganization).toBe("org-default"); + }); +}); diff --git a/apps/web/__tests__/unit/upload-health.test.ts b/apps/web/__tests__/unit/upload-health.test.ts new file mode 100644 index 00000000000..34d75b5c2f6 --- /dev/null +++ b/apps/web/__tests__/unit/upload-health.test.ts @@ -0,0 +1,138 @@ +import { Data, Effect } from "effect"; +import { describe, expect, it, vi } from "vitest"; +import { + checkUploadStorage, + MAX_UPLOAD_PROBE_BYTES, + UploadProbeBody, + UploadProbeLimiter, + UploadProbeTooLarge, +} from "@/lib/upload-health"; + +class TestStorageError extends Data.TaggedError("TestStorageError")<{ + message: string; +}> {} + +describe("upload measurement", () => { + it("times body transfer separately from cold-start and storage latency", () => { + let now = 12_000; + const probe = new UploadProbeBody(() => now); + now += 4000; + probe.append(new Uint8Array(1024)); + now += 800; + probe.append(new Uint8Array(1024)); + now += 30_000; + expect(probe.result()).toEqual({ bytes: 2048, bodyReadMs: 800 }); + }); + + it("reports buffered or empty bodies as inconclusive timing", () => { + const probe = new UploadProbeBody(() => 10); + expect(probe.result()).toEqual({ bytes: 0, bodyReadMs: 0 }); + probe.append(new Uint8Array(1024)); + expect(probe.result()).toEqual({ bytes: 1024, bodyReadMs: 0 }); + }); + + it("bounds streamed bodies even when Content-Length is absent or false", () => { + const probe = new UploadProbeBody(); + probe.append(new Uint8Array(MAX_UPLOAD_PROBE_BYTES)); + expect(() => probe.append(new Uint8Array(1))).toThrow(UploadProbeTooLarge); + expect(probe.result().bytes).toBe(MAX_UPLOAD_PROBE_BYTES); + expect(probe.sample().byteLength).toBe(1024); + }); + + it("does not retain the upload or an alias of its storage sample", () => { + const probe = new UploadProbeBody(); + const bytes = new Uint8Array(2048).fill(7); + probe.append(bytes); + bytes.fill(0); + expect(probe.sample()).toEqual(new Uint8Array(1024).fill(7)); + }); +}); + +describe("selected-storage health check", () => { + const fixture = () => ({ + userId: "owner", + bytes: new Uint8Array([1, 2, 3]), + put: vi.fn((): Effect.Effect => Effect.void), + remove: vi.fn((): Effect.Effect => Effect.void), + onCleanupFailure: vi.fn(), + uuid: () => "test-id", + }); + + it("writes under the authenticated user's prefix and cleans up", async () => { + const deps = fixture(); + await Effect.runPromise(checkUploadStorage(deps)); + expect(deps.put).toHaveBeenCalledWith( + "owner/.cap-upload-health/test-id", + deps.bytes, + ); + expect(deps.remove).toHaveBeenCalledWith( + "owner/.cap-upload-health/test-id", + ); + }); + + it("keeps a successful write healthy when cleanup fails", async () => { + const deps = fixture(); + deps.remove.mockReturnValue( + Effect.fail(new TestStorageError({ message: "delete denied" })), + ); + await expect( + Effect.runPromise(checkUploadStorage(deps)), + ).resolves.toBeUndefined(); + expect(deps.onCleanupFailure).toHaveBeenCalledOnce(); + }); + + it("reports an unavailable storage destination and still attempts cleanup", async () => { + const deps = fixture(); + deps.put.mockReturnValue( + Effect.fail(new TestStorageError({ message: "write failed" })), + ); + await expect(Effect.runPromise(checkUploadStorage(deps))).rejects.toThrow( + "write failed", + ); + expect(deps.remove).toHaveBeenCalledOnce(); + }); +}); + +describe("storage effect execution", () => { + it("executes cleanup after the write effect fails", async () => { + const events: string[] = []; + await expect( + Effect.runPromise( + checkUploadStorage({ + userId: "owner", + bytes: new Uint8Array([1]), + put: () => + Effect.sync(() => events.push("write")).pipe( + Effect.andThen( + Effect.fail(new TestStorageError({ message: "failed" })), + ), + ), + remove: () => + Effect.sync(() => { + events.push("delete"); + }), + onCleanupFailure: () => {}, + }), + ), + ).rejects.toThrow("failed"); + expect(events).toEqual(["write", "delete"]); + }); +}); + +describe("upload probe rate limit", () => { + it("allows two samples and a retry, then rejects repeated checks", () => { + const limiter = new UploadProbeLimiter(); + for (let i = 0; i < 4; i++) expect(limiter.admit("owner", 0)).toBe(true); + expect(limiter.admit("owner", 59_999)).toBe(false); + expect(limiter.admit("other", 59_999)).toBe(true); + expect(limiter.admit("owner", 60_000)).toBe(true); + }); + + it("bounds its per-process state and expires old entries", () => { + const limiter = new UploadProbeLimiter(); + for (let i = 0; i < 1024; i++) + expect(limiter.admit(String(i), 0)).toBe(true); + expect(limiter.admit("extra", 0)).toBe(false); + expect(limiter.admit("extra", 60_000)).toBe(true); + }); +}); diff --git a/apps/web/app/api/desktop/[...route]/video.ts b/apps/web/app/api/desktop/[...route]/video.ts index 8565e429fd9..25981a2379e 100644 --- a/apps/web/app/api/desktop/[...route]/video.ts +++ b/apps/web/app/api/desktop/[...route]/video.ts @@ -21,6 +21,10 @@ import { Effect, Option } from "effect"; import { Hono } from "hono"; import { after } from "next/server"; import { z } from "zod"; +import { + mergeUserOrganizationSelections, + selectDesktopRecordingOrganization, +} from "@/lib/desktop-recording-organization"; import { invalidateGoogleDriveStorageQuotaCache } from "@/lib/google-drive-storage-quota"; import { maybeStartLiveTranscription } from "@/lib/live-transcribe"; import { runPromise } from "@/lib/server"; @@ -35,28 +39,6 @@ import { withAuth } from "../../utils"; export const app = new Hono().use(withAuth); -type UserOrganizationSelection = { - id: Organisation.OrganisationId; - name: string; - createdAt: Date; -}; - -function mergeUserOrganizationSelections( - ...rowSets: UserOrganizationSelection[][] -) { - const organizationsById = new Map(); - - for (const rows of rowSets) { - for (const row of rows) { - organizationsById.set(row.id, row); - } - } - - return Array.from(organizationsById.values()) - .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()) - .map(({ createdAt, ...organization }) => organization); -} - app.get("/new-id", (c) => c.json({ id: nanoId() })); app.get( @@ -239,35 +221,22 @@ app.get( userOrganizations.push({ id: organizationId, name: "My Organization" }); } - const userOrgIds = userOrganizations.map((org) => org.id); - - let videoOrgId: Organisation.OrganisationId; - // Desktop persists orgId in settings and keeps sending it after the - // user leaves the org, so an unknown orgId falls back to the - // default/first org below instead of hard-failing every upload. - if (orgId && userOrgIds.includes(orgId)) { - videoOrgId = orgId; - } else if (user.defaultOrgId) { - // User's defaultOrgId is no longer valid, switch to first available org - if (!userOrgIds.includes(user.defaultOrgId)) { - if (!userOrganizations[0]) - return c.json({ error: "no_valid_org" }, { status: 403 }); - - videoOrgId = userOrganizations[0].id; - - // Update user's defaultOrgId to the new valid org - await db() - .update(users) - .set({ - defaultOrgId: videoOrgId, - }) - .where(eq(users.id, user.id)); - } else videoOrgId = user.defaultOrgId; - } else { - // No orgId provided and no defaultOrgId, use first available org - if (!userOrganizations[0]) - return c.json({ error: "no_valid_org" }, { status: 403 }); - videoOrgId = userOrganizations[0].id; + const videoOrgId = selectDesktopRecordingOrganization( + userOrganizations, + orgId, + user.defaultOrgId, + ); + if (!videoOrgId) + return c.json({ error: "no_valid_org" }, { status: 403 }); + if ( + orgId !== videoOrgId && + user.defaultOrgId && + !userOrganizations.some((org) => org.id === user.defaultOrgId) + ) { + await db() + .update(users) + .set({ defaultOrgId: videoOrgId }) + .where(eq(users.id, user.id)); } const idToUse = Video.VideoId.make( @@ -391,6 +360,7 @@ app.get( return c.json({ id: idToUse, + organizationId: videoOrgId, // All deprecated user_id: user.id, aws_region: "n/a", diff --git a/apps/web/app/api/desktop/upload-health/route.ts b/apps/web/app/api/desktop/upload-health/route.ts new file mode 100644 index 00000000000..725c997c74b --- /dev/null +++ b/apps/web/app/api/desktop/upload-health/route.ts @@ -0,0 +1,166 @@ +import { serverEnv } from "@cap/env"; +import { Storage } from "@cap/web-backend"; +import { CurrentUser, HttpAuthMiddleware, Organisation } from "@cap/web-domain"; +import { + HttpApi, + HttpApiBuilder, + HttpApiEndpoint, + HttpApiError, + HttpApiGroup, + HttpApiSchema, + HttpServerRequest, +} from "@effect/platform"; +import { Effect, Layer, Schema, Stream } from "effect"; +import { getDesktopRecordingOrganization } from "@/lib/desktop-recording-organization"; +import { isRateLimited, RATE_LIMIT_IDS } from "@/lib/rate-limit"; +import { apiToHandler } from "@/lib/server"; +import { + checkUploadStorage, + MAX_UPLOAD_PROBE_BYTES, + UploadProbeBody, + UploadProbeLimiter, + UploadProbeTooLarge, +} from "@/lib/upload-health"; + +class ProbeTooLarge extends HttpApiSchema.EmptyError()({ + tag: "ProbeTooLarge", + status: 413, +}) {} + +class ProbeRateLimited extends HttpApiSchema.EmptyError()({ + tag: "ProbeRateLimited", + status: 429, +}) {} + +class Api extends HttpApi.make("DesktopUploadHealth").add( + HttpApiGroup.make("root") + .add( + HttpApiEndpoint.get("warm")`/api/desktop/upload-health`.addSuccess( + Schema.Struct({ ok: Schema.Literal(true) }), + ), + ) + .add( + HttpApiEndpoint.post("probe")`/api/desktop/upload-health` + .setUrlParams( + Schema.Struct({ + checkStorage: Schema.optional(Schema.Literal("true")), + orgId: Schema.optional(Organisation.OrganisationId), + }), + ) + .addSuccess( + Schema.Struct({ + bytes: Schema.Number, + bodyReadMs: Schema.Number, + organizationId: Organisation.OrganisationId, + }), + ) + .addError(ProbeTooLarge) + .addError(ProbeRateLimited) + .addError(HttpApiError.BadRequest) + .addError(HttpApiError.Forbidden) + .addError(HttpApiError.ServiceUnavailable) + .addError(HttpApiError.RequestTimeout), + ) + .middleware(HttpAuthMiddleware), +) {} + +const limiter = new UploadProbeLimiter(); + +const ApiLive = HttpApiBuilder.api(Api).pipe( + Layer.provide( + HttpApiBuilder.group(Api, "root", (handlers) => + handlers + .handle("warm", () => Effect.succeed({ ok: true as const })) + .handle("probe", ({ urlParams }) => + Effect.gen(function* () { + const user = yield* CurrentUser; + const request = yield* HttpServerRequest.HttpServerRequest; + const apiKey = request.headers.authorization?.split(" ")[1]; + if (apiKey?.length !== 36) { + const site = request.headers["sec-fetch-site"]; + if ( + site !== undefined + ? site !== "same-origin" + : request.headers.origin !== + new URL(serverEnv().WEB_URL).origin + ) { + return yield* new HttpApiError.Forbidden(); + } + } + if ( + !limiter.admit(user.id) || + (yield* Effect.promise(() => + isRateLimited(RATE_LIMIT_IDS.DESKTOP_UPLOAD_HEALTH, { + key: user.id, + headers: new Headers(request.headers), + }), + )) + ) { + return yield* new ProbeRateLimited(); + } + if ( + Number(request.headers["content-length"]) > MAX_UPLOAD_PROBE_BYTES + ) { + return yield* new ProbeTooLarge(); + } + const probe = new UploadProbeBody(); + yield* request.stream.pipe( + Stream.runForEach((chunk) => + Effect.try({ + try: () => probe.append(chunk), + catch: (error) => + error instanceof UploadProbeTooLarge + ? new ProbeTooLarge() + : new HttpApiError.BadRequest(), + }), + ), + Effect.catchTag( + "RequestError", + () => new HttpApiError.BadRequest(), + ), + ); + const result = probe.result(); + if (!result.bytes) return yield* new HttpApiError.BadRequest(); + const organizationId = yield* Effect.tryPromise({ + try: () => + getDesktopRecordingOrganization(user.id, urlParams.orgId), + catch: () => new HttpApiError.ServiceUnavailable(), + }); + if (!organizationId) + return yield* new HttpApiError.ServiceUnavailable(); + if (urlParams.checkStorage) { + const storage = yield* Storage; + const { access } = yield* storage + .getWritableAccessForUser(user.id, organizationId) + .pipe( + Effect.mapError(() => new HttpApiError.ServiceUnavailable()), + ); + yield* checkUploadStorage({ + userId: user.id, + bytes: probe.sample(), + put: (key, bytes) => + access.putProbeObject(key, Buffer.from(bytes)), + remove: (key) => access.deleteObject(key), + onCleanupFailure: () => + console.warn("Upload-health probe cleanup failed"), + }).pipe( + Effect.mapError(() => new HttpApiError.ServiceUnavailable()), + ); + } + return { ...result, organizationId }; + }).pipe( + Effect.timeoutFail({ + duration: "25 seconds", + onTimeout: () => new HttpApiError.RequestTimeout(), + }), + ), + ), + ), + ), +); + +const handler = apiToHandler(ApiLive); +export const GET = handler; +export const POST = handler; +export const OPTIONS = handler; +export const maxDuration = 30; diff --git a/apps/web/lib/desktop-recording-organization.ts b/apps/web/lib/desktop-recording-organization.ts new file mode 100644 index 00000000000..25690027988 --- /dev/null +++ b/apps/web/lib/desktop-recording-organization.ts @@ -0,0 +1,87 @@ +import { db } from "@cap/database"; +import { + organizationMembers, + organizations, + users, +} from "@cap/database/schema"; +import type { Organisation, User } from "@cap/web-domain"; +import { and, eq, isNull } from "drizzle-orm"; + +type UserOrganizationSelection = { + id: Organisation.OrganisationId; + name: string; + createdAt: Date; +}; + +export function mergeUserOrganizationSelections( + ...rowSets: UserOrganizationSelection[][] +) { + const byId = new Map(); + for (const rows of rowSets) { + for (const row of rows) byId.set(row.id, row); + } + return Array.from(byId.values()) + .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()) + .map(({ createdAt, ...organization }) => organization); +} + +export function selectDesktopRecordingOrganization( + available: Array<{ id: Organisation.OrganisationId }>, + requested?: Organisation.OrganisationId, + defaultId?: Organisation.OrganisationId | null, +) { + return ( + [requested, defaultId].find( + (id): id is Organisation.OrganisationId => + !!id && available.some((organization) => organization.id === id), + ) ?? available[0]?.id + ); +} + +export async function getDesktopRecordingOrganization( + userId: User.UserId, + requested?: Organisation.OrganisationId, +) { + const [[user], owned, member] = await Promise.all([ + db() + .select({ defaultOrgId: users.defaultOrgId }) + .from(users) + .where(eq(users.id, userId)) + .limit(1), + db() + .select({ + id: organizations.id, + name: organizations.name, + createdAt: organizations.createdAt, + }) + .from(organizations) + .where( + and( + eq(organizations.ownerId, userId), + isNull(organizations.tombstoneAt), + ), + ), + db() + .select({ + id: organizations.id, + name: organizations.name, + createdAt: organizations.createdAt, + }) + .from(organizationMembers) + .innerJoin( + organizations, + eq(organizations.id, organizationMembers.organizationId), + ) + .where( + and( + eq(organizationMembers.userId, userId), + isNull(organizations.tombstoneAt), + ), + ), + ]); + return selectDesktopRecordingOrganization( + mergeUserOrganizationSelections(owned, member), + requested, + user?.defaultOrgId, + ); +} diff --git a/apps/web/lib/rate-limit.ts b/apps/web/lib/rate-limit.ts index bd28a25c4f0..5ea454cbc22 100644 --- a/apps/web/lib/rate-limit.ts +++ b/apps/web/lib/rate-limit.ts @@ -95,6 +95,7 @@ export const RATE_LIMIT_IDS = { GUEST_CHECKOUT: "rl_guest_checkout", /** Unauthed desktop log → Discord forwarding (spam). Suggested: 10 / 1m per IP. */ DESKTOP_LOGS: "rl_desktop_logs", + DESKTOP_UPLOAD_HEALTH: "rl_desktop_upload_health", /** Unauthed docs Ask AI questions (Anthropic cost). Suggested: 10 / 1m per IP. */ DOCS_ASK: "rl_docs_ask", } as const; diff --git a/apps/web/lib/upload-health.ts b/apps/web/lib/upload-health.ts new file mode 100644 index 00000000000..8377e36a0ce --- /dev/null +++ b/apps/web/lib/upload-health.ts @@ -0,0 +1,86 @@ +import { Effect } from "effect"; + +export const MAX_UPLOAD_PROBE_BYTES = 8 * 1024 * 1024; + +export class UploadProbeTooLarge extends Error {} + +export class UploadProbeBody { + private startedAt: number | undefined; + private finishedAt: number | undefined; + private bytes = 0; + private storageSample = new Uint8Array(); + + constructor(private readonly now: () => number = () => performance.now()) {} + + append(chunk: Uint8Array) { + if (this.bytes + chunk.byteLength > MAX_UPLOAD_PROBE_BYTES) { + throw new UploadProbeTooLarge(); + } + if (!chunk.byteLength) return; + const now = this.now(); + this.startedAt ??= now; + this.finishedAt = now; + if (!this.bytes) this.storageSample = chunk.slice(0, 1024); + this.bytes += chunk.byteLength; + } + + result() { + return { + bytes: this.bytes, + bodyReadMs: Math.max(0, (this.finishedAt ?? 0) - (this.startedAt ?? 0)), + }; + } + + sample() { + return this.storageSample; + } +} + +export function checkUploadStorage({ + userId, + bytes, + put, + remove, + onCleanupFailure, + uuid = () => crypto.randomUUID(), +}: { + userId: string; + bytes: Uint8Array; + put: (key: string, bytes: Uint8Array) => Effect.Effect; + remove: (key: string) => Effect.Effect; + onCleanupFailure: (error: unknown) => void; + uuid?: () => string; +}) { + const key = `${userId}/.cap-upload-health/${uuid()}`; + return put(key, bytes).pipe( + Effect.timeout("10 seconds"), + Effect.ensuring( + remove(key).pipe( + Effect.timeout("3 seconds"), + Effect.catchAll((error) => Effect.sync(() => onCleanupFailure(error))), + ), + ), + ); +} + +export class UploadProbeLimiter { + private readonly windows = new Map< + string, + { until: number; count: number } + >(); + + admit(userId: string, now = Date.now()) { + for (const [key, window] of this.windows) { + if (window.until <= now) this.windows.delete(key); + } + const window = this.windows.get(userId); + if (window) { + if (window.count >= 4) return false; + window.count++; + return true; + } + if (this.windows.size >= 1024) return false; + this.windows.set(userId, { until: now + 60_000, count: 1 }); + return true; + } +} diff --git a/crates/cap-muxer/src/main.rs b/crates/cap-muxer/src/main.rs index fc7092ff31b..73629a7aed6 100644 --- a/crates/cap-muxer/src/main.rs +++ b/crates/cap-muxer/src/main.rs @@ -454,6 +454,11 @@ fn open_output(state: &mut State, params: &StartParams) -> Result, Duration)>>, + video_tx: flume::Sender, Duration)>>, encoder: Arc>, encoder_handle: Option>>, } @@ -247,11 +243,11 @@ impl Muxer for MacOSFragmentedM4SMuxer { std::thread::sleep(Duration::from_millis(50)); match state.video_tx.try_send(None) { Ok(()) => return, - Err(std::sync::mpsc::TrySendError::Disconnected(_)) => { + Err(flume::TrySendError::Disconnected(_)) => { trace!("M4S encoder channel closed during stop retry"); return; } - Err(std::sync::mpsc::TrySendError::Full(_)) => {} + Err(flume::TrySendError::Full(_)) => {} } } warn!("M4S encoder channel still full after retries, finish() will deliver sentinel"); @@ -288,7 +284,7 @@ impl MacOSFragmentedM4SMuxer { ); let (video_tx, video_rx) = - sync_channel::, Duration)>>(buffer_size); + flume::bounded::, Duration)>>(buffer_size); let (ready_tx, ready_rx) = sync_channel::>(1); let encoder_config = SegmentedVideoEncoderConfig { @@ -514,18 +510,18 @@ impl VideoMuxer for MacOSFragmentedM4SMuxer { } if let Some(state) = &self.state { - match state - .video_tx - .try_send(Some((frame.sample_buf, adjusted_timestamp))) - { + match state.video_tx.send_timeout( + Some((frame.sample_buf, adjusted_timestamp)), + Duration::from_millis(super::core::STALL_BUDGET_MS), + ) { Ok(()) => { self.frame_drops.record_frame(); } Err(e) => match e { - std::sync::mpsc::TrySendError::Full(_) => { + flume::SendTimeoutError::Timeout(_) => { self.frame_drops.record_drop(); } - std::sync::mpsc::TrySendError::Disconnected(_) => { + flume::SendTimeoutError::Disconnected(_) => { return Err(anyhow!("M4S encoder channel disconnected")); } }, @@ -636,11 +632,11 @@ impl Muxer for MacOSFragmentedM4SCameraMuxer { std::thread::sleep(Duration::from_millis(50)); match state.video_tx.try_send(None) { Ok(()) => return, - Err(std::sync::mpsc::TrySendError::Disconnected(_)) => { + Err(flume::TrySendError::Disconnected(_)) => { trace!("M4S camera encoder channel closed during stop retry"); return; } - Err(std::sync::mpsc::TrySendError::Full(_)) => {} + Err(flume::TrySendError::Full(_)) => {} } } warn!( @@ -679,7 +675,7 @@ impl MacOSFragmentedM4SCameraMuxer { ); let (video_tx, video_rx) = - sync_channel::, Duration)>>(buffer_size); + flume::bounded::, Duration)>>(buffer_size); let (ready_tx, ready_rx) = sync_channel::>(1); let encoder_config = SegmentedVideoEncoderConfig { @@ -860,18 +856,18 @@ impl VideoMuxer for MacOSFragmentedM4SCameraMuxer { } if let Some(state) = &self.state { - match state - .video_tx - .try_send(Some((frame.sample_buf, adjusted_timestamp))) - { + match state.video_tx.send_timeout( + Some((frame.sample_buf, adjusted_timestamp)), + Duration::from_millis(super::core::STALL_BUDGET_MS), + ) { Ok(()) => { self.frame_drops.record_frame(); } Err(e) => match e { - std::sync::mpsc::TrySendError::Full(_) => { + flume::SendTimeoutError::Timeout(_) => { self.frame_drops.record_drop(); } - std::sync::mpsc::TrySendError::Disconnected(_) => { + flume::SendTimeoutError::Disconnected(_) => { return Err(anyhow!("M4S camera encoder channel disconnected")); } }, diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs index 9afbb643412..256989abc8e 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs @@ -107,7 +107,7 @@ impl FrameDropTracker { type VideoFrameItem = Option<(cidre::arc::R, Duration)>; struct EncoderState { - video_tx: std::sync::mpsc::SyncSender, + video_tx: flume::Sender, encoder_handle: Option>>, } @@ -205,11 +205,11 @@ impl Muxer for OutOfProcessFragmentedM4SMuxer { std::thread::sleep(Duration::from_millis(50)); match state.video_tx.try_send(None) { Ok(()) => return, - Err(std::sync::mpsc::TrySendError::Disconnected(_)) => { + Err(flume::TrySendError::Disconnected(_)) => { trace!("OOP M4S encoder channel closed during stop retry"); return; } - Err(std::sync::mpsc::TrySendError::Full(_)) => {} + Err(flume::TrySendError::Full(_)) => {} } } warn!( @@ -261,7 +261,7 @@ impl OutOfProcessFragmentedM4SMuxer { let bin_path = resolve_muxer_binary().with_context(|| "cap-muxer binary not found for OOP muxer")?; - let (video_tx, video_rx) = std::sync::mpsc::sync_channel::(buffer_size); + let (video_tx, video_rx) = flume::bounded::(buffer_size); let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::>(1); let base_path = self.base_path.clone(); @@ -659,18 +659,18 @@ impl VideoMuxer for OutOfProcessFragmentedM4SMuxer { } if let Some(state) = &self.state { - match state - .video_tx - .try_send(Some((frame.sample_buf, adjusted_timestamp))) - { + match state.video_tx.send_timeout( + Some((frame.sample_buf, adjusted_timestamp)), + Duration::from_millis(super::core::STALL_BUDGET_MS), + ) { Ok(()) => { self.frame_drops.record_frame(); } Err(e) => match e { - std::sync::mpsc::TrySendError::Full(_) => { + flume::SendTimeoutError::Timeout(_) => { self.frame_drops.record_drop(); } - std::sync::mpsc::TrySendError::Disconnected(_) => { + flume::SendTimeoutError::Disconnected(_) => { return Err(anyhow!("OOP M4S encoder channel disconnected")); } }, diff --git a/crates/recording/tests/oop_muxer.rs b/crates/recording/tests/oop_muxer.rs index bd7b889bfbc..dc26743cd14 100644 --- a/crates/recording/tests/oop_muxer.rs +++ b/crates/recording/tests/oop_muxer.rs @@ -52,6 +52,104 @@ fn minimal_video_config(output_dir: &std::path::Path, extradata: Vec) -> Mux } } +#[test] +fn sparse_subprocess_flushes_recoverable_media_without_extra_keyframes() { + use cap_enc_ffmpeg::{ + fragmented_mp4::read_fragment_prefix, + h264::{H264EncoderBuilder, H264Preset}, + }; + use cap_media_info::{Pixel, VideoInfo}; + use std::time::{Duration, Instant}; + + ffmpeg::init().unwrap(); + let directory = TempDir::new().unwrap(); + let output = directory.path().join("video"); + let info = VideoInfo { + pixel_format: Pixel::NV12, + width: TEST_WIDTH, + height: TEST_HEIGHT, + time_base: ffmpeg::Rational(1, 90_000), + frame_rate: ffmpeg::Rational(TEST_FPS as i32, 1), + }; + let mut encoder = H264EncoderBuilder::new(info) + .with_preset(H264Preset::Ultrafast) + .with_encoder_priority_override(&["libx264"]) + .build_standalone() + .unwrap(); + let config = minimal_video_config(&output, encoder.extradata()); + let mut packets = Vec::new(); + for index in 0..37 { + let mut frame = ffmpeg::frame::Video::new(Pixel::NV12, TEST_WIDTH, TEST_HEIGHT); + for plane in 0..frame.planes() { + frame.data_mut(plane).fill(128); + } + encoder + .encode_frame( + frame, + Duration::from_micros(index * 1_000_000 / 3), + |packet| { + packets.push(packet); + Ok(()) + }, + ) + .unwrap(); + } + encoder + .flush(|packet| { + packets.push(packet); + Ok(()) + }) + .unwrap(); + assert_eq!(packets.len(), 37); + assert_eq!( + packets.iter().filter(|packet| packet.is_keyframe).count(), + 1 + ); + let mut subprocess = MuxerSubprocess::spawn(setup_muxer_binary(), config, None).unwrap(); + for packet in &packets { + subprocess + .write_video_packet( + packet.pts, + packet.dts, + packet.duration.max(0) as u64, + packet.is_keyframe, + &packet.data, + ) + .unwrap(); + } + let init = output.join("init.mp4"); + let pending = output.join("segment_001.m4s.tmp"); + let started = Instant::now(); + loop { + if read_fragment_prefix(&init, &pending) + .is_ok_and(|prefix| prefix.duration >= Duration::from_secs(10)) + { + break; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "complete fragments were not flushed before EOF" + ); + std::thread::sleep(Duration::from_millis(10)); + } + let report = subprocess.finish().unwrap(); + assert_eq!(report.packets_written, 37); + let mut bytes = std::fs::read(init).unwrap(); + bytes.extend(std::fs::read(output.join("segment_001.m4s")).unwrap()); + let assembled = directory.path().join("assembled.mp4"); + std::fs::write(&assembled, bytes).unwrap(); + let mut input = ffmpeg::format::input(&assembled).unwrap(); + let output_packets: Vec<_> = input.packets().map(|(_, packet)| packet).collect(); + assert_eq!(output_packets.len(), 37); + assert_eq!( + output_packets + .iter() + .filter(|packet| packet.is_key()) + .count(), + 1 + ); +} + #[test] fn subprocess_spawns_and_finishes_cleanly_without_packets() { let bin = setup_muxer_binary(); diff --git a/crates/verify/expectations.toml b/crates/verify/expectations.toml index ea47214b2ef..94233df4a7f 100644 --- a/crates/verify/expectations.toml +++ b/crates/verify/expectations.toml @@ -361,23 +361,3 @@ hosts = ["github-hosted", "local"] bound = { count = 1, screen-pictures = 1 } reason = "NEW (ScreenCaptureKit idle resume, see above): the first 15 fps picture after the 3 s hold is missing when the resume takes longer than its 67 ms; runners 5 of 6 runs (flaky because the 70 to 82 ms resume straddles the 67 ms picture). Bound: the one hold's picture" link = "https://github.com/CapSoftware/Cap/actions/runs/38745219281" -# Crash safety (crates/verify/tests/crash.rs). Measured on Blacksmith runners -# (run 38850661948: macos-15 libx264, ubuntu-2404, windows-2025) with the suite -# merged into main 3ec0c9497; each fix has its own proof branch -# ci/fixproof-crash- (runs 38850662667, 38850661576, 38850663436, -# 38850662102), which removes exactly its entries. Fixes: #2640 (bug A), -# #2641 (bug B1), #2643 (bug C), #2644 (bug E). - -[[xfail]] -case = "crash/recorder-kill/inproc/static-screen" -class = "loss-time" -bound = { count = 1, secs = [2.5, 20.5] } -reason = "NEW (bug B1): fragments are cut only at keyframes and every recording encoder's keyframe interval is a frame count (g = 2 s x nominal fps; libx264 keyint 60 here, VideoToolbox 12), so when the screen delivers 3 frames per second a fragment spans 20 s of capture and a crash after 11 s loses all of it (3 of 3 platforms; macOS runners use libx264). Bound: at most one 20 s fragment of capture" -link = "https://github.com/CapSoftware/Cap/pull/2641" - -[[xfail]] -case = "crash/recorder-kill/inproc/static-screen" -class = "missing" -bound = { count = 1 } -reason = "NEW (bug B1, see loss-time): no fragment of the static recording is complete when the recorder dies, so recovery has nothing (no frames)" -link = "https://github.com/CapSoftware/Cap/pull/2641" diff --git a/crates/verify/tests/crash.rs b/crates/verify/tests/crash.rs index 33c3b8c2061..b916c42881c 100644 --- a/crates/verify/tests/crash.rs +++ b/crates/verify/tests/crash.rs @@ -16,9 +16,9 @@ //! against what is physically on disk after the failure: every fragment whose //! box structure is complete must come back frame for frame at its capture //! time. Frames lost to a failure must be the newest ones and fit in the -//! fragment being written: one GOP (`GOP_SECS` x fps frames, the encoder's -//! keyframe interval, so the longest a segment can be when frames arrive at -//! the nominal rate) plus `IN_FLIGHT` frames still inside the pipeline. +//! timed fragment being written plus `IN_FLIGHT` frames still inside the +//! pipeline. A segment can contain several recoverable moof/mdat pairs +//! without introducing extra keyframes. mod common; @@ -46,11 +46,10 @@ use cap_verify::{ const FPS: u32 = 30; const WIDTH: u32 = 640; const HEIGHT: u32 = 360; -/// `DEFAULT_KEYFRAME_INTERVAL_SECS` in crates/enc-ffmpeg/src/video/h264.rs: -/// the encoder emits a keyframe every `GOP_SECS * fps` frames and both DASH -/// writers cut a segment only at a keyframe. -const GOP_SECS: f64 = 2.0; -const GOP: u32 = (GOP_SECS as u32) * FPS; +/// Studio flushes complete moof/mdat pairs at this capture-time interval. +/// The encoder GOP and the enclosing segment can be longer on a static screen. +const FRAGMENT_SECS: f64 = 2.0; +const FRAGMENT_FRAMES: u32 = (FRAGMENT_SECS as u32) * FPS; /// Frames that can still be inside the pipeline when the process dies: the /// one the encoder holds back to stamp real durations (segmented_stream.rs /// `queue_frame`), the one being encoded, and the one in the muxer channel. @@ -182,7 +181,10 @@ mod source { Arm::InProcess => { builder .build::>( - MacOSFragmentedM4SMuxerConfig::default(), + MacOSFragmentedM4SMuxerConfig { + timed_fragments: true, + ..Default::default() + }, ) .await } @@ -245,6 +247,7 @@ mod source { // The Studio display config (capture_pipeline.rs, X11Capture): // 2 s segments. .build::>(SegmentedVideoMuxerConfig { + timed_fragments: true, segment_duration: Duration::from_secs(2), ..Default::default() }) @@ -443,10 +446,11 @@ fn read_truth(path: &Path) -> anyhow::Result<(Vec, bool, Vec)> { continue; }; // A kill can cut the last line short. - if let (Ok(index), Ok(secs)) = (index.parse(), secs.parse()) { - if line.len() > 2 && secs_complete(line) { - handed.push(Handed { index, secs }); - } + if let (Ok(index), Ok(secs)) = (index.parse(), secs.parse()) + && line.len() > 2 + && secs_complete(line) + { + handed.push(Handed { index, secs }); } } Some("stop") => stopped = true, @@ -639,35 +643,86 @@ fn wait_for_exit(pids: &[u32], case: &mut Case) { } } -/// Top-level ISO BMFF boxes span the file exactly and include moof + mdat: -/// the writer finished this fragment (independent of recovery.rs). -fn fragment_complete(path: &Path) -> bool { - let Ok(data) = std::fs::read(path) else { - return false; +fn box_bounds(data: &[u8], at: usize) -> Option<(usize, usize)> { + let header = data.get(at..at.checked_add(8)?)?; + let size = u32::from_be_bytes(header[..4].try_into().ok()?); + let (size, header_size) = match size { + 0 => return None, + 1 => ( + usize::try_from(u64::from_be_bytes( + data.get(at.checked_add(8)?..at.checked_add(16)?)? + .try_into() + .ok()?, + )) + .ok()?, + 16, + ), + size => (size as usize, 8), }; - let mut at = 0usize; - let (mut moof, mut mdat) = (false, false); - while at + 8 <= data.len() { - let size32 = u32::from_be_bytes(data[at..at + 4].try_into().unwrap()) as usize; - let kind = &data[at + 4..at + 8]; - let size = match size32 { - 0 => return false, - 1 => { - if at + 16 > data.len() { - return false; + let end = at.checked_add(size)?; + (size >= header_size && end <= data.len()).then_some((at + header_size, end)) +} + +fn moof_sample_count(data: &[u8]) -> Option { + let mut at = 0; + let mut samples = 0usize; + while at < data.len() { + let (payload, end) = box_bounds(data, at)?; + if &data[at + 4..at + 8] == b"traf" { + let mut child = payload; + while child < end { + let (run, next) = box_bounds(&data[..end], child)?; + if &data[child + 4..child + 8] == b"trun" { + let count = u32::from_be_bytes(data[run..next].get(4..8)?.try_into().ok()?); + samples = samples.checked_add(count as usize)?; } - u64::from_be_bytes(data[at + 8..at + 16].try_into().unwrap()) as usize + child = next; } - n => n, - }; - if size < 8 { - return false; } - moof |= kind == b"moof"; - mdat |= kind == b"mdat"; - at += size; + at = end; + } + Some(samples) +} + +fn complete_prefix(data: &[u8]) -> Option<(usize, Vec)> { + let mut at = 0; + let mut complete = 0; + let mut pending = None; + let mut fragments = Vec::new(); + while let Some((payload, end)) = box_bounds(data, at) { + match &data[at + 4..at + 8] { + b"moof" => { + if pending.is_some() { + break; + } + let Some(samples) = + moof_sample_count(&data[payload..end]).filter(|count| *count > 0) + else { + break; + }; + pending = Some(samples); + } + b"mdat" => { + let Some(samples) = pending.take() else { + break; + }; + fragments.push(samples); + complete = end; + } + _ => {} + } + at = end; } - at == data.len() && moof && mdat + (complete > 0).then_some((complete, fragments)) +} + +fn has_complete_fragment(path: &Path) -> bool { + std::fs::read(path).is_ok_and(|data| complete_prefix(&data).is_some()) +} + +fn fragment_complete(path: &Path) -> bool { + std::fs::read(path) + .is_ok_and(|data| complete_prefix(&data).is_some_and(|(length, _)| length == data.len())) } fn segment_index(name: &str) -> Option { @@ -678,9 +733,8 @@ fn segment_index(name: &str) -> Option { .ok() } -/// Physically complete fragments of one writer directory (finished -/// `segment_NNN.m4s`, or a finished `.m4s.tmp` the writer had not renamed -/// yet), in index order. +/// Segments containing complete moof/mdat pairs, including a verified prefix +/// of an open `.m4s.tmp`, in index order. fn complete_fragments(dir: &Path) -> Vec { let Ok(entries) = std::fs::read_dir(dir) else { return Vec::new(); @@ -694,7 +748,7 @@ fn complete_fragments(dir: &Path) -> Vec { let Some(index) = segment_index(name) else { continue; }; - if !fragment_complete(&path) { + if !has_complete_fragment(&path) { continue; } // Prefer the renamed file when both exist. @@ -735,10 +789,22 @@ fn decode_fragments( if fragments.is_empty() { return Ok(Vec::new()); } - let mut joined = std::fs::read(dir.join("init.mp4")).context("init.mp4")?; + let mut bytes = Vec::new(); for fragment in fragments { - joined.extend(std::fs::read(fragment)?); + let data = std::fs::read(fragment)?; + let (length, _) = complete_prefix(&data).context("no complete video fragment")?; + bytes.extend_from_slice(&data[..length]); } + decode_fragment_bytes(dir, &bytes, scratch) +} + +fn decode_fragment_bytes( + dir: &Path, + bytes: &[u8], + scratch: &Path, +) -> anyhow::Result> { + let mut joined = std::fs::read(dir.join("init.mp4")).context("init.mp4")?; + joined.extend_from_slice(bytes); let path = scratch.join(format!( "joined-{}.mp4", dir.file_name().unwrap_or_default().to_string_lossy() @@ -772,11 +838,18 @@ fn fragments_on_disk(display: &Path, scratch: &Path) -> anyhow::Result(), + "video sample tables and decoded frame count differ" + ); + let mut offset = 0; + for count in samples { + out.push(frames[offset..offset + count].to_vec()); + offset += count; + } } } Ok(out) @@ -838,7 +911,7 @@ fn on_disk(display: &Path, scratch: &Path) -> anyhow::Result> { /// A fragment covers the capture time from its first frame to the first /// frame of the next one, and the encoders must start a new one once -/// `GOP_SECS` have passed: the span may exceed `GOP_SECS` only by the +/// `FRAGMENT_SECS` have passed: the span may exceed `FRAGMENT_SECS` only by the /// interval before the frame that closed it. fn check_fragment_spans(case: &mut Case, fragments: &[Vec<(u32, f64)>], handed: &[Handed]) { let time: BTreeMap = handed.iter().map(|h| (h.index, h.secs)).collect(); @@ -856,11 +929,11 @@ fn check_fragment_spans(case: &mut Case, fragments: &[Vec<(u32, f64)>], handed: }; let span = next - start; worst = worst.max(span); - if span > GOP_SECS + (next - end) + 1e-6 { + if span > FRAGMENT_SECS + (next - end) + 1e-6 { case.problem( "fragment-span", format!( - "fragment of frames {}..={} covers {span:.2}s of capture, more than {GOP_SECS}s plus the {:.2}s before the frame that closed it", + "fragment of frames {}..={} covers {span:.2}s of capture, more than {FRAGMENT_SECS}s plus the {:.2}s before the frame that closed it", first.0, last.0, next - end @@ -1044,7 +1117,7 @@ fn check_recovery( ), ); } - let bound = GOP + IN_FLIGHT; + let bound = FRAGMENT_FRAMES + IN_FLIGHT; // In time: one fragment of capture time, plus the in-flight frames and // the frame that closes the fragment at the delivered interval (a static // screen delivers far fewer frames than the nominal rate). @@ -1052,7 +1125,7 @@ fn check_recovery( .windows(2) .map(|w| w[1].secs - w[0].secs) .fold(1.0 / f64::from(FPS), f64::max); - let bound_secs = GOP_SECS + f64::from(IN_FLIGHT + 1) * interval; + let bound_secs = FRAGMENT_SECS + f64::from(IN_FLIGHT + 1) * interval; case.note(format!( "handed {} frames, complete on disk up to {:?}, lost tail {lost_tail} frames / {lost_secs:.2}s (bound {bound} frames / {bound_secs:.2}s)", handed.len(), @@ -1063,7 +1136,7 @@ fn check_recovery( case.problem( "loss", format!( - "the crash lost the last {lost_tail} handed frames, more than one fragment ({GOP}) plus {IN_FLIGHT} in flight" + "the crash lost the last {lost_tail} handed frames, more than one fragment ({FRAGMENT_FRAMES}) plus {IN_FLIGHT} in flight" ), ); } @@ -1072,7 +1145,7 @@ fn check_recovery( case.problem( "loss-time", format!( - "the crash lost {lost_secs:.2}s of capture, more than one {GOP_SECS}s fragment plus {} frames in flight ({bound_secs:.2}s at the delivered interval)", IN_FLIGHT + 1 + "the crash lost {lost_secs:.2}s of capture, more than one {FRAGMENT_SECS}s fragment plus {} frames in flight ({bound_secs:.2}s at the delivered interval)", IN_FLIGHT + 1 ), ); } @@ -1148,7 +1221,7 @@ const STATIC_STEP: u32 = 10; const STATIC_SECONDS: f64 = 14.0; const STATIC_KILL_SECS: f64 = 11.0; -/// Kills at 2*GOP + k for k = 0..=12 and reports the first k at which the +/// Kills at 2*FRAGMENT_FRAMES + k for k = 0..=12 and reports the first k at which the /// second fragment was complete on disk. fn writer_latency(arm: Arm, case: &mut Case) -> anyhow::Result<()> { let mut first_complete = None; @@ -1159,14 +1232,14 @@ fn writer_latency(arm: Arm, case: &mut Case) -> anyhow::Result<()> { let display = scratch.join("display"); let truth = scratch.join("truth.txt"); let mut child = ChildRecorder::spawn(&display, &truth, arm, 5.0, 1)?; - let n = 2 * GOP + k; + let n = 2 * FRAGMENT_FRAMES + k; child.wait_until(Duration::from_secs(40), |handed, _| { handed.last().is_some_and(|h| h.index >= n) })?; let orphans = child.kill()?; wait_for_exit(&orphans, case); let physical = on_disk(&display, &scratch)?; - let done = physical.iter().any(|f| f.0 == 2 * GOP - 1); + let done = physical.iter().any(|f| f.0 == 2 * FRAGMENT_FRAMES - 1); case.note(format!( "kill after frame {n}: fragment 2 complete = {done}" )); @@ -1197,18 +1270,18 @@ fn kill_points(seed: u64) -> Vec<(String, KillPoint)> { let mut points = Vec::new(); // Mid-fragment, at seeded random frames past the first fragment. for k in 0..2 { - let lo = GOP + 15; + let lo = FRAGMENT_FRAMES + 15; let index = lo + (rng.next_u64() % u64::from(total - lo - 10)) as u32; points.push((format!("random{k}"), KillPoint::Frame(index))); } // Around the second rollover: the keyframe that closes fragment 2 is - // frame 2*GOP; the writer renames the segment when the packet after it + // frame 2*FRAGMENT_FRAMES; the writer renames the segment when the packet after it // arrives (the encoder holds one back), and the manifest learns about // it on its next 10-frame scan. for offset in [0u32, 2, 5] { points.push(( format!("rollover+{offset}"), - KillPoint::Frame(2 * GOP + offset), + KillPoint::Frame(2 * FRAGMENT_FRAMES + offset), )); } for delay in [0u64, 40] { @@ -1290,7 +1363,7 @@ fn muxer_kill(rt: &tokio::runtime::Runtime, case: &mut Case) -> anyhow::Result<( let scratch = scratch_root().join("muxer-kill"); fresh(&scratch)?; let display = scratch.join("display"); - let kill_at = GOP + GOP / 2; + let kill_at = FRAGMENT_FRAMES + FRAGMENT_FRAMES / 2; let me = std::process::id(); let killed = std::sync::Arc::new(std::sync::Mutex::new(None::)); let killed_hook = killed.clone(); @@ -1349,7 +1422,7 @@ fn muxer_kill(rt: &tokio::runtime::Runtime, case: &mut Case) -> anyhow::Result<( recovered.len(), run.handed.len() )); - if gap > 2 * GOP + IN_FLIGHT { + if gap > 2 * FRAGMENT_FRAMES + IN_FLIGHT { case.problem( "loss", format!( @@ -1576,7 +1649,7 @@ fn crash_safety() { settle(case); } // A mostly static screen: one frame every STATIC_STEP slots. A - // fragment must still close every GOP_SECS of capture time. + // fragment must still close every FRAGMENT_SECS of capture time. let id = format!("crash/recorder-kill/{}/static-screen", arm.label()); if report::selected(&id) { let mut case = Case::new(&id); @@ -1730,3 +1803,47 @@ fn pending_fragment_bound_rejects_other_complete_frame_loss() { assert!(case.classes.contains("dropped-complete")); assert!(!case.classes.contains("loss")); } + +#[test] +fn complete_prefix_keeps_whole_moof_pairs_before_every_truncated_tail() { + fn boxed(kind: &[u8; 4], payload: &[u8]) -> Vec { + let mut bytes = ((payload.len() + 8) as u32).to_be_bytes().to_vec(); + bytes.extend(kind); + bytes.extend(payload); + bytes + } + fn fragment(samples: u32) -> Vec { + let mut run = 0u32.to_be_bytes().to_vec(); + run.extend(samples.to_be_bytes()); + let mut bytes = boxed(b"moof", &boxed(b"traf", &boxed(b"trun", &run))); + bytes.extend(boxed(b"mdat", &[1, 2, 3, 4])); + bytes + } + let first = fragment(6); + let next = fragment(5); + for end in 0..next.len() { + let mut bytes = first.clone(); + bytes.extend(&next[..end]); + assert_eq!(complete_prefix(&bytes), Some((first.len(), vec![6]))); + } + let mut both = first.clone(); + both.extend(next); + assert_eq!(complete_prefix(&both), Some((both.len(), vec![6, 5]))); + assert!(complete_prefix(&first[..first.len() - 1]).is_none()); + let short_run = boxed(b"moof", &boxed(b"traf", &boxed(b"trun", &[0; 4]))); + assert!(complete_prefix(&short_run).is_none()); +} + +#[test] +fn complete_prefix_rejects_missing_moof_and_invalid_box_sizes() { + assert!(complete_prefix(&[]).is_none()); + assert!(complete_prefix(&[0; 16]).is_none()); + let mut data = 8u32.to_be_bytes().to_vec(); + data.extend(b"mdat"); + assert!(complete_prefix(&data).is_none()); + data[..4].copy_from_slice(&u32::MAX.to_be_bytes()); + assert!(complete_prefix(&data).is_none()); + data[..4].copy_from_slice(&1u32.to_be_bytes()); + data.extend(u64::MAX.to_be_bytes()); + assert!(complete_prefix(&data).is_none()); +} diff --git a/packages/web-backend/src/Storage/GoogleDrive.ts b/packages/web-backend/src/Storage/GoogleDrive.ts index ead4134c34e..9d99bfda484 100644 --- a/packages/web-backend/src/Storage/GoogleDrive.ts +++ b/packages/web-backend/src/Storage/GoogleDrive.ts @@ -92,6 +92,7 @@ export type CreateGoogleDriveUploadInput = { contentType: string; contentLength?: number; videoTitle?: string; + skipRecordingFolders?: boolean; }; const normalizeContentType = (contentType?: string | null) => @@ -1014,6 +1015,7 @@ const getGoogleDriveUploadParentId = ( tokenStore?: GoogleDriveTokenStore, ) => Effect.gen(function* () { + if (input.skipRecordingFolders) return config.folderId; const folderParts = getDriveFolderParts(input.key, config); let parentId = config.folderId; const pathParts: string[] = []; diff --git a/packages/web-backend/src/Storage/index.ts b/packages/web-backend/src/Storage/index.ts index 10455252b30..9394bc6a4d7 100644 --- a/packages/web-backend/src/Storage/index.ts +++ b/packages/web-backend/src/Storage/index.ts @@ -462,6 +462,10 @@ const makeS3Access = (s3: S3BucketAccess) => ({ body: Parameters[1], fields?: Parameters[2], ) => mapStorageError(s3.putObject(key, body, fields)).pipe(Effect.asVoid), + putProbeObject: (key: string, body: Uint8Array) => + mapStorageError( + s3.putObject(key, body, { contentType: "application/octet-stream" }), + ).pipe(Effect.asVoid), copyObject: ( source: string, key: string, @@ -809,6 +813,56 @@ const makeGoogleDriveAccess = ({ ), }; + const putObject = ( + key: string, + body: string | Uint8Array | ArrayBuffer, + fields?: { contentType?: string; contentLength?: number }, + probe = false, + ) => + Effect.gen(function* () { + const contentType = fields?.contentType ?? "application/octet-stream"; + const contentLength = + fields?.contentLength ?? + (typeof body === "string" + ? new TextEncoder().encode(body).byteLength + : body.byteLength); + const uploadUrl = yield* createGoogleDriveResumableUpload( + repo, + config, + { + integrationId, + ownerId, + videoId: probe ? null : parseObjectKeyVideoId(key), + skipRecordingFolders: probe, + key, + contentType, + contentLength, + }, + tokenStore, + ).pipe(mapStorageError); + const uploadBody = + body instanceof Uint8Array ? new Uint8Array(body).buffer : body; + const response = yield* Effect.tryPromise({ + try: () => + fetch(uploadUrl, { + method: "PUT", + headers: getGoogleDriveUploadHeaders(contentType, contentLength), + body: uploadBody, + }), + catch: (cause) => new StorageDomain.StorageError({ cause }), + }); + if (!response.ok) { + return yield* Effect.fail( + new StorageDomain.StorageError({ + cause: new Error(`Google Drive upload failed: ${response.status}`), + }), + ); + } + yield* mapStorageError( + repo.markObjectComplete(integrationId, key, contentLength), + ); + }); + return { provider: "googleDrive" as const, bucketName: "google-drive", @@ -904,54 +958,9 @@ const makeGoogleDriveAccess = ({ key: string, body: string | Uint8Array | ArrayBuffer, fields?: { contentType?: string; contentLength?: number }, - ) => - Effect.gen(function* () { - const contentType = fields?.contentType ?? "application/octet-stream"; - const contentLength = - fields?.contentLength ?? - (typeof body === "string" - ? new TextEncoder().encode(body).byteLength - : body.byteLength); - const uploadUrl = yield* createGoogleDriveResumableUpload( - repo, - config, - { - integrationId, - ownerId, - videoId: parseVideoIdFromObjectKey(key).pipe( - Option.map((id) => id as Video.VideoId), - Option.getOrNull, - ), - key, - contentType, - contentLength, - }, - tokenStore, - ).pipe(mapStorageError); - const uploadBody = - body instanceof Uint8Array ? new Uint8Array(body).buffer : body; - const response = yield* Effect.tryPromise({ - try: () => - fetch(uploadUrl, { - method: "PUT", - headers: getGoogleDriveUploadHeaders(contentType, contentLength), - body: uploadBody, - }), - catch: (cause) => new StorageDomain.StorageError({ cause }), - }); - if (!response.ok) { - return yield* Effect.fail( - new StorageDomain.StorageError({ - cause: new Error( - `Google Drive upload failed: ${response.status}`, - ), - }), - ); - } - yield* mapStorageError( - repo.markObjectComplete(integrationId, key, contentLength), - ); - }), + ) => putObject(key, body, fields), + putProbeObject: (key: string, body: Uint8Array) => + putObject(key, body, undefined, true), copyObject, copyObjectForRecording, deleteObject,