diff --git a/scripts/extract-traces.test.ts b/scripts/extract-traces.test.ts index 6664fa67..0ded1aa1 100644 --- a/scripts/extract-traces.test.ts +++ b/scripts/extract-traces.test.ts @@ -173,10 +173,10 @@ describe("live requirements.md", () => { ); const defined = countDefinedRequirements(md); - expect(defined.UR).toBe(73); + expect(defined.UR).toBe(74); expect(defined.IR).toBe(32); - expect(defined.DR).toBe(156); + expect(defined.DR).toBe(162); expect(defined.JA).toBe(35); - expect(defined.total).toBe(296); + expect(defined.total).toBe(303); }); }); diff --git a/src-tauri/src/commands/download/mod.rs b/src-tauri/src/commands/download/mod.rs index 16e065a7..6c4d3026 100644 --- a/src-tauri/src/commands/download/mod.rs +++ b/src-tauri/src/commands/download/mod.rs @@ -845,7 +845,19 @@ pub async fn get_downloads( Ok(DownloadsResponse { downloads, stats }) } -/// Pause a download +/// Pause a download. +/// +/// Writing `status = 'paused'` is only half of it, and used to be all of it: the +/// streaming task knew nothing about the row and kept running, then overwrote it +/// with `completed`/`failed` when it finished. The row flicked to "paused" and +/// undid itself — the reported "pause does not work". Signalling the worker is +/// what actually stops the bytes; it leaves the `.part` file in place so +/// [`resume_download`] can continue from it. +/// +/// A queued (not yet started) download has no worker to signal, and the status +/// write alone is enough — the pump skips anything that is not `pending`. +/// +/// TRACES: UR-055 | DR-168 #[tauri::command] #[specta::specta] pub async fn pause_download( @@ -858,19 +870,34 @@ pub async fn pause_download( }; let query = Query::with_params( - "UPDATE downloads SET status = 'paused' WHERE id = ? AND status = 'downloading'", + "UPDATE downloads SET status = 'paused' WHERE id = ? AND status IN ('downloading', 'pending')", vec![QueryParam::Int64(download_id)], ); db_service.execute(query).await.map_err(|e| e.to_string())?; + + let was_running = crate::download::stop::signal(download_id); + info!( + "[pause] Download {} paused (in flight: {})", + download_id, was_running + ); Ok(()) } -/// Resume a paused download +/// Resume a paused download. +/// +/// Flipping the row back to `pending` is likewise not enough on its own: the +/// pump is not a poller, it runs when something calls it, so a resumed download +/// sat untouched until some unrelated event happened to pump the queue. That is +/// the other half of "resume does not work". +/// +/// TRACES: UR-055 | DR-168 #[tauri::command] #[specta::specta] pub async fn resume_download( + app: tauri::AppHandle, db: State<'_, DatabaseWrapper>, + download_manager: State<'_, DownloadManagerWrapper>, download_id: i64, ) -> Result<(), String> { let db_service = { @@ -879,11 +906,22 @@ pub async fn resume_download( }; let query = Query::with_params( - "UPDATE downloads SET status = 'pending' WHERE id = ? AND status = 'paused'", + "UPDATE downloads SET status = 'pending', error_message = NULL WHERE id = ? AND status IN ('paused', 'failed')", vec![QueryParam::Int64(download_id)], ); db_service.execute(query).await.map_err(|e| e.to_string())?; + + // Drop any stale stop flag before the pump can start this id again, or the + // resumed run would read the pause that stopped it and halt immediately. + crate::download::stop::clear(download_id); + + let active_downloads = { + let manager = download_manager.0.lock().map_err(|e| e.to_string())?; + manager.get_active_downloads() + }; + pump_download_queue(app, db_service, active_downloads).await; + Ok(()) } @@ -923,6 +961,13 @@ pub async fn cancel_download( .await .map_err(|e| e.to_string())?; + // Stop the worker if this download is actually running. Without this the + // task keeps streaming into a `.part` file whose `downloads` row has just + // been deleted — bytes with nothing pointing at them, and the file below is + // removed while still being written to. (DR-168) + crate::download::stop::signal(download_id); + crate::download::stop::clear(download_id); + // Unregister from download manager (in case it was active) { let manager = download_manager.0.lock().map_err(|e| e.to_string())?; @@ -934,10 +979,12 @@ pub async fn cancel_download( ); } - // Delete partial file if exists + // Delete the partial file, and any completed file, if present. Both go + // through `partial_path` so this cannot drift from what the worker writes — + // it did, and every cancelled download leaked its partial. (DR-169) if let Some(path) = file_path { - let partial_path = format!("{}.part", path); - let _ = std::fs::remove_file(&partial_path); // Ignore errors + let target = std::path::PathBuf::from(&path); + let _ = std::fs::remove_file(crate::download::worker::partial_path(&target)); } Ok(()) @@ -1244,8 +1291,6 @@ pub async fn enqueue_video_downloads( download_ids: Vec, target_dir: String, ) -> Result<(), String> { - use crate::repository::MediaRepository; - let repo = repository.0.get(&handle).ok_or("Repository not found")?; let db_service = { @@ -1270,10 +1315,12 @@ pub async fn enqueue_video_downloads( } }; - // Build the transcode URL (pure URL builder, no server round-trip). - let stream_url = repo - .as_ref() - .get_video_download_url(&item_id, &quality, None); + // Build the download URL, resolving the source's audio codec first so a + // track this device cannot decode is re-encoded on the way down rather + // than saved as a silent file (DR-167). + let stream_url = + crate::repository::resolve_video_download_url(repo.as_ref(), &item_id, &quality, None) + .await; let update_query = Query::with_params( "UPDATE downloads SET status = 'pending', stream_url = ?, target_dir = ? WHERE id = ?", @@ -1530,7 +1577,11 @@ fn spawn_download_worker( let _ = progress_app.emit("download-event", event); }; - let result = worker.download(&task, on_progress).await; + // Registering returns a fresh flag, so a download resumed after a pause + // does not inherit the stop that ended its previous run. (DR-168) + let stop_flag = crate::download::stop::register(download_id); + let result = worker.download(&task, &stop_flag, on_progress).await; + crate::download::stop::clear(download_id); // Free the slot before pumping so the next download can take it. if let Ok(mut active) = active_downloads.lock() { @@ -1601,6 +1652,17 @@ fn spawn_download_worker( Err(e) => error!(" Completed event emit failed: {:?}", e), } } + // A pause or cancel is not a failure. The row already says `paused` + // (or the row is gone, for a cancel), and overwriting that with + // `failed` is what made a pause look like an error and stranded the + // download outside the resumable set. The `.part` file is deliberately + // left alone — it is what the resume continues from. (DR-168) + Err(e) if e.is_stopped() => { + info!( + "[pump] Download {} stopped by request; partial file kept for resume", + download_id + ); + } Err(e) => { error!("Download failed: {:?}", e); @@ -1848,17 +1910,26 @@ pub async fn clear_stale_downloads( Arc::new(database.service()) }; - // Get file paths for stale downloads (pending/paused/failed) + // Ids as well as paths: a stale row may still have a worker attached (a + // 'downloading' row that was paused mid-flight is 'paused' here), and + // deleting the row without stopping the task leaves it writing to a file we + // are about to remove. (DR-168) let file_query = Query::with_params( - "SELECT file_path FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')", + "SELECT id, file_path FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')", vec![QueryParam::String(user_id.clone())], ); - let file_paths: Vec = db_service - .query_many(file_query, |row| row.get(0)) + let stale: Vec<(i64, String)> = db_service + .query_many(file_query, |row| Ok((row.get(0)?, row.get(1)?))) .await .map_err(|e| e.to_string())?; + for (id, _) in &stale { + crate::download::stop::signal(*id); + crate::download::stop::clear(*id); + } + let file_paths: Vec = stale.into_iter().map(|(_, path)| path).collect(); + // Delete all pending, paused, and failed downloads (but keep completed ones) let delete_query = Query::with_params( "DELETE FROM downloads WHERE user_id = ? AND status IN ('pending', 'paused', 'failed')", @@ -1870,10 +1941,12 @@ pub async fn clear_stale_downloads( .await .map_err(|e| e.to_string())?; - // Delete any partial files + // Delete any partial files, via the shared helper so this cannot drift from + // what the worker actually writes. (DR-169) for path in file_paths { - let _ = std::fs::remove_file(&path); - let _ = std::fs::remove_file(format!("{}.part", path)); + let target = std::path::PathBuf::from(&path); + let _ = std::fs::remove_file(&target); + let _ = std::fs::remove_file(crate::download::worker::partial_path(&target)); } Ok(deleted_count as i64) diff --git a/src-tauri/src/download/mod.rs b/src-tauri/src/download/mod.rs index fc8b309e..a9a87088 100644 --- a/src-tauri/src/download/mod.rs +++ b/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; diff --git a/src-tauri/src/download/stop.rs b/src-tauri/src/download/stop.rs new file mode 100644 index 00000000..463ccb3b --- /dev/null +++ b/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>> { + static REGISTRY: OnceLock>>> = 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 { + 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` 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); + } +} diff --git a/src-tauri/src/download/worker.rs b/src-tauri/src/download/worker.rs index 1d556b6e..ead2e04e 100644 --- a/src-tauri/src/download/worker.rs +++ b/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( &self, task: &DownloadTask, + stop: &AtomicBool, on_progress: F, ) -> Result 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( &self, task: &DownloadTask, + stop: &AtomicBool, on_progress: &F, ) -> Result 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::().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()); } }