diff --git a/src-tauri/src/download.rs b/src-tauri/src/download.rs index bc08f6d..dee9300 100644 --- a/src-tauri/src/download.rs +++ b/src-tauri/src/download.rs @@ -469,10 +469,10 @@ async fn download_file( // 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 max_attempts = payload.max_tries.max(1) as usize; + // `max_tries` is the user-facing retry count. Attempts include the first + // try plus those configured retries. + let max_retries = payload.max_tries as usize; + let max_attempts = max_retries + 1; 'url: for url in &payload.urls { let mut strike = 0_usize; let mut attempts = 0_usize; @@ -513,7 +513,7 @@ async fn download_file( } let transient = crate::retry::is_transient_network_error(&error); - let strikes_left = strike < crate::retry::MAX_RETRIES; + let strikes_left = strike < max_retries; if transient && strikes_left { // Transient: announce `Retrying`, back off, then retry. diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 2fa38d5..01a0b47 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2472,9 +2472,7 @@ pub(crate) async fn start_media_download_internal( let config_path = config_file.into_temp_path(); use crate::ipc::DownloadStateEvent; - use crate::retry::{ - backoff_and_emit_cancel, is_transient_network_error, BackoffOutcome, MAX_RETRIES, - }; + use crate::retry::{backoff_and_emit_cancel, is_transient_network_error, BackoffOutcome}; const STDERR_TAIL: usize = 2048; @@ -2504,10 +2502,11 @@ pub(crate) async fn start_media_download_internal( // user-installed tools, or platform-specific executable aliases. let trusted_path = crate::platform::trusted_system_path()?; + let max_retries = max_tries.unwrap_or(0).max(0) as usize; let mut strike = 0_usize; let mut processing_started = false; - while strike <= MAX_RETRIES { + while strike <= max_retries { let ytdlp_path = resolve_bundled_binary_path(&app_handle, "yt-dlp")?; let mut cmd = app_handle .shell() @@ -2520,7 +2519,7 @@ pub(crate) async fn start_media_download_internal( .arg("--socket-timeout") .arg("20") .arg("--retries") - .arg("3") + .arg(max_retries.to_string()) .arg("--extractor-retries") .arg("3") .arg("--ffmpeg-location") @@ -2569,10 +2568,6 @@ pub(crate) async fn start_media_download_internal( } } - if let Some(tries) = max_tries { - cmd = cmd.arg("--retries").arg(tries.to_string()); - } - if let Some(loc) = config_location.as_ref() { cmd = cmd.arg("--config-location").arg(loc); } @@ -2781,7 +2776,7 @@ pub(crate) async fn start_media_download_internal( }; let transient = is_transient_network_error(&failure_reason); - let strikes_left = strike < MAX_RETRIES; + let strikes_left = strike < max_retries; if !(transient && strikes_left) { return Err(failure_reason); } diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 5182b61..11541e5 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -1,5 +1,5 @@ use crate::ipc::{DownloadStateEvent, DownloadStatus, QueueDirection}; -use crate::retry::{backoff_and_emit, is_transient_network_error, BackoffOutcome, MAX_RETRIES}; +use crate::retry::{backoff_and_emit, is_transient_network_error, BackoffOutcome}; use log; use serde::Deserialize; use serde_json; @@ -194,6 +194,33 @@ impl QueueManager { self.semaphore.clone().acquire_owned().await.ok() } + async fn acquire_permit_after_retirement(&self) -> Option { + loop { + let permit = self.acquire_permit().await?; + if self.retire_slot_if_needed() { + permit.forget(); + continue; + } + return Some(permit); + } + } + + fn retire_slot_if_needed(&self) -> bool { + let mut debt = self.slots_to_retire.load(Ordering::Relaxed); + while debt > 0 { + match self.slots_to_retire.compare_exchange_weak( + debt, + debt - 1, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => return true, + Err(actual) => debt = actual, + } + } + false + } + /// Park an already-acquired permit under `id`. pub async fn park_permit(&self, id: &str, permit: OwnedSemaphorePermit) { self.active_permits @@ -214,7 +241,7 @@ impl QueueManager { return false; } - let permit = match self.acquire_permit().await { + let permit = match self.acquire_permit_after_retirement().await { Some(p) => p, None => return false, }; @@ -325,34 +352,10 @@ impl QueueManager { continue; } // (2) Acquire a slot. - let permit_opt = self.semaphore.clone().acquire_owned().await.ok(); - let permit = match permit_opt { + let permit = match self.acquire_permit_after_retirement().await { Some(p) => p, None => break, // Semaphore closed, exit dispatcher }; - // (3) CAS retirement — never underflows to usize::MAX. - let mut retired = false; - let mut debt = self.slots_to_retire.load(Ordering::Relaxed); - while debt > 0 { - match self.slots_to_retire.compare_exchange_weak( - debt, - debt - 1, - Ordering::Relaxed, - Ordering::Relaxed, - ) { - Ok(_) => { - retired = true; - break; - } - Err(actual) => { - debt = actual; - } - } - } - if retired { - drop(permit); - continue; - } // (4) Re-pop under lock — guards against racing removals between // waking from Notify and acquiring the permit. let task = match self.pending.lock().await.pop_front() { @@ -612,20 +615,6 @@ impl QueueManager { 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_retryable_aria2_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)) @@ -634,6 +623,20 @@ impl QueueManager { } let mut payload = payload.unwrap(); + let strike = { + let mut strikes = self.aria2_retry_strikes.lock().await; + let entry = strikes.entry(id.clone()).or_insert(0); + *entry + }; + + let transient = is_retryable_aria2_error(&error); + let strikes_left = strike < automatic_retry_limit(payload.max_tries); + if !(transient && strikes_left) { + self.apply_completion(&id, PendingOutcome::Error(error)) + .await; + return; + } + if is_aria2_range_mode_error(&error) { log::warn!( "aria2 range mode [{}]: server rejected bounded chunk ranges; restarting with a single connection", @@ -850,6 +853,14 @@ impl QueueManager { } } +fn automatic_retry_limit(max_tries: Option) -> usize { + max_tries.unwrap_or(0).max(0) as usize +} + +fn aria2_attempt_limit(max_tries: Option) -> u32 { + (automatic_retry_limit(max_tries) + 1) as u32 +} + fn is_retryable_aria2_error(error: &str) -> bool { is_transient_network_error(error) || is_aria2_range_mode_error(error) } @@ -1080,7 +1091,7 @@ impl SidecarSpawner for ProductionSpawner { "max-connection-per-server".to_string(), serde_json::json!(conn.to_string()), ); - let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let mt = aria2_attempt_limit(payload.max_tries); 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")); @@ -1149,7 +1160,7 @@ impl SidecarSpawner for ProductionSpawner { // aria2 unavailable — fall back to native coordinator. log::warn!("aria2 addUri failed, falling back to native: {}", e); let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; - let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let mt = automatic_retry_limit(payload.max_tries) as u32; let safe_filename = crate::download_ownership::canonical_download_filename(&payload.filename); state @@ -1241,7 +1252,7 @@ impl SidecarSpawner for ProductionSpawner { async fn run_native(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { let state = self.app_handle.state::(); let download_id = uuid::Uuid::parse_str(id).map_err(|e| e.to_string())?; - let mt = payload.max_tries.unwrap_or(1).max(1) as u32; + let mt = automatic_retry_limit(payload.max_tries) as u32; let resolved_dest = crate::resolve_path(&payload.destination, &self.app_handle); let safe_filename = crate::download_ownership::canonical_download_filename(&payload.filename); diff --git a/src-tauri/tests/download_engine.rs b/src-tauri/tests/download_engine.rs index 06ec819..c5d2121 100644 --- a/src-tauri/tests/download_engine.rs +++ b/src-tauri/tests/download_engine.rs @@ -427,7 +427,7 @@ async fn reports_terminal_http_errors_after_retry_budget() { }; assert!(error.contains("500 Internal Server Error")); - assert_eq!(server.state.requests.load(Ordering::SeqCst), 2); + assert_eq!(server.state.requests.load(Ordering::SeqCst), 3); assert!(!output_path.exists()); } diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index e23d6ab..d0fd46f 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -231,22 +231,62 @@ async fn shrink_converges_to_target_without_killing_active() { // Shrink to 2 while 4 are "active" (permits parked). mgr_arc.set_capacity(2); - // Release active permits. Debt is 2; two releases retire both. + // Release active permits. Debt is 2; two releases retire both, but the + // remaining active count still equals the shrunken target. mgr_arc.release_permit("t0").await; mgr_arc.release_permit("t1").await; tokio::time::sleep(Duration::from_millis(100)).await; - // Debt was 2; two releases retired both. The 2 remaining pending tasks - // (t4, t5) can now dispatch since debt is 0 and slots freed. + assert_eq!( + spawner.native_calls.load(Ordering::SeqCst), + 4, + "pending tasks must not dispatch while active count already meets the shrunken target" + ); + + mgr_arc.release_permit("t2").await; + mgr_arc.release_permit("t3").await; + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( spawner.native_calls.load(Ordering::SeqCst), 6, - "shrink converges: after debt exhausted, remaining pending dispatch" + "pending tasks dispatch after active count falls below the shrunken target" ); handle.abort(); } +#[tokio::test] +async fn aria2_resume_waits_for_shrunk_capacity() { + let (mgr, _spawner) = make_manager(3); + let mgr_arc = Arc::new(mgr); + + let active = mgr_arc.acquire_permit().await.unwrap(); + mgr_arc.park_permit("active", active).await; + let paused = mgr_arc.acquire_permit().await.unwrap(); + mgr_arc.park_permit("paused", paused).await; + mgr_arc.release_permit("paused").await; + + mgr_arc.set_capacity(1); + + let waiter = { + let mgr_clone = Arc::clone(&mgr_arc); + tokio::spawn(async move { mgr_clone.ensure_aria2_permit("paused").await }) + }; + + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + !waiter.is_finished(), + "paused resume must wait while current active count already meets shrunk limit" + ); + + mgr_arc.release_permit("active").await; + assert!(timeout(Duration::from_secs(1), waiter) + .await + .expect("resume permit should unblock after active transfer exits") + .expect("resume task should not panic")); +} + fn aria2_task(id: &str) -> QueuedTask { QueuedTask { id: id.to_string(),