fix(queue): prevent stale resume waiters from hanging

src-tauri/src/lib.rs

@@ -4584,6 +4584,7 @@ async fn resume_download(
                             &id_clone,
                             &queue_id,
                             lifecycle_generation,
+                            control_epoch,
                         )
                         .await
                 };
@@ -4791,6 +4792,7 @@ async fn resume_download(
                             &id_clone,
                             &queue_id_clone,
                             lifecycle_generation,
+                            resume_epoch,
                         )
                         .await
                 };

src-tauri/src/queue.rs

@@ -472,6 +472,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
         };
         if released {
             self.aria2_retry_cancelled.lock().await.remove(id);
+            self.notify.notify_waiters();
         }
     }
 
@@ -619,6 +620,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
         let mut epochs = self.aria2_control_epochs.lock().await;
         let epoch = epochs.get(id).copied().unwrap_or_default().wrapping_add(1);
         epochs.insert(id.to_string(), epoch);
+        self.notify.notify_waiters();
         epoch
     }
 
@@ -817,17 +819,28 @@ impl<R: tauri::Runtime> QueueManager<R> {
         id: &str,
         queue_id: &str,
         lifecycle_generation: u64,
+        control_epoch: u64,
     ) -> Option<OwnedSemaphorePermit> {
         loop {
-            if !self.is_registered_generation(id, lifecycle_generation).await {
+            if !self.is_registered_generation(id, lifecycle_generation).await
+                || self.is_aria2_retry_cancelled(id).await
+                || !self.is_aria2_control_epoch_current(id, control_epoch).await
+            {
                 return None;
             }
             let notified = self.notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
             if let Some(permit) = self
                 .try_reserve_queue_slot(id, queue_id, lifecycle_generation)
                 .await
             {
-                if self.is_registered_generation(id, lifecycle_generation).await {
+                if self.is_registered_generation(id, lifecycle_generation).await
+                    && !self.is_aria2_retry_cancelled(id).await
+                    && self
+                        .is_aria2_control_epoch_current(id, control_epoch)
+                        .await
+                {
                     return Some(permit);
                 }
                 self.release_queue_reservation_for_generation(id, lifecycle_generation)
@@ -1141,6 +1154,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
                 return false;
             }
             let notified = self.notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
             let generation = self
                 .registered_lifecycle_generation(id)
                 .await
@@ -1301,6 +1316,8 @@ impl<R: tauri::Runtime> QueueManager<R> {
     pub async fn run_dispatcher(self: Arc<Self>) {
         loop {
             let notified = self.notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
             if let Some((permit, task)) = self.try_admit_next_task().await {
                 // Admission owns the global and per-queue reservation before
                 // this task is spawned. Keep the dispatcher free to admit
@@ -1623,6 +1640,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
             .await
             .insert(id.to_string());
         self.aria2_retry_cancel_notify.notify_waiters();
+        self.notify.notify_waiters();
     }
 
     pub async fn allow_aria2_retries(&self, id: &str) {

src-tauri/tests/queue_manager.rs

@@ -916,6 +916,75 @@ async fn aria2_resume_waits_for_shrunk_capacity() {
         .expect("resume task should not panic"));
 }
 
+#[tokio::test]
+async fn cancelled_resume_waiter_exits_without_capacity_release() {
+    let (mgr, _spawner) = make_manager(1);
+    let manager = Arc::new(mgr);
+    manager
+        .reserve_enqueue_generation("cancelled-resume", 1)
+        .await
+        .unwrap();
+    let blocker = manager.acquire_permit().await.unwrap();
+    manager.park_permit("blocker", blocker).await;
+
+    let waiter = {
+        let manager = Arc::clone(&manager);
+        tokio::spawn(async move {
+            manager
+                .acquire_aria2_permit_candidate_for_queue("cancelled-resume", "main", 1, 0)
+                .await
+                .is_none()
+        })
+    };
+    tokio::task::yield_now().await;
+    assert!(
+        !waiter.is_finished(),
+        "resume worker should be waiting for capacity"
+    );
+
+    manager.cancel_aria2_retries("cancelled-resume").await;
+    assert!(timeout(Duration::from_secs(1), waiter)
+        .await
+        .expect("cancellation should wake the waiting resume worker")
+        .expect("resume worker should not panic"));
+    manager.release_permit("blocker").await;
+}
+
+#[tokio::test]
+async fn stale_control_epoch_wakes_a_waiting_resume_worker() {
+    let (mgr, _spawner) = make_manager(1);
+    let manager = Arc::new(mgr);
+    manager
+        .reserve_enqueue_generation("stale-resume", 1)
+        .await
+        .unwrap();
+    let control_epoch = manager.next_aria2_control_epoch("stale-resume").await;
+    let blocker = manager.acquire_permit().await.unwrap();
+    manager.park_permit("blocker", blocker).await;
+
+    let waiter = {
+        let manager = Arc::clone(&manager);
+        tokio::spawn(async move {
+            manager
+                .acquire_aria2_permit_candidate_for_queue("stale-resume", "main", 1, control_epoch)
+                .await
+                .is_none()
+        })
+    };
+    tokio::task::yield_now().await;
+    assert!(
+        !waiter.is_finished(),
+        "resume worker should be waiting for capacity"
+    );
+
+    manager.next_aria2_control_epoch("stale-resume").await;
+    assert!(timeout(Duration::from_secs(1), waiter)
+        .await
+        .expect("a newer control epoch should wake the stale worker")
+        .expect("resume worker should not panic"));
+    manager.release_permit("blocker").await;
+}
+
 #[tokio::test]
 async fn dispatcher_skips_a_full_front_queue_for_later_eligible_work() {
     let (manager, _spawner) = make_manager(2);
@@ -1146,7 +1215,7 @@ async fn stale_queue_resume_reservation_is_removed_when_activation_fails() {
         .unwrap();
     assert!(previous.is_none());
     let candidate = manager
-        .acquire_aria2_permit_candidate_for_queue("candidate", "queue-a", 1)
+        .acquire_aria2_permit_candidate_for_queue("candidate", "queue-a", 1, 0)
         .await
         .expect("candidate should reserve the queue slot");