diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index aaa3b9d..6be0bf1 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -238,6 +238,12 @@ fn import_legacy_data( } } } + if portable { + // Sanitize before importing as well as sanitizing the legacy + // source afterward. A crash between those two operations must + // not leave raw transfer credentials in the portable database. + sanitize_download_strings(&mut legacy.downloads)?; + } merge_legacy_data(connection, legacy)?; if migration_complete { connection @@ -779,6 +785,19 @@ fn mark_portable_download_unresumable(object: &mut serde_json::Map Result<(), String> { + for data in downloads { + let mut value: Value = serde_json::from_str(data).map_err(|error| { + format!("failed to decode download for portable sanitization: {error}") + })?; + remove_persisted_transfer_secrets(&mut value); + *data = serde_json::to_string(&value).map_err(|error| { + format!("failed to encode download for portable sanitization: {error}") + })?; + } + Ok(()) +} + fn sanitize_persisted_downloads(connection: &mut Connection) -> Result<(), String> { let transaction = connection .transaction() diff --git a/src-tauri/src/download.rs b/src-tauri/src/download.rs index 683dc4f..12e58f3 100644 --- a/src-tauri/src/download.rs +++ b/src-tauri/src/download.rs @@ -142,7 +142,9 @@ async fn run_coordinator( mut media_rx: mpsc::Receiver, ) { let mut active_media = HashMap::)>::new(); - let mut pending_media_acks = HashMap::)>::new(); + let mut cancelled_media_generations = HashMap::::new(); + let mut pending_media_acks = + HashMap::<(String, u64), tokio::sync::oneshot::Sender<()>>::new(); let mut pending_captured_urls = Vec::::new(); let mut frontend_ready = false; @@ -180,24 +182,53 @@ async fn run_coordinator( }; match command { MediaCmd::Register { id, lifecycle_generation, cancel_tx } => { - if let Some((_, previous)) = active_media.insert(id, (lifecycle_generation, cancel_tx)) { + if active_media + .get(&id) + .is_some_and(|(generation, _)| *generation > lifecycle_generation) + { + let _ = cancel_tx.send(true); + continue; + } + let pending_cancel = cancelled_media_generations.get(&id).copied(); + if pending_cancel.is_some_and(|generation| generation < lifecycle_generation) { + cancelled_media_generations.remove(&id); + } + let cancelled = pending_cancel.is_some_and(|generation| generation >= lifecycle_generation); + if let Some((_, previous)) = active_media.insert(id.clone(), (lifecycle_generation, cancel_tx)) { let _ = previous.send(true); } + if cancelled { + if let Some((_, cancel_tx)) = active_media.get(&id) { + let _ = cancel_tx.send(true); + } + if pending_cancel == Some(lifecycle_generation) { + cancelled_media_generations.remove(&id); + } + } } MediaCmd::Pause { id, lifecycle_generation } => { if active_media.get(&id).is_some_and(|(generation, _)| *generation == lifecycle_generation) { if let Some((_, cancel_tx)) = active_media.remove(&id) { let _ = cancel_tx.send(true); } + } else { + cancelled_media_generations + .entry(id) + .and_modify(|generation| *generation = (*generation).max(lifecycle_generation)) + .or_insert(lifecycle_generation); } } MediaCmd::PauseWithAck { id, lifecycle_generation, ack } => { if active_media.get(&id).is_some_and(|(generation, _)| *generation == lifecycle_generation) { if let Some((_, cancel_tx)) = active_media.remove(&id) { let _ = cancel_tx.send(true); - pending_media_acks.insert(id, (lifecycle_generation, ack)); + pending_media_acks.insert((id, lifecycle_generation), ack); } } else { + cancelled_media_generations + .entry(id) + .and_modify(|generation| *generation = (*generation).max(lifecycle_generation)) + .or_insert(lifecycle_generation); let _ = ack.send(()); } } @@ -205,10 +236,16 @@ async fn run_coordinator( if active_media.get(&id).is_some_and(|(generation, _)| *generation == lifecycle_generation) { active_media.remove(&id); } - if pending_media_acks.get(&id).is_some_and(|(generation, _)| *generation == lifecycle_generation) { - if let Some((_, ack)) = pending_media_acks.remove(&id) { - let _ = ack.send(()); - } + if let Some(ack) = + pending_media_acks.remove(&(id.clone(), lifecycle_generation)) + { + let _ = ack.send(()); + } + if cancelled_media_generations + .get(&id) + .is_some_and(|generation| *generation <= lifecycle_generation) + { + cancelled_media_generations.remove(&id); } } } @@ -317,4 +354,97 @@ mod tests { .unwrap(); assert!(*new_cancel.borrow_and_update()); } + + #[tokio::test] + async fn pause_before_media_registration_cancels_the_late_lifecycle() { + let (coordinator, _events) = DownloadCoordinator::spawn_headless(); + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + + coordinator + .pause_media_with_ack("late-media".to_string(), 7, ack_tx) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(1), ack_rx) + .await + .unwrap() + .unwrap(); + + let mut cancel_rx = coordinator + .register_media("late-media".to_string(), 7) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(1), async { + while !*cancel_rx.borrow_and_update() { + cancel_rx.changed().await.unwrap(); + } + }) + .await + .expect("late media registration was not cancelled"); + } + + #[tokio::test] + async fn stale_media_registration_cannot_replace_a_newer_lifecycle() { + let (coordinator, _events) = DownloadCoordinator::spawn_headless(); + let mut new_cancel = coordinator + .register_media("same-id".to_string(), 2) + .await + .unwrap(); + let mut old_cancel = coordinator + .register_media("same-id".to_string(), 1) + .await + .unwrap(); + + tokio::time::timeout(Duration::from_secs(1), old_cancel.changed()) + .await + .unwrap() + .unwrap(); + assert!(*old_cancel.borrow_and_update()); + + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + coordinator + .pause_media_with_ack("same-id".to_string(), 2, ack_tx) + .await + .unwrap(); + coordinator.finish_media("same-id".to_string(), 2).await; + tokio::time::timeout(Duration::from_secs(1), ack_rx) + .await + .unwrap() + .unwrap(); + assert!(*new_cancel.borrow_and_update()); + } + + #[tokio::test] + async fn media_pause_ack_is_preserved_across_lifecycle_replacement() { + let (coordinator, _events) = DownloadCoordinator::spawn_headless(); + let _old_cancel = coordinator + .register_media("same-id".to_string(), 1) + .await + .unwrap(); + let (old_ack_tx, old_ack_rx) = tokio::sync::oneshot::channel(); + coordinator + .pause_media_with_ack("same-id".to_string(), 1, old_ack_tx) + .await + .unwrap(); + + let _new_cancel = coordinator + .register_media("same-id".to_string(), 2) + .await + .unwrap(); + let (new_ack_tx, new_ack_rx) = tokio::sync::oneshot::channel(); + coordinator + .pause_media_with_ack("same-id".to_string(), 2, new_ack_tx) + .await + .unwrap(); + + coordinator.finish_media("same-id".to_string(), 1).await; + tokio::time::timeout(Duration::from_secs(1), old_ack_rx) + .await + .unwrap() + .unwrap(); + coordinator.finish_media("same-id".to_string(), 2).await; + tokio::time::timeout(Duration::from_secs(1), new_ack_rx) + .await + .unwrap() + .unwrap(); + } } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 2f24580..89c6f4d 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3357,11 +3357,11 @@ async fn pause_download( let _control_guard = state.queue_manager.acquire_aria2_control(&id).await; let active_kind = state.queue_manager.active_kind(&id).await; - let media_lifecycle_generation = state + let registered_lifecycle_generation = state .queue_manager .registered_lifecycle_generation(&id) - .await - .unwrap_or_default(); + .await; + let media_lifecycle_generation = registered_lifecycle_generation.unwrap_or_default(); let removed_pending = state.queue_manager.remove_from_pending(&id).await; let gid = state.queue_manager.aria2_gid_for_download(&id); @@ -3379,17 +3379,68 @@ async fn pause_download( log::info!("aria2 pause [{}]: gid {} was already paused", id, gid); } "active" | "waiting" => { - state.queue_manager.next_aria2_control_epoch(&id).await; state.queue_manager.cancel_aria2_retries(&id).await; - let result = rpc_call( + let pause_result = match 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}"))?; - ensure_aria2_gid_result("forcePause", gid, &result)?; + { + Ok(result) => ensure_aria2_gid_result("forcePause", gid, &result), + Err(error) => Err(format!("failed to pause aria2 gid {gid}: {error}")), + }; + if let Err(pause_error) = pause_result { + match aria2_download_status( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + gid, + ) + .await + { + Ok(status) if status == "paused" => { + log::info!( + "aria2 pause [{}]: forcePause failed but gid {} is paused", + id, + gid + ); + } + Ok(status) if status == "complete" => { + state + .queue_manager + .apply_completion_locked(&id, crate::queue::PendingOutcome::Complete) + .await; + return Ok(()); + } + Ok(status) if matches!(status.as_str(), "error" | "removed") => { + let terminal_error = format!( + "cannot pause aria2 gid {gid}: {pause_error}; daemon reports terminal state {status}" + ); + state + .queue_manager + .apply_completion_locked( + &id, + crate::queue::PendingOutcome::Error(terminal_error.clone()), + ) + .await; + return Err(terminal_error); + } + Ok(status) => { + state.queue_manager.allow_aria2_retries(&id).await; + return Err(format!( + "{pause_error}; aria2 gid {gid} is still {status}" + )); + } + Err(status_error) => { + state.queue_manager.allow_aria2_retries(&id).await; + return Err(format!( + "{pause_error}; failed to verify aria2 gid {gid}: {status_error}" + )); + } + } + } + state.queue_manager.next_aria2_control_epoch(&id).await; log::info!("aria2 pause [{}]: gid {} paused", id, gid); } "complete" => { @@ -3442,6 +3493,13 @@ async fn pause_download( return Ok(()); } + // A terminal runner may have already released its lifecycle while this + // command was waiting for the per-download control lock. Treat pause as + // an idempotent no-op instead of emitting a stale Paused state. + if active_kind.is_none() && !removed_pending && registered_lifecycle_generation.is_none() { + return Ok(()); + } + 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; @@ -3460,11 +3518,19 @@ async fn pause_download( rx.await .map_err(|_| "download worker stopped without acknowledging pause".to_string())?; - if !matches!(active_kind, Some(crate::queue::TaskKind::Media)) { - state.queue_manager.release_permit(&id).await; - } - if removed_pending || matches!(active_kind, Some(crate::queue::TaskKind::Aria2)) { - state.queue_manager.release_registered_id(&id).await; + state.queue_manager.release_permit(&id).await; + if registered_lifecycle_generation.is_some() + || removed_pending + || matches!(active_kind, Some(crate::queue::TaskKind::Aria2)) + { + if let Some(generation) = registered_lifecycle_generation { + state + .queue_manager + .release_registered_id_for_generation(&id, generation) + .await; + } else { + state.queue_manager.release_registered_id(&id).await; + } } use tauri::Emitter; let _ = app_handle.emit( @@ -3726,11 +3792,12 @@ async fn remove_download( .unwrap_or_default(); state.queue_manager.remove_from_pending(&id).await; - state.queue_manager.next_aria2_control_epoch(&id).await; - state.queue_manager.cancel_aria2_retries(&id).await; - let gid = state.queue_manager.aria2_gid_for_download(&id); if let Some(gid) = gid.as_deref() { + // Keep the current epoch and mapping alive until daemon removal is + // confirmed. If removal fails, terminal events from this lifecycle + // must still be accepted so the permit and mapping can be cleaned up. + state.queue_manager.cancel_aria2_retries(&id).await; let removal_result = async { force_remove_aria2_gid( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), @@ -3750,11 +3817,17 @@ async fn remove_download( state.queue_manager.allow_aria2_retries(&id).await; return Err(error); } + state.queue_manager.next_aria2_control_epoch(&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; log::info!("aria2 remove [{}]: gid {} stopped and forgotten", id, gid); } else { + // There may be an addUri or retry handoff in flight with no mapped + // GID yet. Invalidate that pending lifecycle before acknowledging the + // removal so its late result cannot resurrect the download. + state.queue_manager.next_aria2_control_epoch(&id).await; + state.queue_manager.cancel_aria2_retries(&id).await; let (tx, rx) = tokio::sync::oneshot::channel(); if matches!(active_kind, Some(crate::queue::TaskKind::Media)) { state @@ -3766,9 +3839,7 @@ async fn remove_download( } let _ = rx.await; - if !matches!(active_kind, Some(crate::queue::TaskKind::Media)) { - state.queue_manager.release_permit(&id).await; - } + state.queue_manager.release_permit(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; state.queue_manager.forget_aria2_gid(&id).await; } @@ -3881,12 +3952,13 @@ async fn detach_download_for_reconfigure( .await .unwrap_or_default(); state.queue_manager.remove_from_pending(&id).await; - state.queue_manager.next_aria2_control_epoch(&id).await; - state.queue_manager.cancel_aria2_retries(&id).await; - let gid = state.queue_manager.aria2_gid_for_download(&id); if let Some(gid) = gid.as_deref() { + // Do not invalidate the mapped lifecycle until the daemon confirms + // the detach. A failed pause must leave terminal-event reconciliation + // and permit ownership intact. + state.queue_manager.cancel_aria2_retries(&id).await; let removal_result = async { let pause_res = rpc_call( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), @@ -3914,12 +3986,16 @@ async fn detach_download_for_reconfigure( state.queue_manager.allow_aria2_retries(&id).await; return Err(error); } + state.queue_manager.next_aria2_control_epoch(&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; state.queue_manager.release_registered_id(&id).await; log::info!("aria2 detach [{}]: gid {} stopped and forgotten", id, gid); } else { + // Invalidate a queued/addUri lifecycle that has not published a GID. + state.queue_manager.next_aria2_control_epoch(&id).await; + state.queue_manager.cancel_aria2_retries(&id).await; let (tx, rx) = tokio::sync::oneshot::channel(); if matches!(active_kind, Some(crate::queue::TaskKind::Media)) { state @@ -3931,9 +4007,7 @@ async fn detach_download_for_reconfigure( } let _ = rx.await; // Wait for the writer to stop - if !matches!(active_kind, Some(crate::queue::TaskKind::Media)) { - state.queue_manager.release_permit(&id).await; - } + state.queue_manager.release_permit(&id).await; state.queue_manager.clear_aria2_retry_state(&id).await; state.queue_manager.forget_aria2_gid(&id).await; state.queue_manager.release_registered_id(&id).await; diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index faf1718..39d8cef 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -277,7 +277,7 @@ impl QueueManager { == Some(generation) } - async fn release_registered_id_for_generation(&self, id: &str, generation: u64) { + pub(crate) async fn release_registered_id_for_generation(&self, id: &str, generation: u64) { let released = { let mut registered = self.registered_ids.lock().await; let mut generations = self.registered_lifecycle_generations.lock().await; @@ -362,7 +362,7 @@ impl QueueManager { pub async fn commit_reserved_enqueue( &self, - task: QueuedTask, + mut task: QueuedTask, generation: u64, ) -> Result<(), String> { let id = task.id.clone(); @@ -373,6 +373,7 @@ impl QueueManager { { return Err("Download enqueue was superseded by a newer user action".to_string()); } + task.lifecycle_generation = generation; self.pending.lock().await.push_back(task); self.emit_state(id, DownloadStatus::Queued); self.notify.notify_one(); @@ -685,6 +686,20 @@ impl QueueManager { async fn dispatch_one(self: Arc, permit: OwnedSemaphorePermit, task: QueuedTask) { let id = task.id.clone(); let lifecycle_generation = task.lifecycle_generation; + // Serialize activation with pause/remove/detach. The dispatcher can + // pop a task just as a control command arrives; registering the + // permit and active kind under the same guard prevents that command + // from observing a half-started lifecycle. + let control_guard = self.acquire_aria2_control(&id).await; + if !self + .is_registered_generation(&id, lifecycle_generation) + .await + { + drop(control_guard); + drop(permit); + return; + } + // Park the permit BEFORE spawning. Uniform parking: // aria2's RPC returns instantly, so the permit must outlive the // dispatch_one call. Media runners release on exit. @@ -694,21 +709,30 @@ impl QueueManager { .lock() .await .insert(id.clone(), task.kind.clone()); + + let aria2_lifecycle_epoch = if matches!(&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() + .await + .insert(id.clone(), task.payload.clone()); + self.aria2_retry_strikes.lock().await.remove(&id); + Some(lifecycle_epoch) + } else { + None + }; self.emit_state(&id, DownloadStatus::Downloading); + drop(control_guard); 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() - .await - .insert(id.clone(), task.payload.clone()); - self.aria2_retry_strikes.lock().await.remove(&id); + let lifecycle_epoch = aria2_lifecycle_epoch + .expect("aria2 dispatch must initialize a control epoch"); match self.spawner.add_uri(&id, &task.payload).await { Ok(gid) => { let control_guard = self.acquire_aria2_control(&id).await; @@ -793,6 +817,10 @@ impl QueueManager { lifecycle_generation: u64, outcome: Result<(), String>, ) { + // Completion and pause/remove/detach must observe one ordered + // lifecycle. Without this guard a terminal media outcome could emit + // Completed, then a racing pause could emit Paused afterward. + let _control_guard = self.acquire_aria2_control(id).await; if !self.is_registered_generation(id, lifecycle_generation).await { log::info!( "media runner [{}]: ignoring stale terminal outcome for lifecycle {}", @@ -1900,25 +1928,29 @@ impl SidecarSpawner for ProductionSpawner { .download_coordinator .register_media(id.to_string(), lifecycle_generation) .await?; - let outcome = crate::start_media_download_internal( - self.app_handle.clone(), - id, - payload.url.clone(), - payload.destination.clone(), - payload.filename.clone(), - payload.format_selector.clone(), - payload.cookie_source.clone(), - payload.speed_limit.clone(), - payload.username.clone(), - payload.password.clone(), - payload.headers.clone(), - payload.cookies.clone(), - payload.proxy.clone(), - payload.user_agent.clone(), - payload.max_tries, - &mut cancel_rx, - ) - .await; + let outcome = if *cancel_rx.borrow() { + Err(crate::queue::MEDIA_RUN_CANCELLED.to_string()) + } else { + crate::start_media_download_internal( + self.app_handle.clone(), + id, + payload.url.clone(), + payload.destination.clone(), + payload.filename.clone(), + payload.format_selector.clone(), + payload.cookie_source.clone(), + payload.speed_limit.clone(), + payload.username.clone(), + payload.password.clone(), + payload.headers.clone(), + payload.cookies.clone(), + payload.proxy.clone(), + payload.user_agent.clone(), + payload.max_tries, + &mut cancel_rx, + ) + .await + }; if let Ok(path) = outcome.as_ref() { let _ = crate::download_ownership::set_primary_path(&self.app_handle, id, path); if let Some(file_name) = path.file_name().and_then(|name| name.to_str()) {