fix(downloads): harden lifecycle cancellation races

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<String, Value
     }
 }
 
+fn sanitize_download_strings(downloads: &mut [String]) -> 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()

src-tauri/src/download.rs

@@ -142,7 +142,9 @@ async fn run_coordinator(
     mut media_rx: mpsc::Receiver<MediaCmd>,
 ) {
     let mut active_media = HashMap::<String, (u64, watch::Sender<bool>)>::new();
-    let mut pending_media_acks = HashMap::<String, (u64, tokio::sync::oneshot::Sender<()>)>::new();
+    let mut cancelled_media_generations = HashMap::<String, u64>::new();
+    let mut pending_media_acks =
+        HashMap::<(String, u64), tokio::sync::oneshot::Sender<()>>::new();
     let mut pending_captured_urls = Vec::<String>::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();
+    }
 }

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;

src-tauri/src/queue.rs

@@ -277,7 +277,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
             == 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<R: tauri::Runtime> QueueManager<R> {
 
     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<R: tauri::Runtime> QueueManager<R> {
         {
             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<R: tauri::Runtime> QueueManager<R> {
     async fn dispatch_one(self: Arc<Self>, 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<R: tauri::Runtime> QueueManager<R> {
             .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<R: tauri::Runtime> QueueManager<R> {
         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()) {