//! 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; use tokio::fs; use tokio::io::AsyncWriteExt; use super::DownloadTask; /// Download worker that handles individual download tasks pub struct DownloadWorker { /// HTTP client for downloads client: reqwest::Client, /// Maximum retry attempts max_retries: u32, } impl DownloadWorker { pub fn new() -> Self { let client = reqwest::Client::builder() .timeout(Duration::from_secs(300)) // 5 minute timeout .https_only(true) .build() .expect("Failed to create HTTP client"); Self { client, max_retries: 3, } } /// 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 F: Fn(u64, Option) + Send + Sync, { let mut retries = 0; loop { 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; let delay = Self::exponential_backoff(retries); warn!( "Download failed (attempt {}/{}), retrying in {:?}: {}", retries, self.max_retries, delay, e ); tokio::time::sleep(delay).await; } Err(e) => return Err(e), } } } /// Attempt a single download async fn try_download( &self, task: &DownloadTask, stop: &AtomicBool, on_progress: &F, ) -> Result where F: Fn(u64, Option) + Send + Sync, { // Create parent directories if let Some(parent) = task.target_path.parent() { fs::create_dir_all(parent) .await .map_err(|e| DownloadError::FileSystem(e.to_string()))?; } // Check for partial download 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 { 0 }; // Build HTTP request with Range header for resume support let mut request = self.client.get(&task.url); if existing_bytes > 0 { request = request.header("Range", format!("bytes={}-", existing_bytes)); } // Send request let response = request .send() .await .map_err(|e| DownloadError::Network(e.to_string()))?; // Check status if !response.status().is_success() && response.status().as_u16() != 206 { return Err(DownloadError::Http(response.status().as_u16())); } // 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| len + resume_from); // 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 } .map_err(|e| DownloadError::FileSystem(e.to_string()))?; // Stream download with progress tracking 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) .await .map_err(|e| DownloadError::FileSystem(e.to_string()))?; downloaded += chunk.len() as u64; // Emit progress every 500ms or every MB if last_progress_emit.elapsed() > Duration::from_millis(500) || downloaded.is_multiple_of(1024 * 1024) { last_progress_emit = std::time::Instant::now(); on_progress(downloaded, _total_bytes); } } file.sync_all() .await .map_err(|e| DownloadError::FileSystem(e.to_string()))?; // Move from .part to final location fs::rename(&temp_path, &task.target_path) .await .map_err(|e| DownloadError::FileSystem(e.to_string()))?; Ok(DownloadResult { bytes_downloaded: downloaded, }) } /// Calculate exponential backoff delay fn exponential_backoff(retry_count: u32) -> Duration { let base_delay = 5; // 5 seconds let delay_secs = base_delay * 3u64.pow(retry_count - 1); // 5s, 15s, 45s Duration::from_secs(delay_secs) } } /// 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 { pub bytes_downloaded: u64, } /// Download error types #[derive(Debug)] 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 { /// Check if this error is retryable fn is_retryable(&self) -> bool { match self { 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 { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { 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"), } } } impl std::error::Error for DownloadError {} #[cfg(test)] 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!( DownloadWorker::exponential_backoff(1), Duration::from_secs(5) ); assert_eq!( DownloadWorker::exponential_backoff(2), Duration::from_secs(15) ); assert_eq!( DownloadWorker::exponential_backoff(3), Duration::from_secs(45) ); } #[test] fn test_error_retryable() { assert!(DownloadError::Network("timeout".to_string()).is_retryable()); assert!(DownloadError::Http(500).is_retryable()); 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()); } }