From a5c22526590cf676dacd4bd10964a499f12050c5 Mon Sep 17 00:00:00 2001 From: Rohith Pariki Date: Sun, 13 Sep 2026 01:18:00 +0530 Subject: [PATCH 1/3] 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/3] 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); From 0eec123d526227ed26f46d4308f9ad9eb0a46378 Mon Sep 17 00:00:00 2001 From: Rohith Pariki Date: Wed, 16 Sep 2026 17:45:09 +0530 Subject: [PATCH 3/3] Fix CI failures and implement review feedback --- src/infra/queue/dispatch.rs | 112 +++++++++--------------------------- src/infra/queue/mod.rs | 29 +++++++--- 2 files changed, 47 insertions(+), 94 deletions(-) diff --git a/src/infra/queue/dispatch.rs b/src/infra/queue/dispatch.rs index a0156cfa..51e7e1d6 100644 --- a/src/infra/queue/dispatch.rs +++ b/src/infra/queue/dispatch.rs @@ -332,30 +332,17 @@ 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_clone = Arc::clone(app); + let app_for_spawn = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); 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_clone, fetch_id, result, &name).await; + finish_decoded_fetch(&app_for_spawn, 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(); - } - } + attach_abort_handle(app, fetch_id, handle.abort_handle()).await; true } @@ -371,30 +358,17 @@ 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_clone = Arc::clone(app); + let app_for_spawn = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); 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_clone, fetch_id, result, &name).await; + finish_decoded_fetch(&app_for_spawn, 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(); - } - } + attach_abort_handle(app, fetch_id, handle.abort_handle()).await; true } @@ -412,30 +386,17 @@ 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_clone = Arc::clone(app); + let app_for_spawn = Arc::clone(app); let uri = uri.to_string(); let name = track.name.clone(); 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_clone, fetch_id, result, &name).await; + finish_decoded_fetch(&app_for_spawn, 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(); - } - } + attach_abort_handle(app, fetch_id, handle.abort_handle()).await; true } @@ -773,6 +734,22 @@ async fn suspended_context_player(app: &Arc>) -> Option>, + fetch_id: u64, + abort_handle: tokio::task::AbortHandle, +) { + 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)); + return; + } + } + abort_handle.abort(); +} + #[cfg(feature = "audio-decode-queue")] async fn release_librespot(app: &Arc>) { #[cfg(feature = "streaming")] @@ -1232,6 +1209,7 @@ mod tests { use std::time::SystemTime; #[cfg(any( + feature = "queue-download", feature = "streaming", not(all(feature = "qobuz", feature = "subsonic")) ))] @@ -1633,43 +1611,5 @@ mod tests { ); } - #[cfg(feature = "queue-download")] - #[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 8258ae37..c506e2aa 100644 --- a/src/infra/queue/mod.rs +++ b/src/infra/queue/mod.rs @@ -411,14 +411,7 @@ pub fn restage( Ok(()) } -/// A queued *decoded* track playing through the shared [`LocalPlayer`] sink -/// (local file, Subsonic, or YouTube). Kept separate from the per-source -/// `*_playback` structs so the underlying context is preserved for resume. -/// -/// Gated on exactly those three sources, not `audio-decode`: they are the ones -/// [`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. +/// Aborts a background queue download when dropped. #[cfg(feature = "queue-download")] pub struct DownloadAbortHandle(pub tokio::task::AbortHandle); @@ -429,6 +422,26 @@ impl Drop for DownloadAbortHandle { } } +#[cfg(all(test, feature = "queue-download"))] +#[tokio::test] +async fn dropping_the_abort_handle_cancels_the_download_task() { + let handle = tokio::spawn(async { + tokio::time::sleep(std::time::Duration::from_secs(3600)).await; + }); + let abort_handle = DownloadAbortHandle(handle.abort_handle()); + drop(abort_handle); + let res = handle.await; + assert!(res.unwrap_err().is_cancelled()); +} + +/// A queued *decoded* track playing through the shared [`LocalPlayer`] sink +/// (local file, Subsonic, or YouTube). Kept separate from the per-source +/// `*_playback` structs so the underlying context is preserved for resume. +/// +/// Gated on exactly those three sources, not `audio-decode`: they are the ones +/// [`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(feature = "audio-decode-queue")] pub struct DecodedQueuePlayback { /// The output-device sink. Shared (`Arc::ptr_eq`) with the suspended context's