diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 24256cf..a735663 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3240,6 +3240,7 @@ async fn pause_download( ) -> Result<(), String> { log::info!("pause_download called for id: {}", id); + let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; let active_kind = state.queue_manager.active_kind(&id).await; let removed_pending = state.queue_manager.remove_from_pending(&id).await; @@ -3306,7 +3307,9 @@ async fn pause_download( } if matches!(active_kind, Some(crate::queue::TaskKind::Aria2)) { + state.queue_manager.next_aria2_control_epoch(&id).await; state.queue_manager.cancel_aria2_retries(&id).await; + state.queue_manager.clear_aria2_retry_state(&id).await; } let (tx, rx) = tokio::sync::oneshot::channel(); @@ -3341,11 +3344,16 @@ async fn resume_download( state: tauri::State<'_, AppState>, id: String, ) -> Result { + let control_guard = state.queue_manager.acquire_aria2_control(&id).await; let Some(gid) = state.queue_manager.aria2_gid_for_download(&id) else { log::info!( "aria2 resume [{}]: no mapped gid; re-enqueue is permitted", id ); + state.queue_manager.next_aria2_control_epoch(&id).await; + state.queue_manager.cancel_aria2_retries(&id).await; + state.queue_manager.clear_aria2_retry_state(&id).await; + state.queue_manager.release_permit(&id).await; state.queue_manager.release_registered_id(&id).await; return Ok(false); }; @@ -3372,11 +3380,13 @@ async fn resume_download( let gid_clone = gid.clone(); let app_handle_clone = app_handle.clone(); + drop(control_guard); tauri::async_runtime::spawn(async move { let acquired = queue_manager.ensure_aria2_permit(&id_clone).await; if !acquired { return; } + let _control_guard = queue_manager.acquire_aria2_control(&id_clone).await; if queue_manager.is_aria2_retry_cancelled(&id_clone).await || !queue_manager .is_aria2_control_epoch_current(&id_clone, control_epoch) @@ -3448,18 +3458,42 @@ async fn resume_download( } log::info!("aria2 resume [{}]: unpaused gid {}", id_clone, gid_clone); }); - return Ok(true); + Ok(true) } "active" | "waiting" => { + let resume_epoch = state.queue_manager.current_aria2_control_epoch(&id).await; + drop(control_guard); state.queue_manager.ensure_aria2_permit(&id).await; - log::info!( - "aria2 resume [{}]: gid {} already {}; no duplicate job created", - id, - gid, - status - ); + let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; + let still_current = state.queue_manager.is_registered(&id).await + && !state.queue_manager.is_aria2_retry_cancelled(&id).await + && state + .queue_manager + .is_aria2_control_epoch_current(&id, resume_epoch) + .await + && state.queue_manager.aria2_gid_for_download(&id).as_deref() == Some(gid.as_str()); + if still_current { + log::info!( + "aria2 resume [{}]: gid {} already {}; no duplicate job created", + id, + gid, + status + ); + use tauri::Emitter; + let _ = app_handle.emit( + "download-state", + crate::ipc::DownloadStateEvent::new( + id, + crate::ipc::DownloadStatus::Downloading, + ), + ); + } + Ok(true) } "complete" | "error" | "removed" => { + drop(control_guard); + state.queue_manager.next_aria2_control_epoch(&id).await; + state.queue_manager.cancel_aria2_retries(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; state.queue_manager.forget_aria2_gid(&id).await; state.queue_manager.release_permit(&id).await; @@ -3470,19 +3504,13 @@ async fn resume_download( gid, status ); - return Ok(false); + Ok(false) } other => { - return Err(format!("aria2 gid {gid} returned unknown status {other}")); + drop(control_guard); + Err(format!("aria2 gid {gid} returned unknown status {other}")) } } - - use tauri::Emitter; - let _ = app_handle.emit( - "download-state", - crate::ipc::DownloadStateEvent::new(id, crate::ipc::DownloadStatus::Downloading), - ); - Ok(true) } #[tauri::command] @@ -3496,6 +3524,7 @@ async fn remove_download( log::info!("remove_download called for id: {}", id); let preserve_resumable = preserve_resumable.unwrap_or(false); let primary_path = crate::download_ownership::primary_path_for_id(&app_handle, &id)?; + let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; let active_kind = state.queue_manager.active_kind(&id).await; state.queue_manager.remove_from_pending(&id).await; @@ -3647,6 +3676,7 @@ async fn detach_download_for_reconfigure( id: String, ) -> Result<(), String> { log::info!("detach_download_for_reconfigure called for id: {}", id); + let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; let active_kind = state.queue_manager.active_kind(&id).await; state.queue_manager.remove_from_pending(&id).await; state.queue_manager.next_aria2_control_epoch(&id).await; @@ -6161,7 +6191,12 @@ pub fn run() { let mut seen_ids = std::collections::HashSet::new(); for status_info in active_arr { let gid = status_info.get("gid").and_then(|s| s.as_str()).unwrap_or(""); - let id = poll_mgr.aria2_gids.read().unwrap().get(gid).cloned(); + let id = poll_mgr + .aria2_gids + .read() + .unwrap() + .get(gid) + .map(|mapping| mapping.id.clone()); if let Some(id) = id { seen_ids.insert(id.clone()); let status = status_info.get("status").and_then(|value| value.as_str()).unwrap_or(""); diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 08a87e6..f33502d 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -4,16 +4,51 @@ use log; use serde::Deserialize; use serde_json; use std::collections::{HashMap, HashSet, VecDeque}; +use std::future::Future; +use std::pin::Pin; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use tauri::{AppHandle, Manager}; -use tokio::sync::{Mutex, Notify, OwnedSemaphorePermit, Semaphore}; +use tokio::sync::{Mutex, Notify, OwnedMutexGuard, OwnedSemaphorePermit, Semaphore}; use ts_rs::TS; /// Default capacity when no setting is read yet. pub const DEFAULT_MAX_CONCURRENT: usize = 3; pub const MEDIA_RUN_CANCELLED: &str = "__firelink_media_run_cancelled__"; +type Aria2ControlLocks = Arc>>>>; + +#[derive(Debug, Clone)] +pub struct Aria2GidMapping { + pub id: String, + pub epoch: u64, +} + +/// Owns one per-download control lock and removes its idle map entry when the +/// last operation for that download finishes. +pub struct Aria2ControlGuard { + locks: Aria2ControlLocks, + id: String, + lock: Arc>, + guard: Option>, +} + +impl Drop for Aria2ControlGuard { + fn drop(&mut self) { + // Release the async mutex before inspecting Arc ownership. The map + // entry and this guard are then the only strong references when no + // waiter is pending, so the entry can be removed safely. + self.guard.take(); + let mut locks = self.locks.lock().unwrap_or_else(|error| error.into_inner()); + let should_remove = locks.get(&self.id).is_some_and(|candidate| { + Arc::ptr_eq(candidate, &self.lock) && Arc::strong_count(&self.lock) == 2 + }); + if should_remove { + locks.remove(&self.id); + } + } +} + /// Outcome of an aria2 completion that arrived before its gid was stored. /// Carries the outcome so the correct state emit survives the race. #[derive(Debug, Clone)] @@ -83,11 +118,6 @@ pub trait SidecarSpawner: Send + Sync + 'static { Err("aria2 connection refresh is unavailable".to_string()) } - /// Leave an aria2 transfer paused after a refresh races with a user pause. - async fn pause_uri(&self, _gid: &str) -> Result<(), String> { - Err("aria2 pause is unavailable".to_string()) - } - /// Run a media download to completion. The permit is parked for the full /// duration; release is handled by QueueManager on the runner's exit. async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String>; @@ -107,7 +137,7 @@ pub struct QueueManager { notify: Notify, /// aria2 gid -> download id map (shared with the WS poller). - pub aria2_gids: Arc>>, + pub aria2_gids: Arc>>, /// gid -> buffered (id_placeholder, outcome) for completions that arrived /// before the gid was stored. Drained by `remember_gid`. @@ -121,9 +151,27 @@ pub struct QueueManager { /// Download ids whose aria2 retry loop must not create another job. aria2_retry_cancelled: Mutex>, + /// Download ids with a retry worker currently sleeping or re-adding a gid. + /// A duplicate aria2 error event must not create a second worker. + aria2_retry_inflight: Mutex>, + /// The gid whose terminal event initiated each in-flight retry. + aria2_retrying_gids: Mutex>, + /// Gids whose terminal events must be ignored after a lifecycle transition. + /// This is bounded so a long-lived daemon cannot grow the set indefinitely. + aria2_ignored_gids: Mutex>, /// Wakes retry backoff workers when a pause/remove action cancels them. aria2_retry_cancel_notify: Notify, + /// Serializes control RPCs for one download (pause, resume, refresh, and + /// retry handoff) without blocking control operations for other downloads. + aria2_control_locks: Aria2ControlLocks, + + /// Serializes GID mapping transitions with early WebSocket event + /// buffering. The RwLock protects individual map access; this lock makes + /// map replacement, ignored-GID retirement, and pending-event draining a + /// single state transition. + aria2_gid_state: Mutex<()>, + /// Monotonic per-download aria2 control generation. Long-running queued /// resume tasks capture this and abort when a later pause/remove wins. aria2_control_epochs: Mutex>, @@ -163,7 +211,12 @@ impl QueueManager { aria2_payloads: Mutex::new(HashMap::new()), aria2_retry_strikes: Mutex::new(HashMap::new()), aria2_retry_cancelled: Mutex::new(HashSet::new()), + aria2_retry_inflight: Mutex::new(HashMap::new()), + aria2_retrying_gids: Mutex::new(HashSet::new()), + aria2_ignored_gids: Mutex::new(VecDeque::new()), aria2_retry_cancel_notify: Notify::new(), + aria2_control_locks: Arc::new(StdMutex::new(HashMap::new())), + aria2_gid_state: Mutex::new(()), aria2_control_epochs: Mutex::new(HashMap::new()), spawner, app_handle, @@ -184,6 +237,10 @@ impl QueueManager { /// Explicitly release a backend registry id (e.g. on un-resumable false paths, removals, or detach). pub async fn release_registered_id(&self, id: &str) { self.registered_ids.lock().await.remove(id); + // A released lifecycle cannot be resumed by a delayed retry worker. + // Epoch checks remain the authoritative guard; removing this marker + // prevents terminal downloads from accumulating cancellation entries. + self.aria2_retry_cancelled.lock().await.remove(id); } pub async fn is_registered(&self, id: &str) -> bool { @@ -322,6 +379,29 @@ impl QueueManager { self.aria2_retry_cancelled.lock().await.contains(id) } + /// Serialize control RPCs for one download while allowing unrelated + /// downloads to pause, resume, or refresh concurrently. + pub async fn acquire_aria2_control(&self, id: &str) -> Aria2ControlGuard { + let lock = { + let mut locks = self + .aria2_control_locks + .lock() + .unwrap_or_else(|error| error.into_inner()); + Arc::clone( + locks + .entry(id.to_string()) + .or_insert_with(|| Arc::new(Mutex::new(()))), + ) + }; + let guard = lock.clone().lock_owned().await; + Aria2ControlGuard { + locks: Arc::clone(&self.aria2_control_locks), + id: id.to_string(), + lock, + guard: Some(guard), + } + } + pub async fn has_aria2_retry_state(&self, id: &str) -> bool { self.aria2_retry_strikes.lock().await.contains_key(id) } @@ -430,11 +510,14 @@ impl QueueManager { }; for id in ids_to_fail { - self.apply_completion( - &id, - PendingOutcome::Error("Aria2 WebSocket connection lost".to_string()), - ) - .await; + let _control_guard = self.acquire_aria2_control(&id).await; + if matches!(self.active_kind(&id).await, Some(TaskKind::Aria2)) { + self.apply_completion_locked( + &id, + PendingOutcome::Error("Aria2 WebSocket connection lost".to_string()), + ) + .await; + } } } @@ -532,6 +615,11 @@ impl QueueManager { match task.kind { TaskKind::Aria2 => { + // Every backend aria2 dispatch starts a new control lifecycle. + // This invalidates retry workers left behind by a previous + // failed or cancelled lifecycle before retry cancellation is + // made reusable for the new task. + let lifecycle_epoch = self.next_aria2_control_epoch(&id).await; self.aria2_retry_cancelled.lock().await.remove(&id); self.aria2_payloads .lock() @@ -540,8 +628,14 @@ impl QueueManager { self.aria2_retry_strikes.lock().await.remove(&id); match self.spawner.add_uri(&id, &task.payload).await { Ok(gid) => { + let control_guard = self.acquire_aria2_control(&id).await; let cancelled = self.aria2_retry_cancelled.lock().await.contains(&id); - if cancelled || !self.is_registered(&id).await { + let current_lifecycle = self + .is_aria2_control_epoch_current(&id, lifecycle_epoch) + .await + && self.is_registered(&id).await; + if cancelled || !current_lifecycle { + drop(control_guard); log::info!( "aria2 dispatch cancellation [{}]: removing late gid {}", id, @@ -555,16 +649,38 @@ impl QueueManager { error ); } - self.clear_aria2_retry_state(&id).await; - self.release_permit(&id).await; + self.ignore_aria2_gid(&gid).await; + if current_lifecycle { + self.clear_aria2_retry_state(&id).await; + self.release_permit(&id).await; + } return; } - self.remember_gid(id.clone(), gid).await; + let buffered_outcome = self.remember_gid(id.clone(), gid.clone()).await; + drop(control_guard); + if let Some(outcome) = buffered_outcome { + self.handle_aria2_event(&gid, outcome).await; + } } Err(error) => { - self.clear_aria2_retry_state(&id).await; - self.emit_failed(&id, error); - self.release_permit(&id).await; + let _control_guard = self.acquire_aria2_control(&id).await; + let current_lifecycle = self + .is_aria2_control_epoch_current(&id, lifecycle_epoch) + .await + && self.is_registered(&id).await; + if current_lifecycle { + self.next_aria2_control_epoch(&id).await; + self.cancel_aria2_retries(&id).await; + self.clear_aria2_retry_state(&id).await; + self.release_permit(&id).await; + self.release_registered_id(&id).await; + self.emit_failed(&id, error); + } else { + log::info!( + "aria2 dispatch [{}]: ignoring stale addUri failure after a newer lifecycle took ownership", + id + ); + } } } } @@ -606,39 +722,73 @@ impl QueueManager { .emit("download-state", DownloadStateEvent::failed(id, error)); } - /// Store gid -> id, then reconcile any buffered completion for that gid. - pub async fn remember_gid(&self, id: String, gid: String) { - { - let mut gids = self.aria2_gids.write().unwrap(); - gids.retain(|existing_gid, existing_id| { - let keep = existing_id != &id || existing_gid == &gid; - if !keep { - log::warn!( - "aria2 gid transition [{}]: dropping stale mapping {} before storing {}", - id, - existing_gid, - gid - ); - } - keep - }); - gids.insert(gid.clone(), id.clone()); - } + /// Store gid -> id and return any buffered terminal event for the caller + /// to reconcile against the correct event path. In particular, buffered + /// errors must still pass through transient retry classification. + pub async fn remember_gid(&self, id: String, gid: String) -> Option { + let epoch = self.current_aria2_control_epoch(&id).await; + let buffered_outcome = { + let _gid_state = self.aria2_gid_state.lock().await; + let mut replaced_gids = Vec::new(); + { + let mut gids = self.aria2_gids.write().unwrap(); + gids.retain(|existing_gid, existing_id| { + let keep = existing_id.id != id.as_str() || existing_gid == &gid; + if !keep { + replaced_gids.push(existing_gid.clone()); + log::warn!( + "aria2 gid transition [{}]: dropping stale mapping {} before storing {}", + id, + existing_gid, + gid + ); + } + keep + }); + gids.insert( + gid.clone(), + Aria2GidMapping { + id: id.clone(), + epoch, + }, + ); + } + + self.unignore_aria2_gid_locked(&gid).await; + for replaced_gid in &replaced_gids { + self.ignore_aria2_gid_locked(replaced_gid).await; + } + let mut buffered = self.pending_completion.lock().await; + for replaced_gid in &replaced_gids { + buffered.remove(replaced_gid); + } + buffered.remove(&gid).map(|(_buf_id, outcome)| outcome) + }; log::info!("aria2 gid transition [{}]: mapped {}", id, gid); - let buffered = self.pending_completion.lock().await.remove(&gid); - if let Some((_buf_id, outcome)) = buffered { - self.apply_completion(&id, outcome).await; - } + buffered_outcome } /// Apply an aria2 completion outcome: release permit + emit state. pub async fn apply_completion(&self, id: &str, outcome: PendingOutcome) { + let _control_guard = self.acquire_aria2_control(id).await; + self.apply_completion_locked(id, outcome).await; + } + + /// Apply a completion while the caller owns the download control lock. + /// Keeping the epoch transition and terminal cleanup under that lock + /// prevents an old WebSocket event from completing a newer lifecycle. + async fn apply_completion_locked(&self, id: &str, outcome: PendingOutcome) { + // A terminal event invalidates every delayed retry or control worker + // from the previous lifecycle before releasing its permit. + self.next_aria2_control_epoch(id).await; + self.cancel_aria2_retries(id).await; match outcome { PendingOutcome::Complete => { self.clear_aria2_retry_state(id).await; self.forget_aria2_gid(id).await; - self.emit_state(id, DownloadStatus::Completed); self.release_registered_id(id).await; + self.release_permit(id).await; + self.emit_state(id, DownloadStatus::Completed); } PendingOutcome::Error(error) => { if error.to_ascii_lowercase().contains("checksum") { @@ -656,11 +806,11 @@ impl QueueManager { self.clear_aria2_retry_state(id).await; self.forget_aria2_gid(id).await; - self.emit_failed(id, error); self.release_registered_id(id).await; + self.release_permit(id).await; + self.emit_failed(id, error); } } - self.release_permit(id).await; } pub async fn clear_aria2_retry_state(&self, id: &str) { @@ -680,12 +830,55 @@ impl QueueManager { self.aria2_retry_cancelled.lock().await.remove(id); } + async fn finish_aria2_retry(&self, id: &str, gid: &str, retry_epoch: u64) { + self.release_aria2_retry_inflight(id, retry_epoch).await; + self.aria2_retrying_gids.lock().await.remove(gid); + } + + async fn release_aria2_retry_inflight(&self, id: &str, retry_epoch: u64) { + let mut inflight = self.aria2_retry_inflight.lock().await; + if inflight.get(id).copied() == Some(retry_epoch) { + inflight.remove(id); + } + } + + async fn ignore_aria2_gid(&self, gid: &str) { + let _gid_state = self.aria2_gid_state.lock().await; + self.ignore_aria2_gid_locked(gid).await; + } + + async fn ignore_aria2_gid_locked(&self, gid: &str) { + const MAX_IGNORED_GIDS: usize = 1024; + let mut ignored = self.aria2_ignored_gids.lock().await; + if !ignored.iter().any(|known| known == gid) { + ignored.push_back(gid.to_string()); + } + while ignored.len() > MAX_IGNORED_GIDS { + ignored.pop_front(); + } + } + + async fn unignore_aria2_gid_locked(&self, gid: &str) { + self.aria2_ignored_gids + .lock() + .await + .retain(|known| known != gid); + } + + async fn is_aria2_gid_ignored_locked(&self, gid: &str) -> bool { + self.aria2_ignored_gids + .lock() + .await + .iter() + .any(|known| known == gid) + } + pub fn aria2_gid_for_download(&self, id: &str) -> Option { self.aria2_gids .read() .unwrap() .iter() - .find_map(|(gid, download_id)| (download_id == id).then(|| gid.clone())) + .find_map(|(gid, mapping)| (mapping.id == id).then(|| gid.clone())) } pub fn aria2_gid_mappings(&self) -> Vec<(String, String)> { @@ -693,7 +886,7 @@ impl QueueManager { .read() .unwrap() .iter() - .map(|(gid, id)| (gid.clone(), id.clone())) + .map(|(gid, mapping)| (gid.clone(), mapping.id.clone())) .collect() } @@ -701,6 +894,7 @@ impl QueueManager { /// persistent connection-pool collapse or a true zero-progress stall. /// The transfer keeps its gid, partial file, and queue permit. pub async fn refresh_aria2_connections(&self, id: &str, gid: &str) -> Result<(), String> { + let _control_guard = self.acquire_aria2_control(id).await; if self.aria2_gid_for_download(id).as_deref() != Some(gid) || !self.is_registered(id).await || self.is_aria2_retry_cancelled(id).await @@ -716,7 +910,11 @@ impl QueueManager { && self.is_aria2_control_epoch_current(id, epoch).await && self.aria2_gid_for_download(id).as_deref() == Some(gid); if !still_current { - let _ = self.spawner.pause_uri(gid).await; + log::info!( + "aria2 connection refresh [{}]: control state changed while refreshing gid {}; leaving the newer action in charge", + id, + gid + ); } Ok(()) } @@ -724,11 +922,12 @@ impl QueueManager { /// Remove every gid mapping for a download and discard buffered terminal /// events for those gids. Returns the most recently encountered gid. pub async fn forget_aria2_gid(&self, id: &str) -> Option { + let _gid_state = self.aria2_gid_state.lock().await; let removed = { let mut gids = self.aria2_gids.write().unwrap(); let removed: Vec = gids .iter() - .filter(|(_, download_id)| *download_id == id) + .filter(|(_, mapping)| mapping.id == id) .map(|(gid, _)| gid.clone()) .collect(); for gid in &removed { @@ -741,6 +940,10 @@ impl QueueManager { return None; } + for gid in &removed { + self.ignore_aria2_gid_locked(gid).await; + } + let mut buffered = self.pending_completion.lock().await; for gid in &removed { buffered.remove(gid); @@ -752,21 +955,66 @@ impl QueueManager { /// 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 = { + fn handle_aria2_download_error( + self: &Arc, + gid: String, + error: String, + ) -> Pin + Send + 'static>> { + let this = Arc::clone(self); + Box::pin(async move { + this.handle_aria2_download_error_inner(&gid, error).await; + }) + } + + /// Resolve a WebSocket event against the GID map, or buffer it while the + /// map transition is still in flight. The state lock closes the window in + /// which an event could be inserted after remember_gid drained it. + async fn map_or_buffer_aria2_event( + &self, + gid: &str, + outcome: PendingOutcome, + ) -> Option<(Aria2GidMapping, PendingOutcome)> { + let _gid_state = self.aria2_gid_state.lock().await; + if self.is_aria2_gid_ignored_locked(gid).await { + return None; + } + let mapping = { let gids = self.aria2_gids.read().unwrap(); gids.get(gid).cloned() }; - let id = match id { - Some(id) => id, - None => { - self.pending_completion.lock().await.insert( - gid.to_string(), - (String::new(), PendingOutcome::Error(error)), - ); - return; - } + if let Some(mapping) = mapping { + return Some((mapping, outcome)); + } + self.pending_completion + .lock() + .await + .insert(gid.to_string(), (String::new(), outcome)); + None + } + + async fn handle_aria2_download_error_inner(self: &Arc, gid: &str, error: String) { + let Some((mapping, PendingOutcome::Error(error))) = self + .map_or_buffer_aria2_event(gid, PendingOutcome::Error(error)) + .await + else { + return; }; + + let _control_guard = self.acquire_aria2_control(&mapping.id).await; + let current_mapping = { + let gids = self.aria2_gids.read().unwrap(); + gids.get(gid).cloned() + }; + if current_mapping + .as_ref() + .is_none_or(|current| current.id != mapping.id || current.epoch != mapping.epoch) + || !self + .is_aria2_control_epoch_current(&mapping.id, mapping.epoch) + .await + { + return; + } + let id = mapping.id; if self.aria2_retry_cancelled.lock().await.contains(&id) { log::info!( "aria2 retry cancellation [{}]: ignoring error for gid {} during removal", @@ -776,9 +1024,26 @@ impl QueueManager { return; } + if self.aria2_retrying_gids.lock().await.contains(gid) { + log::debug!( + "aria2 retry [{}]: ignoring duplicate error event for retrying gid {}", + id, + gid + ); + return; + } + + if self.aria2_retry_inflight.lock().await.contains_key(&id) { + log::debug!( + "aria2 retry [{}]: ignoring duplicate error event while retry handoff is in flight", + id + ); + return; + } + let payload = self.aria2_payloads.lock().await.get(&id).cloned(); if payload.is_none() { - self.apply_completion(&id, PendingOutcome::Error(error)) + self.apply_completion_locked(&id, PendingOutcome::Error(error)) .await; return; } @@ -793,11 +1058,26 @@ impl QueueManager { 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)) + self.apply_completion_locked(&id, PendingOutcome::Error(error)) .await; return; } + self.aria2_retrying_gids + .lock() + .await + .insert(gid.to_string()); + let retry_epoch = self.current_aria2_control_epoch(&id).await; + let already_inflight = { + let mut inflight = self.aria2_retry_inflight.lock().await; + inflight.insert(id.clone(), retry_epoch).is_some() + }; + if already_inflight { + self.aria2_retrying_gids.lock().await.remove(gid); + return; + } + let retry_gid = gid.to_string(); + if is_aria2_range_mode_error(&error) { log::warn!( "aria2 range mode [{}]: server rejected bounded chunk ranges; restarting with a single connection", @@ -835,44 +1115,46 @@ impl QueueManager { notified.await; } }; - let outcome = backoff_and_emit( - strike, - error_for_emit, - retry_cancel, - |reason| { - use tauri::Emitter; - let _ = this.app_handle.emit( - "download-state", - DownloadStateEvent::retrying(&id_for_task, reason), - ); - }, - ) + let outcome = backoff_and_emit(strike, error_for_emit, retry_cancel, |reason| { + use tauri::Emitter; + let _ = this.app_handle.emit( + "download-state", + DownloadStateEvent::retrying(&id_for_task, reason), + ); + }) .await; if outcome == BackoffOutcome::Aborted { + this.finish_aria2_retry(&id_for_task, &retry_gid, retry_epoch) + .await; return; } - if !this.active_permits.lock().await.contains_key(&id_for_task) { - return; - } - if this - .aria2_retry_cancelled - .lock() - .await - .contains(&id_for_task) + if !this.active_permits.lock().await.contains_key(&id_for_task) + || this.is_aria2_retry_cancelled(&id_for_task).await + || !this + .is_aria2_control_epoch_current(&id_for_task, retry_epoch) + .await + || !this.is_registered(&id_for_task).await + || this.aria2_gid_for_download(&id_for_task).as_deref() != Some(retry_gid.as_str()) { + this.finish_aria2_retry(&id_for_task, &retry_gid, retry_epoch) + .await; return; } match this.spawner.add_uri(&id_for_task, &payload).await { Ok(new_gid) => { - if this - .aria2_retry_cancelled - .lock() - .await - .contains(&id_for_task) - { + let control_guard = this.acquire_aria2_control(&id_for_task).await; + let stale = this.is_aria2_retry_cancelled(&id_for_task).await + || !this + .is_aria2_control_epoch_current(&id_for_task, retry_epoch) + .await + || !this.is_registered(&id_for_task).await + || this.aria2_gid_for_download(&id_for_task).as_deref() + != Some(retry_gid.as_str()); + if stale { + drop(control_guard); if let Err(error) = this.spawner.remove_uri(&new_gid).await { log::error!( "aria2 retry cancellation [{}]: failed to remove late gid {}: {}", @@ -882,19 +1164,13 @@ impl QueueManager { ); } else { log::info!( - "aria2 retry cancellation [{}]: removed late gid {}", + "aria2 retry cancellation [{}]: removed stale gid {}", id_for_task, new_gid ); - return; } - let retained_gid = new_gid.clone(); - this.remember_gid(id_for_task.clone(), new_gid).await; - log::warn!( - "aria2 retry cancellation [{}]: retained late gid {} mapping for remove retry", - id_for_task, - retained_gid - ); + this.finish_aria2_retry(&id_for_task, &retry_gid, retry_epoch) + .await; return; } this.aria2_retry_strikes @@ -902,10 +1178,36 @@ impl QueueManager { .await .insert(id_for_task.clone(), strike + 1); this.emit_state(&id_for_task, DownloadStatus::Downloading); - this.remember_gid(id_for_task.clone(), new_gid).await; + // Stop suppressing events for the id before exposing the + // new gid. The old gid remains marked as retrying until + // remember_gid atomically replaces its mapping, so a + // duplicate old event is still ignored while a genuine + // new-gid error is allowed through. + this.release_aria2_retry_inflight(&id_for_task, retry_epoch) + .await; + let new_gid_for_event = new_gid.clone(); + let buffered_outcome = this.remember_gid(id_for_task.clone(), new_gid).await; + this.aria2_retrying_gids.lock().await.remove(&retry_gid); + drop(control_guard); + if let Some(outcome) = buffered_outcome { + this.handle_aria2_event(&new_gid_for_event, outcome).await; + } } Err(retry_error) => { - this.apply_completion(&id_for_task, PendingOutcome::Error(retry_error)) + let control_guard = this.acquire_aria2_control(&id_for_task).await; + let stale = this.is_aria2_retry_cancelled(&id_for_task).await + || !this + .is_aria2_control_epoch_current(&id_for_task, retry_epoch) + .await; + if !stale { + this.apply_completion_locked( + &id_for_task, + PendingOutcome::Error(retry_error), + ) + .await; + } + drop(control_guard); + this.finish_aria2_retry(&id_for_task, &retry_gid, retry_epoch) .await; } } @@ -915,28 +1217,33 @@ impl QueueManager { /// 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)); - } - } - } + if let PendingOutcome::Error(error) = outcome { + self.handle_aria2_download_error(gid.to_string(), error) + .await; + return; } + let Some((mapping, outcome)) = self.map_or_buffer_aria2_event(gid, outcome).await else { + return; + }; + + let _control_guard = self.acquire_aria2_control(&mapping.id).await; + if self.aria2_retrying_gids.lock().await.contains(gid) { + return; + } + let current_mapping = { + let gids = self.aria2_gids.read().unwrap(); + gids.get(gid).cloned() + }; + if current_mapping + .as_ref() + .is_none_or(|current| current.id != mapping.id || current.epoch != mapping.epoch) + || !self + .is_aria2_control_epoch_current(&mapping.id, mapping.epoch) + .await + { + return; + } + self.apply_completion_locked(&mapping.id, outcome).await; } /// Reorder a pending task up or down. Returns the new pending order. @@ -1400,40 +1707,17 @@ impl SidecarSpawner for ProductionSpawner { let state = self.app_handle.state::(); let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); let secret = &state.aria2_secret; - let paused = crate::rpc_call( - port, - secret, - "aria2.forcePause", - serde_json::json!([gid]), - ) - .await - .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; + let paused = crate::rpc_call(port, secret, "aria2.forcePause", serde_json::json!([gid])) + .await + .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; crate::ensure_aria2_gid_result("forcePause", gid, &paused)?; - let resumed = crate::rpc_call( - port, - secret, - "aria2.unpause", - serde_json::json!([gid]), - ) - .await - .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; + let resumed = crate::rpc_call(port, secret, "aria2.unpause", serde_json::json!([gid])) + .await + .map_err(|error| format!("failed to refresh aria2 gid {gid}: {error}"))?; crate::ensure_aria2_gid_result("unpause", gid, &resumed) } - async fn pause_uri(&self, gid: &str) -> Result<(), String> { - let state = self.app_handle.state::(); - let result = crate::rpc_call( - state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), - &state.aria2_secret, - "aria2.forcePause", - serde_json::json!([gid]), - ) - .await - .map_err(|error| format!("failed to pause aria2 gid {gid}: {error}"))?; - crate::ensure_aria2_gid_result("forcePause", gid, &result) - } - async fn run_media(&self, id: &str, payload: &SpawnPayload) -> Result<(), String> { let state = self.app_handle.state::(); let mut cancel_rx = state diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index b5e5057..406d020 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/src-tauri/tests/queue_manager.rs @@ -16,13 +16,29 @@ struct CountingSpawner { struct DelayedAria2Spawner { gid_tx: tokio::sync::Mutex>>, + add_uri_calls: AtomicUsize, remove_uri_calls: AtomicUsize, } +struct FailFirstAria2Spawner { + add_uri_calls: AtomicUsize, + fail_first: std::sync::atomic::AtomicBool, +} + +impl FailFirstAria2Spawner { + fn new() -> Self { + Self { + add_uri_calls: AtomicUsize::new(0), + fail_first: std::sync::atomic::AtomicBool::new(true), + } + } +} + impl DelayedAria2Spawner { fn new(gid_tx: tokio::sync::oneshot::Sender<()>) -> Self { Self { gid_tx: tokio::sync::Mutex::new(Some(gid_tx)), + add_uri_calls: AtomicUsize::new(0), remove_uri_calls: AtomicUsize::new(0), } } @@ -31,10 +47,14 @@ impl DelayedAria2Spawner { #[async_trait::async_trait] impl SidecarSpawner for DelayedAria2Spawner { async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { - let tx = self.gid_tx.lock().await.take().expect("gid release sender"); - let _ = tx.send(()); - tokio::time::sleep(Duration::from_millis(50)).await; - Ok("late-gid".to_string()) + let call = self.add_uri_calls.fetch_add(1, Ordering::SeqCst) + 1; + if let Some(tx) = self.gid_tx.lock().await.take() { + let _ = tx.send(()); + tokio::time::sleep(Duration::from_millis(50)).await; + Ok("late-gid".to_string()) + } else { + Ok(format!("gid-{call}")) + } } async fn remove_uri(&self, gid: &str) -> Result<(), String> { @@ -48,6 +68,29 @@ impl SidecarSpawner for DelayedAria2Spawner { } } +#[async_trait::async_trait] +impl SidecarSpawner for FailFirstAria2Spawner { + async fn add_uri(&self, _id: &str, _payload: &SpawnPayload) -> Result { + let call = self.add_uri_calls.fetch_add(1, Ordering::SeqCst) + 1; + if self + .fail_first + .swap(false, std::sync::atomic::Ordering::SeqCst) + { + Err("initial aria2 RPC failure".to_string()) + } else { + Ok(format!("gid-{call}")) + } + } + + async fn remove_uri(&self, _gid: &str) -> Result<(), String> { + Ok(()) + } + + async fn run_media(&self, _id: &str, _payload: &SpawnPayload) -> Result<(), String> { + unreachable!("media is not used by fail-first aria2 tests") + } +} + impl CountingSpawner { fn new() -> Self { Self { @@ -195,6 +238,41 @@ async fn aria2_control_epoch_invalidates_stale_resume_workers() { assert!(mgr.is_aria2_control_epoch_current("a", pause).await); } +#[tokio::test] +async fn stale_terminal_event_cannot_complete_a_newer_control_epoch() { + use firelink_lib::queue::PendingOutcome; + + let (mgr, _spawner) = make_manager(1); + let manager = Arc::new(mgr); + manager.push(aria2_task("stale-event")).await.unwrap(); + let permit = manager.acquire_permit().await.expect("permit"); + manager.park_permit("stale-event", permit).await; + + let old_epoch = manager.next_aria2_control_epoch("stale-event").await; + manager + .remember_gid("stale-event".to_string(), "gid-old".to_string()) + .await; + manager.next_aria2_control_epoch("stale-event").await; + + manager + .handle_aria2_event("gid-old", PendingOutcome::Complete) + .await; + + assert!( + !manager + .is_aria2_control_epoch_current("stale-event", old_epoch) + .await + ); + assert_eq!( + manager.available_permits(), + 0, + "a terminal event from an older epoch must not release the newer lifecycle permit" + ); + manager.forget_aria2_gid("stale-event").await; + manager.release_permit("stale-event").await; + manager.release_registered_id("stale-event").await; +} + #[tokio::test] async fn forgetting_aria2_gid_clears_mapping_without_releasing_twice() { let (mgr, _spawner) = make_manager(1); @@ -565,6 +643,296 @@ async fn transient_aria2_error_reissues_after_backoff() { dispatcher.abort(); } +#[tokio::test] +async fn duplicate_transient_events_schedule_only_one_retry_worker() { + use firelink_lib::queue::PendingOutcome; + + let (mgr, spawner) = make_manager(1); + let manager = Arc::new(mgr); + let mut task = aria2_task("duplicate-retry"); + task.payload.max_tries = Some(1); + manager.push(task).await.unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + tokio::time::sleep(Duration::from_millis(100)).await; + + let error = PendingOutcome::Error( + "aria2 error code 1: Failed to receive data, cause: protocol error".to_string(), + ); + manager.handle_aria2_event("gid-1", error.clone()).await; + manager.handle_aria2_event("gid-1", error).await; + + timeout(Duration::from_secs(4), async { + loop { + if spawner.add_uri_calls.load(Ordering::SeqCst) >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("one retry should be issued"); + assert_eq!( + spawner.add_uri_calls.load(Ordering::SeqCst), + 2, + "duplicate terminal events must not create duplicate aria2 jobs" + ); + + manager + .handle_aria2_event( + "gid-2", + PendingOutcome::Error("HTTP 404 Not Found".to_string()), + ) + .await; + dispatcher.abort(); +} + +#[tokio::test] +async fn stale_retry_worker_cannot_reenter_after_new_control_epoch() { + use firelink_lib::queue::PendingOutcome; + + let (mgr, spawner) = make_manager(1); + let manager = Arc::new(mgr); + let mut task = aria2_task("stale-retry"); + task.payload.max_tries = Some(1); + manager.push(task).await.unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + tokio::time::sleep(Duration::from_millis(100)).await; + manager + .handle_aria2_event( + "gid-1", + PendingOutcome::Error( + "aria2 error code 1: Failed to receive data, cause: protocol error".to_string(), + ), + ) + .await; + + // Simulate a newer pause/resume lifecycle while the old worker is in its + // cancel-safe backoff. Clearing the reusable cancellation flag must not + // revive the worker because its control epoch is stale. + manager.next_aria2_control_epoch("stale-retry").await; + manager.allow_aria2_retries("stale-retry").await; + tokio::time::sleep(Duration::from_secs(3)).await; + + assert_eq!( + spawner.add_uri_calls.load(Ordering::SeqCst), + 1, + "a retry worker from an older lifecycle must not add a new gid" + ); + manager.release_permit("stale-retry").await; + dispatcher.abort(); +} + +#[tokio::test] +async fn completion_event_for_retrying_gid_cannot_release_new_lifecycle_permit() { + use firelink_lib::queue::PendingOutcome; + + let (mgr, spawner) = make_manager(1); + let manager = Arc::new(mgr); + let mut task = aria2_task("retry-complete-race"); + task.payload.max_tries = Some(1); + manager.push(task).await.unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + tokio::time::sleep(Duration::from_millis(100)).await; + manager + .handle_aria2_event( + "gid-1", + PendingOutcome::Error( + "aria2 error code 1: Failed to receive data, cause: protocol error".to_string(), + ), + ) + .await; + manager + .handle_aria2_event("gid-1", PendingOutcome::Complete) + .await; + assert_eq!( + manager.available_permits(), + 0, + "a duplicate completion for the retrying gid must not free the permit" + ); + + timeout(Duration::from_secs(4), async { + loop { + if spawner.add_uri_calls.load(Ordering::SeqCst) >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("retry should create the next gid"); + manager + .handle_aria2_event("gid-2", PendingOutcome::Complete) + .await; + assert_eq!(manager.available_permits(), 1); + dispatcher.abort(); +} + +#[tokio::test] +async fn initial_aria2_add_failure_releases_registry_for_restart() { + let app = mock_builder() + .build(mock_context(noop_assets())) + .expect("mock app"); + let spawner = Arc::new(FailFirstAria2Spawner::new()); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + spawner.clone(), + )); + manager + .push_with_generation(aria2_task("initial-failure"), 1) + .await + .unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + timeout(Duration::from_secs(1), async { + loop { + if !manager.is_registered("initial-failure").await { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("failed initial addUri must release the registry id"); + + manager + .push_with_generation(aria2_task("initial-failure"), 2) + .await + .expect("the same download must be restartable after initial add failure"); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 2); + manager.release_permit("initial-failure").await; + dispatcher.abort(); +} + +#[tokio::test] +async fn late_initial_gid_cannot_attach_to_a_newer_lifecycle() { + let app = mock_builder() + .build(mock_context(noop_assets())) + .expect("mock app"); + let (gid_started_tx, gid_started_rx) = tokio::sync::oneshot::channel(); + let spawner = Arc::new(DelayedAria2Spawner::new(gid_started_tx)); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + spawner.clone(), + )); + manager + .push_with_generation(aria2_task("dispatch-race"), 1) + .await + .unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + gid_started_rx.await.expect("first addUri should start"); + + // Model pause while the first addUri is still resolving, followed by a + // new enqueue for the same frontend download id. + manager.next_aria2_control_epoch("dispatch-race").await; + manager.cancel_aria2_retries("dispatch-race").await; + manager.clear_aria2_retry_state("dispatch-race").await; + manager.release_permit("dispatch-race").await; + manager.release_registered_id("dispatch-race").await; + manager + .push_with_generation(aria2_task("dispatch-race"), 2) + .await + .unwrap(); + + timeout(Duration::from_secs(1), async { + loop { + if manager.aria2_gid_for_download("dispatch-race").as_deref() == Some("gid-2") { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("new lifecycle should own the mapped gid"); + tokio::time::sleep(Duration::from_millis(100)).await; + + assert_eq!(spawner.add_uri_calls.load(Ordering::SeqCst), 2); + assert_eq!( + spawner.remove_uri_calls.load(Ordering::SeqCst), + 1, + "the late gid from the old lifecycle must be removed" + ); + assert_eq!( + manager.aria2_gid_for_download("dispatch-race").as_deref(), + Some("gid-2") + ); + manager.release_permit("dispatch-race").await; + dispatcher.abort(); +} + +#[tokio::test] +async fn transient_error_buffered_before_gid_mapping_still_retries() { + use firelink_lib::queue::PendingOutcome; + + let app = mock_builder() + .build(mock_context(noop_assets())) + .expect("mock app"); + let (gid_started_tx, gid_started_rx) = tokio::sync::oneshot::channel(); + let spawner = Arc::new(DelayedAria2Spawner::new(gid_started_tx)); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + spawner.clone(), + )); + let mut task = aria2_task("early-error"); + task.payload.max_tries = Some(1); + manager.push(task).await.unwrap(); + + let dispatcher = { + let manager = Arc::clone(&manager); + tokio::spawn(async move { manager.run_dispatcher().await }) + }; + gid_started_rx.await.expect("initial addUri should start"); + manager + .handle_aria2_event( + "late-gid", + PendingOutcome::Error( + "aria2 error code 1: Failed to receive data, cause: protocol error".to_string(), + ), + ) + .await; + + timeout(Duration::from_secs(4), async { + loop { + if spawner.add_uri_calls.load(Ordering::SeqCst) >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("the buffered transient error must enter the retry loop"); + + manager + .handle_aria2_event( + "gid-2", + PendingOutcome::Error("HTTP 404 Not Found".to_string()), + ) + .await; + dispatcher.abort(); +} + #[tokio::test] async fn gid_completion_before_store_buffers_and_reconciles() { use firelink_lib::queue::PendingOutcome;