fix(offline): cancellable downloads + stable sidebar progress toast (#694)

* fix(sidebar): keep offline-download toast from squishing in a short window

The toast lives in the sidebar nav flex column; without flex-shrink: 0 the
column compressed it vertically when the main window was small. The label
now also ellipsis-truncates instead of overflowing on a narrow sidebar.

* fix(offline): make offline downloads cancellable down to the Rust transfer

A running offline download could not be stopped — the sidebar X button only
dropped not-yet-started tracks between batches of 8, and the Rust transfer had
no cancellation path at all, so in-flight HTTP streams always ran to completion.

Add an offline_cancel_flags() registry (mirroring sync_cancel_flags for the
device-sync side) plus additive cancel_offline_downloads / clear_offline_cancel
commands. download_track_offline takes an optional download_id, checks the flag
right after acquiring its semaphore slot, and threads it through
finalize_streamed_download / stream_to_file so an in-flight stream aborts at the
next chunk — the partial .part file is cleaned up by the existing error path.

* fix(offline): cancel per-track and clear the sidebar toast immediately

downloadAlbum tags each run with a downloadId, checks for cancellation before
every track instead of once per 8-track batch (which never re-ran for albums of
8 or fewer tracks), and persists tracks that finished before the cancel so they
are not orphaned on disk. cancelDownload / cancelAllDownloads drop every job for
the album and call cancel_offline_downloads so Rust aborts the in-flight
transfers — the toast disappears at once instead of lingering on stuck rows.

Adds offlineJobStore cancellation tests.

* docs(changelog): offline download cancel button + toast sizing fixes
This commit is contained in:
Frank Stellmacher
2026-05-14 20:08:08 +02:00
committed by GitHub
parent 946528350c
commit b4c8ed4b65
12 changed files with 316 additions and 28 deletions
+1 -1
View File
@@ -69,7 +69,7 @@ pub async fn download_track_hot_cache(
// Stream directly to a .part file; rename on success to avoid partial files.
let part_path = file_path.with_extension(format!("{suffix}.part"));
if let Err(e) = stream_to_file(response, &part_path).await {
if let Err(e) = stream_to_file(response, &part_path, None).await {
let _ = tokio::fs::remove_file(&part_path).await;
return Err(e);
}
+90 -8
View File
@@ -1,8 +1,11 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tauri::Manager;
use psysonic_analysis::analysis_cache;
use psysonic_analysis::analysis_runtime::enqueue_analysis_seed;
use crate::DownloadSemaphore;
use crate::{offline_cancel_flags, DownloadSemaphore};
use crate::file_transfer::{finalize_streamed_download, subsonic_http_client};
@@ -48,12 +51,16 @@ pub(crate) async fn read_seed_bytes_if_needed(
/// the cached path on hit, otherwise issues a GET via `client`, streams to
/// `<cache_dir>/<track_id>.<suffix>` via a `.part` file, and returns the
/// final path. Caller is responsible for the semaphore + analysis seeding.
///
/// `cancel`, when supplied, aborts the in-flight stream with `Err("CANCELLED")`
/// (the `.part` file is cleaned up); `None` means the download is not cancellable.
pub(crate) async fn download_track_to_cache_dir(
cache_dir: &std::path::Path,
track_id: &str,
suffix: &str,
url: &str,
client: &reqwest::Client,
cancel: Option<&AtomicBool>,
) -> Result<std::path::PathBuf, String> {
tokio::fs::create_dir_all(cache_dir)
.await
@@ -70,7 +77,7 @@ pub(crate) async fn download_track_to_cache_dir(
}
let part_path = file_path.with_extension(format!("{suffix}.part"));
finalize_streamed_download(response, &file_path, &part_path).await?;
finalize_streamed_download(response, &file_path, &part_path, cancel).await?;
Ok(file_path)
}
@@ -98,12 +105,14 @@ pub(crate) fn resolve_offline_cache_dir(
/// Returns the absolute file path so TypeScript can store it and later
/// construct a `psysonic-local://<path>` URL for the audio engine.
#[tauri::command]
#[allow(clippy::too_many_arguments)] // Tauri command surface — args map 1:1 to the JS call.
pub async fn download_track_offline(
track_id: String,
server_id: String,
url: String,
suffix: String,
custom_dir: Option<String>,
download_id: Option<String>,
dl_sem: tauri::State<'_, DownloadSemaphore>,
app: tauri::AppHandle,
) -> Result<String, String> {
@@ -122,19 +131,68 @@ pub async fn download_track_offline(
return Ok(path_str);
}
// Resolve this download's cancellation flag. A missing `download_id` (e.g.
// an older caller) simply means the download cannot be cancelled.
let cancel_flag: Option<Arc<AtomicBool>> = download_id.as_deref().and_then(|id| {
offline_cancel_flags().lock().ok().map(|mut flags| {
flags
.entry(id.to_string())
.or_insert_with(|| Arc::new(AtomicBool::new(false)))
.clone()
})
});
// Acquire a download slot. The permit is held for the duration of the HTTP transfer
// and released automatically when this function returns (success or error).
let _permit = dl_sem.acquire().await.map_err(|e| e.to_string())?;
// Cancelled while parked on the semaphore — bail before opening a connection.
if cancel_flag.as_ref().is_some_and(|f| f.load(Ordering::Relaxed)) {
return Err("CANCELLED".to_string());
}
let client = subsonic_http_client(std::time::Duration::from_secs(120))?;
let final_path =
download_track_to_cache_dir(&cache_dir, &track_id, &suffix, &url, &client).await?;
let final_path = download_track_to_cache_dir(
&cache_dir,
&track_id,
&suffix,
&url,
&client,
cancel_flag.as_deref(),
)
.await?;
enqueue_analysis_seed_from_file(&app, &track_id, &final_path).await;
Ok(path_str)
}
/// Marks the given offline-download ids as cancelled. In-flight
/// `download_track_offline` calls abort their HTTP stream at the next chunk
/// boundary; ones still parked on the download semaphore bail as soon as they
/// acquire a slot. Mirrors `cancel_device_sync` for the device-sync side.
#[tauri::command]
pub fn cancel_offline_downloads(download_ids: Vec<String>) {
if let Ok(mut flags) = offline_cancel_flags().lock() {
for id in download_ids {
flags
.entry(id)
.or_insert_with(|| Arc::new(AtomicBool::new(false)))
.store(true, Ordering::Relaxed);
}
}
}
/// Drops a finished download's cancellation flag so the registry does not grow
/// across a long session. The frontend calls this once an album/playlist
/// download settles (completed or cancelled).
#[tauri::command]
pub fn clear_offline_cancel(download_id: String) {
if let Ok(mut flags) = offline_cancel_flags().lock() {
flags.remove(&download_id);
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -156,7 +214,7 @@ mod tests {
let client = subsonic_http_client(std::time::Duration::from_secs(5)).unwrap();
let url = format!("{}/stream/track-1", server.uri());
let path = download_track_to_cache_dir(&cache_dir, "track-1", "flac", &url, &client)
let path = download_track_to_cache_dir(&cache_dir, "track-1", "flac", &url, &client, None)
.await
.unwrap();
assert!(path.exists());
@@ -176,7 +234,7 @@ mod tests {
let client = subsonic_http_client(std::time::Duration::from_secs(5)).unwrap();
let url = format!("{}/should-not-be-hit", server.uri());
let path = download_track_to_cache_dir(&cache_dir, "track-1", "flac", &url, &client)
let path = download_track_to_cache_dir(&cache_dir, "track-1", "flac", &url, &client, None)
.await
.unwrap();
assert_eq!(path, pre_existing);
@@ -197,7 +255,7 @@ mod tests {
let client = subsonic_http_client(std::time::Duration::from_secs(5)).unwrap();
let url = format!("{}/stream/missing", server.uri());
let err = download_track_to_cache_dir(&cache_dir, "missing", "flac", &url, &client)
let err = download_track_to_cache_dir(&cache_dir, "missing", "flac", &url, &client, None)
.await
.unwrap_err();
assert!(err.contains("HTTP 404"), "got {err}");
@@ -221,12 +279,36 @@ mod tests {
let client = subsonic_http_client(std::time::Duration::from_secs(5)).unwrap();
let url = format!("{}/track", server.uri());
download_track_to_cache_dir(&cache_dir, "t", "mp3", &url, &client)
download_track_to_cache_dir(&cache_dir, "t", "mp3", &url, &client, None)
.await
.unwrap();
assert!(cache_dir.join("t.mp3").exists());
}
#[tokio::test(flavor = "multi_thread")]
async fn download_to_cache_dir_aborts_and_cleans_up_when_cancelled() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(wm_path("/stream/track-1"))
.respond_with(ResponseTemplate::new(200).set_body_bytes(b"flac body bytes".to_vec()))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let cache_dir = dir.path().join("psysonic-offline").join("server-A");
let client = subsonic_http_client(std::time::Duration::from_secs(5)).unwrap();
let url = format!("{}/stream/track-1", server.uri());
let cancel = AtomicBool::new(true);
let err =
download_track_to_cache_dir(&cache_dir, "track-1", "flac", &url, &client, Some(&cancel))
.await
.unwrap_err();
assert_eq!(err, "CANCELLED");
assert!(!cache_dir.join("track-1.flac").exists(), "no final file on cancel");
assert!(!cache_dir.join("track-1.flac.part").exists(), "no .part orphan on cancel");
}
// ── delete_offline_track_with_boundary (AppHandle-free) ─────────────────
#[tokio::test(flavor = "multi_thread")]
@@ -1,4 +1,5 @@
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
/// Build a reqwest client with the standard Subsonic UA and a single overall timeout.
@@ -14,9 +15,15 @@ pub fn subsonic_http_client(timeout: Duration) -> Result<reqwest::Client, String
/// Streams an HTTP response body to `dest_path` in chunks. Never buffers the full
/// file in memory — keeps RAM flat regardless of file size.
///
/// When `cancel` is supplied, the flag is checked before each chunk write: a set
/// flag aborts the transfer with `Err("CANCELLED")`, leaving the partial
/// `dest_path` for the caller to clean up. `None` means the transfer cannot be
/// cancelled (device-sync / hot-cache callers).
pub async fn stream_to_file(
response: reqwest::Response,
dest_path: &Path,
cancel: Option<&AtomicBool>,
) -> Result<(), String> {
use futures_util::StreamExt;
use tokio::io::AsyncWriteExt;
@@ -26,6 +33,9 @@ pub async fn stream_to_file(
.map_err(|e| e.to_string())?;
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
if cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
return Err("CANCELLED".to_string());
}
let chunk = chunk.map_err(|e| e.to_string())?;
file.write_all(&chunk).await.map_err(|e| e.to_string())?;
}
@@ -35,7 +45,8 @@ pub async fn stream_to_file(
/// Streams `response` to `part_path`, then renames `part_path` → `dest_path`.
/// On any failure the partial `.part` file is best-effort removed so it does
/// not linger on disk. Caller must ensure `dest_path.parent()` exists.
/// not linger on disk — this includes a `cancel`-triggered abort. Caller must
/// ensure `dest_path.parent()` exists.
///
/// Note vs. previous inline implementations: the offline/device single-track
/// flows used to leave a `.part` orphan if the final rename failed. This helper
@@ -44,8 +55,9 @@ pub async fn finalize_streamed_download(
response: reqwest::Response,
dest_path: &Path,
part_path: &Path,
cancel: Option<&AtomicBool>,
) -> Result<(), String> {
if let Err(e) = stream_to_file(response, part_path).await {
if let Err(e) = stream_to_file(response, part_path, cancel).await {
let _ = tokio::fs::remove_file(part_path).await;
return Err(e);
}
@@ -97,7 +109,7 @@ mod tests {
let response = reqwest::get(format!("{}/track.flac", server.uri()))
.await
.unwrap();
stream_to_file(response, &dest).await.unwrap();
stream_to_file(response, &dest, None).await.unwrap();
let written = std::fs::read(&dest).unwrap();
assert_eq!(written, body);
@@ -117,7 +129,7 @@ mod tests {
let response = reqwest::get(format!("{}/empty", server.uri()))
.await
.unwrap();
stream_to_file(response, &dest).await.unwrap();
stream_to_file(response, &dest, None).await.unwrap();
assert!(dest.exists());
assert_eq!(std::fs::metadata(&dest).unwrap().len(), 0);
}
@@ -136,7 +148,7 @@ mod tests {
let response = reqwest::get(format!("{}/x", server.uri()))
.await
.unwrap();
let result = stream_to_file(response, &dest).await;
let result = stream_to_file(response, &dest, None).await;
assert!(result.is_err(), "create on missing parent dir must err");
}
@@ -159,7 +171,7 @@ mod tests {
.await
.unwrap();
finalize_streamed_download(response, &dest, &part).await.unwrap();
finalize_streamed_download(response, &dest, &part, None).await.unwrap();
assert!(dest.exists(), "dest file must exist after success");
assert!(!part.exists(), "part file must not linger");
assert_eq!(std::fs::read(&dest).unwrap(), body);
@@ -186,9 +198,54 @@ mod tests {
.await
.unwrap();
let result = finalize_streamed_download(response, &dest, &part).await;
let result = finalize_streamed_download(response, &dest, &part, None).await;
assert!(result.is_err(), "rename onto existing directory must fail");
assert!(!part.exists(), "part file must be cleaned up after rename failure");
assert!(dest.is_dir(), "the blocker directory itself stays untouched");
}
// ── cancellation ──────────────────────────────────────────────────────────
#[tokio::test(flavor = "multi_thread")]
async fn stream_to_file_aborts_when_cancel_flag_is_already_set() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(wm_path("/track"))
.respond_with(ResponseTemplate::new(200).set_body_bytes(b"body bytes".to_vec()))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("track.flac");
let response = reqwest::get(format!("{}/track", server.uri()))
.await
.unwrap();
let cancel = AtomicBool::new(true);
let result = stream_to_file(response, &dest, Some(&cancel)).await;
assert_eq!(result.unwrap_err(), "CANCELLED");
}
#[tokio::test(flavor = "multi_thread")]
async fn finalize_cleans_up_part_when_cancelled() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(wm_path("/track"))
.respond_with(ResponseTemplate::new(200).set_body_bytes(b"body bytes".to_vec()))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let dest = dir.path().join("track.flac");
let part = dest.with_extension("flac.part");
let response = reqwest::get(format!("{}/track", server.uri()))
.await
.unwrap();
let cancel = AtomicBool::new(true);
let result = finalize_streamed_download(response, &dest, &part, Some(&cancel)).await;
assert_eq!(result.unwrap_err(), "CANCELLED");
assert!(!part.exists(), "cancelled transfer must not leave a .part orphan");
assert!(!dest.exists(), "cancelled transfer must not produce the final file");
}
}
@@ -26,3 +26,13 @@ pub fn sync_cancel_flags() -> &'static Mutex<HashMap<String, Arc<AtomicBool>>> {
static FLAGS: OnceLock<Mutex<HashMap<String, Arc<AtomicBool>>>> = OnceLock::new();
FLAGS.get_or_init(|| Mutex::new(HashMap::new()))
}
/// Per-download cancellation flags for offline album/playlist downloads,
/// keyed by the frontend-supplied download id. Each `download_track_offline`
/// call checks its flag (once after acquiring a slot, then on every chunk
/// while streaming); `cancel_offline_downloads` flips it. Mirrors
/// [`sync_cancel_flags`] for the device-sync side.
pub fn offline_cancel_flags() -> &'static Mutex<HashMap<String, Arc<AtomicBool>>> {
static FLAGS: OnceLock<Mutex<HashMap<String, Arc<AtomicBool>>>> = OnceLock::new();
FLAGS.get_or_init(|| Mutex::new(HashMap::new()))
}
@@ -513,7 +513,7 @@ pub async fn sync_batch_to_device(
};
let part_path = dest_path.with_extension(format!("{}.part", track.suffix));
if let Err(e) = finalize_streamed_download(response, &dest_path, &part_path).await {
if let Err(e) = finalize_streamed_download(response, &dest_path, &part_path, None).await {
f.fetch_add(1, Ordering::Relaxed);
let _ = app2.emit("device:sync:progress", serde_json::json!({
"jobId": job, "trackId": track.id, "status": "error",
@@ -347,7 +347,7 @@ pub(crate) async fn sync_download_one_track(
return Err(format!("HTTP {}", response.status().as_u16()));
}
let part_path = dest_path.with_extension(format!("{}.part", suffix));
finalize_streamed_download(response, dest_path, &part_path).await?;
finalize_streamed_download(response, dest_path, &part_path, None).await?;
Ok(true)
}