dtourolleandClaude Opus 5
e271874b1d
The download worker built its HTTP client with `Client::timeout(300s)`, which in reqwest is a total deadline that runs until the response body has finished. Every transfer longer than five minutes was cut off mid-body as "error decoding response body" and retried. A transcode ignores `Range`, so each retry restarted from byte zero, met the same deadline, and after three attempts the download failed — no feature film at transcode speed ever completed on a device whose audio must be re-encoded, and a large direct copy limped through in five-minute slices with a backoff between each. A connect timeout plus a read timeout that resets on every chunk catches a dead connection without capping how long a healthy transfer may run. Red first: a loopback server dribbling a body three times longer than the timeout failed with the old client (and burned the whole retry budget) and passes now; a second test hangs the socket and shows the stall is still detected. DR-289, UT-251. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
582 lines
22 KiB
Rust
582 lines
22 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,
|
||
}
|
||
|
||
/// How long a transfer may go without receiving a single byte before it is
|
||
/// treated as dead and retried. Generous because a transcode download waits on
|
||
/// ffmpeg, which pauses when Jellyfin throttles it.
|
||
const STALL_TIMEOUT: Duration = Duration::from_secs(60);
|
||
|
||
/// How long to wait for the TCP/TLS handshake before giving up.
|
||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
|
||
|
||
impl DownloadWorker {
|
||
pub fn new() -> Self {
|
||
Self::with_stall_timeout(STALL_TIMEOUT, true)
|
||
}
|
||
|
||
/// Build a worker whose HTTP client gives up on a transfer that receives
|
||
/// nothing for `stall`. `https_only` is relaxed only by tests, which serve
|
||
/// from a loopback socket.
|
||
///
|
||
/// The timeouts are a *connect* timeout and a *read* timeout — never
|
||
/// `Client::timeout`. That one is a total deadline that runs until the body
|
||
/// has finished, and it was set to five minutes: every download longer than
|
||
/// that was cut off mid-body with "error decoding response body", then
|
||
/// retried. A transcode ignores `Range`, so each retry restarted from byte
|
||
/// zero, ran into the same five minutes, and after three attempts the
|
||
/// download failed — which is why no feature film at transcode speed ever
|
||
/// completed on a device that needs the audio re-encoded. A read timeout
|
||
/// resets on every chunk, so it catches a dead connection without putting a
|
||
/// ceiling on how long a healthy transfer may run.
|
||
///
|
||
/// TRACES: UR-071 | DR-289 | UT-251
|
||
fn with_stall_timeout(stall: Duration, https_only: bool) -> Self {
|
||
let client = reqwest::Client::builder()
|
||
.connect_timeout(CONNECT_TIMEOUT)
|
||
.read_timeout(stall)
|
||
.https_only(https_only)
|
||
.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());
|
||
}
|
||
}
|
||
|
||
/// Transfers that outlive the timeout. These drive the real `reqwest` client
|
||
/// against a loopback socket because the defect lived in how that client was
|
||
/// configured, not in any code of ours a mock could stand in for.
|
||
#[cfg(test)]
|
||
mod timeout_tests {
|
||
use super::*;
|
||
use tokio::net::TcpListener;
|
||
|
||
/// Serve one HTTP/1.1 response of `chunks` bodies of `chunk_len` bytes,
|
||
/// pausing `gap` between them. `hang_after` chunks, the server stops sending
|
||
/// and never closes — a stalled connection.
|
||
async fn dribbling_server(
|
||
chunks: usize,
|
||
chunk_len: usize,
|
||
gap: Duration,
|
||
hang_after: Option<usize>,
|
||
) -> String {
|
||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||
let addr = listener.local_addr().unwrap();
|
||
tokio::spawn(async move {
|
||
let (mut sock, _) = listener.accept().await.unwrap();
|
||
// Drain the request head; we answer the same thing regardless.
|
||
let mut buf = [0u8; 4096];
|
||
let _ = tokio::io::AsyncReadExt::read(&mut sock, &mut buf).await;
|
||
let head = format!(
|
||
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
|
||
chunks * chunk_len
|
||
);
|
||
if sock.write_all(head.as_bytes()).await.is_err() {
|
||
return;
|
||
}
|
||
let body = vec![b'x'; chunk_len];
|
||
for i in 0..chunks {
|
||
if hang_after == Some(i) {
|
||
// Hold the socket open forever without writing.
|
||
tokio::time::sleep(Duration::from_secs(3600)).await;
|
||
}
|
||
// The client hanging up (as it does once it times out) is not
|
||
// the server's failure to report.
|
||
if sock.write_all(&body).await.is_err() || sock.flush().await.is_err() {
|
||
return;
|
||
}
|
||
tokio::time::sleep(gap).await;
|
||
}
|
||
});
|
||
format!("http://{}/file.bin", addr)
|
||
}
|
||
|
||
/// A download that takes longer than the timeout but never stalls must
|
||
/// finish. The worker set `Client::timeout`, which in reqwest is a *total*
|
||
/// deadline covering the body, so every transfer longer than five minutes —
|
||
/// any film at transcode speed — was cut off with "error decoding response
|
||
/// body", retried from byte zero (a transcode ignores `Range`), and cut off
|
||
/// again until the retry budget ran out.
|
||
///
|
||
/// TRACES: UR-071 | DR-289 | UT-251
|
||
#[tokio::test]
|
||
async fn test_download_longer_than_the_stall_timeout_completes_when_bytes_keep_flowing() {
|
||
let stall = Duration::from_millis(400);
|
||
// 12 chunks × 100 ms ≈ 1.2 s of transfer, three times the stall timeout,
|
||
// with every gap comfortably inside it.
|
||
let url = dribbling_server(12, 1024, Duration::from_millis(100), None).await;
|
||
let dir = tempfile::tempdir().unwrap();
|
||
let task = DownloadTask {
|
||
url,
|
||
target_path: dir.path().join("file.bin"),
|
||
};
|
||
|
||
let worker = DownloadWorker::with_stall_timeout(stall, false);
|
||
let result = worker
|
||
.download(&task, &AtomicBool::new(false), |_, _| {})
|
||
.await;
|
||
|
||
let result = result
|
||
.unwrap_or_else(|e| panic!("a transfer that never stalls must not time out: {e:?}"));
|
||
assert_eq!(result.bytes_downloaded, 12 * 1024);
|
||
assert!(task.target_path.exists());
|
||
}
|
||
|
||
/// The converse: a connection that goes silent is still given up on, so
|
||
/// dropping the total deadline did not turn a dead wifi link into a download
|
||
/// that hangs forever with no retry.
|
||
///
|
||
/// TRACES: UR-071 | DR-289 | UT-251
|
||
#[tokio::test]
|
||
async fn test_download_that_stalls_is_given_up_on() {
|
||
let stall = Duration::from_millis(300);
|
||
let url = dribbling_server(4, 1024, Duration::from_millis(10), Some(2)).await;
|
||
let dir = tempfile::tempdir().unwrap();
|
||
let task = DownloadTask {
|
||
url,
|
||
target_path: dir.path().join("file.bin"),
|
||
};
|
||
|
||
let worker = DownloadWorker::with_stall_timeout(stall, false);
|
||
// `download()` retries with 5 s/15 s/45 s backoff; a single attempt is
|
||
// what proves the stall is detected.
|
||
let started = std::time::Instant::now();
|
||
let result = worker
|
||
.try_download(&task, &AtomicBool::new(false), &|_, _| {})
|
||
.await;
|
||
|
||
assert!(
|
||
matches!(result, Err(DownloadError::Network(_))),
|
||
"a stalled transfer must fail as a network error: {result:?}"
|
||
);
|
||
assert!(
|
||
started.elapsed() < Duration::from_secs(5),
|
||
"the stall must be detected promptly, took {:?}",
|
||
started.elapsed()
|
||
);
|
||
}
|
||
}
|