From a5c22526590cf676dacd4bd10964a499f12050c5 Mon Sep 17 00:00:00 2001 From: Rohith Pariki Date: Sun, 13 Sep 2026 01:18:00 +0530 Subject: [PATCH 1/2] feat: cancel superseded native-queue downloads (fixes #491) --- src/infra/queue/dispatch.rs | 105 ++++++++++++++++++++++++++++++++---- src/infra/queue/mod.rs | 15 ++++++ 2 files changed, 111 insertions(+), 9 deletions(-) diff --git a/src/infra/queue/dispatch.rs b/src/infra/queue/dispatch.rs index 7b2c979c..6213577e 100644 --- a/src/infra/queue/dispatch.rs +++ b/src/infra/queue/dispatch.rs @@ -377,15 +377,30 @@ async fn play_queued_subsonic(app: &Arc>, track: &TrackInfo, uri: &st let fetch_id = publish_pending_decoded(app, &player, track).await; // Fetch off the IoEvent pump: awaiting the download here would freeze every // other event (skips included, for every source) for its whole duration. - let app = Arc::clone(app); + let app_clone = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let result = crate::infra::subsonic::dispatch::download_for_queue(&source, &uri) .await .map(|tmp| (tmp, None)); - finish_decoded_fetch(&app, fetch_id, result, &name).await; + finish_decoded_fetch(&app_clone, fetch_id, result, &name).await; }); + + let abort_handle = handle.abort_handle(); + { + let mut guard = app.lock().await; + let mut injected = false; + if let Some(crate::infra::queue::QueueNowPlaying::Decoded(ref mut d)) = guard.queue_now { + if d.fetch_id == fetch_id { + d.abort_handle = Some(crate::infra::queue::DownloadAbortHandle(abort_handle.clone())); + injected = true; + } + } + if !injected { + abort_handle.abort(); + } + } true } @@ -401,15 +416,30 @@ async fn play_queued_qobuz(app: &Arc>, track: &TrackInfo, uri: &str) let fetch_id = publish_pending_decoded(app, &player, track).await; let quality = app.lock().await.user_config.behavior.qobuz_quality; // Fetch off the IoEvent pump, like Subsonic: a Qobuz track is a long download. - let app = Arc::clone(app); + let app_clone = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let result = crate::infra::qobuz::dispatch::download_for_queue(&source, &uri, quality) .await .map(|(tmp, label)| (tmp, Some(label))); - finish_decoded_fetch(&app, fetch_id, result, &name).await; + finish_decoded_fetch(&app_clone, fetch_id, result, &name).await; }); + + let abort_handle = handle.abort_handle(); + { + let mut guard = app.lock().await; + let mut injected = false; + if let Some(crate::infra::queue::QueueNowPlaying::Decoded(ref mut d)) = guard.queue_now { + if d.fetch_id == fetch_id { + d.abort_handle = Some(crate::infra::queue::DownloadAbortHandle(abort_handle.clone())); + injected = true; + } + } + if !injected { + abort_handle.abort(); + } + } true } @@ -427,15 +457,30 @@ async fn play_queued_youtube(app: &Arc>, track: &TrackInfo, uri: &str let source = crate::infra::youtube::dispatch::build_source(app).await; // Fetch off the IoEvent pump: awaiting yt-dlp here would freeze every other // event (skips included, for every source) for its whole duration. - let app = Arc::clone(app); + let app_clone = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let result = crate::infra::youtube::dispatch::download_for_queue(&source, &uri) .await .map(|tmp| (tmp, None)); - finish_decoded_fetch(&app, fetch_id, result, &name).await; + finish_decoded_fetch(&app_clone, fetch_id, result, &name).await; }); + + let abort_handle = handle.abort_handle(); + { + let mut guard = app.lock().await; + let mut injected = false; + if let Some(crate::infra::queue::QueueNowPlaying::Decoded(ref mut d)) = guard.queue_now { + if d.fetch_id == fetch_id { + d.abort_handle = Some(crate::infra::queue::DownloadAbortHandle(abort_handle.clone())); + injected = true; + } + } + if !injected { + abort_handle.abort(); + } + } true } @@ -621,6 +666,8 @@ async fn publish_pending_decoded( fetch_id, #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] tempfile: None, + #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] + abort_handle: None, quality: None, })); fetch_id @@ -1730,4 +1777,44 @@ mod tests { "the queue slot now owns playback" ); } + + #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] + #[tokio::test] + async fn test_queue_skip_aborts_pending_download() { + let app = test_app(); + let track_info = track("subsonic:track:1", "Track 1"); + let player = Arc::new(crate::infra::audio::LocalPlayer::new().unwrap()); + + // 1. Publish the slot + let fetch_id = publish_pending_decoded(&app, &player, &track_info).await; + + // 2. Spawn a dummy sleeping task + let handle = tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + }); + + // 3. Inject abort handle + let abort_handle = handle.abort_handle(); + { + let mut guard = app.lock().await; + if let Some(crate::infra::queue::QueueNowPlaying::Decoded(ref mut d)) = guard.queue_now { + if d.fetch_id == fetch_id { + d.abort_handle = Some(crate::infra::queue::DownloadAbortHandle(abort_handle.clone())); + } + } + } + + // Verify the task is alive + assert!(!handle.is_finished()); + + // 4. Simulate a skip by clearing the slot + { + app.lock().await.queue_now = None; + } + + // 5. Wait a beat and verify the task was aborted + let result = handle.await; + assert!(result.is_err()); + assert!(result.unwrap_err().is_cancelled()); + } } diff --git a/src/infra/queue/mod.rs b/src/infra/queue/mod.rs index 55ae4d18..f543c682 100644 --- a/src/infra/queue/mod.rs +++ b/src/infra/queue/mod.rs @@ -434,6 +434,16 @@ pub fn restage( /// [`dispatch::try_play_queued`] can play. Internet radio pulls `audio-decode` /// in as well, but a live stream is never a queue item, so a radio-only build /// can never construct this. + +#[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] +pub struct DownloadAbortHandle(pub tokio::task::AbortHandle); + +#[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] +impl Drop for DownloadAbortHandle { + fn drop(&mut self) { + self.0.abort(); + } +} #[cfg(any( feature = "local-files", feature = "subsonic", @@ -468,6 +478,11 @@ pub struct DecodedQueuePlayback { #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] #[allow(dead_code)] pub tempfile: Option, + /// The handle to abort the background download task if the slot is cleared + /// or replaced before the download completes. + #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] + #[allow(dead_code)] + pub abort_handle: Option, /// The delivered audio format of a downloaded track (Qobuz, e.g. /// `FLAC 24/96`), shown after the artists in the playbar. pub quality: Option, From e996825b97aaf80f05db19153cacb91033333fb3 Mon Sep 17 00:00:00 2001 From: Rohith Pariki Date: Sun, 13 Sep 2026 01:29:48 +0530 Subject: [PATCH 2/2] fix: initialize abort_handle in play_decoded --- src/infra/queue/dispatch.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/infra/queue/dispatch.rs b/src/infra/queue/dispatch.rs index 6213577e..c83b740b 100644 --- a/src/infra/queue/dispatch.rs +++ b/src/infra/queue/dispatch.rs @@ -766,6 +766,8 @@ async fn publish_decoded( fetch_id: next_fetch_id(), #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] tempfile, + #[cfg(any(feature = "subsonic", feature = "qobuz", feature = "youtube"))] + abort_handle: None, quality: None, })); guard.set_status_message(format!("\u{266a} {name} (queue)"), 4);