diff --git a/templates/clips/desktop/src-tauri/Cargo.lock b/templates/clips/desktop/src-tauri/Cargo.lock index 907774909f..19efb91774 100644 --- a/templates/clips/desktop/src-tauri/Cargo.lock +++ b/templates/clips/desktop/src-tauri/Cargo.lock @@ -5303,6 +5303,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2fac314a64dc9a36e61a9eb4261a5e9bbfbc922b27e518af97bc32b926cf967" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -6002,9 +6013,21 @@ dependencies = [ "pin-project-lite", "signal-hook-registry", "socket2 0.6.3", + "tokio-macros", "windows-sys 0.61.2", ] +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.0", +] + [[package]] name = "tokio-native-tls" version = "0.3.1" diff --git a/templates/clips/desktop/src-tauri/Cargo.toml b/templates/clips/desktop/src-tauri/Cargo.toml index fe76297b25..b114eb7000 100644 --- a/templates/clips/desktop/src-tauri/Cargo.toml +++ b/templates/clips/desktop/src-tauri/Cargo.toml @@ -67,7 +67,13 @@ reqwest = { version = "0.12", default-features = false, features = [ "rustls-tls", "json", ] } -tokio = { version = "1", features = ["rt", "rt-multi-thread", "time", "sync"] } +tokio = { version = "1", features = [ + "rt", + "rt-multi-thread", + "time", + "sync", + "macros", +] } chrono = { version = "0.4", features = ["serde"] } # Verify the integrity of the Whisper model we download from HuggingFace. sha2 = "0.10" diff --git a/templates/clips/desktop/src-tauri/src/lib.rs b/templates/clips/desktop/src-tauri/src/lib.rs index 85d6578fc9..9c5a8c90c2 100644 --- a/templates/clips/desktop/src-tauri/src/lib.rs +++ b/templates/clips/desktop/src-tauri/src/lib.rs @@ -161,6 +161,7 @@ pub fn run() { native_screen::native_fullscreen_pending_uploads, native_screen::native_fullscreen_recover_orphaned_uploads, native_screen::native_fullscreen_recording_retry_upload, + native_screen::native_fullscreen_recording_cancel_retry, native_screen::native_fullscreen_recording_mark_upload_error, native_screen::native_fullscreen_recording_clear_upload, native_screen::native_fullscreen_recording_dismiss_upload, diff --git a/templates/clips/desktop/src-tauri/src/native_screen.rs b/templates/clips/desktop/src-tauri/src/native_screen.rs index 505d5330b5..e227c0e4c5 100644 --- a/templates/clips/desktop/src-tauri/src/native_screen.rs +++ b/templates/clips/desktop/src-tauri/src/native_screen.rs @@ -211,9 +211,11 @@ struct NativeUploadResumeResponse { next_chunk_index: Option, attempt_id: Option, upload_generation_id: Option, + reason: Option, + retry_after_ms: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] struct NativeUploadResetResponse { upload_mode: Option, @@ -226,6 +228,16 @@ impl NativeUploadResetResponse { } } +fn accept_native_retry_reset( + reset: NativeUploadResetResponse, + cancelled: bool, +) -> Result { + if cancelled { + return Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()); + } + Ok(reset) +} + impl NativeUploadMode { pub(crate) fn from_option(value: Option) -> Self { match value.as_deref() { @@ -956,6 +968,8 @@ fn clear_recording_active(app: &AppHandle) { static LAST_NATIVE_UPLOAD_FINISHED: OnceLock>> = OnceLock::new(); static CLAIMED_NATIVE_UPLOAD_OPEN: OnceLock>> = OnceLock::new(); +static CANCELLED_NATIVE_UPLOAD_RETRIES: OnceLock>> = OnceLock::new(); +const NATIVE_UPLOAD_RETRY_CANCELLED: &str = "native recording upload retry cancelled"; fn last_native_upload_finished() -> &'static Mutex> { LAST_NATIVE_UPLOAD_FINISHED.get_or_init(|| Mutex::new(None)) @@ -965,6 +979,36 @@ fn claimed_native_upload_open() -> &'static Mutex> { CLAIMED_NATIVE_UPLOAD_OPEN.get_or_init(|| Mutex::new(None)) } +fn cancelled_native_upload_retries() -> &'static Mutex> { + CANCELLED_NATIVE_UPLOAD_RETRIES.get_or_init(|| Mutex::new(BTreeSet::new())) +} + +fn native_upload_retry_cancelled(recording_id: &str) -> bool { + cancelled_native_upload_retries() + .lock() + .map(|cancelled| cancelled.contains(recording_id)) + .unwrap_or(true) +} + +fn clear_native_upload_retry_cancelled(recording_id: &str) { + if let Ok(mut cancelled) = cancelled_native_upload_retries().lock() { + cancelled.remove(recording_id); + } +} + +fn take_native_upload_retry_cancelled(recording_id: &str) -> bool { + cancelled_native_upload_retries() + .lock() + .map(|mut cancelled| cancelled.remove(recording_id)) + .unwrap_or(true) +} + +async fn wait_for_native_upload_retry_cancel(recording_id: &str) { + while !native_upload_retry_cancelled(recording_id) { + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + fn reset_native_upload_completion_state() { if let Ok(mut last) = last_native_upload_finished().lock() { *last = None; @@ -2130,6 +2174,7 @@ pub async fn native_fullscreen_recording_stop_and_upload( match result { Ok(result) => { + clear_native_upload_retry_cancelled(&recording_id); if !result.verification_pending { clear_saved_recording_after_success(&app, &saved); } @@ -3636,6 +3681,10 @@ pub async fn native_fullscreen_recording_retry_upload( auth_token: Option, cookie: Option, ) -> Result { + if take_native_upload_retry_cancelled(&recording_id) { + emit_native_upload_progress(&app, "paused", "Retry cancelled", None, None); + return Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()); + } let mut saved = read_saved_recording_metadata(&app, &recording_id)?; saved.server_url = server_url.trim_end_matches('/').to_string(); saved.last_attempt_at = Some(now_iso()); @@ -3658,6 +3707,7 @@ pub async fn native_fullscreen_recording_retry_upload( // second click cannot steal an upload session already owned by this // local recording. let retry_plan = match get_native_retry_upload_plan( + &app, &saved.server_url, &saved.recording_id, prepared.bytes, @@ -3670,16 +3720,18 @@ pub async fn native_fullscreen_recording_retry_upload( { Ok(plan) => plan, Err(err) => { - interrupt_native_retry_upload( - &saved.server_url, - &saved.recording_id, - &err, - Some(&claimed_attempt_id), - None, - &auth_token, - &cookie, - ) - .await; + if err != NATIVE_UPLOAD_RETRY_CANCELLED { + interrupt_native_retry_upload( + &saved.server_url, + &saved.recording_id, + &err, + Some(&claimed_attempt_id), + None, + &auth_token, + &cookie, + ) + .await; + } cleanup_prepared_saved_recording_files(&prepared, retry_combined_path); return Err(err); } @@ -3753,19 +3805,25 @@ pub async fn native_fullscreen_recording_retry_upload( { Ok(reset) => reset, Err(err) => { - interrupt_native_retry_upload( - &saved.server_url, - &saved.recording_id, - &err, - active_attempt_id.as_deref(), - active_upload_generation_id.as_deref(), - &auth_token, - &cookie, - ) - .await; + if err != NATIVE_UPLOAD_RETRY_CANCELLED { + interrupt_native_retry_upload( + &saved.server_url, + &saved.recording_id, + &err, + active_attempt_id.as_deref(), + active_upload_generation_id.as_deref(), + &auth_token, + &cookie, + ) + .await; + } return Err(err); } }; + let reset = accept_native_retry_reset( + reset, + native_upload_retry_cancelled(&saved.recording_id), + )?; (reset.mode(), None, reset.upload_generation_id) } NativeRetryUploadPlan::Reconcile => unreachable!("handled above"), @@ -3814,6 +3872,10 @@ pub async fn native_fullscreen_recording_retry_upload( { Ok(reset) => { interruption_upload_generation_id = reset.upload_generation_id.clone(); + let reset = accept_native_retry_reset( + reset, + native_upload_retry_cancelled(&saved.recording_id), + )?; upload_prepared_recording_file( &app, &prepared, @@ -3839,16 +3901,18 @@ pub async fn native_fullscreen_recording_retry_upload( upload_result }; if let Err(err) = &upload_result { - interrupt_native_retry_upload( - &saved.server_url, - &saved.recording_id, - err, - replay_attempt_id.as_deref(), - interruption_upload_generation_id.as_deref(), - &auth_token, - &cookie, - ) - .await; + if err != NATIVE_UPLOAD_RETRY_CANCELLED { + interrupt_native_retry_upload( + &saved.server_url, + &saved.recording_id, + err, + replay_attempt_id.as_deref(), + interruption_upload_generation_id.as_deref(), + &auth_token, + &cookie, + ) + .await; + } } cleanup_prepared_saved_recording_files(&prepared, retry_combined_path); upload_result @@ -3863,6 +3927,11 @@ pub async fn native_fullscreen_recording_retry_upload( Ok(result) } Err(err) => { + clear_native_upload_retry_cancelled(&recording_id); + if err == NATIVE_UPLOAD_RETRY_CANCELLED { + emit_native_upload_progress(&app, "paused", "Retry cancelled", None, None); + return Err(err); + } if is_moov_corrupt_error(&err) { saved.corrupt = true; } @@ -3878,6 +3947,15 @@ pub async fn native_fullscreen_recording_retry_upload( } } +#[tauri::command] +pub fn native_fullscreen_recording_cancel_retry(recording_id: String) -> Result<(), String> { + cancelled_native_upload_retries() + .lock() + .map_err(|_| "native upload retry cancellation state is unavailable".to_string())? + .insert(recording_id); + Ok(()) +} + #[tauri::command] pub async fn native_fullscreen_recording_mark_upload_error( app: AppHandle, @@ -5554,7 +5632,8 @@ async fn upload_prepared_recording_file( let mut buffer = vec![0_u8; UPLOAD_CHUNK_BYTES]; file.read_exact(&mut buffer) .map_err(|e| format!("native recording read failed: {e}"))?; - send_upload_post_with_attempt( + tokio::select! { + result = send_upload_post_with_attempt( &client, &server_url, &recording_id, @@ -5575,8 +5654,11 @@ async fn upload_prepared_recording_file( upload_attempt_id, upload_generation_id, buffer, - ) - .await?; + ) => result, + _ = wait_for_native_upload_retry_cancel(&recording_id) => { + Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()) + } + }?; emit_native_upload_progress( app, "uploading", @@ -5599,7 +5681,8 @@ async fn upload_prepared_recording_file( None, Some(streaming_full_chunks as f32 / total_posts as f32), ); - verification_pending = send_upload_post_with_attempt( + verification_pending = tokio::select! { + result = send_upload_post_with_attempt( &client, &server_url, &recording_id, @@ -5620,8 +5703,11 @@ async fn upload_prepared_recording_file( upload_attempt_id, upload_generation_id, final_body, - ) - .await?; + ) => result, + _ = wait_for_native_upload_retry_cancel(&recording_id) => { + Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()) + } + }?; } else { for index in 0..total_chunks { let mut buffer = vec![0_u8; UPLOAD_CHUNK_BYTES]; @@ -5632,7 +5718,8 @@ async fn upload_prepared_recording_file( return Err("Native recording ended before all chunks were read.".into()); } buffer.truncate(read); - send_upload_post_with_attempt( + tokio::select! { + result = send_upload_post_with_attempt( &client, &server_url, &recording_id, @@ -5653,8 +5740,11 @@ async fn upload_prepared_recording_file( upload_attempt_id, upload_generation_id, buffer, - ) - .await?; + ) => result, + _ = wait_for_native_upload_retry_cancel(&recording_id) => { + Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()) + } + }?; emit_native_upload_progress( app, "uploading", @@ -5671,7 +5761,8 @@ async fn upload_prepared_recording_file( None, Some(total_chunks as f32 / total_posts as f32), ); - verification_pending = send_upload_post_with_attempt( + verification_pending = tokio::select! { + result = send_upload_post_with_attempt( &client, &server_url, &recording_id, @@ -5692,8 +5783,11 @@ async fn upload_prepared_recording_file( upload_attempt_id, upload_generation_id, Vec::new(), - ) - .await?; + ) => result, + _ = wait_for_native_upload_retry_cancel(&recording_id) => { + Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()) + } + }?; } emit_native_upload_progress(app, "opening", "Uploading clip", None, Some(1.0)); @@ -5708,6 +5802,7 @@ async fn upload_prepared_recording_file( } async fn get_native_retry_upload_plan( + app: &AppHandle, server_url: &str, recording_id: &str, local_bytes: u64, @@ -5735,30 +5830,104 @@ async fn get_native_retry_upload_plan( if !cookie.trim().is_empty() { request = request.header("Cookie", cookie.trim()); } - let response = request - .send() - .await - .map_err(|e| format!("native recording resume check failed: {e}"))?; - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - if !status.is_success() { - return Err(format!( - "native recording resume check returned {status}: {}", - body.chars().take(400).collect::() + let deadline = tokio::time::Instant::now() + Duration::from_secs(5 * 60); + loop { + if native_upload_retry_cancelled(recording_id) { + return Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()); + } + let response = tokio::select! { + response = request + .try_clone() + .ok_or_else(|| "native recording resume request could not be retried".to_string())? + .send() => response.map_err(|e| format!("native recording resume check failed: {e}"))?, + _ = wait_for_native_upload_retry_cancel(recording_id) => { + return Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()); + } + }; + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + let parsed = serde_json::from_str::(&body); + if !status.is_success() { + if let Ok(conflict) = &parsed { + if let Some(delay) = native_retry_conflict_delay(conflict) { + if tokio::time::Instant::now() + delay <= deadline { + emit_native_upload_progress( + app, + "uploading", + "Waiting for prior retry", + None, + None, + ); + tokio::select! { + _ = tokio::time::sleep(delay) => {} + _ = wait_for_native_upload_retry_cancel(recording_id) => { + return Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()); + } + } + continue; + } + return Err( + "Another upload retry is still active. Wait a moment and try again" + .to_string(), + ); + } + if conflict.reason.as_deref() == Some("retry_claim_liveness_unavailable") { + return Err( + "Clips could not verify whether another retry is active".to_string() + ); + } + } + return Err(format!("native recording resume check failed ({status})")); + } + let response = parsed.map_err(|_| { + "native recording resume check returned an unreadable response".to_string() + })?; + if response.resumable && response.attempt_id.as_deref() != Some(claimed_attempt_id) { + return Err( + "native recording resume check did not acknowledge its attempt claim".to_string(), + ); + } + let recovery_enabled = response.recovery_enabled; + let rollback_attempt_id = response.attempt_id.clone(); + let rollback_generation_id = response.upload_generation_id.clone(); + return Ok(preserve_native_retry_fence_during_rollback( + plan_native_retry_upload(response, local_bytes, exact_local_stream), + recovery_enabled, + rollback_attempt_id, + rollback_generation_id, )); } - let response: NativeUploadResumeResponse = serde_json::from_str(&body) - .map_err(|_| "native recording resume check returned an unreadable response".to_string())?; - if response.resumable && response.attempt_id.as_deref() != Some(claimed_attempt_id) { - return Err( - "native recording resume check did not acknowledge its attempt claim".to_string(), - ); +} + +fn preserve_native_retry_fence_during_rollback( + mut plan: NativeRetryUploadPlan, + recovery_enabled: bool, + acknowledged_attempt_id: Option, + upload_generation_id: Option, +) -> NativeRetryUploadPlan { + if !recovery_enabled { + if let NativeRetryUploadPlan::Restart { + attempt_id, + upload_generation_id: planned_generation_id, + } = &mut plan + { + *attempt_id = acknowledged_attempt_id; + *planned_generation_id = upload_generation_id; + } } - Ok(plan_native_retry_upload( - response, - local_bytes, - exact_local_stream, - )) + plan +} + +fn native_retry_conflict_delay(response: &NativeUploadResumeResponse) -> Option { + if response.resumable + || !response.recovery_enabled + || response.reason.as_deref() != Some("retry_already_active") + { + return None; + } + response + .retry_after_ms + .map(|delay| Duration::from_millis(delay.clamp(250, 30_000))) } #[derive(Deserialize)] @@ -5944,14 +6113,55 @@ fn native_retry_interruption_payload( #[cfg(test)] mod native_retry_upload_plan_tests { use super::{ - is_native_upload_restart_required, is_native_upload_unfenced_restart_required, - native_replay_attempt_id, native_retry_attempt_id, native_retry_interruption_payload, - plan_native_retry_upload, saved_native_retry_attempt_id, upload_url, - NativeFullscreenUploadResult, NativeRetryUploadPlan, NativeUploadResumeResponse, - NATIVE_UPLOAD_RESTART_REQUIRED, NATIVE_UPLOAD_UNFENCED_RESTART_REQUIRED, - UPLOAD_CHUNK_BYTES, + accept_native_retry_reset, is_native_upload_restart_required, + is_native_upload_unfenced_restart_required, native_fullscreen_recording_cancel_retry, + native_replay_attempt_id, native_retry_attempt_id, native_retry_conflict_delay, + native_retry_interruption_payload, native_upload_retry_cancelled, plan_native_retry_upload, + preserve_native_retry_fence_during_rollback, saved_native_retry_attempt_id, + take_native_upload_retry_cancelled, upload_url, NativeFullscreenUploadResult, + NativeRetryUploadPlan, NativeUploadResetResponse, NativeUploadResumeResponse, + NATIVE_UPLOAD_RESTART_REQUIRED, NATIVE_UPLOAD_RETRY_CANCELLED, + NATIVE_UPLOAD_UNFENCED_RESTART_REQUIRED, UPLOAD_CHUNK_BYTES, }; + #[test] + fn consumes_a_cancellation_that_arrives_before_retry_startup() { + let recording_id = "pre-start-cancel-recording".to_string(); + assert!(!take_native_upload_retry_cancelled(&recording_id)); + + native_fullscreen_recording_cancel_retry(recording_id.clone()) + .expect("record pre-start cancellation"); + assert!(native_upload_retry_cancelled(&recording_id)); + assert!(take_native_upload_retry_cancelled(&recording_id)); + assert!(!native_upload_retry_cancelled(&recording_id)); + } + + #[test] + fn cancellation_after_a_committed_reset_preserves_the_authoritative_response() { + let reset = NativeUploadResetResponse { + upload_mode: Some("streaming".to_string()), + upload_generation_id: Some("generation-after-reset".to_string()), + }; + + assert_eq!( + accept_native_retry_reset(reset, true), + Err(NATIVE_UPLOAD_RETRY_CANCELLED.to_string()) + ); + assert_eq!( + accept_native_retry_reset( + NativeUploadResetResponse { + upload_mode: Some("streaming".to_string()), + upload_generation_id: Some("generation-after-reset".to_string()), + }, + false, + ) + .expect("retry may re-enter the committed reset fence") + .upload_generation_id + .as_deref(), + Some("generation-after-reset") + ); + } + fn response(bytes_received: u64, next_chunk_index: u64) -> NativeUploadResumeResponse { NativeUploadResumeResponse { resumable: true, @@ -5962,6 +6172,8 @@ mod native_retry_upload_plan_tests { next_chunk_index: Some(next_chunk_index), attempt_id: Some("attempt-1".to_string()), upload_generation_id: Some("generation-1".to_string()), + reason: None, + retry_after_ms: None, } } @@ -6013,6 +6225,8 @@ mod native_retry_upload_plan_tests { next_chunk_index: None, attempt_id: Some("ignored-attempt".to_string()), upload_generation_id: Some("ignored-generation".to_string()), + reason: Some("feature_disabled".to_string()), + retry_after_ms: None, }, UPLOAD_CHUNK_BYTES as u64, true, @@ -6027,6 +6241,66 @@ mod native_retry_upload_plan_tests { )); } + #[test] + fn preserves_an_existing_fence_when_resumable_retry_is_disabled() { + let plan = preserve_native_retry_fence_during_rollback( + NativeRetryUploadPlan::Restart { + attempt_id: None, + upload_generation_id: None, + }, + false, + Some("attempt-1".to_string()), + Some("generation-1".to_string()), + ); + assert!(matches!( + plan, + NativeRetryUploadPlan::Restart { + attempt_id: Some(attempt_id), + upload_generation_id: Some(generation_id), + } if attempt_id == "attempt-1" && generation_id == "generation-1" + )); + } + + #[test] + fn keeps_an_unacknowledged_legacy_restart_unfenced_when_resumable_retry_is_disabled() { + let plan = preserve_native_retry_fence_during_rollback( + NativeRetryUploadPlan::Restart { + attempt_id: None, + upload_generation_id: None, + }, + false, + None, + None, + ); + assert!(matches!( + plan, + NativeRetryUploadPlan::Restart { + attempt_id: None, + upload_generation_id: None, + } + )); + } + + #[test] + fn preserves_an_acknowledged_legacy_attempt_without_a_generation() { + let plan = preserve_native_retry_fence_during_rollback( + NativeRetryUploadPlan::Restart { + attempt_id: None, + upload_generation_id: None, + }, + false, + Some("attempt-1".to_string()), + None, + ); + assert!(matches!( + plan, + NativeRetryUploadPlan::Restart { + attempt_id: Some(attempt_id), + upload_generation_id: None, + } if attempt_id == "attempt-1" + )); + } + #[test] fn reconciles_terminal_resume_without_an_attempt_echo() { let terminal = plan_native_retry_upload( @@ -6039,6 +6313,8 @@ mod native_retry_upload_plan_tests { next_chunk_index: None, attempt_id: None, upload_generation_id: None, + reason: None, + retry_after_ms: None, }, UPLOAD_CHUNK_BYTES as u64, true, @@ -6064,6 +6340,8 @@ mod native_retry_upload_plan_tests { next_chunk_index: Some(0), attempt_id: Some(claimed_attempt_id.clone()), upload_generation_id: Some("generation-1".to_string()), + reason: None, + retry_after_ms: None, }, UPLOAD_CHUNK_BYTES as u64, true, @@ -6108,6 +6386,30 @@ mod native_retry_upload_plan_tests { assert_eq!(saved_attempt_id.as_deref(), Some(first.as_str())); } + #[test] + fn waits_only_for_a_typed_bounded_retry_conflict() { + let conflict = NativeUploadResumeResponse { + resumable: false, + recovery_enabled: true, + status: Some("uploading".to_string()), + upload_mode: None, + bytes_received: None, + next_chunk_index: None, + attempt_id: None, + upload_generation_id: None, + reason: Some("retry_already_active".to_string()), + retry_after_ms: Some(60_000), + }; + assert_eq!( + native_retry_conflict_delay(&conflict), + Some(std::time::Duration::from_secs(30)) + ); + + let mut untyped = conflict; + untyped.retry_after_ms = None; + assert_eq!(native_retry_conflict_delay(&untyped), None); + } + #[test] fn reports_retry_interruptions_with_or_without_a_fencing_claim() { let unfenced = native_retry_interruption_payload("upload failed", None, None); diff --git a/templates/clips/desktop/src/app.tsx b/templates/clips/desktop/src/app.tsx index d308b67d19..b6535aca5e 100644 --- a/templates/clips/desktop/src/app.tsx +++ b/templates/clips/desktop/src/app.tsx @@ -160,6 +160,10 @@ interface PendingNativeUpload { type PendingDesktopUpload = PendingNativeUpload | PendingBrowserRecordingUpload; +type NativeUploadProgress = { + message?: string; +}; + type PopoverView = | "recorder" | "memory" @@ -967,6 +971,26 @@ export function App() { const [retryingUploadStatus, setRetryingUploadStatus] = useState< string | null >(null); + const retryUploadAbortRef = useRef(null); + const retryingUploadKindRef = useRef( + null, + ); + useEffect(() => { + if (!retryingUploadId) return; + let disposed = false; + let unlisten: (() => void) | null = null; + listen("clips:native-upload-progress", (event) => { + const message = event.payload?.message?.trim(); + if (message) setRetryingUploadStatus(message); + }).then((cleanup) => { + if (disposed) cleanup(); + else unlisten = cleanup; + }); + return () => { + disposed = true; + unlisten?.(); + }; + }, [retryingUploadId]); const [exportingUploadId, setExportingUploadId] = useState( null, ); @@ -2873,6 +2897,9 @@ export function App() { const targetServerUrl = serverUrlForPendingUpload(upload, serverUrl); setRecError(null); setRetryingUploadId(upload.recordingId); + const abortController = new AbortController(); + retryUploadAbortRef.current = abortController; + retryingUploadKindRef.current = upload.kind; try { const authToken = loadDesktopAuthToken(targetServerUrl); if (upload.kind === "native") { @@ -2898,13 +2925,16 @@ export function App() { recordingId: upload.recordingId, serverUrl: targetServerUrl, authToken, + signal: abortController.signal, onRecoveryDecision: ({ action, progress }) => { setRetryingUploadStatus( action === "resume" ? `Resuming ยท ${Math.round(progress * 100)}% already uploaded` - : action === "restart" - ? "Restarting upload" - : "Finishing upload", + : action === "wait" + ? "Waiting for prior retry" + : action === "restart" + ? "Restarting upload" + : "Finishing upload", ); }, }); @@ -2918,6 +2948,14 @@ export function App() { emit("clips:popover-visible", false).catch(() => {}); } catch (err) { const message = err instanceof Error ? err.message : String(err); + if ( + abortController.signal.aborted || + (err instanceof DOMException && err.name === "AbortError") || + message === "native recording upload retry cancelled" + ) { + await loadPendingUploads(); + return; + } console.error("[clips-tray] retry saved upload failed:", err); setRecError( isStorageSetupFailureMessage(message) @@ -2926,11 +2964,28 @@ export function App() { ); await loadPendingUploads(); } finally { + if (retryUploadAbortRef.current === abortController) { + retryUploadAbortRef.current = null; + retryingUploadKindRef.current = null; + } setRetryingUploadId(null); setRetryingUploadStatus(null); } } + function cancelPendingUploadRetry(upload: PendingDesktopUpload) { + if (retryingUploadId !== upload.recordingId) return; + retryUploadAbortRef.current?.abort(); + if (retryingUploadKindRef.current === "native") { + invoke("native_fullscreen_recording_cancel_retry", { + recordingId: upload.recordingId, + }).catch((err) => { + console.error("[clips-tray] cancel saved upload retry failed:", err); + }); + } + setRetryingUploadStatus("Cancelling retry"); + } + async function exportPendingUpload(upload: PendingDesktopUpload) { if (retryingUploadId || exportingUploadId || dismissingUploadId) return; setRecError(null); @@ -3637,22 +3692,26 @@ export function App() { const showCameraRow = mode !== "screen"; // screen-only has no camera const showSourceRow = mode !== "camera"; // camera-only has no screen source - const pendingUploadBanner = recordingStopFinalizing ? ( - - ) : pendingUploads.length > 0 ? ( - openVideoStorageSetup(upload.serverUrl)} - /> - ) : null; + const pendingUploadBanner = + authStatus === "authed" ? ( + recordingStopFinalizing ? ( + + ) : pendingUploads.length > 0 ? ( + openVideoStorageSetup(upload.serverUrl)} + /> + ) : null + ) : null; async function copyRewindAgentPrompt() { try { @@ -4352,6 +4411,7 @@ function PendingUploadBanner({ dismissingUploadId, onExport, onRetry, + onCancelRetry, onDismiss, onOpenFolder, onConnectStorage, @@ -4363,6 +4423,7 @@ function PendingUploadBanner({ dismissingUploadId: string | null; onExport: (upload: PendingDesktopUpload) => void; onRetry: (upload: PendingDesktopUpload) => void; + onCancelRetry: (upload: PendingDesktopUpload) => void; onDismiss: (upload: PendingDesktopUpload) => void; onOpenFolder: (upload: PendingDesktopUpload) => void; onConnectStorage: (upload: PendingDesktopUpload) => void; @@ -4371,6 +4432,8 @@ function PendingUploadBanner({ if (!latest) return null; const retrying = retryingUploadId === latest.recordingId; + const retryCancelling = + retrying && retryingUploadStatus === "Cancelling retry"; const storageSetupFailure = isStorageSetupFailureMessage(latest.lastError); const canOpenFolder = latest.kind === "native" && !!latest.folderPath; @@ -4477,12 +4540,22 @@ function PendingUploadBanner({ )} diff --git a/templates/clips/desktop/src/lib/recorder.ts b/templates/clips/desktop/src/lib/recorder.ts index fe3bc77028..8db1162dfe 100644 --- a/templates/clips/desktop/src/lib/recorder.ts +++ b/templates/clips/desktop/src/lib/recorder.ts @@ -104,6 +104,7 @@ import { planStreamingRecovery, retryAttemptIdAfterRestartSignal, retryAttemptIdAfterResumeResponse, + retryConflictDelay, type UploadResumeResponse, } from "./upload-recovery"; import { @@ -1061,6 +1062,7 @@ async function postBackupChunk( url: string, blob: Blob, authToken?: string, + signal?: AbortSignal, ): Promise { const res = await fetch(url, { method: "POST", @@ -1070,6 +1072,7 @@ async function postBackupChunk( ), credentials: "include", body: blob, + signal, }); const body = await res.text().catch(() => ""); if (!res.ok) { @@ -1103,6 +1106,7 @@ async function resetBrowserRecordingBackupUpload( authToken?: string, attemptId?: string, uploadGenerationId?: string, + signal?: AbortSignal, ): Promise<{ uploadMode: UploadMode; uploadGenerationId?: string }> { const res = await fetch( `${meta.serverUrl.replace(/\/+$/, "")}/api/uploads/${meta.recordingId}/reset-chunks`, @@ -1120,6 +1124,7 @@ async function resetBrowserRecordingBackupUpload( ...(attemptId ? { attemptId } : {}), ...(uploadGenerationId ? { uploadGenerationId } : {}), }), + signal, }, ); if (!res.ok) { @@ -1147,32 +1152,60 @@ async function getBrowserRecordingUploadResume( meta: BrowserRecordingBackupMeta, attemptId: string, authToken?: string, + onWaiting?: (delayMs: number) => void, + signal?: AbortSignal, ): Promise { const resumeUrl = new URL( `${meta.serverUrl.replace(/\/+$/, "")}/api/uploads/${meta.recordingId}/resume`, ); resumeUrl.searchParams.set("attemptId", attemptId); - const res = await fetch(resumeUrl, { - method: "GET", - headers: buildRetryHeaders("application/json", authToken), - credentials: "include", - }); - const body = await res.text().catch(() => ""); - if (!res.ok) { - throw new Error( - `Upload resume check failed (${res.status}): ${body.slice(0, 200)}`, - ); - } - let parsed: UploadResumeResponse; - try { - parsed = JSON.parse(body) as UploadResumeResponse; - } catch { - throw new Error("Upload resume check returned an unreadable response"); - } - if (parsed.resumable && parsed.attemptId !== attemptId) { - throw new Error("Upload resume check returned a mismatched retry token"); + const deadline = Date.now() + 5 * 60_000; + for (;;) { + const res = await fetch(resumeUrl, { + method: "GET", + headers: buildRetryHeaders("application/json", authToken), + credentials: "include", + signal, + }); + let body: string; + try { + body = await res.text(); + } catch { + throw new Error("Upload resume check response could not be read"); + } + let parsed: UploadResumeResponse; + try { + parsed = JSON.parse(body) as UploadResumeResponse; + } catch { + throw new Error("Upload resume check returned an unreadable response"); + } + if (!res.ok) { + const delayMs = retryConflictDelay(parsed); + if (delayMs !== null && Date.now() + delayMs <= deadline) { + onWaiting?.(delayMs); + await abortableWait(delayMs, signal); + continue; + } + if (!parsed.resumable && parsed.reason === "retry_already_active") { + throw new Error( + "Another upload retry is still active. Wait a moment and try again.", + ); + } + if ( + !parsed.resumable && + parsed.reason === "retry_claim_liveness_unavailable" + ) { + throw new Error( + "Clips could not verify whether another retry is active. Your local clip is safe; try again.", + ); + } + throw new Error(`Upload resume check failed (${res.status})`); + } + if (parsed.resumable && parsed.attemptId !== attemptId) { + throw new Error("Upload resume check returned a mismatched retry token"); + } + return parsed; } - return parsed; } async function replayBrowserBackupToResumableSession( @@ -1185,6 +1218,7 @@ async function replayBrowserBackupToResumableSession( bytesReceived: 0, nextChunkIndex: 0, }, + signal?: AbortSignal, ): Promise { // The backup is stored in raw MediaRecorder blobs, which have arbitrary // boundaries. A resumable provider needs every non-final request aligned, @@ -1217,6 +1251,7 @@ async function replayBrowserBackupToResumableSession( }), body, authToken, + signal, ); } @@ -1243,6 +1278,7 @@ async function replayBrowserBackupToResumableSession( }), finalBody, authToken, + signal, ); } @@ -1250,8 +1286,9 @@ export async function retryBrowserRecordingBackup(input: { recordingId: string; serverUrl?: string; authToken?: string; + signal?: AbortSignal; onRecoveryDecision?: (decision: { - action: "resume" | "restart" | "reconcile"; + action: "wait" | "resume" | "restart" | "reconcile"; progress: number; }) => void; }): Promise<{ recordingId: string; viewUrl: string }> { @@ -1284,6 +1321,8 @@ export async function retryBrowserRecordingBackup(input: { meta, activeAttemptId, input.authToken, + () => input.onRecoveryDecision?.({ action: "wait", progress: 0 }), + input.signal, ); const recoveryPlan = planStreamingRecovery({ response: resumeResponse, @@ -1294,7 +1333,7 @@ export async function retryBrowserRecordingBackup(input: { activeAttemptId, resumeResponse, ); - activeUploadGenerationId = resumeResponse.resumable + activeUploadGenerationId = activeAttemptId ? resumeResponse.uploadGenerationId : undefined; if (recoveryPlan.action === "reconcile") { @@ -1343,6 +1382,7 @@ export async function retryBrowserRecordingBackup(input: { input.authToken, activeAttemptId, activeUploadGenerationId, + input.signal, ); uploadMode = reset.uploadMode; activeUploadGenerationId = reset.uploadGenerationId; @@ -1358,6 +1398,7 @@ export async function retryBrowserRecordingBackup(input: { activeAttemptId, activeUploadGenerationId, resumeFrom, + input.signal, ); } catch (err) { if (err instanceof UploadRestartRequiredError) { @@ -1365,9 +1406,6 @@ export async function retryBrowserRecordingBackup(input: { activeAttemptId, err.recoveryEnabled, ); - if (err.recoveryEnabled === false) { - activeUploadGenerationId = undefined; - } input.onRecoveryDecision?.({ action: "restart", progress: 0 }); console.info("[clips-recorder] restarting expired upload session", { recordingId: meta.recordingId, @@ -1378,6 +1416,7 @@ export async function retryBrowserRecordingBackup(input: { input.authToken, activeAttemptId, activeUploadGenerationId, + input.signal, ); uploadMode = reset.uploadMode; activeUploadGenerationId = reset.uploadGenerationId; @@ -1388,6 +1427,8 @@ export async function retryBrowserRecordingBackup(input: { input.authToken, activeAttemptId, activeUploadGenerationId, + undefined, + input.signal, ); } } else if ( @@ -1439,6 +1480,7 @@ export async function retryBrowserRecordingBackup(input: { }), chunk.blob, input.authToken, + input.signal, ); } @@ -1466,6 +1508,7 @@ export async function retryBrowserRecordingBackup(input: { finalChunkUrl, new Blob([], { type: meta.mimeType }), input.authToken, + input.signal, ); const receiptStatus = verifyFinalizeReceipt(receipt, meta); if (receiptStatus === "processing") { @@ -1498,6 +1541,12 @@ export async function retryBrowserRecordingBackup(input: { await deleteBrowserRecordingBackup(meta.recordingId); return { recordingId: meta.recordingId, viewUrl: `/r/${meta.recordingId}` }; } catch (err) { + if ( + input.signal?.aborted || + (err instanceof DOMException && err.name === "AbortError") + ) { + throw err; + } const message = err instanceof Error ? err.message : String(err); if ( await recoverAcceptedRecordingAfterFinalizeError({ @@ -1638,6 +1687,23 @@ function wait(ms: number): Promise { return new Promise((resolve) => window.setTimeout(resolve, ms)); } +function abortableWait(ms: number, signal?: AbortSignal): Promise { + if (!signal) return wait(ms); + if (signal.aborted) + return Promise.reject(new DOMException("Aborted", "AbortError")); + return new Promise((resolve, reject) => { + const timer = window.setTimeout(() => { + signal.removeEventListener("abort", onAbort); + resolve(); + }, ms); + const onAbort = () => { + window.clearTimeout(timer); + reject(new DOMException("Aborted", "AbortError")); + }; + signal.addEventListener("abort", onAbort, { once: true }); + }); +} + interface NativeFullscreenUploadResult { recordingId: string; durationMs: number; diff --git a/templates/clips/desktop/src/lib/upload-recovery.test.ts b/templates/clips/desktop/src/lib/upload-recovery.test.ts index eb9dd3d301..d7bdba6fc0 100644 --- a/templates/clips/desktop/src/lib/upload-recovery.test.ts +++ b/templates/clips/desktop/src/lib/upload-recovery.test.ts @@ -3,17 +3,61 @@ import { describe, expect, it } from "vitest"; import { buildStreamingReplayPlan, planStreamingRecovery, + retryConflictDelay, retryAttemptIdAfterRestartSignal, retryAttemptIdAfterResumeResponse, } from "./upload-recovery"; const CHUNK_BYTES = 3_932_160; -describe("retryAttemptIdAfterRestartSignal", () => { - it("drops the retry claim only when the server disables recovery", () => { +describe("retryConflictDelay", () => { + it("accepts only a typed, bounded active-retry delay", () => { expect( - retryAttemptIdAfterRestartSignal("attempt-1", false), - ).toBeUndefined(); + retryConflictDelay({ + resumable: false, + recoveryEnabled: true, + status: "uploading", + reason: "retry_already_active", + retryAfterMs: 1_500, + }), + ).toBe(1_500); + expect( + retryConflictDelay({ + resumable: false, + recoveryEnabled: true, + status: "uploading", + reason: "retry_already_active", + retryAfterMs: 60_000, + }), + ).toBe(30_000); + }); + + it("rejects untyped and unreadable conflict delays", () => { + expect( + retryConflictDelay({ + resumable: false, + recoveryEnabled: true, + status: "uploading", + reason: "retry_already_active", + }), + ).toBeNull(); + expect( + retryConflictDelay({ + resumable: false, + recoveryEnabled: true, + status: "uploading", + reason: "upload_state_changed", + retryAfterMs: 1_000, + }), + ).toBeNull(); + }); +}); + +describe("retryAttemptIdAfterRestartSignal", () => { + it("preserves an existing retry claim through a flag rollback", () => { + expect(retryAttemptIdAfterRestartSignal("attempt-1", false)).toBe( + "attempt-1", + ); expect(retryAttemptIdAfterRestartSignal("attempt-1", true)).toBe( "attempt-1", ); @@ -48,6 +92,30 @@ describe("retryAttemptIdAfterResumeResponse", () => { }), ).toBeUndefined(); }); + + it("preserves a server-acknowledged claim while the flag is rolled back", () => { + expect( + retryAttemptIdAfterResumeResponse("attempt-1", { + resumable: false, + recoveryEnabled: false, + status: "uploading", + reason: "feature_disabled", + attemptId: "attempt-1", + uploadGenerationId: "generation-1", + }), + ).toBe("attempt-1"); + }); + + it("drops an unacknowledged legacy claim while the flag is rolled back", () => { + expect( + retryAttemptIdAfterResumeResponse("attempt-1", { + resumable: false, + recoveryEnabled: false, + status: "uploading", + reason: "feature_disabled", + }), + ).toBeUndefined(); + }); }); describe("planStreamingRecovery", () => { diff --git a/templates/clips/desktop/src/lib/upload-recovery.ts b/templates/clips/desktop/src/lib/upload-recovery.ts index a8f160ee82..a10f476bb0 100644 --- a/templates/clips/desktop/src/lib/upload-recovery.ts +++ b/templates/clips/desktop/src/lib/upload-recovery.ts @@ -17,8 +17,25 @@ export type UploadResumeResponse = videoUrl?: string | null; reason?: string; attemptId?: string; + uploadGenerationId?: string; + retryAfterMs?: number; }; +export function retryConflictDelay( + response: UploadResumeResponse, +): number | null { + if ( + response.resumable || + response.recoveryEnabled !== true || + response.reason !== "retry_already_active" || + !Number.isSafeInteger(response.retryAfterMs) || + (response.retryAfterMs ?? 0) <= 0 + ) { + return null; + } + return Math.min(Math.max(response.retryAfterMs!, 250), 30_000); +} + export type StreamingRecoveryPlan = | { action: "resume"; @@ -38,15 +55,16 @@ export interface StreamingReplayRequest { export function retryAttemptIdAfterRestartSignal( attemptId: string | undefined, - recoveryEnabled: unknown, + _recoveryEnabled: unknown, ): string | undefined { - return recoveryEnabled === false ? undefined : attemptId; + return attemptId; } export function retryAttemptIdAfterResumeResponse( attemptId: string | undefined, response: UploadResumeResponse, ): string | undefined { + if (response.recoveryEnabled === false) return response.attemptId; return response.resumable && response.attemptId === attemptId ? attemptId : undefined; diff --git a/templates/clips/server/lib/resumable-session.test.ts b/templates/clips/server/lib/resumable-session.test.ts new file mode 100644 index 0000000000..f255b31d4f --- /dev/null +++ b/templates/clips/server/lib/resumable-session.test.ts @@ -0,0 +1,50 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mockCompareAndSetAppState = vi.hoisted(() => vi.fn()); + +vi.mock("@agent-native/core/application-state", () => ({ + compareAndSetAppState: (...args: unknown[]) => + mockCompareAndSetAppState(...args), + deleteAppState: vi.fn(), + readAppState: vi.fn(), + writeAppState: vi.fn(), +})); + +import { compareAndSetResumableSession } from "./resumable-session"; + +describe("compareAndSetResumableSession", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockCompareAndSetAppState.mockResolvedValue(true); + }); + + it("fences settlement to the exact recording generation and session snapshot", async () => { + const expected = { + providerId: "s3", + sessionId: "session-a", + meta: { uploadId: "upload-a" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }; + const next = { + ...expected, + meta: { ...expected.meta, completedPart: 3 }, + bytesUploaded: 125, + lastCommittedIndex: 3, + }; + + await expect( + compareAndSetResumableSession( + "recording-a", + expected, + next, + "generation-a", + ), + ).resolves.toBe(true); + expect(mockCompareAndSetAppState).toHaveBeenCalledWith( + "resumable-session-recording-a-generation-a", + expected, + next, + ); + }); +}); diff --git a/templates/clips/server/lib/resumable-session.ts b/templates/clips/server/lib/resumable-session.ts index 06b41ff7c3..28590ea452 100644 --- a/templates/clips/server/lib/resumable-session.ts +++ b/templates/clips/server/lib/resumable-session.ts @@ -1,4 +1,5 @@ import { + compareAndSetAppState, deleteAppState, readAppState, writeAppState, @@ -10,6 +11,20 @@ export interface StoredResumableSession { meta: Record; bytesUploaded: number; lastCommittedIndex?: number; + providerClosed?: boolean; +} + +export async function compareAndSetResumableSession( + recordingId: string, + expected: StoredResumableSession, + next: StoredResumableSession, + generationId?: string | null, +): Promise { + return compareAndSetAppState( + key(recordingId, generationId), + expected as unknown as Record, + next as unknown as Record, + ); } const key = (recordingId: string, generationId?: string | null) => diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.test.ts b/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.test.ts index bb52b73d0c..3449ca81c3 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.test.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.test.ts @@ -21,6 +21,7 @@ const mockSumRecordingChunkBytes = vi.hoisted(() => vi.fn()); const mockGetResumableSession = vi.hoisted(() => vi.fn()); const mockDeleteResumableSession = vi.hoisted(() => vi.fn()); const mockSetResumableSession = vi.hoisted(() => vi.fn()); +const mockCompareAndSetResumableSession = vi.hoisted(() => vi.fn()); const mockRelayChunk = vi.hoisted(() => vi.fn()); const mockAbortSession = vi.hoisted(() => vi.fn()); const mockResolveResumableUploadProvider = vi.hoisted(() => vi.fn()); @@ -122,6 +123,8 @@ vi.mock("../../../../lib/recordings.js", () => ({ })); vi.mock("../../../../lib/resumable-session.js", () => ({ + compareAndSetResumableSession: (...args: unknown[]) => + mockCompareAndSetResumableSession(...args), deleteResumableSession: (...args: unknown[]) => mockDeleteResumableSession(...args), getResumableSession: (...args: unknown[]) => mockGetResumableSession(...args), @@ -192,6 +195,7 @@ describe("/api/uploads/:recordingId/chunk route", () => { mockGetResumableSession.mockResolvedValue(null); mockDeleteResumableSession.mockResolvedValue(undefined); mockSetResumableSession.mockResolvedValue(undefined); + mockCompareAndSetResumableSession.mockResolvedValue(true); mockIsStreamingUploadDisabled.mockReturnValue(false); mockShouldRejectVideoUploadWithoutStorage.mockResolvedValue(false); mockAllowsSqlRecordingChunkScratch.mockReturnValue(true); @@ -311,7 +315,7 @@ describe("/api/uploads/:recordingId/chunk route", () => { expect(mockWriteAppState).not.toHaveBeenCalled(); }); - it("forces a full restart when the retry flag switches off between chunks", async () => { + it("preserves a fenced retry when the retry flag switches off between chunks", async () => { mockGetResumableSession.mockResolvedValue({ providerId: "s3", sessionId: "sess-1", @@ -349,15 +353,101 @@ describe("/api/uploads/:recordingId/chunk route", () => { body: new Uint8Array([4, 5, 6]), }); await expect(handler({} as any)).resolves.toEqual({ - ok: false, - error: "Resumable upload retry is disabled.", - restartRequired: true, - recoveryEnabled: false, + ok: true, + finalized: false, + index: 1, + bytes: 3, }); - expect(mockSetResponseStatus).toHaveBeenLastCalledWith({}, 409); - expect(mockReadRawBody).toHaveBeenCalledOnce(); - expect(mockRelayChunk).toHaveBeenCalledOnce(); - expect(mockRenewUploadLease).toHaveBeenCalledTimes(3); + expect(mockReadRawBody).toHaveBeenCalledTimes(2); + expect(mockRelayChunk).toHaveBeenCalledTimes(2); + expect(mockRenewUploadLease).toHaveBeenCalledTimes(6); + }); + + it("heartbeats a fenced retry while a provider relay is still in flight", async () => { + vi.useFakeTimers(); + try { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-1", + meta: { objectKey: "clips/rec-1.webm" }, + bytesUploaded: 0, + lastCommittedIndex: -1, + }); + let finishRelay!: (value: { ok: boolean; status: number }) => void; + mockRelayChunk.mockImplementationOnce( + () => + new Promise((resolve) => { + finishRelay = resolve; + }), + ); + setRequest({ + query: { + index: "0", + mimeType: "video/webm", + attemptId: "retry-attempt", + }, + body: new Uint8Array([1]), + }); + + const pending = handler({} as any); + await vi.advanceTimersByTimeAsync(10_000); + expect(mockRenewUploadLease).toHaveBeenCalledTimes(3); + finishRelay({ ok: true, status: 308 }); + await expect(pending).resolves.toEqual( + expect.objectContaining({ ok: true, finalized: false }), + ); + const renewalsAfterRelay = mockRenewUploadLease.mock.calls.length; + await vi.advanceTimersByTimeAsync(30_000); + expect(mockRenewUploadLease).toHaveBeenCalledTimes(renewalsAfterRelay); + } finally { + vi.useRealTimers(); + } + }); + + it("fails loudly when a provider relay loses its fenced retry claim", async () => { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-1", + meta: { objectKey: "clips/rec-1.webm" }, + bytesUploaded: 0, + lastCommittedIndex: -1, + }); + mockRenewUploadLease + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: false, staleAttempt: true }); + let finishRelay!: (value: { ok: boolean; status: number }) => void; + mockRelayChunk.mockImplementationOnce( + () => + new Promise((resolve) => { + finishRelay = resolve; + }), + ); + vi.useFakeTimers(); + try { + setRequest({ + query: { + index: "0", + mimeType: "video/webm", + attemptId: "retry-attempt", + }, + body: new Uint8Array([1]), + }); + const pending = handler({} as any); + await vi.advanceTimersByTimeAsync(10_000); + finishRelay({ ok: true, status: 308 }); + await expect(pending).resolves.toEqual( + expect.objectContaining({ staleAttempt: true }), + ); + expect(mockCompareAndSetResumableSession).toHaveBeenCalledWith( + "rec-1", + expect.objectContaining({ bytesUploaded: 0 }), + expect.objectContaining({ bytesUploaded: 1, lastCommittedIndex: 0 }), + null, + ); + } finally { + vi.useRealTimers(); + } }); it("stores in-order chunks and advances upload progress state", async () => { @@ -929,8 +1019,15 @@ describe("/api/uploads/:recordingId/chunk route", () => { bytes, { mimeType: "video/webm" }, ); - expect(mockSetResumableSession).toHaveBeenCalledWith( + expect(mockCompareAndSetResumableSession).toHaveBeenCalledWith( "rec-1", + { + providerId: "s3", + sessionId: "sess-1", + meta: { objectKey: "clips/rec-1.webm" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }, { providerId: "s3", sessionId: "sess-1", @@ -943,7 +1040,7 @@ describe("/api/uploads/:recordingId/chunk route", () => { expect(mockFinalizeRun).not.toHaveBeenCalled(); }); - it("aborts and surfaces a provider error on the final resumable chunk", async () => { + it("defers destructive cleanup when a final provider call throws", async () => { mockGetResumableSession.mockResolvedValue({ providerId: "s3", sessionId: "sess-final", @@ -970,20 +1067,50 @@ describe("/api/uploads/:recordingId/chunk route", () => { try { await expect(handler({} as any)).resolves.toEqual({ ok: false, - error: "Final chunk upload failed: S3 staging object read failed (500)", + error: + "Chunk upload outcome is unknown: S3 staging object read failed (500)", + restartRequired: true, }); } finally { consoleError.mockRestore(); } - expect(mockAbortSession).toHaveBeenCalledWith({ - sessionId: "sess-final", - meta: { objectKey: "clips/rec-1.webm" }, - }); - expect(mockDeleteResumableSession).toHaveBeenCalledWith("rec-1", null); + expect(mockAbortSession).not.toHaveBeenCalled(); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); expect(mockFinalizeRun).not.toHaveBeenCalled(); }); + it("forces a retired-generation restart for an ambiguous ordinary chunk", async () => { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-ordinary", + meta: {}, + bytesUploaded: 100, + lastCommittedIndex: 2, + }); + mockRelayChunk.mockRejectedValueOnce(new Error("connection reset")); + setRequest({ + query: { index: "3", mimeType: "video/webm" }, + body: new Uint8Array([1]), + }); + const consoleError = vi + .spyOn(console, "error") + .mockImplementation(() => undefined); + + try { + await expect(handler({} as any)).resolves.toEqual({ + ok: false, + error: "Chunk upload outcome is unknown: connection reset", + restartRequired: true, + }); + } finally { + consoleError.mockRestore(); + } + expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); + expect(mockAbortSession).not.toHaveBeenCalled(); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); + }); + it("acks a replayed resumable chunk without re-uploading to the provider", async () => { mockGetResumableSession.mockResolvedValue({ providerId: "s3", @@ -1021,7 +1148,7 @@ describe("/api/uploads/:recordingId/chunk route", () => { expect(mockFinalizeRun).not.toHaveBeenCalled(); }); - it("retires an expired provider session so the desktop can restart safely", async () => { + it("reports an expired provider session without destroying its live generation", async () => { mockGetResumableSession.mockResolvedValue({ providerId: "s3", sessionId: "expired-session", @@ -1041,10 +1168,251 @@ describe("/api/uploads/:recordingId/chunk route", () => { restartRequired: true, }); expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); - expect(mockDeleteResumableSession).toHaveBeenCalledWith("rec-1", null); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); expect(mockSetResumableSession).not.toHaveBeenCalled(); }); + it("returns stale without cleanup when a failed provider response loses ownership", async () => { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-final", + meta: {}, + bytesUploaded: 100, + lastCommittedIndex: 2, + }); + mockRenewUploadLease + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: false, staleAttempt: true }); + mockRelayChunk.mockResolvedValueOnce({ ok: false, status: 500 }); + setRequest({ + query: { + index: "3", + isFinal: "1", + mimeType: "video/webm", + attemptId: "attempt-a", + }, + body: new Uint8Array([1]), + }); + + await expect(handler({} as any)).resolves.toEqual( + expect.objectContaining({ staleAttempt: true }), + ); + expect(mockAbortSession).not.toHaveBeenCalled(); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); + }); + + it("settles an accepted close sentinel before returning stale ownership", async () => { + vi.useFakeTimers(); + try { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-close", + meta: { uploadId: "upload-1" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }); + mockRenewUploadLease + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: false, staleAttempt: true }); + let finishRelay!: (value: { + ok: boolean; + status: number; + updatedMeta: Record; + }) => void; + mockRelayChunk.mockImplementationOnce( + () => + new Promise((resolve) => { + finishRelay = resolve; + }), + ); + setRequest({ + query: { + index: "3", + isFinal: "1", + mimeType: "video/webm", + attemptId: "attempt-a", + }, + }); + + const pending = handler({} as any); + await vi.advanceTimersByTimeAsync(10_000); + finishRelay({ + ok: true, + status: 200, + updatedMeta: { completedPart: 3 }, + }); + await expect(pending).resolves.toEqual( + expect.objectContaining({ staleAttempt: true }), + ); + expect(mockCompareAndSetResumableSession).toHaveBeenCalledWith( + "rec-1", + expect.objectContaining({ sessionId: "sess-close" }), + expect.objectContaining({ + providerClosed: true, + meta: { uploadId: "upload-1", completedPart: 3 }, + }), + null, + ); + expect(mockFinalizeRun).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("forces a retired-generation restart for an ambiguous close sentinel", async () => { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-close", + meta: {}, + bytesUploaded: 100, + lastCommittedIndex: 2, + }); + mockRelayChunk.mockRejectedValueOnce( + new Error("response connection closed"), + ); + setRequest({ + query: { index: "3", isFinal: "1", mimeType: "video/webm" }, + }); + const consoleError = vi + .spyOn(console, "error") + .mockImplementation(() => undefined); + + try { + await expect(handler({} as any)).resolves.toEqual({ + ok: false, + error: + "Resumable session close outcome is unknown: response connection closed", + restartRequired: true, + }); + } finally { + consoleError.mockRestore(); + } + expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); + expect(mockAbortSession).not.toHaveBeenCalled(); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); + }); + + it("settles accepted final data before returning stale ownership", async () => { + vi.useFakeTimers(); + try { + mockGetResumableSession.mockResolvedValue({ + providerId: "s3", + sessionId: "sess-final", + meta: { uploadId: "upload-1" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }); + mockRenewUploadLease + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: true }) + .mockResolvedValueOnce({ held: false, staleAttempt: true }); + let finishRelay!: (value: { + ok: boolean; + status: number; + updatedMeta: Record; + }) => void; + mockRelayChunk.mockImplementationOnce( + () => + new Promise((resolve) => { + finishRelay = resolve; + }), + ); + setRequest({ + query: { + index: "3", + isFinal: "1", + mimeType: "video/webm", + attemptId: "attempt-a", + }, + body: new Uint8Array([1, 2, 3]), + }); + + const pending = handler({} as any); + await vi.advanceTimersByTimeAsync(10_000); + finishRelay({ + ok: true, + status: 200, + updatedMeta: { completedPart: 3 }, + }); + await expect(pending).resolves.toEqual( + expect.objectContaining({ staleAttempt: true }), + ); + expect(mockCompareAndSetResumableSession).toHaveBeenCalledWith( + "rec-1", + expect.objectContaining({ bytesUploaded: 100 }), + expect.objectContaining({ + bytesUploaded: 103, + lastCommittedIndex: 3, + meta: { uploadId: "upload-1", completedPart: 3 }, + }), + null, + ); + expect(mockFinalizeRun).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("accepts an already-reconciled CAS loss for the same session", async () => { + const initial = { + providerId: "s3", + sessionId: "sess-1", + meta: { uploadId: "upload-1" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }; + mockGetResumableSession + .mockResolvedValueOnce(initial) + .mockResolvedValueOnce({ + ...initial, + meta: { uploadId: "upload-1", completedPart: 3 }, + bytesUploaded: 101, + lastCommittedIndex: 3, + }); + mockCompareAndSetResumableSession.mockResolvedValueOnce(false); + mockRelayChunk.mockResolvedValueOnce({ + ok: true, + status: 308, + updatedMeta: { completedPart: 3 }, + }); + setRequest({ + query: { index: "3", mimeType: "video/webm" }, + body: new Uint8Array([1]), + }); + + await expect(handler({} as any)).resolves.toEqual( + expect.objectContaining({ ok: true, finalized: false }), + ); + }); + + it("forces restart when accepted state contradicts the same stored session", async () => { + const initial = { + providerId: "s3", + sessionId: "sess-1", + meta: { uploadId: "upload-1" }, + bytesUploaded: 100, + lastCommittedIndex: 2, + }; + mockGetResumableSession + .mockResolvedValueOnce(initial) + .mockResolvedValueOnce(initial); + mockCompareAndSetResumableSession.mockResolvedValueOnce(false); + mockRelayChunk.mockResolvedValueOnce({ ok: true, status: 308 }); + setRequest({ + query: { index: "3", mimeType: "video/webm" }, + body: new Uint8Array([1]), + }); + + await expect(handler({} as any)).resolves.toEqual({ + ok: false, + error: "Accepted provider state could not be reconciled safely.", + restartRequired: true, + }); + expect(mockFinalizeRun).not.toHaveBeenCalled(); + }); + it("keeps replacement-generation scratch when a stale writer loses its lease", async () => { (mockSelectRows.rows[0] as Record).uploadGenerationId = "generation-a"; @@ -1116,17 +1484,14 @@ describe("/api/uploads/:recordingId/chunk route", () => { await expect(handler({} as any)).resolves.toEqual( expect.objectContaining({ restartRequired: true }), ); - expect(mockDeleteResumableSession).toHaveBeenCalledWith( - "rec-1", - "generation-a", - ); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); expect(mockDeleteResumableSession).not.toHaveBeenCalledWith( "rec-1", "generation-b", ); }); - it("does not restore a replacement session after a delayed old provider success", async () => { + it("settles a delayed provider success before returning stale ownership", async () => { (mockSelectRows.rows[0] as Record).uploadGenerationId = "generation-a"; mockGetResumableSession.mockResolvedValue({ @@ -1159,11 +1524,19 @@ describe("/api/uploads/:recordingId/chunk route", () => { await expect(handler({} as any)).resolves.toEqual( expect.objectContaining({ ok: false }), ); - expect(mockSetResumableSession).not.toHaveBeenCalled(); - expect(mockSetResumableSession).not.toHaveBeenCalledWith( + expect(mockCompareAndSetResumableSession).toHaveBeenCalledWith( "rec-1", - expect.anything(), - "generation-b", + expect.objectContaining({ + sessionId: "old-session", + bytesUploaded: 100, + lastCommittedIndex: 0, + }), + expect.objectContaining({ + sessionId: "old-session", + bytesUploaded: 101, + lastCommittedIndex: 1, + }), + "generation-a", ); }); diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.ts b/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.ts index bedb82fe34..83fe833cad 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/chunk.post.ts @@ -17,7 +17,6 @@ import { readAppState, writeAppState, } from "@agent-native/core/application-state"; -import { isFeatureFlagEnabled } from "@agent-native/core/feature-flags"; import { runWithRequestContext } from "@agent-native/core/server"; import { track } from "@agent-native/core/tracking"; import { normalizeChunkUploadNumber } from "@shared/recording-core.js"; @@ -35,7 +34,6 @@ import { } from "h3"; import finalizeRecording from "../../../../../actions/finalize-recording.js"; -import { UPLOAD_RETRY_RESUME_FLAG } from "../../../../../shared/feature-flags.js"; import { getDb, schema } from "../../../../db/index.js"; import { debugLog } from "../../../../lib/debug.js"; import { @@ -47,15 +45,16 @@ import { ownerEmailMatches, } from "../../../../lib/recordings.js"; import { - deleteResumableSession, + compareAndSetResumableSession, getResumableSession, - setResumableSession, type StoredResumableSession, } from "../../../../lib/resumable-session.js"; -import { abortResumableUploadSession } from "../../../../lib/resumable-upload-cleanup.js"; import { resolveResumableUploadProvider } from "../../../../lib/resumable-upload-provider.js"; import { isStreamingUploadDisabled } from "../../../../lib/streaming-upload-mode.js"; -import { renewUploadLease } from "../../../../lib/upload-lease.js"; +import { + renewUploadLease, + type UploadLeaseResult, +} from "../../../../lib/upload-lease.js"; import { allowsSqlRecordingChunkScratch, shouldRejectVideoUploadWithoutStorage, @@ -68,6 +67,42 @@ const RECORDING_TOO_LARGE_REASON = `Recording exceeds the ${Math.round(MAX_RECOR // are base64 encoded by the gateway and effectively cap out around 4.5 MB. // Keep our own cap lower so dev/local failures match production. const MAX_CHUNK_BYTES = 4 * 1024 * 1024; +const RETRY_OWNERSHIP_HEARTBEAT_MS = 10 * 1000; + +async function relayWithRetryOwnershipHeartbeat( + recordingId: string, + attemptId: string | null, + generationId: string | null, + relay: () => Promise, +): Promise<{ result: T; ownershipFailure: UploadLeaseResult | Error | null }> { + if (attemptId === null) + return { result: await relay(), ownershipFailure: null }; + let ownershipFailure: UploadLeaseResult | Error | null = null; + let pending: Promise | null = null; + const heartbeat = () => { + if (pending || ownershipFailure) return; + pending = renewUploadLease(recordingId, { attemptId, generationId }) + .then((lease) => { + if (!lease.held) ownershipFailure = lease; + }) + .catch((error) => { + ownershipFailure = + error instanceof Error ? error : new Error(String(error)); + }) + .finally(() => { + pending = null; + }); + }; + const timer = setInterval(heartbeat, RETRY_OWNERSHIP_HEARTBEAT_MS); + let result: T; + try { + result = await relay(); + } finally { + clearInterval(timer); + await pending; + } + return { result: result!, ownershipFailure }; +} const ALLOWED_RECORDING_MIME_TYPES = new Set([ "video/webm", @@ -239,23 +274,6 @@ export default defineEventHandler(async (event: H3Event) => { } debugLog("[chunk] resolved owner:", ownerEmail); - if ( - attemptId !== null && - !(await isFeatureFlagEnabled(UPLOAD_RETRY_RESUME_FLAG, { - userEmail: ownerEmail, - userKey: ownerEmail, - orgId, - })) - ) { - setResponseStatus(event, 409); - return { - ok: false, - error: "Resumable upload retry is disabled.", - restartRequired: true, - recoveryEnabled: false, - }; - } - return runWithRequestContext({ userEmail: ownerEmail, orgId }, async () => { const db = getDb(); @@ -855,21 +873,55 @@ async function handleResumableChunk( `[resumable-chunk-${recordingId}] resumable session exists - bytesUploaded=${session.bytesUploaded} index=${index} isFinal=${isFinal}`, ); - const cleanupFailedFinalSession = async () => { - const cleaned = await abortResumableUploadSession(session, { - provider: uploadProvider, - label: `resumable-final-${recordingId}`, - }); - if (cleaned) { - await deleteResumableSession(recordingId, uploadGenerationId).catch( - (error) => - console.warn( - `[resumable-chunk-${recordingId}] failed to retire aborted session:`, - error, - ), - ); + const settleAcceptedProviderEffect = async ( + next: StoredResumableSession, + ): Promise<"settled" | "superseded" | "contradictory"> => { + if ( + await compareAndSetResumableSession( + recordingId, + session, + next, + uploadGenerationId, + ) + ) { + session = next; + return "settled"; + } + + const current = await getResumableSession(recordingId, uploadGenerationId); + if (!current || current.sessionId !== session.sessionId) { + return "superseded"; } - return !cleaned; + const metaSettled = Object.entries(next.meta).every( + ([key, value]) => + JSON.stringify(current.meta[key]) === JSON.stringify(value), + ); + if ( + current.bytesUploaded >= next.bytesUploaded && + (current.lastCommittedIndex ?? -1) >= (next.lastCommittedIndex ?? -1) && + (!next.providerClosed || current.providerClosed === true) && + metaSettled + ) { + session = current; + return "settled"; + } + return "contradictory"; + }; + + const settlementFailure = (outcome: "superseded" | "contradictory") => { + setResponseStatus(event, 409); + return outcome === "superseded" + ? { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + } + : { + ok: false, + error: "Accepted provider state could not be reconciled safely.", + restartRequired: true, + }; }; const raw = await readRawBody(event, false); @@ -902,40 +954,95 @@ async function handleResumableChunk( // 0-byte sentinel from the recorder after stop(). All data chunks have // already been PUT to the provider; send Content-Range: bytes */ // to close the session before handing off to finalize-recording. - let closeRes; - try { - closeRes = await uploadProvider.resumable.relayChunk( - { sessionId: session.sessionId, meta: session.meta }, - `bytes */${session.bytesUploaded}`, - new Uint8Array(0), - ); - } catch (error) { - const cleanupFailed = await cleanupFailedFinalSession(); - const detail = error instanceof Error ? error.message : String(error); - console.error( - `[resumable-chunk-${recordingId}] session close threw:`, - error, - ); - setResponseStatus(event, 502); - return { - ok: false, - error: `Resumable session close failed: ${detail}`, - ...(cleanupFailed ? { cleanupFailed: true } : {}), - }; - } - if (!closeRes.ok || closeRes.status === 308) { - console.error( - `[resumable-chunk-${recordingId}] session close failed (${closeRes.status})`, - ); - const cleanupFailed = await cleanupFailedFinalSession(); - setResponseStatus(event, 502); - return { - ok: false, - error: `Resumable session close failed (${closeRes.status})`, - ...(cleanupFailed ? { cleanupFailed: true } : {}), - }; - } - if (closeRes.updatedMeta) { + if (session.providerClosed) { + // A prior close response was accepted but its caller lost ownership. + // The durable marker makes replay a no-op before idempotent finalization. + } else { + let closeRes; + try { + const relayed = await relayWithRetryOwnershipHeartbeat( + recordingId, + attemptId, + uploadGenerationId, + () => + uploadProvider.resumable!.relayChunk( + { sessionId: session.sessionId, meta: session.meta }, + `bytes */${session.bytesUploaded}`, + new Uint8Array(0), + ), + ); + closeRes = relayed.result; + if (closeRes.ok && closeRes.status !== 308) { + const settlement = await settleAcceptedProviderEffect({ + ...session, + ...(closeRes.updatedMeta + ? { meta: { ...session.meta, ...closeRes.updatedMeta } } + : {}), + providerClosed: true, + }); + if (settlement !== "settled") return settlementFailure(settlement); + } + if (relayed.ownershipFailure) { + setResponseStatus(event, 409); + return { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + }; + } + } catch (error) { + const failedCloseLease = await renewUploadLease(recordingId, { + attemptId, + generationId: uploadGenerationId, + }); + if (!failedCloseLease.held) { + setResponseStatus(event, 409); + return { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + }; + } + const detail = error instanceof Error ? error.message : String(error); + console.error( + `[resumable-chunk-${recordingId}] session close threw:`, + error, + ); + setResponseStatus(event, 409); + return { + ok: false, + error: `Resumable session close outcome is unknown: ${detail}`, + restartRequired: true, + }; + } + if (!closeRes.ok || closeRes.status === 308) { + const failedCloseLease = await renewUploadLease(recordingId, { + attemptId, + generationId: uploadGenerationId, + }); + if (!failedCloseLease.held) { + setResponseStatus(event, 409); + return { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + }; + } + console.error( + `[resumable-chunk-${recordingId}] session close failed (${closeRes.status})`, + ); + const restartRequired = + closeRes.status === 404 || closeRes.status === 410; + setResponseStatus(event, restartRequired ? 409 : 502); + return { + ok: false, + error: `Resumable session close failed (${closeRes.status})`, + ...(restartRequired ? { restartRequired: true } : {}), + }; + } const postCloseLease = await renewUploadLease(recordingId, { attemptId, generationId: uploadGenerationId, @@ -947,16 +1054,9 @@ async function handleResumableChunk( error: postCloseLease.failureReason ?? "Recording upload has already failed.", + staleAttempt: true, }; } - await setResumableSession( - recordingId, - { - ...session, - meta: { ...session.meta, ...closeRes.updatedMeta }, - }, - uploadGenerationId, - ); } } else { // Idempotent replay guard: a client retry (after a lost response) can @@ -1002,28 +1102,65 @@ async function handleResumableChunk( const putT0 = Date.now(); let putResult; try { - putResult = await uploadProvider.resumable.relayChunk( - { sessionId: session.sessionId, meta: session.meta }, - contentRange, - bytes, - { mimeType: mimeType.split(";")[0].trim() }, + const relayed = await relayWithRetryOwnershipHeartbeat( + recordingId, + attemptId, + uploadGenerationId, + () => + uploadProvider.resumable!.relayChunk( + { sessionId: session.sessionId, meta: session.meta }, + contentRange, + bytes, + { mimeType: mimeType.split(";")[0].trim() }, + ), ); + putResult = relayed.result; + if (isFinal ? putResult.ok && putResult.status !== 308 : putResult.ok) { + const settlement = await settleAcceptedProviderEffect({ + ...session, + ...(putResult.updatedMeta + ? { meta: { ...session.meta, ...putResult.updatedMeta } } + : {}), + bytesUploaded: start + bytes.byteLength, + lastCommittedIndex: index, + }); + if (settlement !== "settled") return settlementFailure(settlement); + finalizedSourceSizeBytes = start + bytes.byteLength; + } + if (relayed.ownershipFailure) { + setResponseStatus(event, 409); + return { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + }; + } } catch (error) { - if (isFinal) { - const cleanupFailed = await cleanupFailedFinalSession(); - const detail = error instanceof Error ? error.message : String(error); - console.error( - `[resumable-chunk-${recordingId}] final chunk upload threw:`, - error, - ); - setResponseStatus(event, 502); + const failedUploadLease = await renewUploadLease(recordingId, { + attemptId, + generationId: uploadGenerationId, + }); + if (!failedUploadLease.held) { + setResponseStatus(event, 409); return { ok: false, - error: `Final chunk upload failed: ${detail}`, - ...(cleanupFailed ? { cleanupFailed: true } : {}), + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, }; } - throw error; + const detail = error instanceof Error ? error.message : String(error); + console.error( + `[resumable-chunk-${recordingId}] provider response was ambiguous:`, + error, + ); + setResponseStatus(event, 409); + return { + ok: false, + error: `Chunk upload outcome is unknown: ${detail}`, + restartRequired: true, + }; } console.log( `[resumable-chunk-${recordingId}] PUT ${Date.now() - putT0}ms status=${putResult.status} range="${contentRange}"`, @@ -1033,22 +1170,26 @@ async function handleResumableChunk( ? putResult.ok && putResult.status !== 308 : putResult.ok; if (!resultOk) { + const failedUploadLease = await renewUploadLease(recordingId, { + attemptId, + generationId: uploadGenerationId, + }); + if (!failedUploadLease.held) { + setResponseStatus(event, 409); + return { + ok: false, + error: + "Upload retry ownership was lost while the provider was responding.", + staleAttempt: true, + }; + } const restartRequired = putResult.status === 404 || putResult.status === 410; - const cleanupFailed = isFinal - ? await cleanupFailedFinalSession() - : false; - if (restartRequired && !cleanupFailed) { - await deleteResumableSession(recordingId, uploadGenerationId).catch( - () => {}, - ); - } setResponseStatus(event, restartRequired ? 409 : 502); return { ok: false, error: `Chunk upload failed (${putResult.status})`, ...(restartRequired ? { restartRequired: true } : {}), - ...(cleanupFailed ? { cleanupFailed: true } : {}), }; } @@ -1065,20 +1206,6 @@ async function handleResumableChunk( "Recording upload has already failed.", }; } - await setResumableSession( - recordingId, - { - ...session, - ...(putResult.updatedMeta - ? { meta: { ...session.meta, ...putResult.updatedMeta } } - : {}), - bytesUploaded: start + bytes.byteLength, - lastCommittedIndex: index, - }, - uploadGenerationId, - ); - finalizedSourceSizeBytes = start + bytes.byteLength; - if (!isFinal) { return { ok: true, finalized: false, index, bytes: bytes.byteLength }; } diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.test.ts b/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.test.ts index 6b18398f0d..83fd7b422e 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.test.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.test.ts @@ -263,7 +263,7 @@ describe("/api/uploads/:recordingId/reset-chunks route", () => { expect(mockDeleteResumableSession).not.toHaveBeenCalled(); }); - it("clears a recovery claim when the flag is disabled mid-retry", async () => { + it("does not clear a recovery claim when the flag is disabled mid-retry", async () => { mockIsFeatureFlagEnabled.mockResolvedValue(false); mockExistingRecording.current.uploadAttemptId = "old-attempt"; mockReadBody.mockResolvedValue({ @@ -273,14 +273,32 @@ describe("/api/uploads/:recordingId/reset-chunks route", () => { }); await expect(handler({} as any)).resolves.toEqual( - expect.objectContaining({ ok: true, uploadGenerationId: null }), + expect.objectContaining({ staleAttempt: true }), ); - expect(mockUpdateSets).toContainEqual( + expect(mockUpdateSets).toHaveLength(0); + }); + + it("preserves a fenced retry through flag disable when the client echoes its claim", async () => { + mockIsFeatureFlagEnabled.mockResolvedValue(false); + mockExistingRecording.current.uploadAttemptId = "old-attempt"; + mockExistingRecording.current.uploadGenerationId = "generation-old"; + mockReadBody.mockResolvedValue({ + attemptId: "old-attempt", + uploadGenerationId: "generation-old", + }); + + await expect(handler({} as any)).resolves.toEqual( expect.objectContaining({ - uploadAttemptId: null, - uploadGenerationId: null, + ok: true, + uploadGenerationId: expect.any(String), }), ); + expect(mockUpdateSets).toContainEqual( + expect.objectContaining({ uploadGenerationId: expect.any(String) }), + ); + expect(mockUpdateSets.some((set) => set.uploadAttemptId === null)).toBe( + false, + ); }); it("recreates a resumable session for a browser backup retry", async () => { diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.ts b/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.ts index a4ee5a736c..4cf5834e8a 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/reset-chunks.post.ts @@ -238,7 +238,10 @@ export default defineEventHandler(async (event: H3Event) => { : null; const existingAttemptId = existing.uploadAttemptId ?? null; const existingGenerationId = existing.uploadGenerationId ?? null; - if (recoveryEnabled && existingAttemptId !== requestedAttemptId) { + if ( + existingAttemptId !== null && + existingAttemptId !== requestedAttemptId + ) { setResponseStatus(event, 409); return { error: "A newer upload retry is already active.", @@ -251,7 +254,10 @@ export default defineEventHandler(async (event: H3Event) => { body.uploadGenerationId.length <= 128 ? body.uploadGenerationId : null; - if (recoveryEnabled && existingGenerationId !== requestedGenerationId) { + if ( + existingGenerationId !== null && + existingGenerationId !== requestedGenerationId + ) { setResponseStatus(event, 409); return { error: "A newer upload generation is already active.", @@ -281,10 +287,12 @@ export default defineEventHandler(async (event: H3Event) => { // fence. Retry claims always opt in; legacy reset callers keep the null // generation wire contract until they are upgraded. const useGenerationFence = - recoveryEnabled && - (requestedAttemptId !== null || - requestedGenerationId !== null || - body?.useGenerationFence === true); + existingAttemptId !== null || + existingGenerationId !== null || + (recoveryEnabled && + (requestedAttemptId !== null || + requestedGenerationId !== null || + body?.useGenerationFence === true)); const nextGenerationId = useGenerationFence ? randomUUID() : null; const uploadStateKey = `recording-upload-${recordingId}`; const uploadStateSnapshot = await readAppState(uploadStateKey); @@ -314,7 +322,9 @@ export default defineEventHandler(async (event: H3Event) => { failureReason: null, uploadProgress: 0, uploadGenerationId: nextGenerationId, - ...(!recoveryEnabled ? { uploadAttemptId: null } : {}), + ...(!recoveryEnabled && existingAttemptId === null + ? { uploadAttemptId: null } + : {}), uploadLeaseExpiresAt: uploadLeaseExpiry(), updatedAt: now, }) @@ -503,8 +513,9 @@ export default defineEventHandler(async (event: H3Event) => { } } + const preservedAttemptId = existingAttemptId; const resetLease = await renewUploadLease(recordingId, { - attemptId: recoveryEnabled ? existingAttemptId : null, + attemptId: preservedAttemptId, generationId: nextGenerationId, }); if (!resetLease.held) { @@ -528,7 +539,7 @@ export default defineEventHandler(async (event: H3Event) => { progress: 0, chunksReceived: 0, bytesReceived: 0, - uploadAttemptId: recoveryEnabled ? existingAttemptId : null, + uploadAttemptId: preservedAttemptId, uploadGenerationId: nextGenerationId, maxBytes: MAX_RECORDING_UPLOAD_BYTES, updatedAt: now, @@ -550,8 +561,7 @@ export default defineEventHandler(async (event: H3Event) => { ); if ( current?.status !== "uploading" || - (current.uploadAttemptId ?? null) !== - (recoveryEnabled ? existingAttemptId : null) || + (current.uploadAttemptId ?? null) !== preservedAttemptId || (current.uploadGenerationId ?? null) !== nextGenerationId ) { setResponseStatus(event, 409); diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.test.ts b/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.test.ts index 7505613df7..e70f7003fd 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.test.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.test.ts @@ -2,6 +2,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const mockRenewUploadLease = vi.hoisted(() => vi.fn()); const mockGetResumableSession = vi.hoisted(() => vi.fn()); +const mockDeleteResumableSession = vi.hoisted(() => vi.fn()); +const mockAbortResumableUploadSession = vi.hoisted(() => vi.fn()); const mockListRecordingChunkKeys = vi.hoisted(() => vi.fn()); const mockSumRecordingChunkBytes = vi.hoisted(() => vi.fn()); const mockReadAppState = vi.hoisted(() => vi.fn()); @@ -11,6 +13,7 @@ const mockSetResponseStatus = vi.hoisted(() => vi.fn()); const mockGetQuery = vi.hoisted(() => vi.fn()); const mockIsFeatureFlagEnabled = vi.hoisted(() => vi.fn()); const mockUpdateRows = vi.hoisted(() => ({ rows: [{ id: "rec-1" }] })); +const mockUpdateSets = vi.hoisted(() => [] as Array>); const mockSelectRows = vi.hoisted(() => ({ rows: [] as Array>, })); @@ -24,7 +27,10 @@ const mockDb = vi.hoisted(() => ({ }), update: vi.fn(() => { const builder = { - set: vi.fn(() => builder), + set: vi.fn((values: Record) => { + mockUpdateSets.push(values); + return builder; + }), where: vi.fn(() => builder), returning: vi.fn(async () => mockUpdateRows.rows), }; @@ -48,10 +54,15 @@ vi.mock("@agent-native/core/server", () => ({ runWithRequestContext: (_ctx: unknown, fn: () => unknown) => fn(), })); +vi.mock("node:crypto", () => ({ + randomUUID: () => "generation-2", +})); + vi.mock("drizzle-orm", () => ({ and: vi.fn(() => "and"), eq: vi.fn(() => "eq"), isNull: vi.fn(() => "is-null"), + lte: vi.fn(() => "lte"), })); vi.mock("h3", () => ({ @@ -89,10 +100,18 @@ vi.mock("../../../../lib/recording-upload-state.js", () => ({ })); vi.mock("../../../../lib/resumable-session.js", () => ({ + deleteResumableSession: (...args: unknown[]) => + mockDeleteResumableSession(...args), getResumableSession: (...args: unknown[]) => mockGetResumableSession(...args), })); +vi.mock("../../../../lib/resumable-upload-cleanup.js", () => ({ + abortResumableUploadSession: (...args: unknown[]) => + mockAbortResumableUploadSession(...args), +})); + vi.mock("../../../../lib/upload-lease.js", () => ({ + UPLOAD_LEASE_MS: 60 * 60 * 1000, renewUploadLease: (...args: unknown[]) => mockRenewUploadLease(...args), uploadLeaseExpiry: () => "2099-01-01T00:00:00.000Z", })); @@ -106,30 +125,56 @@ describe("/api/uploads/:recordingId/resume route", () => { mockGetQuery.mockReturnValue({ attemptId: "client-attempt-0001" }); mockRenewUploadLease.mockResolvedValue({ held: true }); mockGetResumableSession.mockResolvedValue(null); + mockAbortResumableUploadSession.mockResolvedValue(true); + mockDeleteResumableSession.mockResolvedValue(undefined); mockListRecordingChunkKeys.mockResolvedValue([]); mockSumRecordingChunkBytes.mockResolvedValue(0); mockReadAppState.mockResolvedValue({ progress: 50 }); mockWriteAppState.mockResolvedValue(undefined); mockCompareAndSetAppState.mockResolvedValue(true); mockUpdateRows.rows = [{ id: "rec-1" }]; + mockUpdateSets.length = 0; mockIsFeatureFlagEnabled.mockResolvedValue(true); }); - it("leaves upload state untouched when resumable retry is disabled", async () => { + it("leaves legacy upload state untouched when resumable retry is disabled", async () => { mockIsFeatureFlagEnabled.mockResolvedValue(false); await expect(handler({} as any)).resolves.toEqual({ recoveryEnabled: false, resumable: false, recordingId: "rec-1", - status: null, + status: "uploading", reason: "feature_disabled", }); - expect(mockDb.select).not.toHaveBeenCalled(); + expect(mockDb.select).toHaveBeenCalledOnce(); expect(mockDb.update).not.toHaveBeenCalled(); expect(mockWriteAppState).not.toHaveBeenCalled(); }); + it("returns an existing fence when the flag is disabled", async () => { + mockIsFeatureFlagEnabled.mockResolvedValue(false); + mockSelectRows.rows = [ + { + id: "rec-1", + status: "uploading", + uploadAttemptId: "client-attempt-0001", + uploadGenerationId: "generation-1", + }, + ]; + + await expect(handler({} as any)).resolves.toEqual({ + recoveryEnabled: false, + resumable: false, + recordingId: "rec-1", + status: "uploading", + reason: "feature_disabled", + attemptId: "client-attempt-0001", + uploadGenerationId: "generation-1", + }); + expect(mockDb.update).not.toHaveBeenCalled(); + }); + it("reports the provider's committed offset for a streaming upload", async () => { mockGetResumableSession.mockResolvedValue({ bytesUploaded: 4_194_304, @@ -223,6 +268,22 @@ describe("/api/uploads/:recordingId/resume route", () => { expect(mockDb.update).toHaveBeenCalledOnce(); }); + it("accepts an interruption with its detailed retryable reason", async () => { + mockSelectRows.rows = [ + { + id: "rec-1", + status: "failed", + failureReason: + "Upload was interrupted. The local recording is safe; retry from the Clips desktop app. Last error: network changed", + }, + ]; + + await expect(handler({} as any)).resolves.toEqual( + expect.objectContaining({ resumable: true, status: "uploading" }), + ); + expect(mockDb.update).toHaveBeenCalledOnce(); + }); + it("claims a restart token when the prior provider session is gone", async () => { mockSelectRows.rows = [ { @@ -254,6 +315,7 @@ describe("/api/uploads/:recordingId/resume route", () => { recordingId: "rec-1", status: "uploading", reason: "retry_already_active", + retryAfterMs: 250, }); expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); expect(mockWriteAppState).not.toHaveBeenCalled(); @@ -321,12 +383,41 @@ describe("/api/uploads/:recordingId/resume route", () => { }); }); - it("does not let a different claim steal an active retry", async () => { + it("returns a bounded typed conflict for a live different retry claim", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-21T12:00:00.000Z")); + try { + mockSelectRows.rows = [ + { + id: "rec-1", + status: "uploading", + uploadAttemptId: "active-attempt-0001", + uploadLeaseExpiresAt: "2026-08-21T12:59:59.000Z", + }, + ]; + + await expect(handler({} as any)).resolves.toEqual({ + resumable: false, + recoveryEnabled: true, + recordingId: "rec-1", + status: "uploading", + reason: "retry_already_active", + retryAfterMs: 29_000, + }); + expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); + expect(mockDb.update).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("fails loudly when a different retry claim has unreadable liveness", async () => { mockSelectRows.rows = [ { id: "rec-1", status: "uploading", uploadAttemptId: "active-attempt-0001", + uploadLeaseExpiresAt: "not-a-timestamp", }, ]; @@ -335,12 +426,108 @@ describe("/api/uploads/:recordingId/resume route", () => { recoveryEnabled: true, recordingId: "rec-1", status: "uploading", - reason: "retry_already_active", + reason: "retry_claim_liveness_unavailable", }); expect(mockSetResponseStatus).toHaveBeenCalledWith({}, 409); expect(mockDb.update).not.toHaveBeenCalled(); }); + it("invalidates an expired claim's provider session before restarting it", async () => { + mockSelectRows.rows = [ + { + id: "rec-1", + status: "uploading", + uploadAttemptId: "stale-attempt-0001", + uploadGenerationId: "generation-1", + uploadLeaseExpiresAt: "2000-01-01T00:00:00.000Z", + }, + ]; + mockGetResumableSession.mockResolvedValue({ + bytesUploaded: 7_864_320, + lastCommittedIndex: 1, + }); + + await expect(handler({} as any)).resolves.toEqual( + expect.objectContaining({ + resumable: true, + attemptId: "client-attempt-0001", + uploadGenerationId: "generation-2", + uploadMode: "buffered", + bytesReceived: 0, + nextChunkIndex: 0, + }), + ); + expect(mockCompareAndSetAppState).toHaveBeenCalledWith( + "recording-upload-rec-1", + expect.anything(), + expect.objectContaining({ + uploadAttemptId: "client-attempt-0001", + uploadGenerationId: "generation-2", + }), + ); + expect(mockAbortResumableUploadSession).toHaveBeenCalledOnce(); + expect(mockDeleteResumableSession).toHaveBeenCalledWith( + "rec-1", + "generation-1", + ); + }); + + it("keeps the replacement generation fenced when retired-session cleanup fails", async () => { + mockSelectRows.rows = [ + { + id: "rec-1", + status: "uploading", + uploadAttemptId: "stale-attempt-0001", + uploadGenerationId: "generation-1", + uploadLeaseExpiresAt: "2000-01-01T00:00:00.000Z", + }, + ]; + mockGetResumableSession.mockResolvedValue({ + bytesUploaded: 7_864_320, + lastCommittedIndex: 1, + }); + mockAbortResumableUploadSession.mockResolvedValue(false); + + await expect(handler({} as any)).resolves.toEqual({ + resumable: false, + recoveryEnabled: true, + recordingId: "rec-1", + status: "uploading", + reason: "stale_provider_session_invalidation_failed", + }); + expect(mockUpdateSets).toHaveLength(1); + expect(mockUpdateSets[0]).toEqual( + expect.objectContaining({ + uploadAttemptId: "client-attempt-0001", + uploadGenerationId: "generation-2", + }), + ); + expect(mockDeleteResumableSession).not.toHaveBeenCalled(); + expect(mockCompareAndSetAppState).not.toHaveBeenCalled(); + }); + + it("loses a stale-claim takeover race without publishing resume state", async () => { + mockSelectRows.rows = [ + { + id: "rec-1", + status: "uploading", + uploadAttemptId: "stale-attempt-0001", + uploadLeaseExpiresAt: "2000-01-01T00:00:00.000Z", + }, + ]; + mockUpdateRows.rows = []; + + await expect(handler({} as any)).resolves.toEqual({ + resumable: false, + recoveryEnabled: true, + recordingId: "rec-1", + status: "uploading", + reason: "retry_already_active", + retryAfterMs: 250, + }); + expect(mockWriteAppState).not.toHaveBeenCalled(); + }); + it("lets the same claim re-read its offset after a lost response", async () => { mockSelectRows.rows = [ { diff --git a/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.ts b/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.ts index de3244b895..eb5949eca4 100644 --- a/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.ts +++ b/templates/clips/server/routes/api/uploads/[recordingId]/resume.get.ts @@ -11,6 +11,8 @@ * Route: GET /api/uploads/:recordingId/resume */ +import { randomUUID } from "node:crypto"; + import { compareAndSetAppState, readAppState, @@ -18,7 +20,7 @@ import { } from "@agent-native/core/application-state"; import { isFeatureFlagEnabled } from "@agent-native/core/feature-flags"; import { runWithRequestContext } from "@agent-native/core/server"; -import { and, eq, isNull } from "drizzle-orm"; +import { and, eq, isNull, lte } from "drizzle-orm"; import { createError, defineEventHandler, @@ -40,12 +42,32 @@ import { getEventOwnerContext, ownerEmailMatches, } from "../../../../lib/recordings.js"; -import { getResumableSession } from "../../../../lib/resumable-session.js"; import { - isRetryableUploadInterruption, - RETRYABLE_UPLOAD_INTERRUPTION_REASON, -} from "../../../../lib/upload-interruption.js"; -import { uploadLeaseExpiry } from "../../../../lib/upload-lease.js"; + deleteResumableSession, + getResumableSession, +} from "../../../../lib/resumable-session.js"; +import { abortResumableUploadSession } from "../../../../lib/resumable-upload-cleanup.js"; +import { isRetryableUploadInterruption } from "../../../../lib/upload-interruption.js"; +import { + UPLOAD_LEASE_MS, + uploadLeaseExpiry, +} from "../../../../lib/upload-lease.js"; + +const RETRY_CLAIM_LIVENESS_MS = 30 * 1000; +const RETRY_CLAIM_RETRY_AFTER_MIN_MS = 250; + +function retryClaimRetryAfterMs( + lastHeartbeatMs: number, + nowMs: number, +): number { + return Math.max( + RETRY_CLAIM_RETRY_AFTER_MIN_MS, + Math.min( + RETRY_CLAIM_LIVENESS_MS, + lastHeartbeatMs + RETRY_CLAIM_LIVENESS_MS - nowMs, + ), + ); +} export default defineEventHandler(async (event: H3Event) => { setResponseHeader(event, "Cache-Control", "private, max-age=0, no-store"); @@ -85,13 +107,51 @@ export default defineEventHandler(async (event: H3Event) => { orgId, }); if (!recoveryEnabled) { - return { - recoveryEnabled: false, - resumable: false, - recordingId, - status: null, - reason: "feature_disabled", - }; + return runWithRequestContext({ userEmail: ownerEmail, orgId }, async () => { + const [recording] = await getDb() + .select({ + status: schema.recordings.status, + uploadAttemptId: schema.recordings.uploadAttemptId, + uploadGenerationId: schema.recordings.uploadGenerationId, + }) + .from(schema.recordings) + .where( + and( + eq(schema.recordings.id, recordingId), + ownerEmailMatches(schema.recordings.ownerEmail, ownerEmail), + ), + ); + if (!recording) { + setResponseStatus(event, 404); + return { error: "Recording not found" }; + } + if ( + recording.uploadAttemptId && + recording.uploadAttemptId !== requestedAttemptId + ) { + setResponseStatus(event, 409); + return { + recoveryEnabled: false, + resumable: false, + recordingId, + status: recording.status, + reason: "retry_already_active", + }; + } + return { + recoveryEnabled: false, + resumable: false, + recordingId, + status: recording.status ?? null, + reason: "feature_disabled", + ...(recording.uploadAttemptId + ? { attemptId: recording.uploadAttemptId } + : {}), + ...(recording.uploadGenerationId + ? { uploadGenerationId: recording.uploadGenerationId } + : {}), + }; + }); } return runWithRequestContext({ userEmail: ownerEmail, orgId }, async () => { @@ -104,6 +164,7 @@ export default defineEventHandler(async (event: H3Event) => { uploadProgress: schema.recordings.uploadProgress, uploadAttemptId: schema.recordings.uploadAttemptId, uploadGenerationId: schema.recordings.uploadGenerationId, + uploadLeaseExpiresAt: schema.recordings.uploadLeaseExpiresAt, }) .from(schema.recordings) .where( @@ -121,19 +182,38 @@ export default defineEventHandler(async (event: H3Event) => { // Legacy rows keep their null generation and unscoped scratch. A reset // upgrades them by installing a fresh generation before it deletes data. const existingGenerationId = recording.uploadGenerationId ?? null; - const generationId = existingGenerationId; - const session = generationId + let generationId = existingGenerationId; + let session = generationId ? await getResumableSession(recordingId, generationId) : await getResumableSession(recordingId); const retryableFailure = recording.status === "failed" && isRetryableUploadInterruption(recording.failureReason); const existingAttemptId = recording.uploadAttemptId ?? null; - if ( + const nowMs = Date.now(); + const now = new Date(nowMs).toISOString(); + const staleLeaseThreshold = new Date( + nowMs + UPLOAD_LEASE_MS - RETRY_CLAIM_LIVENESS_MS, + ).toISOString(); + const claimLeaseExpiryMs = Date.parse(recording.uploadLeaseExpiresAt ?? ""); + const claimHeartbeatMs = claimLeaseExpiryMs - UPLOAD_LEASE_MS; + const differentRetryClaim = recording.status === "uploading" && existingAttemptId !== null && - existingAttemptId !== requestedAttemptId - ) { + existingAttemptId !== requestedAttemptId; + if (differentRetryClaim && !Number.isFinite(claimHeartbeatMs)) { + setResponseStatus(event, 409); + return { + resumable: false, + recoveryEnabled: true, + recordingId, + status: "uploading", + reason: "retry_claim_liveness_unavailable", + }; + } + const differentLiveRetryClaim = + differentRetryClaim && claimHeartbeatMs > nowMs - RETRY_CLAIM_LIVENESS_MS; + if (differentLiveRetryClaim) { setResponseStatus(event, 409); return { resumable: false, @@ -141,6 +221,7 @@ export default defineEventHandler(async (event: H3Event) => { recordingId, status: "uploading", reason: "retry_already_active", + retryAfterMs: retryClaimRetryAfterMs(claimHeartbeatMs, nowMs), }; } if (recording.status !== "uploading" && !retryableFailure) { @@ -158,15 +239,21 @@ export default defineEventHandler(async (event: H3Event) => { const uploadStateRaw = await readAppState(uploadStateKey); const uploadState = uploadStateRaw ?? {}; const attemptId = requestedAttemptId; - const now = new Date().toISOString(); + const takingOverStaleRetryClaim = differentRetryClaim; + const claimedGenerationId = takingOverStaleRetryClaim + ? randomUUID() + : generationId; + const claimedLeaseExpiry = uploadLeaseExpiry(nowMs); const claimed = await getDb() .update(schema.recordings) .set({ status: "uploading", failureReason: null, uploadAttemptId: attemptId, - ...(generationId ? { uploadGenerationId: generationId } : {}), - uploadLeaseExpiresAt: uploadLeaseExpiry(), + ...(claimedGenerationId + ? { uploadGenerationId: claimedGenerationId } + : {}), + uploadLeaseExpiresAt: claimedLeaseExpiry, updatedAt: now, }) .where( @@ -177,10 +264,7 @@ export default defineEventHandler(async (event: H3Event) => { ? eq(schema.recordings.status, "failed") : eq(schema.recordings.status, "uploading"), retryableFailure - ? eq( - schema.recordings.failureReason, - RETRYABLE_UPLOAD_INTERRUPTION_REASON, - ) + ? eq(schema.recordings.failureReason, recording.failureReason!) : undefined, existingAttemptId === null ? isNull(schema.recordings.uploadAttemptId) @@ -188,6 +272,9 @@ export default defineEventHandler(async (event: H3Event) => { existingGenerationId === null ? isNull(schema.recordings.uploadGenerationId) : eq(schema.recordings.uploadGenerationId, existingGenerationId), + takingOverStaleRetryClaim + ? lte(schema.recordings.uploadLeaseExpiresAt, staleLeaseThreshold) + : undefined, ), ) .returning({ id: schema.recordings.id }); @@ -200,9 +287,29 @@ export default defineEventHandler(async (event: H3Event) => { recordingId, status: "uploading", reason: "retry_already_active", + retryAfterMs: 250, }; } + if (takingOverStaleRetryClaim && session) { + const invalidated = await abortResumableUploadSession(session, { + label: `upload-resume-takeover-${recordingId}`, + }); + if (!invalidated) { + setResponseStatus(event, 409); + return { + resumable: false, + recoveryEnabled: true, + recordingId, + status: "uploading", + reason: "stale_provider_session_invalidation_failed", + }; + } + await deleteResumableSession(recordingId, generationId); + session = null; + } + generationId = claimedGenerationId; + const uploadStateUpdated = await compareAndSetAppState( uploadStateKey, uploadStateRaw,