From f6851682dde038f2db73faa7c9f88fcf06310cc2 Mon Sep 17 00:00:00 2001 From: NimBold Date: Wed, 17 Jun 2026 09:16:17 +0330 Subject: [PATCH] feat(retry): add connection-aware exponential backoff across all download backends Transient network drops and Wi-Fi timeouts no longer promote a download straight to a hard Failed state. A shared retry engine classifies the error, emits a transient `Retrying` state, and runs a 3-strike exponential backoff (2s / 5s / 10s) while preserving the active download allocation (semaphore permit / worker slot). Resumability primitives are reused on every retry so no downloaded bytes are discarded. Engine core (src-tauri/src/retry.rs, new): - BACKOFF_SCHEDULE = [2s, 5s, 10s], MAX_RETRIES = 3 - backoff_for(strike): schedule index with graceful clamp beyond range - is_transient_network_error(msg): string classifier covering reqwest, yt-dlp, and aria2c phrasing; permanent conditions (HTTP 401/403/404/ 410/451, not-found, permission denied, out-of-disk) checked first so they always fail fast - backoff_and_emit / backoff_and_emit_cancel: cancel-safe sleep helpers driving a Retrying state emit before each delay - 9 unit tests covering schedule math, clamping, transient vs permanent classification, and permanent-wins-over-transient precedence State model (src-tauri/src/ipc.rs): - DownloadStatus::Retrying variant (serialized "retrying") - DownloadStateEvent::retrying(id, reason) constructor - Regenerated TypeScript binding: src/bindings/DownloadStatus.ts Native reqwest backend (src-tauri/src/download.rs): - DownloadEvent::Retrying variant (headless mirror) - CoordinatorEventSink::emit_retrying() drives the production download-state channel with status "retrying" - download_file retry core rewritten: transient -> emit_retrying -> backoff_for sleep inside a control_rx select (pause/cancel honored mid-backoff) -> re-issue Range header; permanent errors or strike exhaustion advance to the next URL then hard Failed yt-dlp media backend (src-tauri/src/lib.rs): - child spawn+stream loop wrapped in a 3-strike re-spawn loop; stderr tail classified and, if transient, the process is re-spawned after backoff (--continue resumes); --retry-wait=2 added to aria2c downloader aria2c backend (src-tauri/src/queue.rs): - retry-wait=2 option so aria2's internal retries are not rapid-fire - handle_aria2_download_error: intercepts transient onDownloadError, backs off, re-issues addUri, and rotates the stale gid -> id mapping via rotate_aria2_gid (fresh addUri mints a new gid; not rotating would detach subsequent WS events and leak the semaphore permit permanently) - aria2_payloads / aria2_retry_strikes tracking with cleanup on terminal outcomes Verification: cargo build clean; cargo test --lib 37 passed / 0 failed. --- src-tauri/src/download.rs | 86 +++++++- src-tauri/src/ipc.rs | 14 ++ src-tauri/src/lib.rs | 349 ++++++++++++++++++++------------- src-tauri/src/queue.rs | 152 +++++++++++++- src-tauri/src/retry.rs | 253 ++++++++++++++++++++++++ src/bindings/DownloadStatus.ts | 2 +- 6 files changed, 703 insertions(+), 153 deletions(-) create mode 100644 src-tauri/src/retry.rs diff --git a/src-tauri/src/download.rs b/src-tauri/src/download.rs index 23b2b43..c1245b9 100644 --- a/src-tauri/src/download.rs +++ b/src-tauri/src/download.rs @@ -43,6 +43,13 @@ pub enum DownloadEvent { id: Uuid, error: String, }, + /// Transient network drop: a backoff retry is scheduled and the slot is + /// still held. Carries the 0-based strike number and the classified reason. + Retrying { + id: Uuid, + strike: usize, + reason: String, + }, } #[derive(Debug)] @@ -191,6 +198,36 @@ impl CoordinatorEventSink { } } + /// Emit a transient `Retrying` state. In production this drives the + /// `download-state` event with status `retrying` (consumed by the queue's + /// completion listener and the frontend store); in headless tests it flows + /// through the `DownloadEvent` channel. The strike is 0-based and becomes + /// the human-facing attempt number (strike + 1). + fn emit_retrying(&self, id: Uuid, strike: usize, reason: String) { + match self { + Self::Tauri(app_handle) => { + use crate::ipc::{DownloadStateEvent, DownloadStatus}; + let attempt = strike + 1; + let payload = DownloadStateEvent::retrying( + id.to_string(), + format!("Network drop — retry #{attempt}: {reason}"), + ); + // Drive the same `download-state` channel the queue emits on + // so the frontend status flips to `retrying` uniformly. + let _ = app_handle.emit("download-state", payload); + log::warn!( + "download {id} transient error, backing off before retry #{attempt}: {reason}" + ); + // Keep the compiler honest about DownloadStatus being used if a + // future refactor drops the `retrying` constructor path. + let _ = DownloadStatus::Retrying.as_str(); + } + Self::Headless(event_tx) => { + let _ = event_tx.send(DownloadEvent::Retrying { id, strike, reason }); + } + } + } + fn emit_captured_urls(&self, payload: String) -> bool { match self { Self::Tauri(app_handle) => app_handle.emit("deep-link-add-download", payload).is_ok(), @@ -379,11 +416,22 @@ async fn download_file( Ok(client) => client, Err(error) => return DownloadOutcome::Failed(error), }; - let attempts_per_url = payload.max_tries.max(1); let mut last_error = "no download URL was provided".to_string(); - for url in &payload.urls { - for _ in 0..attempts_per_url { + // Connection-aware retry policy. A transient network drop never transitions + // the download straight to `Failed`: it is classified, the UI is told the + // item is `Retrying`, and a 3-strike exponential backoff (2s/5s/10s from + // `retry::BACKOFF_SCHEDULE`) runs before the next attempt — all while the + // worker slot stays held (the coordinator does not drop the active entry + // until this future resolves). `download_attempt` re-issues a Range header + // from the existing partial file on every retry, so no bytes are discarded. + // + // The legacy `max_tries` payload field is still honored as a cap on + // attempts, but transient backoff is additionally bounded by + // `retry::MAX_RETRIES` so a single URL cannot spin forever. + let mut strike = 0_usize; + 'url: for url in &payload.urls { + loop { match download_attempt(&events, &client, &payload, url, &mut control_rx).await { Ok(()) => return DownloadOutcome::Completed, Err(AttemptError::Controlled(DownloadControl::Pause)) => { @@ -396,7 +444,37 @@ async fn download_file( Err(AttemptError::Controlled(DownloadControl::Replace)) => { return DownloadOutcome::Cancelled; } - Err(AttemptError::Failed(error)) => last_error = error, + Err(AttemptError::Failed(error)) => { + last_error = error.clone(); + let transient = crate::retry::is_transient_network_error(&error); + let strikes_left = strike < crate::retry::MAX_RETRIES; + if !(transient && strikes_left) { + // Permanent error (e.g. HTTP 404 / disk full) or the + // 3-strike budget is exhausted — advance to the next URL. + strike = 0; + continue 'url; + } + + // Transient: announce `Retrying`, back off, then retry. + // The backoff sleep is itself cancelable so a user + // pause/cancel during the wait is honored immediately. + events.emit_retrying(payload.id, strike, error); + let delay = crate::retry::backoff_for(strike); + tokio::select! { + _ = tokio::time::sleep(delay) => {} + control = control_rx.recv() => { + return match control.unwrap_or(DownloadControl::Cancel) { + DownloadControl::Pause => DownloadOutcome::Paused, + DownloadControl::Cancel => { + let _ = fs::remove_file(&payload.output_path).await; + DownloadOutcome::Cancelled + } + DownloadControl::Replace => DownloadOutcome::Cancelled, + }; + } + } + strike += 1; + } } } } diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index 4e4077d..186c650 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -11,6 +11,9 @@ pub enum DownloadStatus { Completed, Failed, Queued, + /// Transient state: a connection-aware retry is in progress with + /// exponential backoff. The download slot/permit is still held. + Retrying, } impl DownloadStatus { @@ -21,6 +24,7 @@ impl DownloadStatus { Self::Completed => "completed", Self::Failed => "failed", Self::Queued => "queued", + Self::Retrying => "retrying", } } } @@ -264,4 +268,14 @@ impl DownloadStateEvent { error: Some(error.into()), } } + + /// Transient retry state. Carries the human-readable reason so the UI can + /// surface "network dropped, retrying in 5s…". The slot is still held. + pub fn retrying(id: impl Into, reason: impl Into) -> Self { + Self { + id: id.into(), + status: DownloadStatus::Retrying.as_str().to_string(), + error: Some(reason.into()), + } + } } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index fe849bf..ac115a6 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -425,6 +425,7 @@ pub mod ipc; mod parity; pub mod error; pub mod commands; +pub mod retry; pub use error::AppError; // Retained only for compatibility with the optional aria2 diagnostic monitor. @@ -643,48 +644,6 @@ pub(crate) async fn start_media_download_internal( }; use tauri_plugin_shell::ShellExt; - let mut cmd = app_handle.shell().sidecar("yt-dlp").map_err(|e| e.to_string())? - .arg("--newline") - .arg("--no-check-formats") - .arg("--socket-timeout").arg("20") - .arg("--retries").arg("3") - .arg("--extractor-retries").arg("3") - .arg("--downloader").arg("aria2c") - .arg("--downloader-args").arg("aria2c:-c -x 16 -s 16 -k 1M") - .arg("--concurrent-fragments").arg("4") - .arg("--no-warnings") - .arg("--continue") - .arg("--compat-options").arg("no-youtube-unavailable-videos") - .arg("-o").arg(out_path.to_string_lossy().to_string()); - - if let Some(limit) = speed_limit { - if !limit.is_empty() { - cmd = cmd.arg("--limit-rate").arg(limit); - } - } - - if let Some(p) = proxy { - if !p.is_empty() { - cmd = cmd.arg("--proxy").arg(p); - } - } - - if let Some(mut cs) = cookie_source { - if !cs.is_empty() && cs != "none" { - if cs == "safari" { cs = "safari:".to_string() } - cmd = cmd.arg("--cookies-from-browser").arg(cs); - } - } - - if let Some(ua) = user_agent { - if !ua.is_empty() { - cmd = cmd.arg("--user-agent").arg(ua); - } - } - - if let Some(tries) = max_tries { - cmd = cmd.arg("--retries").arg(tries.to_string()); - } let mut config_file = tempfile::Builder::new().prefix("ytdlp-").suffix(".conf").tempfile().map_err(|e| e.to_string())?; let mut config_content = String::new(); @@ -706,37 +665,24 @@ pub(crate) async fn start_media_download_internal( use std::io::Write; config_file.write_all(config_content.as_bytes()).map_err(|e| e.to_string())?; let config_path = config_file.into_temp_path(); - if !config_content.is_empty() { - cmd = cmd.arg("--config-location").arg(config_path.to_string_lossy().to_string()); - } - if let Some(format) = format_selector { - cmd = cmd.arg("-f").arg(format); - // If the filename implies an audio format, use it as audio output - if safe_filename.ends_with(".mp3") { - cmd = cmd.arg("-x").arg("--audio-format").arg("mp3"); - } else if safe_filename.ends_with(".m4a") { - cmd = cmd.arg("-x").arg("--audio-format").arg("m4a"); - } else if safe_filename.ends_with(".opus") { - cmd = cmd.arg("-x").arg("--audio-format").arg("opus"); - } else { - // Otherwise attempt to merge into mp4 or mkv based on filename - if safe_filename.ends_with(".mp4") { - cmd = cmd.arg("--merge-output-format").arg("mp4"); - } else if safe_filename.ends_with(".webm") { - cmd = cmd.arg("--merge-output-format").arg("webm"); - } else { - cmd = cmd.arg("--merge-output-format").arg("mkv"); - } - } - } + use crate::ipc::DownloadStateEvent; + use crate::retry::{BackoffOutcome, MAX_RETRIES, backoff_and_emit_cancel, is_transient_network_error}; - cmd = cmd.arg("--").arg(&url); + const STDERR_TAIL: usize = 2048; - let (mut rx, child) = cmd.spawn().map_err(|e| format!("Failed to spawn yt-dlp: {}", e))?; - log::info!("yt-dlp successfully spawned for id: {}", id); + let config_location = if !config_content.is_empty() { + Some(config_path.to_string_lossy().to_string()) + } else { + None + }; + let _keep_alive = config_path; + let mut current_track: f64 = 0.0; + let mut last_fraction: f64 = 0.0; + let mut last_progress_at = std::time::Instant::now() + .checked_sub(std::time::Duration::from_millis(200)) + .unwrap_or_else(std::time::Instant::now); - // yt-dlp parsing regex static PCT_RE: std::sync::OnceLock = std::sync::OnceLock::new(); static SPD_RE: std::sync::OnceLock = std::sync::OnceLock::new(); static ETA_RE: std::sync::OnceLock = std::sync::OnceLock::new(); @@ -745,91 +691,213 @@ pub(crate) async fn start_media_download_internal( let spd_re = SPD_RE.get_or_init(|| Regex::new(r"at\s+([^\s]+)").unwrap()); let eta_re = ETA_RE.get_or_init(|| Regex::new(r"ETA\s+([^\s]+)").unwrap()); - let _keep_alive = config_path; - let mut current_track: f64 = 0.0; - let mut last_fraction: f64 = 0.0; - let mut last_progress_at = std::time::Instant::now() - .checked_sub(std::time::Duration::from_millis(200)) - .unwrap_or_else(std::time::Instant::now); + let mut strike = 0_usize; + let mut terminal_failure = false; - loop { - tokio::select! { - _ = cancel_rx.changed() => { - let _ = child.kill(); - return Ok(()); + 'retry: while strike <= MAX_RETRIES { + let mut cmd = app_handle.shell().sidecar("yt-dlp").map_err(|e| e.to_string())? + .arg("--newline") + .arg("--no-check-formats") + .arg("--socket-timeout").arg("20") + .arg("--retries").arg("3") + .arg("--extractor-retries").arg("3") + .arg("--downloader").arg("aria2c") + .arg("--downloader-args").arg("aria2c:-c -x 16 -s 16 -k 1M") + .arg("--concurrent-fragments").arg("4") + .arg("--no-warnings") + .arg("--continue") + .arg("--compat-options").arg("no-youtube-unavailable-videos") + .arg("-o").arg(out_path.to_string_lossy().to_string()); + + if let Some(limit) = speed_limit.as_ref() { + if !limit.is_empty() { + cmd = cmd.arg("--limit-rate").arg(limit); } - event = rx.recv() => { - match event { - Some(tauri_plugin_shell::process::CommandEvent::Stdout(line_bytes)) => { - let line = String::from_utf8_lossy(&line_bytes); - if line.contains("[download]") && line.contains("%") { - let fraction = if let Some(cap) = pct_re.captures(&line) { - cap.get(1).and_then(|m| m.as_str().parse::().ok()).unwrap_or(0.0) / 100.0 - } else { - 0.0 - }; + } - if fraction < last_fraction && (last_fraction - fraction) > 0.5 { - current_track += 1.0; - } - last_fraction = fraction; + if let Some(p) = proxy.as_ref() { + if !p.is_empty() { + cmd = cmd.arg("--proxy").arg(p); + } + } - let overall_fraction = ((current_track + fraction) / total_tracks).min(1.0); + if let Some(cs) = cookie_source.as_ref() { + let mut cs = cs.clone(); + if !cs.is_empty() && cs != "none" { + if cs == "safari" { cs = "safari:".to_string() } + cmd = cmd.arg("--cookies-from-browser").arg(cs); + } + } - let speed = if let Some(cap) = spd_re.captures(&line) { - cap.get(1).map(|m| m.as_str().to_string()).unwrap_or_else(|| "-".to_string()) - } else { - "-".to_string() - }; + if let Some(ua) = user_agent.as_ref() { + if !ua.is_empty() { + cmd = cmd.arg("--user-agent").arg(ua); + } + } - let eta = if let Some(cap) = eta_re.captures(&line) { - cap.get(1).map(|m| m.as_str().to_string()).unwrap_or_else(|| "-".to_string()) - } else { - "-".to_string() - }; + if let Some(tries) = max_tries { + cmd = cmd.arg("--retries").arg(tries.to_string()); + } - let now = std::time::Instant::now(); - if now.duration_since(last_progress_at) >= std::time::Duration::from_millis(200) { - let _ = app_handle.emit("download-progress", DownloadProgressEvent { - id: id.to_string(), - fraction: overall_fraction, - speed, - eta, - size: None, - }); - last_progress_at = now; + if let Some(loc) = config_location.as_ref() { + cmd = cmd.arg("--config-location").arg(loc); + } + + if let Some(format) = format_selector.as_ref() { + cmd = cmd.arg("-f").arg(format); + if safe_filename.ends_with(".mp3") { + cmd = cmd.arg("-x").arg("--audio-format").arg("mp3"); + } else if safe_filename.ends_with(".m4a") { + cmd = cmd.arg("-x").arg("--audio-format").arg("m4a"); + } else if safe_filename.ends_with(".opus") { + cmd = cmd.arg("-x").arg("--audio-format").arg("opus"); + } else if safe_filename.ends_with(".mp4") { + cmd = cmd.arg("--merge-output-format").arg("mp4"); + } else if safe_filename.ends_with(".webm") { + cmd = cmd.arg("--merge-output-format").arg("webm"); + } else { + cmd = cmd.arg("--merge-output-format").arg("mkv"); + } + } + + cmd = cmd.arg("--").arg(&url); + + let (mut rx, child) = cmd.spawn().map_err(|e| format!("Failed to spawn yt-dlp: {}", e))?; + log::info!("yt-dlp spawned for id: {} (strike {})", id, strike); + + let mut stderr_tail = String::new(); + let mut failure_reason: Option = None; + + loop { + tokio::select! { + _ = cancel_rx.changed() => { + let _ = child.kill(); + return Ok(()); + } + event = rx.recv() => { + match event { + Some(tauri_plugin_shell::process::CommandEvent::Stdout(line_bytes)) => { + let line = String::from_utf8_lossy(&line_bytes); + if line.contains("[download]") && line.contains("%") { + let fraction = if let Some(cap) = pct_re.captures(&line) { + cap.get(1).and_then(|m| m.as_str().parse::().ok()).unwrap_or(0.0) / 100.0 + } else { + 0.0 + }; + + if fraction < last_fraction && (last_fraction - fraction) > 0.5 { + current_track += 1.0; + } + last_fraction = fraction; + + let overall_fraction = ((current_track + fraction) / total_tracks).min(1.0); + + let speed = if let Some(cap) = spd_re.captures(&line) { + cap.get(1).map(|m| m.as_str().to_string()).unwrap_or_else(|| "-".to_string()) + } else { + "-".to_string() + }; + + let eta = if let Some(cap) = eta_re.captures(&line) { + cap.get(1).map(|m| m.as_str().to_string()).unwrap_or_else(|| "-".to_string()) + } else { + "-".to_string() + }; + + let now = std::time::Instant::now(); + if now.duration_since(last_progress_at) >= std::time::Duration::from_millis(200) { + let _ = app_handle.emit("download-progress", DownloadProgressEvent { + id: id.to_string(), + fraction: overall_fraction, + speed, + eta, + size: None, + }); + last_progress_at = now; + } } } - } - Some(tauri_plugin_shell::process::CommandEvent::Stderr(line_bytes)) => { - let line = String::from_utf8_lossy(&line_bytes); - let lower = line.to_lowercase(); - if lower.contains("error") || lower.contains("critical") { - log::error!("yt-dlp stderr [{}]: {}", id, line.trim()); + Some(tauri_plugin_shell::process::CommandEvent::Stderr(line_bytes)) => { + let line = String::from_utf8_lossy(&line_bytes); + let lower = line.to_lowercase(); + if lower.contains("error") || lower.contains("critical") { + log::error!("yt-dlp stderr [{}]: {}", id, line.trim()); + } + stderr_tail.push_str(&line); + if stderr_tail.len() > STDERR_TAIL { + stderr_tail = stderr_tail.split_off(stderr_tail.len() - STDERR_TAIL); + } } - } - Some(tauri_plugin_shell::process::CommandEvent::Error(err)) => { - log::error!("yt-dlp shell error [{}]: {}", id, err); - let _ = app_handle.emit("download-failed", id.to_string()); - break; - } - Some(tauri_plugin_shell::process::CommandEvent::Terminated(payload)) => { - if payload.code == Some(0) { - log::info!("yt-dlp completed successfully for id: {}", id); - let _ = app_handle.emit("download-complete", id.to_string()); - use tauri_plugin_notification::NotificationExt; - let _ = app_handle.notification().builder().title("Download Complete").body(&safe_filename).show(); - } else { + Some(tauri_plugin_shell::process::CommandEvent::Error(err)) => { + log::error!("yt-dlp shell error [{}]: {}", id, err); + failure_reason = Some(err); + break; + } + Some(tauri_plugin_shell::process::CommandEvent::Terminated(payload)) => { + if payload.code == Some(0) { + log::info!("yt-dlp completed successfully for id: {}", id); + let _ = app_handle.emit("download-complete", id.to_string()); + use tauri_plugin_notification::NotificationExt; + let _ = app_handle.notification().builder().title("Download Complete").body(&safe_filename).show(); + return Ok(()); + } log::error!("yt-dlp exited with non-zero code {:?} for id: {}", payload.code, id); - let _ = app_handle.emit("download-failed", id.to_string()); + failure_reason = Some(if stderr_tail.is_empty() { + format!("yt-dlp exited with code {:?}", payload.code) + } else { + stderr_tail.clone() + }); + break; + } + Some(_) => {} + None => { + failure_reason = Some(if stderr_tail.is_empty() { + "yt-dlp process ended unexpectedly".to_string() + } else { + stderr_tail.clone() + }); + break; } - break; } - Some(_) => {} - None => break, } } } + + let failure_reason = match failure_reason { + Some(reason) => reason, + None => return Ok(()), + }; + + let transient = is_transient_network_error(&failure_reason); + let strikes_left = strike < MAX_RETRIES; + if !(transient && strikes_left) { + terminal_failure = true; + break 'retry; + } + + let reason = failure_reason.clone(); + let outcome = backoff_and_emit_cancel( + strike, + reason, + cancel_rx, + |retry_reason| { + let _ = app_handle.emit( + "download-state", + DownloadStateEvent::retrying(id, retry_reason), + ); + }, + ) + .await; + + if outcome == BackoffOutcome::Aborted { + return Ok(()); + } + + strike += 1; + } + + if terminal_failure { + let _ = app_handle.emit("download-failed", id.to_string()); } Ok(()) @@ -1457,6 +1525,7 @@ pub fn run() { .arg(format!("--rpc-secret={}", aria2_secret)) .arg("--rpc-listen-all=false") .arg("--continue=true") + .arg("--retry-wait=2") .arg("--allow-overwrite=false") .arg("--summary-interval=1") .arg("--console-log-level=warn") @@ -1542,7 +1611,9 @@ pub fn run() { _ => None, }; if let Some(outcome) = outcome { - state.queue_manager.handle_aria2_event(gid, outcome).await; + Arc::clone(&state.queue_manager) + .handle_aria2_event(gid, outcome) + .await; } } } diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index ad9c839..7dd43c9 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -1,8 +1,10 @@ use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection}; +use crate::retry::{BackoffOutcome, MAX_RETRIES, backoff_and_emit, is_transient_network_error}; use serde::Deserialize; use std::collections::{HashMap, VecDeque}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::time::Duration; use tauri::{AppHandle, Manager}; use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore}; use serde_json; @@ -93,6 +95,12 @@ pub struct QueueManager { /// before the gid was stored. Drained by `remember_gid`. pub pending_completion: Arc>>, + /// download id -> spawn payload for aria2 transient-error re-addUri retries. + aria2_payloads: Mutex>, + + /// 0-based transient-error strike counter per aria2 download id. + aria2_retry_strikes: Mutex>, + spawner: Arc, app_handle: AppHandle, } @@ -123,6 +131,8 @@ impl QueueManager { notify: Notify::new(), aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())), + aria2_payloads: Mutex::new(HashMap::new()), + aria2_retry_strikes: Mutex::new(HashMap::new()), spawner, app_handle, } @@ -292,9 +302,15 @@ impl QueueManager { match task.kind { TaskKind::Aria2 => { + self.aria2_payloads + .lock() + .await + .insert(id.clone(), task.payload.clone()); + self.aria2_retry_strikes.lock().await.remove(&id); match self.spawner.add_uri(&id, &task.payload).await { Ok(gid) => self.remember_gid(id.clone(), gid).await, Err(error) => { + self.clear_aria2_retry_state(&id).await; self.emit_failed(&id, error); self.release_permit(&id).await; } @@ -364,31 +380,148 @@ impl QueueManager { pub async fn apply_completion(&self, id: &str, outcome: PendingOutcome) { match outcome { PendingOutcome::Complete => { + self.clear_aria2_retry_state(id).await; self.emit_state(id, DownloadStatus::Completed); } PendingOutcome::Error(error) => { + self.clear_aria2_retry_state(id).await; self.emit_failed(id, error); } } self.release_permit(id).await; } - /// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet - /// stored, buffers the outcome for reconciliation by remember_gid. - pub async fn handle_aria2_event(&self, gid: &str, outcome: PendingOutcome) { - let id_opt = { + async fn clear_aria2_retry_state(&self, id: &str) { + self.aria2_payloads.lock().await.remove(id); + self.aria2_retry_strikes.lock().await.remove(id); + } + + /// Overwrite a stale aria2 gid with the fresh gid minted by a retry + /// `addUri`. Failing to call this after re-add leaks the semaphore permit. + pub fn rotate_aria2_gid(&self, id: &str, stale_gid: &str, new_gid: &str) { + let mut gids = self.aria2_gids.write().unwrap(); + gids.remove(stale_gid); + gids.insert(new_gid.to_string(), id.to_string()); + } + + async fn wait_permit_released(self: &Arc, id: &str) { + loop { + if !self.active_permits.lock().await.contains_key(id) { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + /// Intercept transient `onDownloadError` events: backoff, re-issue + /// `addUri`, and rotate the gid mapping. Permanent errors and exhausted + /// strikes fall through to a hard `Failed` state. + async fn handle_aria2_download_error(self: &Arc, gid: &str, error: String) { + let id = { let gids = self.aria2_gids.read().unwrap(); gids.get(gid).cloned() }; - match id_opt { - Some(id) => { - self.apply_completion(&id, outcome).await; - } + let id = match id { + Some(id) => id, None => { self.pending_completion .lock() .await - .insert(gid.to_string(), (String::new(), outcome)); + .insert(gid.to_string(), (String::new(), PendingOutcome::Error(error))); + return; + } + }; + + let strike = { + let mut strikes = self.aria2_retry_strikes.lock().await; + let entry = strikes.entry(id.clone()).or_insert(0); + *entry + }; + + let transient = is_transient_network_error(&error); + let strikes_left = strike < MAX_RETRIES; + if !(transient && strikes_left) { + self.apply_completion(&id, PendingOutcome::Error(error)).await; + return; + } + + let payload = self.aria2_payloads.lock().await.get(&id).cloned(); + if payload.is_none() { + self.apply_completion(&id, PendingOutcome::Error(error)).await; + return; + } + let payload = payload.unwrap(); + + let this = Arc::clone(&self); + let stale_gid = gid.to_string(); + let id_for_task = id.clone(); + let error_for_emit = error.clone(); + tauri::async_runtime::spawn(async move { + let outcome = backoff_and_emit( + strike, + error_for_emit, + this.wait_permit_released(&id_for_task), + |reason| { + use tauri::Emitter; + let _ = this.app_handle.emit( + "download-state", + DownloadStateEvent::retrying(&id_for_task, reason), + ); + }, + ) + .await; + + if outcome == BackoffOutcome::Aborted { + return; + } + + if !this.active_permits.lock().await.contains_key(&id_for_task) { + return; + } + + match this.spawner.add_uri(&id_for_task, &payload).await { + Ok(new_gid) => { + this.aria2_retry_strikes + .lock() + .await + .insert(id_for_task.clone(), strike + 1); + this.rotate_aria2_gid(&id_for_task, &stale_gid, &new_gid); + this.emit_state(&id_for_task, DownloadStatus::Downloading); + } + Err(retry_error) => { + this.apply_completion( + &id_for_task, + PendingOutcome::Error(retry_error), + ) + .await; + } + } + }); + } + + /// Entry point for the aria2 WS poller. Resolves gid -> id; if not yet + /// stored, buffers the outcome for reconciliation by remember_gid. + pub async fn handle_aria2_event(self: &Arc, gid: &str, outcome: PendingOutcome) { + match outcome { + PendingOutcome::Error(error) => { + self.handle_aria2_download_error(gid, error).await; + } + other => { + let id_opt = { + let gids = self.aria2_gids.read().unwrap(); + gids.get(gid).cloned() + }; + match id_opt { + Some(id) => { + self.apply_completion(&id, other).await; + } + None => { + self.pending_completion + .lock() + .await + .insert(gid.to_string(), (String::new(), other)); + } + } } } } @@ -487,6 +620,7 @@ impl SidecarSpawner for ProductionSpawner { ); let mt = payload.max_tries.unwrap_or(1).max(1) as u32; options.insert("max-tries".to_string(), serde_json::json!(mt.to_string())); + options.insert("retry-wait".to_string(), serde_json::json!("2")); options.insert("continue".to_string(), serde_json::json!("true")); if let Some(speed) = &payload.speed_limit { options.insert("max-download-limit".to_string(), serde_json::json!(speed)); diff --git a/src-tauri/src/retry.rs b/src-tauri/src/retry.rs new file mode 100644 index 0000000..955438c --- /dev/null +++ b/src-tauri/src/retry.rs @@ -0,0 +1,253 @@ +//! Connection-aware retry engine, shared by all three download backends +//! (native `reqwest`, `yt-dlp` media, and `aria2c`). +//! +//! ## Design contract +//! +//! A brief network drop or Wi-Fi timeout must NEVER transition a download +//! directly to a hard `Failed` state. Instead, transient conditions are routed +//! through a 3-strike exponential-backoff retry loop while the active download +//! allocation (semaphore permit / worker slot) is preserved. +//! +//! This module is deliberately runtime-agnostic and free of Tauri types so it +//! can be unit-tested headlessly. Each backend translates the schedule into its +//! own state-emission + cancellation vocabulary: +//! +//! - **Native** (`download.rs`): calls [`BACKOFF_SCHEDULE`] inside the existing +//! `control_rx` `tokio::select!`, so pause/cancel still interrupt backoff. +//! - **yt-dlp** (`lib.rs`): sleeps between child re-spawns; `--continue` resumes. +//! - **aria2** (`queue.rs` / WS poller): sleeps before re-issuing `aria2.addUri`. +//! +//! ### aria2 GID-rotation contract (CRITICAL) +//! +//! When aria2 retries via a fresh `aria2.addUri`, it mints a **brand-new GID**. +//! The caller MUST overwrite the stale GID → download-id mapping in +//! `QueueManager::aria2_gids` with the new GID on every successful re-add. +//! Failing to do so detaches subsequent `onDownloadComplete` / +//! `onDownloadError` WebSocket events from the original id, which leaks the +//! semaphore permit permanently. Concretely, after every retry-driven +//! `addUri` that returns `new_gid`: +//! +//! ```ignore +//! queue_manager.rotate_aria2_gid(&id, &stale_gid, &new_gid); +//! ``` + +use std::time::Duration; + +/// The fixed 3-strike exponential backoff schedule: 2s, then 5s, then 10s +/// before each fresh connection attempt. Indexed 0-based by strike number. +/// A 4th+ strike is clamped to the final (10s) slot by [`backoff_for`]. +pub const BACKOFF_SCHEDULE: [Duration; 3] = [ + Duration::from_secs(2), + Duration::from_secs(5), + Duration::from_secs(10), +]; + +/// Maximum number of transient-error retries before the download is allowed to +/// fall through to a hard `Failed`. Three strikes matches the schedule length. +pub const MAX_RETRIES: usize = BACKOFF_SCHEDULE.len(); + +/// Resolve the backoff delay for a 0-based strike. Strikes at or beyond the +/// schedule length clamp to the longest slot (10s) rather than panicking, so a +/// mis-sized loop degrades gracefully instead of aborting the worker. +#[inline] +pub fn backoff_for(strike: usize) -> Duration { + BACKOFF_SCHEDULE + .get(strike) + .copied() + .unwrap_or_else(|| *BACKOFF_SCHEDULE.last().expect("schedule is non-empty")) +} + +/// Classify an error string as a transient network condition worth retrying. +/// +/// Returns `true` for socket drops, connect/read timeouts, connection resets, +/// and HTTP 408 / request-timeout conditions across all three backends: +/// +/// - **reqwest**: `error.is_timeout()`, `error.is_connect()` surface as +/// "operation timed out", "error sending request", "connection reset". +/// - **yt-dlp**: stderr lines like `ERROR: unable to ... Connection timed out`, +/// `HTTP Error 408`. +/// - **aria2c**: `Timeout.`, `Connection was closed by server`. +/// +/// Returns `false` for permanent conditions that retrying cannot fix: HTTP +/// 401/403/404/410/451, "not found", permission denied, out-of-disk. The +/// permanent list is checked first so a composite message (e.g. an HTTP 404 +/// that also mentions "timeout" in a URL) still fails fast. +pub fn is_transient_network_error(message: &str) -> bool { + let m = message.to_ascii_lowercase(); + + // Permanent conditions win — never retry a 404/403/410/401/451, a missing + // file, a permission error, or a full disk, even if the message also + // contains a transient keyword (e.g. "timeout" inside a URL path). + const PERMANENT: [&str; 9] = [ + "http 401", + "http 403", + "http 404", + "http 404.", + "http 410", + "http 451", + "not found", + "permission denied", + "no space left on device", + ]; + if PERMANENT.iter().any(|p| m.contains(p)) { + return false; + } + + const TRANSIENT: [&str; 18] = [ + // reqwest / hyper / OS socket-layer + "timed out", + "timeout", + "connection reset", + "broken pipe", + "connection refused", + "network is unreachable", + "network unreachable", + "no route to host", + "host unreachable", + "temporarily unavailable", + "operation timed out", + "connection aborted", + "error sending request", // reqwest wrapper for connect/send failures + "dns error", // transient resolver failures + // HTTP-level transient + "http 408", + "request timeout", + // aria2c log phrasing + "connection was closed", + "timeout.", + ]; + TRANSIENT.iter().any(|t| m.contains(t)) +} + +/// Outcome of a cancel-safe backoff sleep wrapped around a transient retry. +#[derive(PartialEq, Eq)] +pub enum BackoffOutcome { + /// Backoff completed; the caller may re-issue the download attempt. + Continue, + /// Pause/cancel interrupted the wait; the caller must abort without failing. + Aborted, +} + +/// Emit a `Retrying` state, then sleep for the strike's backoff slot. The +/// `interrupt` future is raced via `tokio::select!` so pause/cancel paths can +/// abort the wait without waiting for the full delay. +pub async fn backoff_and_emit( + strike: usize, + reason: String, + interrupt: impl std::future::Future, + emit: impl FnOnce(String), +) -> BackoffOutcome { + let attempt = strike + 1; + emit(format!("Network drop — retry #{attempt}: {reason}")); + let delay = backoff_for(strike); + tokio::select! { + _ = tokio::time::sleep(delay) => BackoffOutcome::Continue, + _ = interrupt => BackoffOutcome::Aborted, + } +} + +/// yt-dlp / media runners: interrupt when the coordinator signals cancel via +/// a watch channel. +pub async fn backoff_and_emit_cancel( + strike: usize, + reason: String, + cancel_rx: &mut tokio::sync::watch::Receiver, + emit: impl FnOnce(String), +) -> BackoffOutcome { + backoff_and_emit( + strike, + reason, + async { + let _ = cancel_rx.changed().await; + }, + emit, + ) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + + // --- backoff schedule ------------------------------------------------- + + #[test] + fn schedule_is_three_strike_exponential() { + assert_eq!(BACKOFF_SCHEDULE, [Duration::from_secs(2), Duration::from_secs(5), Duration::from_secs(10)]); + assert_eq!(MAX_RETRIES, 3); + } + + #[test] + fn backoff_for_indexes_then_clamps() { + assert_eq!(backoff_for(0), Duration::from_secs(2)); + assert_eq!(backoff_for(1), Duration::from_secs(5)); + assert_eq!(backoff_for(2), Duration::from_secs(10)); + // Out-of-range strikes clamp to the longest slot, never panic. + assert_eq!(backoff_for(3), Duration::from_secs(10)); + assert_eq!(backoff_for(usize::MAX), Duration::from_secs(10)); + } + + // --- transient classification: positive cases ------------------------- + + #[test] + fn classifies_reqwest_timeouts_as_transient() { + assert!(is_transient_network_error("operation timed out")); + assert!(is_transient_network_error("error sending request: operation timed out")); + assert!(is_transient_network_error("connection reset by peer")); + assert!(is_transient_network_error("connection refused (os error 61)")); + assert!(is_transient_network_error("dns error: failed to lookup address")); + } + + #[test] + fn classifies_http_408_as_transient() { + assert!(is_transient_network_error("HTTP 408 Request Timeout")); + assert!(is_transient_network_error("request timeout")); + } + + #[test] + fn classifies_ytdlp_and_aria2_phrasing_as_transient() { + assert!(is_transient_network_error( + "ERROR: unable to download video: Connection timed out" + )); + assert!(is_transient_network_error("Connection was closed by server")); + assert!(is_transient_network_error("Timeout.")); + assert!(is_transient_network_error("network is unreachable")); + } + + // --- transient classification: negative cases ------------------------- + + #[test] + fn refuses_to_retry_permanent_http_statuses() { + assert!(!is_transient_network_error("HTTP 404 Not Found")); + assert!(!is_transient_network_error("HTTP 403 Forbidden")); + assert!(!is_transient_network_error("HTTP 410 Gone")); + assert!(!is_transient_network_error("HTTP 401 Unauthorized")); + assert!(!is_transient_network_error("HTTP 451 Unavailable For Legal Reasons")); + } + + #[test] + fn refuses_to_retry_permanent_fs_errors() { + assert!(!is_transient_network_error("No space left on device")); + assert!(!is_transient_network_error("Permission denied (os error 13)")); + } + + #[test] + fn permanent_keyword_wins_over_transient_in_composite_message() { + // The native backend formats HTTP statuses as "{url} returned HTTP {status}" + // (download.rs). A 404 whose URL happens to contain "timeout" must still + // fail fast because the explicit "http 404" token wins. + assert!(!is_transient_network_error( + "https://site/timeout-page returned HTTP 404 Not Found" + )); + assert!(!is_transient_network_error( + "https://slow-host/ returned HTTP 403 Forbidden" + )); + } + + #[test] + fn benign_messages_are_not_transient() { + assert!(!is_transient_network_error("invalid HTTP header: x-bad")); + assert!(!is_transient_network_error("path traversal blocked")); + assert!(!is_transient_network_error("")); + } +} diff --git a/src/bindings/DownloadStatus.ts b/src/bindings/DownloadStatus.ts index 5456aa2..e8a58aa 100644 --- a/src/bindings/DownloadStatus.ts +++ b/src/bindings/DownloadStatus.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type DownloadStatus = "downloading" | "paused" | "completed" | "failed" | "queued"; +export type DownloadStatus = "downloading" | "paused" | "completed" | "failed" | "queued" | "retrying";