diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 6e128ce..62456e7 100644 --- a/src-tauri/src/lib.rs +++ b/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 }; diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 9858f82..480821c 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -472,6 +472,7 @@ impl QueueManager { }; if released { self.aria2_retry_cancelled.lock().await.remove(id); + self.notify.notify_waiters(); } } @@ -619,6 +620,7 @@ impl QueueManager { 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 QueueManager { id: &str, queue_id: &str, lifecycle_generation: u64, + control_epoch: u64, ) -> Option { 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 QueueManager { 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 QueueManager { pub async fn run_dispatcher(self: Arc) { 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 QueueManager { .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) { diff --git a/src-tauri/tests/queue_manager.rs b/src-tauri/tests/queue_manager.rs index a03ebd9..4c0570a 100644 --- a/src-tauri/tests/queue_manager.rs +++ b/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");