fix(downloads): stop libraries mixing, make pause/resume real, reap partials, end bitrate corruption

Four defects behind "downloads still flaky", each with its own cause.

Libraries mixed their media (DR-167). Cached items carry no link back to their
library — library_id and parent_id are NULL on every row — so the library branch
of get_downloaded_items matched `EXISTS (SELECT 1 FROM libraries WHERE id = ?)`,
which asserts only that the library exists and never constrains the item to it.
Opening any downloaded library listed every downloaded top-level item on the
server: films under Music, albums under TV. The query deciding which libraries
appear already had the right rule, so the two disagreed about the same question;
that collection_type <-> item_type mapping is now one constant used by both.

Pause and resume did nothing (DR-168). pause_download wrote status = 'paused'
and stopped there — no cancellation existed anywhere in the download stack, so
the streaming task ran on and overwrote the row with completed/failed when it
finished. The row flicked to "paused" and undid itself. resume_download had the
mirror defect: it flipped the row to 'pending' without pumping, and the pump is
not a poller, so a resumed download sat until some unrelated event pumped the
queue. Adds a per-download stop flag the worker reads between chunks and on
retry, returning Stopped — not retryable, not recorded as a failure, and the
.part file is kept because that is what the resume continues from. Registering
returns a fresh flag so a resumed download does not inherit the pause that
stopped it. Cancel and clear_stale_downloads signal it too, so neither deletes a
file still being written.

Partial files were never reaped (DR-169). The worker named its sidecar with
with_extension("part"), which replaces: movie.mp4 became movie.part. Every
cleanup path deleted "{file_path}.part" — movie.mp4.part. They never matched, so
the partial of every cancelled or failed download stayed on disk forever,
invisible to disk-usage totals because no row pointed at it. One partial_path
helper now serves the writer and the cleaners.

Bitrate downloads corrupted themselves (DR-170). Only `original` asks for
Static=true; every other rung requests a transcode, which Jellyfin serves
chunked with no Content-Length and cannot byte-seek — it ignores Range and
answers 200 with the whole stream, not 206 with the tail. The worker sent the
header whenever a .part existed and appended the body regardless, so each retry
concatenated another full copy onto what was on disk. The file grew past its
real size and would not play, which is why bitrate downloads stayed broken after
the videoBitRate casing fix corrected the request. resume_offset now lets the
response decide: append only on 206, otherwise truncate and take it from the top.

docs/requirements.md also carries DR-171/UT-166, written by a parallel session
working in the same tree; its code lands separately.

src-tauri/src/download/mod.rs

@@ -9,6 +9,7 @@
 pub mod cache;
 pub mod events;
 pub mod network;
+pub mod stop;
 pub mod worker;
 
 use crate::utils::lock::MutexSafe;

src-tauri/src/download/stop.rs

@@ -0,0 +1,150 @@
+//! Stop signalling for in-flight downloads.
+//!
+//! TRACES: UR-055 | DR-168
+//!
+//! Pausing and cancelling used to be database-only: `pause_download` wrote
+//! `status = 'paused'` and nothing else. No cancellation existed anywhere in the
+//! download stack — no token, no flag, no abort — so the streaming task kept
+//! running, kept writing bytes, and on finishing overwrote the row with
+//! `completed` or `failed`. The row flicked to "paused" and then undid itself,
+//! which is precisely the reported "pause does not work".
+//!
+//! This is the missing half: a flag per in-flight download that the worker reads
+//! between chunks. Setting it makes the worker return [`Stopped`] promptly and
+//! leave the `.part` file **intact**, which is what lets a resume pick up from
+//! where it stopped via the existing HTTP Range request.
+//!
+//! Kept as a module-level registry rather than on `DownloadManager` because the
+//! two sides never meet: the command handler holds the manager's lock, while the
+//! worker runs detached inside `tauri::async_runtime::spawn` with no access to
+//! Tauri state. A registry both can reach is the smallest thing that works.
+//!
+//! [`Stopped`]: crate::download::worker::DownloadError::Stopped
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, OnceLock};
+
+use crate::utils::lock::MutexSafe;
+
+/// download id → its stop flag, for downloads currently in flight.
+fn registry() -> &'static Mutex<HashMap<i64, Arc<AtomicBool>>> {
+    static REGISTRY: OnceLock<Mutex<HashMap<i64, Arc<AtomicBool>>>> = OnceLock::new();
+    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+/// Register `download_id` as in-flight and hand back its stop flag.
+///
+/// Called by the worker as it starts. A previous flag for the same id is
+/// replaced, so a download that is paused and later resumed does not inherit the
+/// set flag from its last run and stop immediately.
+pub fn register(download_id: i64) -> Arc<AtomicBool> {
+    let flag = Arc::new(AtomicBool::new(false));
+    registry().lock_safe().insert(download_id, flag.clone());
+    flag
+}
+
+/// Ask an in-flight download to stop.
+///
+/// Returns whether one was actually in flight — the caller uses this to tell a
+/// running download (which will stop shortly) from a merely queued one (which
+/// the database update alone has already handled).
+pub fn signal(download_id: i64) -> bool {
+    match registry().lock_safe().get(&download_id) {
+        Some(flag) => {
+            flag.store(true, Ordering::SeqCst);
+            true
+        }
+        None => false,
+    }
+}
+
+/// Forget a download's flag. Called when its task finishes, however it ended.
+pub fn clear(download_id: i64) {
+    registry().lock_safe().remove(&download_id);
+}
+
+/// Whether a stop has been requested for `download_id`.
+///
+/// The worker reads its own `Arc<AtomicBool>` directly rather than looking the id
+/// up, so this exists for the tests that assert the registry's behaviour.
+#[cfg(test)]
+pub fn is_stopping(download_id: i64) -> bool {
+    registry()
+        .lock_safe()
+        .get(&download_id)
+        .map(|f| f.load(Ordering::SeqCst))
+        .unwrap_or(false)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Ids are per-test so the shared registry cannot leak between them.
+    fn unique_id(seed: i64) -> i64 {
+        900_000 + seed
+    }
+
+    #[test]
+    fn test_a_registered_download_starts_unflagged() {
+        let id = unique_id(1);
+        let flag = register(id);
+        assert!(!flag.load(Ordering::SeqCst));
+        assert!(!is_stopping(id));
+        clear(id);
+    }
+
+    #[test]
+    fn test_signal_sets_the_flag_the_worker_reads() {
+        let id = unique_id(2);
+        let flag = register(id);
+
+        assert!(signal(id), "a registered download reports as in flight");
+        assert!(
+            flag.load(Ordering::SeqCst),
+            "the worker's own handle sees it"
+        );
+        assert!(is_stopping(id));
+
+        clear(id);
+    }
+
+    /// The pump only needs to abort a task that exists; a queued row is handled
+    /// by its database status alone.
+    #[test]
+    fn test_signalling_an_unregistered_download_reports_not_in_flight() {
+        assert!(!signal(unique_id(3)));
+    }
+
+    #[test]
+    fn test_clear_forgets_the_download() {
+        let id = unique_id(4);
+        register(id);
+        signal(id);
+        clear(id);
+
+        assert!(!is_stopping(id));
+        assert!(!signal(id), "a cleared download is no longer in flight");
+    }
+
+    /// The bug this guards: pause sets the flag, and resume re-runs the same
+    /// download id. If registering reused the old flag, the resumed run would see
+    /// a set flag and stop instantly — a download that could never be resumed.
+    #[test]
+    fn test_reregistering_clears_a_previous_stop() {
+        let id = unique_id(5);
+        register(id);
+        signal(id);
+        assert!(is_stopping(id));
+
+        let fresh = register(id);
+        assert!(!fresh.load(Ordering::SeqCst));
+        assert!(
+            !is_stopping(id),
+            "a resumed download must not inherit the pause"
+        );
+
+        clear(id);
+    }
+}

src-tauri/src/download/worker.rs

@@ -1,6 +1,7 @@
 //! Download worker for HTTP streaming with progress tracking and retry logic
 
 use log::warn;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use futures_util::StreamExt;
@@ -31,10 +32,18 @@ impl DownloadWorker {
         }
     }
 
-    /// Download a file with retry logic and progress tracking
+    /// Download a file with retry logic and progress tracking.
+    ///
+    /// `stop` is the pause/cancel flag (see [`crate::download::stop`]). It is
+    /// checked between chunks and again between retries, so a paused download
+    /// stops promptly rather than after its next backoff — up to 45 seconds
+    /// away, which reads as the pause having done nothing.
+    ///
+    /// TRACES: UR-055 | DR-168
     pub async fn download<F>(
         &self,
         task: &DownloadTask,
+        stop: &AtomicBool,
         on_progress: F,
     ) -> Result<DownloadResult, DownloadError>
     where
@@ -43,7 +52,10 @@ impl DownloadWorker {
         let mut retries = 0;
 
         loop {
-            match self.try_download(task, &on_progress).await {
+            if stop.load(Ordering::SeqCst) {
+                return Err(DownloadError::Stopped);
+            }
+            match self.try_download(task, stop, &on_progress).await {
                 Ok(result) => return Ok(result),
                 Err(e) if retries < self.max_retries && e.is_retryable() => {
                     retries += 1;
@@ -63,6 +75,7 @@ impl DownloadWorker {
     async fn try_download<F>(
         &self,
         task: &DownloadTask,
+        stop: &AtomicBool,
         on_progress: &F,
     ) -> Result<DownloadResult, DownloadError>
     where
@@ -76,7 +89,7 @@ impl DownloadWorker {
         }
 
         // Check for partial download
-        let temp_path = task.target_path.with_extension("part");
+        let temp_path = partial_path(&task.target_path);
         let existing_bytes = if temp_path.exists() {
             fs::metadata(&temp_path).await.map(|m| m.len()).unwrap_or(0)
         } else {
@@ -100,22 +113,32 @@ impl DownloadWorker {
             return Err(DownloadError::Http(response.status().as_u16()));
         }
 
-        // Get content length
+        // Did the server actually honour the Range? A transcode does not, and
+        // answers 200 with the whole stream — appending that would duplicate what
+        // we already hold. (DR-170)
+        let resume_from = resume_offset(existing_bytes, response.status().as_u16());
+        if existing_bytes > 0 && resume_from == 0 {
+            warn!(
+                "Server ignored the Range request (HTTP {}) — restarting {} from the beginning \
+                 instead of appending to {} existing bytes",
+                response.status().as_u16(),
+                task.target_path.display(),
+                existing_bytes
+            );
+        }
+
+        // Get content length. Absent on a chunked transcode, which is why progress
+        // for a non-`original` preset has no percentage to show.
         let _total_bytes = response
             .headers()
             .get(reqwest::header::CONTENT_LENGTH)
             .and_then(|v| v.to_str().ok())
             .and_then(|v| v.parse::<u64>().ok())
-            .map(|len| {
-                if existing_bytes > 0 {
-                    len + existing_bytes
-                } else {
-                    len
-                }
-            });
+            .map(|len| len + resume_from);
 
-        // Open file for appending
-        let mut file = if existing_bytes > 0 {
+        // Append only when resuming a range the server agreed to; otherwise
+        // create/truncate so the restarted stream replaces the stale bytes.
+        let mut file = if resume_from > 0 {
             fs::OpenOptions::new().append(true).open(&temp_path).await
         } else {
             fs::File::create(&temp_path).await
@@ -123,11 +146,22 @@ impl DownloadWorker {
         .map_err(|e| DownloadError::FileSystem(e.to_string()))?;
 
         // Stream download with progress tracking
-        let mut downloaded = existing_bytes;
+        let mut downloaded = resume_from;
         let mut stream = response.bytes_stream();
         let mut last_progress_emit = std::time::Instant::now();
 
         while let Some(chunk) = stream.next().await {
+            // Checked before writing, so a paused download stops on a byte
+            // boundary the `.part` file already accounts for — the Range request
+            // on resume then asks for exactly what is missing. Flushing what we
+            // have and leaving the file in place is the whole mechanism behind
+            // "resume", so this must never delete it. (DR-168)
+            if stop.load(Ordering::SeqCst) {
+                let _ = file.flush().await;
+                let _ = file.sync_all().await;
+                return Err(DownloadError::Stopped);
+            }
+
             let chunk = chunk.map_err(|e| DownloadError::Network(e.to_string()))?;
 
             file.write_all(&chunk)
@@ -138,7 +172,7 @@ impl DownloadWorker {
 
             // Emit progress every 500ms or every MB
             if last_progress_emit.elapsed() > Duration::from_millis(500)
-                || downloaded % (1024 * 1024) == 0
+                || downloaded.is_multiple_of(1024 * 1024)
             {
                 last_progress_emit = std::time::Instant::now();
                 on_progress(downloaded, _total_bytes);
@@ -167,6 +201,56 @@ impl DownloadWorker {
     }
 }
 
+/// Where to resume writing a partial download, given how the server answered.
+///
+/// A byte offset of 0 means "start the file again"; anything else means "append
+/// from here".
+///
+/// This is what makes non-`original` downloads survive. Those presets ask
+/// Jellyfin to **transcode**, and a live transcode is chunked with no
+/// `Content-Length` and cannot be byte-seeked: the server ignores `Range` and
+/// answers `200` with the whole stream from the beginning, not `206` with the
+/// requested tail. The worker sent the header and appended the body regardless,
+/// so every retry — and every resume — concatenated a fresh copy of the whole
+/// transcode onto the bytes already on disk. The file grew past its real size
+/// and would not play. Only a `206` actually promises the tail; a `200` means we
+/// must discard what we have and take the stream from the top.
+///
+/// TRACES: UR-071 | DR-170
+pub fn resume_offset(existing_bytes: u64, status: u16) -> u64 {
+    if existing_bytes == 0 {
+        return 0;
+    }
+    // 206 Partial Content is the only answer that honours the Range request.
+    if status == 206 {
+        existing_bytes
+    } else {
+        0
+    }
+}
+
+/// The partial-download sidecar for `target`.
+///
+/// **Appends** `.part` rather than replacing the extension. The worker used
+/// `Path::with_extension("part")`, which replaces: `movie.mp4` became
+/// `movie.part`. Every cleanup path meanwhile deleted `"{file_path}.part"` —
+/// `movie.mp4.part` — so nothing ever matched and the partial file of every
+/// cancelled or failed download was left on disk forever, invisible to the
+/// disk-usage totals because no `downloads` row pointed at it. That is the
+/// reported "failure is not cleaned".
+///
+/// Appending also removes a collision the old form had: `movie.mp4` and
+/// `movie.mkv` both mapped to `movie.part` and would have fought over one file.
+///
+/// One function so the writer and the cleaners cannot disagree again.
+///
+/// TRACES: UR-055 | DR-169
+pub fn partial_path(target: &std::path::Path) -> std::path::PathBuf {
+    let mut s = target.as_os_str().to_os_string();
+    s.push(".part");
+    std::path::PathBuf::from(s)
+}
+
 /// Result of a successful download
 #[derive(Debug)]
 pub struct DownloadResult {
@@ -179,6 +263,10 @@ pub enum DownloadError {
     Network(String),
     Http(u16),
     FileSystem(String),
+    /// The download was asked to stop (paused or cancelled). Not a failure: the
+    /// row's status already says what happened, and the partial file is kept so a
+    /// resume can continue from it.
+    Stopped,
 }
 
 impl DownloadError {
@@ -188,8 +276,16 @@ impl DownloadError {
             DownloadError::Network(_) => true,
             DownloadError::Http(status) => *status >= 500, // Retry server errors
             DownloadError::FileSystem(_) => false,
+            // Retrying would restart the very download the user just paused.
+            DownloadError::Stopped => false,
         }
     }
+
+    /// Whether this outcome means "the user stopped it", rather than a failure to
+    /// record and report.
+    pub fn is_stopped(&self) -> bool {
+        matches!(self, DownloadError::Stopped)
+    }
 }
 
 impl std::fmt::Display for DownloadError {
@@ -198,6 +294,7 @@ impl std::fmt::Display for DownloadError {
             DownloadError::Network(msg) => write!(f, "Network error: {}", msg),
             DownloadError::Http(status) => write!(f, "HTTP error {}", status),
             DownloadError::FileSystem(msg) => write!(f, "File system error: {}", msg),
+            DownloadError::Stopped => write!(f, "Download stopped by request"),
         }
     }
 }
@@ -208,6 +305,64 @@ impl std::error::Error for DownloadError {}
 mod tests {
     use super::*;
 
+    /// The bitrate-download corruption: a transcode ignores `Range` and answers
+    /// `200` with the whole stream. Appending that to the bytes already on disk
+    /// duplicated them, so every retry grew the file past its real size and left
+    /// it unplayable. Only `206` promises the requested tail.
+    ///
+    /// TRACES: UR-071 | DR-170 | UT-164
+    #[test]
+    fn test_resume_offset_only_appends_when_the_server_honoured_the_range() {
+        // Nothing on disk: start at the beginning either way.
+        assert_eq!(resume_offset(0, 200), 0);
+        assert_eq!(resume_offset(0, 206), 0);
+
+        // The server agreed to the range — append to what we have.
+        assert_eq!(resume_offset(5_000, 206), 5_000);
+
+        // The server ignored it and is sending the whole file (a transcode).
+        // Restart, or the bytes are duplicated.
+        assert_eq!(
+            resume_offset(5_000, 200),
+            0,
+            "a 200 carries the whole stream; appending it corrupts the file"
+        );
+    }
+
+    /// The regression: `with_extension` replaced the extension, so the worker
+    /// wrote `movie.part` while every cleanup path deleted `movie.mp4.part`.
+    /// Nothing matched, and partial files accumulated forever.
+    ///
+    /// TRACES: UR-055 | DR-169 | UT-163
+    #[test]
+    fn test_partial_path_appends_rather_than_replacing_the_extension() {
+        use std::path::Path;
+
+        assert_eq!(
+            partial_path(Path::new("/media/movie.mp4")),
+            Path::new("/media/movie.mp4.part"),
+            "the cleanup paths delete \"{{file_path}}.part\"; this must produce it"
+        );
+
+        // Two sources for one title must not fight over a single partial file.
+        assert_ne!(
+            partial_path(Path::new("/media/movie.mp4")),
+            partial_path(Path::new("/media/movie.mkv")),
+        );
+
+        // Extension-less targets still get a sidecar rather than being clobbered.
+        assert_eq!(
+            partial_path(Path::new("/media/track")),
+            Path::new("/media/track.part"),
+        );
+
+        // A dotted name keeps every part of its own name.
+        assert_eq!(
+            partial_path(Path::new("/media/S01.E02.episode.mkv")),
+            Path::new("/media/S01.E02.episode.mkv.part"),
+        );
+    }
+
     #[test]
     fn test_exponential_backoff() {
         assert_eq!(
@@ -231,5 +386,9 @@ mod tests {
         assert!(DownloadError::Http(503).is_retryable());
         assert!(!DownloadError::Http(404).is_retryable());
         assert!(!DownloadError::FileSystem("disk full".to_string()).is_retryable());
+        // Retrying a paused download would restart what the user just stopped.
+        assert!(!DownloadError::Stopped.is_retryable());
+        assert!(DownloadError::Stopped.is_stopped());
+        assert!(!DownloadError::Network("timeout".to_string()).is_stopped());
     }
 }