dtourolle
747ec0161c
Two halves of one failure, either of which is enough to produce an
offline item that never starts.
The worker marked a transfer `completed` without checking it produced
any bytes, so a server that answered 200 with no body — an error page, a
transcode that yielded nothing — renamed a zero-byte `.part` into place
and published it as available offline. That is worse than failing: the
retry budget never applies and the UI shows the item as ready.
The media server then answered a request for that file with a span of
`{ start: 0, end: 0 }`. `end` is inclusive, so `Span::len()` reported
**one** byte: the response declared `Content-Length: 1` and streamed
nothing, which Chromium's media loader waits on forever. The user sees a
downloaded item that just never plays, with nothing explaining why.
A zero-length file has no satisfiable range, so `span_for` now returns
`None` and the server answers 416. An empty transfer is rejected as a
network error, which keeps the `.part` for a resume and lets the existing
retry budget do its job.
439 lines
16 KiB
Rust
439 lines
16 KiB
Rust
//! 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<F>(
|
|
&self,
|
|
task: &DownloadTask,
|
|
stop: &AtomicBool,
|
|
on_progress: F,
|
|
) -> Result<DownloadResult, DownloadError>
|
|
where
|
|
F: Fn(u64, Option<u64>) + 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<F>(
|
|
&self,
|
|
task: &DownloadTask,
|
|
stop: &AtomicBool,
|
|
on_progress: &F,
|
|
) -> Result<DownloadResult, DownloadError>
|
|
where
|
|
F: Fn(u64, Option<u64>) + 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::<u64>().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()))?;
|
|
|
|
// A media file is never legitimately empty, and completing one is worse
|
|
// than failing: the row goes `completed`, the item shows as available
|
|
// offline, and playback then stalls on a file with nothing in it. A
|
|
// server that answered 200 with no body — an error page, a transcode
|
|
// that produced nothing — used to land here. Treat it as the network
|
|
// failure it is so the retry budget applies and the `.part` is kept.
|
|
if let Some(reason) = rejects_as_empty(downloaded) {
|
|
return Err(DownloadError::Network(reason.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
|
|
}
|
|
}
|
|
|
|
/// Why a finished transfer must not be accepted, if it must not be.
|
|
///
|
|
/// A media file is never legitimately empty, and *completing* an empty one is
|
|
/// worse than failing: the row goes `completed`, the item shows as available
|
|
/// offline, and playback later stalls on a file with nothing in it. A server
|
|
/// that answered 200 with no body — an error page, a transcode that produced
|
|
/// nothing — used to land exactly there.
|
|
///
|
|
/// Reported as a network error so the existing retry budget applies and the
|
|
/// `.part` file is kept for a resume.
|
|
///
|
|
/// TRACES: UR-019 | DR-168 | UT-168
|
|
pub fn rejects_as_empty(downloaded: u64) -> Option<&'static str> {
|
|
(downloaded == 0)
|
|
.then_some("server sent an empty body; refusing to complete a zero-byte download")
|
|
}
|
|
|
|
/// 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::*;
|
|
|
|
/// A transfer that produced no bytes must never be marked complete.
|
|
///
|
|
/// Completing it publishes an empty file as playable offline; the media
|
|
/// server then answers a request for it with a 416 and the item simply
|
|
/// never starts, with nothing in the UI explaining why.
|
|
///
|
|
/// TRACES: UR-019 | DR-168 | UT-168
|
|
#[test]
|
|
fn test_a_zero_byte_transfer_is_rejected_rather_than_completed() {
|
|
assert!(
|
|
rejects_as_empty(0).is_some(),
|
|
"a zero-byte download must not be completed"
|
|
);
|
|
assert!(rejects_as_empty(1).is_none());
|
|
assert!(rejects_as_empty(4 * 1024 * 1024).is_none());
|
|
}
|
|
|
|
/// 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());
|
|
}
|
|
}
|