From 7f5e0406b7e7a59957052f9169c2e51f77551d94 Mon Sep 17 00:00:00 2001 From: zackees Date: Sun, 23 Aug 2026 02:10:09 -0700 Subject: [PATCH 1/3] fix(download): resume from the byte offset instead of restarting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes FastLED/fbuild#1370. The 282 MB ARM toolchain could not be provisioned at all on a connection that drops around 90 MB. Every retry started from zero and died in the same band — the reporter measured ~450 MB transferred for zero net progress, five restarts, no toolchain. That is not slow, it is non-terminating: more retries cannot help when each one begins where the last one began. Two changes, and the second matters as much as the first. ## Resume Bytes now stream to a `.part` beside the destination and are appended to across attempts. A retry sends `Range: bytes=-`, so a dropped connection costs only the bytes it did not deliver. The part file is renamed into place only once the body is complete, so no consumer can observe a truncated archive at the real path — and the checksum `staged_install` already runs still verifies the assembled result. Three server behaviors are handled explicitly, because getting any of them wrong corrupts the file rather than failing it: - **206** with `Content-Range` — append, and take the total from that header. `content_length()` on a partial response is what *remains*, so using it would have made the reported percentage restart at 0 on every retry. - **200 despite the range** — legal, and it means the body restarts. The part file is truncated rather than appended to, since the old prefix is no longer the right one. - **416** — the whole resource is already on disk; finalize and verify. A body that ends short of its announced length is now a retryable error instead of a successful download of a truncated file. ## A retry budget that can converge The budget is spent on *stalls*, not attempts. A 282 MB file over a link that dies at 90 MB needs four attempts, and a fixed five-attempt cap would fail a download that was converging perfectly well. An attempt that advances the file resets the counter; five consecutive attempts with no progress stop it, and the error names the byte offset it gave up at, as the issue asked. This is what changes `streaming_download_stops_after_five_truncated_bodies` to six requests: its mock ignores `Range`, so the first attempt makes progress and the five after it do not. Renamed to match the new contract. Also drops a 282 MB `Vec` from the download path — the buffer used to hold the entire archive in memory before a single write. ## Verified - `streaming_download_resumes_from_the_byte_offset_after_a_drop` — a mock that hangs up mid-body then honors the ranged retry. RED/GREEN confirmed: disabling the `Range` header makes it fail. - `streaming_download_gives_up_when_the_server_ignores_range` — proves the no-progress budget terminates, leaves neither a truncated archive nor a part file, and names the offset. - fbuild-packages-fetch 135 tests, workspace clippy `-D warnings`, full `dylint --all` sweep, and the platform-boundary comparison: all clean. ## Not included Cross-invocation resume. `staged_install` wipes its staging directory on entry — deliberately, since a stale partial *extract* must not be trusted — so a part file cannot survive there today. Making the archive resumable across runs means giving it a home outside staging, which is a separate change. Segmented/parallel fetch is likewise out of scope: resume is the part that changes whether the download finishes rather than how fast. Co-Authored-By: Claude Opus 5 (1M context) --- .../fbuild-packages-fetch/src/downloader.rs | 602 +++++++++++++++--- 1 file changed, 520 insertions(+), 82 deletions(-) diff --git a/crates/fbuild-packages-fetch/src/downloader.rs b/crates/fbuild-packages-fetch/src/downloader.rs index 1e60ee9d..25bfe0b1 100644 --- a/crates/fbuild-packages-fetch/src/downloader.rs +++ b/crates/fbuild-packages-fetch/src/downloader.rs @@ -76,7 +76,20 @@ enum DownloadAttemptError { Request(reqwest::Error), HttpStatus(reqwest::StatusCode), Body(reqwest::Error), - BodyStalled { filename: String }, + BodyStalled { + filename: String, + }, + /// The part file on disk could not be opened, written, or flushed. + PartFile { + path: String, + error: String, + }, + /// The body ended before the announced length — a dropped connection that + /// happened to land on a chunk boundary (FastLED/fbuild#1370). + BodyTruncated { + got: u64, + expected: u64, + }, } impl DownloadAttemptError { @@ -85,6 +98,11 @@ impl DownloadAttemptError { Self::Request(error) | Self::Body(error) => is_transient(error), Self::HttpStatus(status) => status.is_server_error(), Self::BodyStalled { .. } => true, + // A truncated body is the failure this retry loop exists for. + Self::BodyTruncated { .. } => true, + // Local disk trouble will not fix itself by asking the server + // again, and retrying would just rewrite the same bytes. + Self::PartFile { .. } => false, } } @@ -104,6 +122,14 @@ impl DownloadAttemptError { CHUNK_READ_TIMEOUT.as_secs(), filename )), + Self::PartFile { path, error } => FbuildError::PackageError(format!( + "failed to write partial download at {}: {}", + path, error + )), + Self::BodyTruncated { got, expected } => FbuildError::PackageError(format!( + "download of {} ended early: got {} of {} bytes", + url, got, expected + )), } } } @@ -120,6 +146,12 @@ impl Display for DownloadAttemptError { CHUNK_READ_TIMEOUT.as_secs(), filename ), + Self::PartFile { path, error } => { + write!(f, "partial-download write error at {path}: {error}") + } + Self::BodyTruncated { got, expected } => { + write!(f, "body ended early: {got} of {expected} bytes") + } } } } @@ -128,17 +160,87 @@ async fn open_attempt( client: &reqwest::Client, url: &str, ) -> std::result::Result { - let response = client - .get(url) + open_attempt_from(client, url, 0) + .await + .map(|opened| opened.response) +} + +/// A response plus what the server agreed to about resuming. +struct OpenedRange { + response: reqwest::Response, + /// Byte offset the body actually starts at. Zero when the server sent a + /// full body, whether or not a range was requested. + starts_at: u64, + /// Total size of the complete resource, when the server disclosed it. + total: Option, +} + +/// GET `url`, asking to resume from `offset` when that is non-zero. +/// +/// A server may decline the range and send the whole body instead — that is +/// legal, and the caller has to notice, because appending a full body onto a +/// partial file would silently corrupt it. `starts_at` reports what the +/// server actually did rather than what was asked for (FastLED/fbuild#1370). +async fn open_attempt_from( + client: &reqwest::Client, + url: &str, + offset: u64, +) -> std::result::Result { + let mut request = client.get(url); + if offset > 0 { + request = request.header(reqwest::header::RANGE, format!("bytes={offset}-")); + } + let response = request .send() .await .map_err(DownloadAttemptError::Request)?; let status = response.status(); - if status.is_success() { - Ok(response) - } else { - Err(DownloadAttemptError::HttpStatus(status)) + + // The whole resource is already on disk: the server has nothing left to + // send. Not an error — the caller finalizes and verifies. + if status == reqwest::StatusCode::RANGE_NOT_SATISFIABLE && offset > 0 { + return Ok(OpenedRange { + response, + starts_at: offset, + total: Some(offset), + }); } + if !status.is_success() { + return Err(DownloadAttemptError::HttpStatus(status)); + } + + if status == reqwest::StatusCode::PARTIAL_CONTENT { + let (starts_at, total) = parse_content_range(&response).unwrap_or((offset, None)); + return Ok(OpenedRange { + response, + starts_at, + total, + }); + } + + // 200 with a full body. If a range was asked for, the server ignored it. + let total = response.content_length(); + Ok(OpenedRange { + response, + starts_at: 0, + total, + }) +} + +/// Parse `Content-Range: bytes -/`. +/// +/// Returns the start offset and the total when the total is a number rather +/// than the `*` an origin is allowed to send. +fn parse_content_range(response: &reqwest::Response) -> Option<(u64, Option)> { + let value = response + .headers() + .get(reqwest::header::CONTENT_RANGE)? + .to_str() + .ok()?; + let spec = value.trim().strip_prefix("bytes ")?; + let (range, total) = spec.split_once('/')?; + let start = range.split_once('-')?.0.trim().parse::().ok()?; + Some((start, total.trim().parse::().ok())) } async fn wait_before_retry( @@ -277,8 +379,26 @@ async fn download_file_with_progress_using( .await } +/// How many attempts in a row may make zero progress before giving up. +/// +/// The retry budget is spent on *stalls*, not on attempts. A 282 MB download +/// on a connection that dies around 90 MB needs four attempts to finish, and +/// counting those against a fixed total would fail a download that was +/// converging fine — which is exactly the shape of FastLED/fbuild#1370, where +/// five restarts moved ~450 MB for zero net progress and could never +/// terminate. An attempt that advances the file resets this counter, so a +/// download that keeps making headway keeps going, and one that is genuinely +/// stuck still stops promptly. +const MAX_STALLED_ATTEMPTS: u32 = 5; + /// [`download_file_with_progress_using`] with the retry durations injected. /// See [`RetryTiming`] for why tests need this instead of paused Tokio time. +/// +/// Bytes land in a `.part` beside the destination and are appended +/// to across retries, so a failed attempt costs only the bytes it did not +/// finish rather than everything downloaded so far. The part file is renamed +/// into place only once the body is complete, so a consumer never observes a +/// truncated archive at the real path. async fn download_file_with_progress_timed( client: &reqwest::Client, url: &str, @@ -288,86 +408,227 @@ async fn download_file_with_progress_timed( ) -> Result<()> { let filename = url.rsplit('/').next().unwrap_or("download").to_string(); let dest_path = dest_dir.join(&filename); + let part_path = dest_dir.join(format!("{filename}.part")); + + // Start from a known state. A part file left by an earlier invocation + // cannot be trusted: nothing proves it came from this URL, and the + // caller wipes its staging directory anyway. + if let Err(error) = tokio::fs::remove_file(&part_path).await { + if error.kind() != std::io::ErrorKind::NotFound { + return Err(FbuildError::PackageError(format!( + "failed to clear partial download at {}: {}", + part_path.display(), + error + ))); + } + } + let mut resume_from: u64 = 0; + let mut stalled: u32 = 0; let mut attempt: u32 = 0; - let buf = loop { + + loop { attempt += 1; - let result: std::result::Result, DownloadAttemptError> = async { - let response = open_attempt(client, url).await?; - let total_bytes = response.content_length(); - let mut downloaded: u64 = 0; - let mut attempt_buf = - Vec::with_capacity(total_bytes.unwrap_or(8 * 1024 * 1024) as usize); - let mut last_report = Instant::now(); - let mut last_pct: u32 = 0; - let mut stream = response; - // FastLED/fbuild#805 CRITICAL: per-chunk deadline. The shared - // `http::client()` already enforces a 300 s total-request timeout, - // but defense-in-depth — wrap each `chunk().await` in a 60 s - // tokio timeout so a stalled mid-download fails *this* attempt - // promptly instead of waiting out the 5 min total. This is what - // the audit calls out specifically: streaming body reads have no - // per-chunk wake-up signal otherwise. - loop { - let chunk = - match tokio::time::timeout(timing.chunk_read_timeout, stream.chunk()).await { - Ok(Ok(Some(chunk))) => chunk, - Ok(Ok(None)) => break, - Ok(Err(error)) => return Err(DownloadAttemptError::Body(error)), - Err(_) => { - return Err(DownloadAttemptError::BodyStalled { - filename: filename.clone(), - }); - } - }; - attempt_buf.extend_from_slice(&chunk); - downloaded += chunk.len() as u64; - - let elapsed = last_report.elapsed().as_secs(); - let current_pct = total_bytes - .map(|total| { - if total > 0 { - (downloaded as f64 / total as f64 * 100.0) as u32 - } else { - 0 - } - }) - .unwrap_or(0); - let pct_jump = current_pct >= last_pct + 10; - - if elapsed >= 15 || pct_jump { - let progress = DownloadProgress { - downloaded, - total_bytes, - filename: filename.clone(), - }; - on_progress(&progress); - last_report = Instant::now(); - last_pct = current_pct; + let started_at = resume_from; + + let outcome = fetch_into_part( + client, + url, + &part_path, + resume_from, + &filename, + on_progress, + timing, + ) + .await; + + resume_from = part_len(&part_path).await; + + match outcome { + Ok(()) => break, + Err(error) => { + // Progress, not attempt count, is what earns another try. + if resume_from > started_at { + stalled = 0; + tracing::info!( + "download {}: attempt {} ended at {} bytes; resuming", + url, + attempt, + resume_from + ); + } else { + stalled += 1; } + + if error.is_retryable() && stalled < MAX_STALLED_ATTEMPTS { + wait_before_retry(url, stalled.max(1), &error, timing).await; + continue; + } + + let _ = tokio::fs::remove_file(&part_path).await; + return Err(FbuildError::PackageError(format!( + "{} (gave up after {} attempts at byte offset {}; \ + {} consecutive attempts made no progress)", + error.into_fbuild_error(url), + attempt, + resume_from, + stalled + ))); } - Ok(attempt_buf) } - .await; + } - match result { - Ok(bytes) => break bytes, - Err(error) if error.is_retryable() && attempt < MAX_ATTEMPTS => { - wait_before_retry(url, attempt, &error, timing).await; - } - Err(error) => return Err(error.into_fbuild_error(url)), + // Windows will not rename over an existing file. + if let Err(error) = tokio::fs::remove_file(&dest_path).await { + if error.kind() != std::io::ErrorKind::NotFound { + return Err(FbuildError::PackageError(format!( + "failed to replace {}: {}", + dest_path.display(), + error + ))); } - }; + } + tokio::fs::rename(&part_path, &dest_path) + .await + .map_err(|e| { + FbuildError::PackageError(format!( + "failed to move completed download into {}: {}", + dest_path.display(), + e + )) + })?; + + tracing::info!("downloaded {} ({} bytes)", filename, resume_from); + Ok(()) +} - tokio::fs::write(&dest_path, &buf).await.map_err(|e| { - FbuildError::PackageError(format!( - "failed to write downloaded file to {}: {}", - dest_path.display(), - e - )) - })?; +/// Bytes already in the part file, or zero when it does not exist. +async fn part_len(part_path: &Path) -> u64 { + tokio::fs::metadata(part_path) + .await + .map(|meta| meta.len()) + .unwrap_or(0) +} + +/// Stream one attempt's worth of body into `part_path`, resuming at `offset`. +/// +/// Appends when the server honors the range and truncates when it does not, +/// so the part file always matches what the server is actually sending. +#[allow(clippy::too_many_arguments)] +async fn fetch_into_part( + client: &reqwest::Client, + url: &str, + part_path: &Path, + offset: u64, + filename: &str, + on_progress: &mut (dyn FnMut(&DownloadProgress) + Send), + timing: RetryTiming, +) -> std::result::Result<(), DownloadAttemptError> { + use tokio::io::AsyncWriteExt; + + let opened = open_attempt_from(client, url, offset).await?; + let OpenedRange { + mut response, + starts_at, + total, + } = opened; + + // The server declined the range and restarted the body. Anything already + // written is now the wrong prefix, so drop it rather than append. + let appending = starts_at == offset && offset > 0; + if !appending && offset > 0 { + tracing::warn!( + "download {}: server ignored the resume request and restarted from 0; \ + discarding {} partial bytes", + url, + offset + ); + } + + let mut file = tokio::fs::OpenOptions::new() + .create(true) + .write(true) + .append(appending) + .truncate(!appending) + .open(part_path) + .await + .map_err(|error| DownloadAttemptError::PartFile { + path: part_path.display().to_string(), + error: error.to_string(), + })?; + + let mut downloaded: u64 = if appending { offset } else { 0 }; + // `content_length()` on a 206 is what remains, not the whole resource, so + // the total has to come from Content-Range when resuming — otherwise the + // percentage the caller renders would restart at 0 on every retry. + let total_bytes = total.or_else(|| response.content_length().map(|len| len + downloaded)); + + let mut last_report = Instant::now(); + let mut last_pct: u32 = 0; + + loop { + let chunk = match tokio::time::timeout(timing.chunk_read_timeout, response.chunk()).await { + Ok(Ok(Some(chunk))) => chunk, + Ok(Ok(None)) => break, + Ok(Err(error)) => return Err(DownloadAttemptError::Body(error)), + Err(_) => { + return Err(DownloadAttemptError::BodyStalled { + filename: filename.to_string(), + }); + } + }; + file.write_all(&chunk) + .await + .map_err(|error| DownloadAttemptError::PartFile { + path: part_path.display().to_string(), + error: error.to_string(), + })?; + downloaded += chunk.len() as u64; + + let elapsed = last_report.elapsed().as_secs(); + let current_pct = total_bytes + .map(|total| { + if total > 0 { + (downloaded as f64 / total as f64 * 100.0) as u32 + } else { + 0 + } + }) + .unwrap_or(0); + let pct_jump = current_pct >= last_pct + 10; + + if elapsed >= 15 || pct_jump { + let progress = DownloadProgress { + downloaded, + total_bytes, + filename: filename.to_string(), + }; + on_progress(&progress); + last_report = Instant::now(); + last_pct = current_pct; + } + } - tracing::info!("downloaded {} ({} bytes)", filename, buf.len()); + // Flush before the caller stats the file to decide whether this attempt + // made progress — buffered bytes would read as a stall. + file.flush() + .await + .map_err(|error| DownloadAttemptError::PartFile { + path: part_path.display().to_string(), + error: error.to_string(), + })?; + + // A short body is a dropped connection that happened to end on a chunk + // boundary. Treat it as a retryable failure so the resume loop continues + // rather than renaming a truncated archive into place. + if let Some(total) = total_bytes { + if downloaded < total { + return Err(DownloadAttemptError::BodyTruncated { + got: downloaded, + expected: total, + }); + } + } Ok(()) } @@ -756,7 +1017,7 @@ mod tests { } #[tokio::test] - async fn streaming_download_stops_after_five_truncated_bodies_without_output() { + async fn streaming_download_stops_after_five_stalled_attempts_without_output() { let _guard = network_test_guard().await; let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ truncated_response(), @@ -776,11 +1037,188 @@ mod tests { .await .expect_err("the fifth truncated response should exhaust retries"); - // See the buffered variant above: both body-read and connection - // errors are retryable transport failures, so only exhaustion of the - // five-attempt budget is stable across platforms. - assert_eq!(request_count.load(Ordering::SeqCst), 5); + // Six, not five, and the extra one is the point of + // FastLED/fbuild#1370: the budget is now five attempts that make *no + // progress*, not five attempts total. The first attempt advances the + // part file from 0 to the truncation point, so it does not spend + // budget; the five after it re-deliver the same prefix (this mock + // ignores `Range`) and do. A download that keeps advancing is no + // longer cut off at a fixed attempt count, which is what let a large + // toolchain fail forever on a connection that could not carry it in + // one stream. + assert_eq!(request_count.load(Ordering::SeqCst), 6); assert!(!temp.path().join("file").exists()); + assert!(!temp.path().join("file.part").exists()); + } + + /// A server that drops the connection partway through the body, then + /// serves the remainder to a ranged retry. + /// + /// This is the shape FastLED/fbuild#1370 reported: a 282 MB download that + /// died in the same 80-98 MB band every time. `honor_range = false` + /// models an origin that ignores `Range` and restarts the body, which is + /// legal and must not corrupt the partial file. + async fn run_resuming_server( + body: &'static [u8], + first_len: usize, + honor_range: bool, + request_count: std::sync::Arc, + ) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + let _ = ready_tx.send(()); + loop { + let (mut stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + + let mut buf = [0u8; 2048]; + let read = stream.read(&mut buf).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buf[..read]).to_string(); + let start = parse_request_range(&request); + + if honor_range && start > 0 && start < body.len() { + let head = format!( + "HTTP/1.1 206 Partial Content\r\n\ + Content-Length: {}\r\n\ + Content-Range: bytes {}-{}/{}\r\n\ + Accept-Ranges: bytes\r\n\r\n", + body.len() - start, + start, + body.len() - 1, + body.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(&body[start..]).await; + } else { + // Announce the whole body but hang up after `first_len`. + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(&body[..first_len]).await; + } + let _ = stream.shutdown().await; + } + }); + ready_rx.await.expect("resuming test server should start"); + port + } + + /// Pull the start offset out of a `Range: bytes=N-` request header. + fn parse_request_range(request: &str) -> usize { + request + .lines() + .find_map(|line| { + let value = line + .strip_prefix("Range:") + .or_else(|| line.strip_prefix("range:"))?; + let spec = value.trim().strip_prefix("bytes=")?; + spec.split('-').next()?.trim().parse::().ok() + }) + .unwrap_or(0) + } + + const RESUME_BODY: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; + + /// The fix for FastLED/fbuild#1370: a dropped connection costs only the + /// bytes it did not deliver. + /// + /// Without resume this needs the server to send a complete body in one + /// attempt, which is exactly what the reporter's connection could not do. + /// With it, two attempts finish the file and the second one asks for the + /// remainder rather than starting over. + #[tokio::test] + async fn streaming_download_resumes_from_the_byte_offset_after_a_drop() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_resuming_server(RESUME_BODY, 10, true, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut seen: Vec = Vec::new(); + let mut progress = |p: &DownloadProgress| seen.push(p.downloaded); + + download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect("a ranged retry should finish the download"); + + assert_eq!( + std::fs::read(temp.path().join("file")).unwrap(), + RESUME_BODY, + "the resumed file must be byte-identical to the source" + ); + assert_eq!( + request_count.load(Ordering::SeqCst), + 2, + "one dropped attempt plus one ranged resume" + ); + assert!( + !temp.path().join("file.part").exists(), + "the part file must be renamed away, not left behind" + ); + assert!( + seen.iter().all(|d| *d <= RESUME_BODY.len() as u64), + "reported progress must stay cumulative rather than restarting: {seen:?}" + ); + } + + /// An origin that ignores `Range` must not corrupt the partial file, and + /// must still terminate rather than looping forever. + /// + /// The second half of #1370 is that retries which make no progress cannot + /// converge. Here every attempt lands on the same byte, so the + /// no-progress budget is what stops it. + #[tokio::test] + async fn streaming_download_gives_up_when_the_server_ignores_range() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_resuming_server(RESUME_BODY, 10, false, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_p: &DownloadProgress| {}; + + let error = + download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect_err("a server that never sends the tail must fail, not hang"); + + let message = error.to_string(); + assert!( + message.contains("byte offset"), + "the failure must name the offset it stopped at, per #1370: {message}" + ); + assert!( + !temp.path().join("file").exists(), + "no truncated archive may be left at the destination" + ); + assert!( + !temp.path().join("file.part").exists(), + "the part file must be cleaned up on hard failure" + ); + // One attempt makes progress (0 -> 10), then every later attempt + // re-sends the same prefix, so the no-progress budget ends it. + assert!( + request_count.load(Ordering::SeqCst) >= MAX_STALLED_ATTEMPTS as usize, + "should have spent the no-progress budget" + ); + } + + #[test] + fn content_range_start_and_total_are_parsed() { + // Exercised through the public shape rather than a Response, which + // cannot be constructed here: the header grammar is the fragile part. + assert_eq!(parse_request_range("Range: bytes=1234-\r\n"), 1234); + assert_eq!(parse_request_range("range: bytes=0-\r\n"), 0); + assert_eq!(parse_request_range("GET / HTTP/1.1\r\n"), 0); } /// Retry timings short enough to run in real time. This test previously From c0463a3aedcf00c4bffc3dfb658d164a560f94b0 Mon Sep 17 00:00:00 2001 From: zackees Date: Sun, 23 Aug 2026 02:28:21 -0700 Subject: [PATCH 2/3] refactor(download): move the downloader tests into their own file The resume work pushed `downloader.rs` to 1320 LOC and tripped the workspace's 1000-LOC gate. The gate grandfathers files that were already over on the base ref, so this was genuinely new. Tests move to `downloader_tests.rs` behind `#[cfg(test)] #[path = ...]`, which is the pattern `compiler.rs` / `compiler_tests.rs` already established. Implementation drops to 702 LOC; no test content changed. Refs FastLED/fbuild#1370 Co-Authored-By: Claude Opus 5 (1M context) --- .../fbuild-packages-fetch/src/downloader.rs | 622 +----------------- .../src/downloader_tests.rs | 619 +++++++++++++++++ 2 files changed, 621 insertions(+), 620 deletions(-) create mode 100644 crates/fbuild-packages-fetch/src/downloader_tests.rs diff --git a/crates/fbuild-packages-fetch/src/downloader.rs b/crates/fbuild-packages-fetch/src/downloader.rs index 25bfe0b1..0b566ebd 100644 --- a/crates/fbuild-packages-fetch/src/downloader.rs +++ b/crates/fbuild-packages-fetch/src/downloader.rs @@ -698,623 +698,5 @@ pub async fn verify_checksum_async(path: &Path, expected: &str) -> Result<()> { } #[cfg(test)] -mod tests { - use super::*; - use std::io::Write; - use std::sync::atomic::{AtomicUsize, Ordering}; - use tempfile::NamedTempFile; - - static NETWORK_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); - - async fn network_test_guard() -> tokio::sync::MutexGuard<'static, ()> { - NETWORK_TEST_LOCK.lock().await - } - - fn named_temp_file() -> NamedTempFile { - NamedTempFile::new_in(fbuild_paths::temp_subdir( - "fbuild-packages-downloader-tests", - )) - .unwrap() - } - - fn test_client() -> reqwest::Client { - fbuild_core::http::client_with_timeout(Duration::from_secs(300)) - } - - #[test] - fn test_verify_checksum_valid() { - let mut f = named_temp_file(); - f.write_all(b"hello world").unwrap(); - f.flush().unwrap(); - - // SHA256 of "hello world" - let expected = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"; - verify_checksum(f.path(), expected).unwrap(); - } - - #[test] - fn test_verify_checksum_invalid() { - let mut f = named_temp_file(); - f.write_all(b"hello world").unwrap(); - f.flush().unwrap(); - - let result = verify_checksum( - f.path(), - "0000000000000000000000000000000000000000000000000000000000000000", - ); - assert!(result.is_err()); - assert!( - result - .unwrap_err() - .to_string() - .contains("checksum mismatch") - ); - } - - // ---- transient-retry tests ---- - - /// Stand up a tiny raw-TCP HTTP server on a loopback port. Reads - /// one request, drops the body, writes whatever 4-line HTTP - /// response the caller queued for that attempt, and closes the - /// connection. The caller pre-queues a Vec of responses, one per - /// attempt; the server pops the next one as each connection - /// comes in. Keeps the deps to tokio (already required). - async fn run_flaky_server( - responses: std::sync::Arc>>, - request_count: std::sync::Arc, - ) -> u16 { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; - - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = listener.local_addr().unwrap().port(); - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); - - tokio::spawn(async move { - // A bound listener is not necessarily being polled yet. Make - // the caller wait until this task has reached its accept loop so - // a retry test cannot burn an attempt during task startup on a - // loaded runner. - let _ = ready_tx.send(()); - loop { - let (mut stream, _) = match listener.accept().await { - Ok(p) => p, - Err(_) => break, - }; - request_count.fetch_add(1, Ordering::SeqCst); - let resp = { - let mut guard = responses.lock().unwrap_or_else(|err| err.into_inner()); - if guard.is_empty() { - break; - } - guard.remove(0) - }; - let mut buf = [0u8; 1024]; - // Read just the request headers — don't care about the - // body for these tests. - // The client under test always writes a request. Do not use - // a paused-clock timeout here: it races the retry backoff and - // can make the mock emit a response before the request task - // has been scheduled on macOS. - let _ = stream.read(&mut buf).await; - let _ = stream.write_all(resp.as_bytes()).await; - let _ = stream.shutdown().await; - } - }); - ready_rx.await.expect("flaky test server task should start"); - port - } - - /// How long `run_stalling_server` withholds the announced body. Only has - /// to comfortably exceed `FAST_RETRY_TIMING.chunk_read_timeout`. - const STALL_DURATION: Duration = Duration::from_secs(30); - - async fn run_stalling_server(request_count: std::sync::Arc) -> u16 { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; - - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = listener.local_addr().unwrap().port(); - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); - - tokio::spawn(async move { - let _ = ready_tx.send(()); - loop { - let (stream, _) = match listener.accept().await { - Ok(pair) => pair, - Err(_) => break, - }; - request_count.fetch_add(1, Ordering::SeqCst); - tokio::spawn(async move { - let mut stream = stream; - let mut request = [0u8; 1024]; - // See `run_flaky_server`: this test owns the client, so - // waiting for its request is deterministic. - let _ = stream.read(&mut request).await; - let _ = stream - .write_all( - b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\n", - ) - .await; - // Announce a body, then never send it. Just needs to - // outlast the client's per-chunk deadline; the task is - // dropped at runtime shutdown, so the test doesn't wait - // on it (FastLED/fbuild#1222 — this runs in real time - // now, not paused time). - tokio::time::sleep(STALL_DURATION).await; - let _ = stream.shutdown().await; - }); - } - }); - ready_rx - .await - .expect("stalling test server task should start"); - port - } - - fn truncated_response() -> &'static str { - "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nConnection: close\r\n\r\nshort" - } - - fn complete_response() -> &'static str { - "HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello" - } - - #[test] - fn retry_policy_is_five_attempts_with_exponential_backoff() { - assert_eq!(MAX_ATTEMPTS, 5); - assert_eq!( - RETRY_BACKOFFS, - &[ - Duration::from_secs(1), - Duration::from_secs(2), - Duration::from_secs(4), - Duration::from_secs(8), - ] - ); - } - - /// #205 nightly STM32 acceptance gate started flaking on - /// `dl.registry.platformio.org` transient errors. A 5xx must - /// trigger a retry, and the retry must succeed. - #[tokio::test] - async fn get_with_retry_retries_on_5xx() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - complete_response(), - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses.clone(), request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let bytes = get_with_retry_using(&test_client(), &url) - .await - .expect("retry should succeed"); - assert_eq!(bytes, b"hello"); - assert_eq!(request_count.load(Ordering::SeqCst), 5); - } - - /// 4xx is deterministic — it must NOT retry. The test queues a - /// single 404; if the implementation retried we'd hit the server's - /// empty-queue branch and the test would hang or panic. - #[tokio::test] - async fn get_with_retry_does_not_retry_on_4xx() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses.clone(), request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/missing"); - let err = get_with_retry_using(&test_client(), &url) - .await - .expect_err("should error"); - assert!( - err.to_string().contains("404"), - "expected 404 in error, got: {err}" - ); - assert_eq!(request_count.load(Ordering::SeqCst), 1); - } - - /// Repeated 5xx exhausts the budget and surfaces the last - /// response. - #[tokio::test] - async fn get_with_retry_gives_up_after_max_attempts() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses.clone(), request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let err = get_with_retry_using(&test_client(), &url) - .await - .expect_err("should give up"); - // Last attempt was a 503; that's what gets surfaced. - assert!( - err.to_string().contains("503"), - "expected last-attempt 503 in error, got: {err}" - ); - assert_eq!(request_count.load(Ordering::SeqCst), 5); - } - - #[tokio::test] - async fn get_with_retry_retries_truncated_bodies_until_attempt_five() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - complete_response(), - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - - let bytes = get_with_retry_using(&test_client(), &url) - .await - .expect("the fifth complete response should succeed"); - - assert_eq!(bytes, b"hello"); - assert_eq!(request_count.load(Ordering::SeqCst), 5); - } - - #[tokio::test] - async fn get_with_retry_stops_after_five_truncated_bodies() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - - let _err = get_with_retry_using(&test_client(), &url) - .await - .expect_err("the fifth truncated response should exhaust retries"); - - // The final transient can surface either while reqwest reads the - // deliberately short body or while it opens that last connection. - // The retry budget, rather than this transport-layer wording, is the - // contract under test. - assert_eq!(request_count.load(Ordering::SeqCst), 5); - } - - #[tokio::test] - async fn streaming_download_retries_truncated_bodies_until_attempt_five() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - complete_response(), - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let temp = tempfile::TempDir::new().unwrap(); - let mut progress = |_progress: &DownloadProgress| {}; - - download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) - .await - .expect("the fifth complete response should succeed"); - - assert_eq!(std::fs::read(temp.path().join("file")).unwrap(), b"hello"); - assert_eq!(request_count.load(Ordering::SeqCst), 5); - } - - #[tokio::test] - async fn streaming_download_stops_after_five_stalled_attempts_without_output() { - let _guard = network_test_guard().await; - let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - truncated_response(), - ])); - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_flaky_server(responses, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let temp = tempfile::TempDir::new().unwrap(); - let mut progress = |_progress: &DownloadProgress| {}; - - let _err = - download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) - .await - .expect_err("the fifth truncated response should exhaust retries"); - - // Six, not five, and the extra one is the point of - // FastLED/fbuild#1370: the budget is now five attempts that make *no - // progress*, not five attempts total. The first attempt advances the - // part file from 0 to the truncation point, so it does not spend - // budget; the five after it re-deliver the same prefix (this mock - // ignores `Range`) and do. A download that keeps advancing is no - // longer cut off at a fixed attempt count, which is what let a large - // toolchain fail forever on a connection that could not carry it in - // one stream. - assert_eq!(request_count.load(Ordering::SeqCst), 6); - assert!(!temp.path().join("file").exists()); - assert!(!temp.path().join("file.part").exists()); - } - - /// A server that drops the connection partway through the body, then - /// serves the remainder to a ranged retry. - /// - /// This is the shape FastLED/fbuild#1370 reported: a 282 MB download that - /// died in the same 80-98 MB band every time. `honor_range = false` - /// models an origin that ignores `Range` and restarts the body, which is - /// legal and must not corrupt the partial file. - async fn run_resuming_server( - body: &'static [u8], - first_len: usize, - honor_range: bool, - request_count: std::sync::Arc, - ) -> u16 { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; - - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let port = listener.local_addr().unwrap().port(); - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); - - tokio::spawn(async move { - let _ = ready_tx.send(()); - loop { - let (mut stream, _) = match listener.accept().await { - Ok(pair) => pair, - Err(_) => break, - }; - request_count.fetch_add(1, Ordering::SeqCst); - - let mut buf = [0u8; 2048]; - let read = stream.read(&mut buf).await.unwrap_or(0); - let request = String::from_utf8_lossy(&buf[..read]).to_string(); - let start = parse_request_range(&request); - - if honor_range && start > 0 && start < body.len() { - let head = format!( - "HTTP/1.1 206 Partial Content\r\n\ - Content-Length: {}\r\n\ - Content-Range: bytes {}-{}/{}\r\n\ - Accept-Ranges: bytes\r\n\r\n", - body.len() - start, - start, - body.len() - 1, - body.len() - ); - let _ = stream.write_all(head.as_bytes()).await; - let _ = stream.write_all(&body[start..]).await; - } else { - // Announce the whole body but hang up after `first_len`. - let head = format!( - "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n\r\n", - body.len() - ); - let _ = stream.write_all(head.as_bytes()).await; - let _ = stream.write_all(&body[..first_len]).await; - } - let _ = stream.shutdown().await; - } - }); - ready_rx.await.expect("resuming test server should start"); - port - } - - /// Pull the start offset out of a `Range: bytes=N-` request header. - fn parse_request_range(request: &str) -> usize { - request - .lines() - .find_map(|line| { - let value = line - .strip_prefix("Range:") - .or_else(|| line.strip_prefix("range:"))?; - let spec = value.trim().strip_prefix("bytes=")?; - spec.split('-').next()?.trim().parse::().ok() - }) - .unwrap_or(0) - } - - const RESUME_BODY: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; - - /// The fix for FastLED/fbuild#1370: a dropped connection costs only the - /// bytes it did not deliver. - /// - /// Without resume this needs the server to send a complete body in one - /// attempt, which is exactly what the reporter's connection could not do. - /// With it, two attempts finish the file and the second one asks for the - /// remainder rather than starting over. - #[tokio::test] - async fn streaming_download_resumes_from_the_byte_offset_after_a_drop() { - let _guard = network_test_guard().await; - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_resuming_server(RESUME_BODY, 10, true, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let temp = tempfile::TempDir::new().unwrap(); - let mut seen: Vec = Vec::new(); - let mut progress = |p: &DownloadProgress| seen.push(p.downloaded); - - download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) - .await - .expect("a ranged retry should finish the download"); - - assert_eq!( - std::fs::read(temp.path().join("file")).unwrap(), - RESUME_BODY, - "the resumed file must be byte-identical to the source" - ); - assert_eq!( - request_count.load(Ordering::SeqCst), - 2, - "one dropped attempt plus one ranged resume" - ); - assert!( - !temp.path().join("file.part").exists(), - "the part file must be renamed away, not left behind" - ); - assert!( - seen.iter().all(|d| *d <= RESUME_BODY.len() as u64), - "reported progress must stay cumulative rather than restarting: {seen:?}" - ); - } - - /// An origin that ignores `Range` must not corrupt the partial file, and - /// must still terminate rather than looping forever. - /// - /// The second half of #1370 is that retries which make no progress cannot - /// converge. Here every attempt lands on the same byte, so the - /// no-progress budget is what stops it. - #[tokio::test] - async fn streaming_download_gives_up_when_the_server_ignores_range() { - let _guard = network_test_guard().await; - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_resuming_server(RESUME_BODY, 10, false, request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let temp = tempfile::TempDir::new().unwrap(); - let mut progress = |_p: &DownloadProgress| {}; - - let error = - download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) - .await - .expect_err("a server that never sends the tail must fail, not hang"); - - let message = error.to_string(); - assert!( - message.contains("byte offset"), - "the failure must name the offset it stopped at, per #1370: {message}" - ); - assert!( - !temp.path().join("file").exists(), - "no truncated archive may be left at the destination" - ); - assert!( - !temp.path().join("file.part").exists(), - "the part file must be cleaned up on hard failure" - ); - // One attempt makes progress (0 -> 10), then every later attempt - // re-sends the same prefix, so the no-progress budget ends it. - assert!( - request_count.load(Ordering::SeqCst) >= MAX_STALLED_ATTEMPTS as usize, - "should have spent the no-progress budget" - ); - } - - #[test] - fn content_range_start_and_total_are_parsed() { - // Exercised through the public shape rather than a Response, which - // cannot be constructed here: the header grammar is the fragile part. - assert_eq!(parse_request_range("Range: bytes=1234-\r\n"), 1234); - assert_eq!(parse_request_range("range: bytes=0-\r\n"), 0); - assert_eq!(parse_request_range("GET / HTTP/1.1\r\n"), 0); - } - - /// Retry timings short enough to run in real time. This test previously - /// used `#[tokio::test(start_paused = true)]` against a real - /// `TcpListener`, which flaked on loaded macOS runners: paused time - /// auto-advances whenever the runtime looks idle, but socket readiness - /// comes from the OS reactor, so the clock could jump past a connection - /// that was about to reach `accept()` — leaving `request_count` short of - /// 5. Real durations remove the race entirely (FastLED/fbuild#1222). - const FAST_RETRY_TIMING: RetryTiming = RetryTiming { - chunk_read_timeout: Duration::from_millis(150), - backoffs: &[ - Duration::from_millis(10), - Duration::from_millis(10), - Duration::from_millis(10), - Duration::from_millis(10), - ], - }; - - #[tokio::test] - async fn streaming_download_retries_chunk_stalls_five_times_without_output() { - let _guard = network_test_guard().await; - let request_count = std::sync::Arc::new(AtomicUsize::new(0)); - let port = run_stalling_server(request_count.clone()).await; - let url = format!("http://127.0.0.1:{port}/file"); - let temp = tempfile::TempDir::new().unwrap(); - let mut progress = |_progress: &DownloadProgress| {}; - - let _err = download_file_with_progress_timed( - &test_client(), - &url, - temp.path(), - &mut progress, - FAST_RETRY_TIMING, - ) - .await - .expect_err("five chunk stalls should exhaust retries"); - - // A stalled body is retryable, as is a connection error while opening - // a retry. The latter can legitimately be the final transient on a - // busy platform, so the stable contract is exhausting all five - // attempts without publishing an output. - assert_eq!(request_count.load(Ordering::SeqCst), 5); - assert!(!temp.path().join("file").exists()); - } - - /// The injected timings are a test seam, not a behavior change: the - /// production path must still carry the real constants. - #[test] - fn production_retry_timing_matches_the_constants() { - assert_eq!( - RetryTiming::PRODUCTION.chunk_read_timeout, - CHUNK_READ_TIMEOUT - ); - assert_eq!(RetryTiming::PRODUCTION.backoffs, RETRY_BACKOFFS); - for attempt in 1..MAX_ATTEMPTS { - assert_eq!( - RetryTiming::PRODUCTION.backoff(attempt), - RETRY_BACKOFFS[(attempt - 1) as usize] - ); - } - } - - #[test] - fn format_download_progress_with_total() { - let p = DownloadProgress { - downloaded: 50 * 1024 * 1024, - total_bytes: Some(150 * 1024 * 1024), - filename: "toolchain.tar.gz".into(), - }; - let msg = p.format_message(); - assert!(msg.contains("50"), "msg: {msg}"); - assert!(msg.contains("150"), "msg: {msg}"); - assert!(msg.contains("33%"), "msg: {msg}"); - } - - #[test] - fn format_download_progress_without_total() { - let p = DownloadProgress { - downloaded: 5 * 1024 * 1024, - total_bytes: None, - filename: "library.zip".into(), - }; - let msg = p.format_message(); - assert!(msg.contains("5"), "msg: {msg}"); - assert!(!msg.contains("%"), "msg: {msg}"); - } - - #[test] - fn format_download_progress_zero() { - let p = DownloadProgress { - downloaded: 0, - total_bytes: Some(100 * 1024 * 1024), - filename: "file.bin".into(), - }; - let msg = p.format_message(); - assert!(msg.contains("0%"), "msg: {msg}"); - } -} +#[path = "downloader_tests.rs"] +mod tests; diff --git a/crates/fbuild-packages-fetch/src/downloader_tests.rs b/crates/fbuild-packages-fetch/src/downloader_tests.rs new file mode 100644 index 00000000..5b2d2955 --- /dev/null +++ b/crates/fbuild-packages-fetch/src/downloader_tests.rs @@ -0,0 +1,619 @@ +//! Tests for [`super`] — the retrying, resumable downloader. +//! +//! Split out of `downloader.rs` to keep that file under the workspace's +//! 1000-LOC limit; `compiler_tests.rs` is the same pattern. + +use super::*; +use std::io::Write; +use std::sync::atomic::{AtomicUsize, Ordering}; +use tempfile::NamedTempFile; + +static NETWORK_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +async fn network_test_guard() -> tokio::sync::MutexGuard<'static, ()> { + NETWORK_TEST_LOCK.lock().await +} + +fn named_temp_file() -> NamedTempFile { + NamedTempFile::new_in(fbuild_paths::temp_subdir( + "fbuild-packages-downloader-tests", + )) + .unwrap() +} + +fn test_client() -> reqwest::Client { + fbuild_core::http::client_with_timeout(Duration::from_secs(300)) +} + +#[test] +fn test_verify_checksum_valid() { + let mut f = named_temp_file(); + f.write_all(b"hello world").unwrap(); + f.flush().unwrap(); + + // SHA256 of "hello world" + let expected = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"; + verify_checksum(f.path(), expected).unwrap(); +} + +#[test] +fn test_verify_checksum_invalid() { + let mut f = named_temp_file(); + f.write_all(b"hello world").unwrap(); + f.flush().unwrap(); + + let result = verify_checksum( + f.path(), + "0000000000000000000000000000000000000000000000000000000000000000", + ); + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .to_string() + .contains("checksum mismatch") + ); +} + +// ---- transient-retry tests ---- + +/// Stand up a tiny raw-TCP HTTP server on a loopback port. Reads +/// one request, drops the body, writes whatever 4-line HTTP +/// response the caller queued for that attempt, and closes the +/// connection. The caller pre-queues a Vec of responses, one per +/// attempt; the server pops the next one as each connection +/// comes in. Keeps the deps to tokio (already required). +async fn run_flaky_server( + responses: std::sync::Arc>>, + request_count: std::sync::Arc, +) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + // A bound listener is not necessarily being polled yet. Make + // the caller wait until this task has reached its accept loop so + // a retry test cannot burn an attempt during task startup on a + // loaded runner. + let _ = ready_tx.send(()); + loop { + let (mut stream, _) = match listener.accept().await { + Ok(p) => p, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + let resp = { + let mut guard = responses.lock().unwrap_or_else(|err| err.into_inner()); + if guard.is_empty() { + break; + } + guard.remove(0) + }; + let mut buf = [0u8; 1024]; + // Read just the request headers — don't care about the + // body for these tests. + // The client under test always writes a request. Do not use + // a paused-clock timeout here: it races the retry backoff and + // can make the mock emit a response before the request task + // has been scheduled on macOS. + let _ = stream.read(&mut buf).await; + let _ = stream.write_all(resp.as_bytes()).await; + let _ = stream.shutdown().await; + } + }); + ready_rx.await.expect("flaky test server task should start"); + port +} + +/// How long `run_stalling_server` withholds the announced body. Only has +/// to comfortably exceed `FAST_RETRY_TIMING.chunk_read_timeout`. +const STALL_DURATION: Duration = Duration::from_secs(30); + +async fn run_stalling_server(request_count: std::sync::Arc) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + let _ = ready_tx.send(()); + loop { + let (stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let mut stream = stream; + let mut request = [0u8; 1024]; + // See `run_flaky_server`: this test owns the client, so + // waiting for its request is deterministic. + let _ = stream.read(&mut request).await; + let _ = stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\n") + .await; + // Announce a body, then never send it. Just needs to + // outlast the client's per-chunk deadline; the task is + // dropped at runtime shutdown, so the test doesn't wait + // on it (FastLED/fbuild#1222 — this runs in real time + // now, not paused time). + tokio::time::sleep(STALL_DURATION).await; + let _ = stream.shutdown().await; + }); + } + }); + ready_rx + .await + .expect("stalling test server task should start"); + port +} + +fn truncated_response() -> &'static str { + "HTTP/1.1 200 OK\r\nContent-Length: 10\r\nConnection: close\r\n\r\nshort" +} + +fn complete_response() -> &'static str { + "HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello" +} + +#[test] +fn retry_policy_is_five_attempts_with_exponential_backoff() { + assert_eq!(MAX_ATTEMPTS, 5); + assert_eq!( + RETRY_BACKOFFS, + &[ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), + Duration::from_secs(8), + ] + ); +} + +/// #205 nightly STM32 acceptance gate started flaking on +/// `dl.registry.platformio.org` transient errors. A 5xx must +/// trigger a retry, and the retry must succeed. +#[tokio::test] +async fn get_with_retry_retries_on_5xx() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + complete_response(), + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses.clone(), request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let bytes = get_with_retry_using(&test_client(), &url) + .await + .expect("retry should succeed"); + assert_eq!(bytes, b"hello"); + assert_eq!(request_count.load(Ordering::SeqCst), 5); +} + +/// 4xx is deterministic — it must NOT retry. The test queues a +/// single 404; if the implementation retried we'd hit the server's +/// empty-queue branch and the test would hang or panic. +#[tokio::test] +async fn get_with_retry_does_not_retry_on_4xx() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses.clone(), request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/missing"); + let err = get_with_retry_using(&test_client(), &url) + .await + .expect_err("should error"); + assert!( + err.to_string().contains("404"), + "expected 404 in error, got: {err}" + ); + assert_eq!(request_count.load(Ordering::SeqCst), 1); +} + +/// Repeated 5xx exhausts the budget and surfaces the last +/// response. +#[tokio::test] +async fn get_with_retry_gives_up_after_max_attempts() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses.clone(), request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let err = get_with_retry_using(&test_client(), &url) + .await + .expect_err("should give up"); + // Last attempt was a 503; that's what gets surfaced. + assert!( + err.to_string().contains("503"), + "expected last-attempt 503 in error, got: {err}" + ); + assert_eq!(request_count.load(Ordering::SeqCst), 5); +} + +#[tokio::test] +async fn get_with_retry_retries_truncated_bodies_until_attempt_five() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + complete_response(), + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + + let bytes = get_with_retry_using(&test_client(), &url) + .await + .expect("the fifth complete response should succeed"); + + assert_eq!(bytes, b"hello"); + assert_eq!(request_count.load(Ordering::SeqCst), 5); +} + +#[tokio::test] +async fn get_with_retry_stops_after_five_truncated_bodies() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + + let _err = get_with_retry_using(&test_client(), &url) + .await + .expect_err("the fifth truncated response should exhaust retries"); + + // The final transient can surface either while reqwest reads the + // deliberately short body or while it opens that last connection. + // The retry budget, rather than this transport-layer wording, is the + // contract under test. + assert_eq!(request_count.load(Ordering::SeqCst), 5); +} + +#[tokio::test] +async fn streaming_download_retries_truncated_bodies_until_attempt_five() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + complete_response(), + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_progress: &DownloadProgress| {}; + + download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect("the fifth complete response should succeed"); + + assert_eq!(std::fs::read(temp.path().join("file")).unwrap(), b"hello"); + assert_eq!(request_count.load(Ordering::SeqCst), 5); +} + +#[tokio::test] +async fn streaming_download_stops_after_five_stalled_attempts_without_output() { + let _guard = network_test_guard().await; + let responses = std::sync::Arc::new(std::sync::Mutex::new(vec![ + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + truncated_response(), + ])); + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_flaky_server(responses, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_progress: &DownloadProgress| {}; + + let _err = download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect_err("the fifth truncated response should exhaust retries"); + + // Six, not five, and the extra one is the point of + // FastLED/fbuild#1370: the budget is now five attempts that make *no + // progress*, not five attempts total. The first attempt advances the + // part file from 0 to the truncation point, so it does not spend + // budget; the five after it re-deliver the same prefix (this mock + // ignores `Range`) and do. A download that keeps advancing is no + // longer cut off at a fixed attempt count, which is what let a large + // toolchain fail forever on a connection that could not carry it in + // one stream. + assert_eq!(request_count.load(Ordering::SeqCst), 6); + assert!(!temp.path().join("file").exists()); + assert!(!temp.path().join("file.part").exists()); +} + +/// A server that drops the connection partway through the body, then +/// serves the remainder to a ranged retry. +/// +/// This is the shape FastLED/fbuild#1370 reported: a 282 MB download that +/// died in the same 80-98 MB band every time. `honor_range = false` +/// models an origin that ignores `Range` and restarts the body, which is +/// legal and must not corrupt the partial file. +async fn run_resuming_server( + body: &'static [u8], + first_len: usize, + honor_range: bool, + request_count: std::sync::Arc, +) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + let _ = ready_tx.send(()); + loop { + let (mut stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + + let mut buf = [0u8; 2048]; + let read = stream.read(&mut buf).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buf[..read]).to_string(); + let start = parse_request_range(&request); + + if honor_range && start > 0 && start < body.len() { + let head = format!( + "HTTP/1.1 206 Partial Content\r\n\ + Content-Length: {}\r\n\ + Content-Range: bytes {}-{}/{}\r\n\ + Accept-Ranges: bytes\r\n\r\n", + body.len() - start, + start, + body.len() - 1, + body.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(&body[start..]).await; + } else { + // Announce the whole body but hang up after `first_len`. + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(&body[..first_len]).await; + } + let _ = stream.shutdown().await; + } + }); + ready_rx.await.expect("resuming test server should start"); + port +} + +/// Pull the start offset out of a `Range: bytes=N-` request header. +fn parse_request_range(request: &str) -> usize { + request + .lines() + .find_map(|line| { + let value = line + .strip_prefix("Range:") + .or_else(|| line.strip_prefix("range:"))?; + let spec = value.trim().strip_prefix("bytes=")?; + spec.split('-').next()?.trim().parse::().ok() + }) + .unwrap_or(0) +} + +const RESUME_BODY: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; + +/// The fix for FastLED/fbuild#1370: a dropped connection costs only the +/// bytes it did not deliver. +/// +/// Without resume this needs the server to send a complete body in one +/// attempt, which is exactly what the reporter's connection could not do. +/// With it, two attempts finish the file and the second one asks for the +/// remainder rather than starting over. +#[tokio::test] +async fn streaming_download_resumes_from_the_byte_offset_after_a_drop() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_resuming_server(RESUME_BODY, 10, true, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut seen: Vec = Vec::new(); + let mut progress = |p: &DownloadProgress| seen.push(p.downloaded); + + download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect("a ranged retry should finish the download"); + + assert_eq!( + std::fs::read(temp.path().join("file")).unwrap(), + RESUME_BODY, + "the resumed file must be byte-identical to the source" + ); + assert_eq!( + request_count.load(Ordering::SeqCst), + 2, + "one dropped attempt plus one ranged resume" + ); + assert!( + !temp.path().join("file.part").exists(), + "the part file must be renamed away, not left behind" + ); + assert!( + seen.iter().all(|d| *d <= RESUME_BODY.len() as u64), + "reported progress must stay cumulative rather than restarting: {seen:?}" + ); +} + +/// An origin that ignores `Range` must not corrupt the partial file, and +/// must still terminate rather than looping forever. +/// +/// The second half of #1370 is that retries which make no progress cannot +/// converge. Here every attempt lands on the same byte, so the +/// no-progress budget is what stops it. +#[tokio::test] +async fn streaming_download_gives_up_when_the_server_ignores_range() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_resuming_server(RESUME_BODY, 10, false, request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_p: &DownloadProgress| {}; + + let error = download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect_err("a server that never sends the tail must fail, not hang"); + + let message = error.to_string(); + assert!( + message.contains("byte offset"), + "the failure must name the offset it stopped at, per #1370: {message}" + ); + assert!( + !temp.path().join("file").exists(), + "no truncated archive may be left at the destination" + ); + assert!( + !temp.path().join("file.part").exists(), + "the part file must be cleaned up on hard failure" + ); + // One attempt makes progress (0 -> 10), then every later attempt + // re-sends the same prefix, so the no-progress budget ends it. + assert!( + request_count.load(Ordering::SeqCst) >= MAX_STALLED_ATTEMPTS as usize, + "should have spent the no-progress budget" + ); +} + +#[test] +fn content_range_start_and_total_are_parsed() { + // Exercised through the public shape rather than a Response, which + // cannot be constructed here: the header grammar is the fragile part. + assert_eq!(parse_request_range("Range: bytes=1234-\r\n"), 1234); + assert_eq!(parse_request_range("range: bytes=0-\r\n"), 0); + assert_eq!(parse_request_range("GET / HTTP/1.1\r\n"), 0); +} + +/// Retry timings short enough to run in real time. This test previously +/// used `#[tokio::test(start_paused = true)]` against a real +/// `TcpListener`, which flaked on loaded macOS runners: paused time +/// auto-advances whenever the runtime looks idle, but socket readiness +/// comes from the OS reactor, so the clock could jump past a connection +/// that was about to reach `accept()` — leaving `request_count` short of +/// 5. Real durations remove the race entirely (FastLED/fbuild#1222). +const FAST_RETRY_TIMING: RetryTiming = RetryTiming { + chunk_read_timeout: Duration::from_millis(150), + backoffs: &[ + Duration::from_millis(10), + Duration::from_millis(10), + Duration::from_millis(10), + Duration::from_millis(10), + ], +}; + +#[tokio::test] +async fn streaming_download_retries_chunk_stalls_five_times_without_output() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_stalling_server(request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_progress: &DownloadProgress| {}; + + let _err = download_file_with_progress_timed( + &test_client(), + &url, + temp.path(), + &mut progress, + FAST_RETRY_TIMING, + ) + .await + .expect_err("five chunk stalls should exhaust retries"); + + // A stalled body is retryable, as is a connection error while opening + // a retry. The latter can legitimately be the final transient on a + // busy platform, so the stable contract is exhausting all five + // attempts without publishing an output. + assert_eq!(request_count.load(Ordering::SeqCst), 5); + assert!(!temp.path().join("file").exists()); +} + +/// The injected timings are a test seam, not a behavior change: the +/// production path must still carry the real constants. +#[test] +fn production_retry_timing_matches_the_constants() { + assert_eq!( + RetryTiming::PRODUCTION.chunk_read_timeout, + CHUNK_READ_TIMEOUT + ); + assert_eq!(RetryTiming::PRODUCTION.backoffs, RETRY_BACKOFFS); + for attempt in 1..MAX_ATTEMPTS { + assert_eq!( + RetryTiming::PRODUCTION.backoff(attempt), + RETRY_BACKOFFS[(attempt - 1) as usize] + ); + } +} + +#[test] +fn format_download_progress_with_total() { + let p = DownloadProgress { + downloaded: 50 * 1024 * 1024, + total_bytes: Some(150 * 1024 * 1024), + filename: "toolchain.tar.gz".into(), + }; + let msg = p.format_message(); + assert!(msg.contains("50"), "msg: {msg}"); + assert!(msg.contains("150"), "msg: {msg}"); + assert!(msg.contains("33%"), "msg: {msg}"); +} + +#[test] +fn format_download_progress_without_total() { + let p = DownloadProgress { + downloaded: 5 * 1024 * 1024, + total_bytes: None, + filename: "library.zip".into(), + }; + let msg = p.format_message(); + assert!(msg.contains("5"), "msg: {msg}"); + assert!(!msg.contains("%"), "msg: {msg}"); +} + +#[test] +fn format_download_progress_zero() { + let p = DownloadProgress { + downloaded: 0, + total_bytes: Some(100 * 1024 * 1024), + filename: "file.bin".into(), + }; + let msg = p.format_message(); + assert!(msg.contains("0%"), "msg: {msg}"); +} From 1a0fd831d0085753d5f5782cfa32c579e67d7f72 Mon Sep 17 00:00:00 2001 From: zackees Date: Sun, 23 Aug 2026 03:11:55 -0700 Subject: [PATCH 3/3] fix(download): never append a 416 body, and bound the retry loop absolutely MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings, all correct, one of which I introduced as a data- corruption bug. ## A 416 error body was appended to the completed file (Critical) `open_attempt_from` returned `Ok` for `416` with `starts_at == offset`, so `fetch_into_part` opened the part file in append mode and streamed the response body onto the end of it. S3, GCS and several CDNs answer `416` with an XML or HTML error document, so those bytes landed in the archive — and because they push the length *past* the expected total, the short-body check could not catch it either. With a checksum the install would fail after a full download; without one, a corrupt archive would be extracted. `OpenedRange` now carries `already_complete`, and `fetch_into_part` returns before touching the file. RED/GREEN confirmed: with the early return removed, the new test shows `InvalidRange` appended to the file. ## The stall budget was not a bound (Major) `stalled` resets on any progress, and one byte counts. A server that drops the connection after a handful of bytes resets it every time, so the loop had no total attempt count, no byte-rate floor, and no deadline — and there is no outer timeout above this function, so that is a wedged install rather than a failed one. The same non-terminating shape #1370 reported, reached from the opposite direction. Added `MAX_TOTAL_ATTEMPTS = 40` as an absolute ceiling, sized so a genuinely converging download still finishes: the reporter's case needs four. The error now says which bound stopped it, alongside the byte offset. ## The header grammar had no test (Major) The test I added covered `parse_request_range` — the *mock's* helper — not `parse_content_range`, the production parser. Split the grammar into `parse_content_range_value(&str)` and tested it directly: `*` totals (legal when the origin does not know the size), a missing `bytes ` prefix, a range with no `/total`, and a non-numeric start. Refs FastLED/fbuild#1370 Co-Authored-By: Claude Opus 5 (1M context) --- .../fbuild-packages-fetch/src/downloader.rs | 61 ++++- .../src/downloader_tests.rs | 235 +++++++++++++++++- 2 files changed, 283 insertions(+), 13 deletions(-) diff --git a/crates/fbuild-packages-fetch/src/downloader.rs b/crates/fbuild-packages-fetch/src/downloader.rs index 0b566ebd..64ecf373 100644 --- a/crates/fbuild-packages-fetch/src/downloader.rs +++ b/crates/fbuild-packages-fetch/src/downloader.rs @@ -173,6 +173,15 @@ struct OpenedRange { starts_at: u64, /// Total size of the complete resource, when the server disclosed it. total: Option, + /// The server has nothing left to send (a `416` answer to our range). + /// + /// Carried as a flag rather than an empty body because a `416` response + /// usually *has* a body — S3, GCS and several CDNs send XML or HTML + /// explaining the error. Streaming that onto the end of an already + /// complete file would corrupt it, and because those bytes push the + /// length past the expected total, the short-body check would not catch + /// it either. + already_complete: bool, } /// GET `url`, asking to resume from `offset` when that is non-zero. @@ -203,6 +212,7 @@ async fn open_attempt_from( response, starts_at: offset, total: Some(offset), + already_complete: true, }); } if !status.is_success() { @@ -215,6 +225,7 @@ async fn open_attempt_from( response, starts_at, total, + already_complete: false, }); } @@ -224,6 +235,7 @@ async fn open_attempt_from( response, starts_at: 0, total, + already_complete: false, }) } @@ -237,6 +249,16 @@ fn parse_content_range(response: &reqwest::Response) -> Option<(u64, Option .get(reqwest::header::CONTENT_RANGE)? .to_str() .ok()?; + parse_content_range_value(value) +} + +/// Parse the value half of `Content-Range: bytes -/`. +/// +/// Split from the header lookup so the grammar — the fragile part — is +/// directly testable. `` may be `*` when the origin does not know the +/// full size, which is legal and yields `None` for the total rather than +/// failing the parse. +fn parse_content_range_value(value: &str) -> Option<(u64, Option)> { let spec = value.trim().strip_prefix("bytes ")?; let (range, total) = spec.split_once('/')?; let start = range.split_once('-')?.0.trim().parse::().ok()?; @@ -391,6 +413,20 @@ async fn download_file_with_progress_using( /// stuck still stops promptly. const MAX_STALLED_ATTEMPTS: u32 = 5; +/// Absolute ceiling on attempts, whatever progress is being made. +/// +/// The stall budget alone is not a bound: a server that drops the connection +/// after a handful of bytes resets it every single time, and the loop would +/// run forever at a byte rate no operator would accept. There is no outer +/// timeout above this function, so "forever" means a wedged install rather +/// than a failed one — the same non-terminating shape FastLED/fbuild#1370 +/// reported, reached from the opposite direction. +/// +/// Sized so a genuinely converging download still finishes: the reporter's +/// 282 MB archive over a link dying at ~90 MB needs four attempts, and this +/// leaves an order of magnitude of headroom. +const MAX_TOTAL_ATTEMPTS: u32 = 40; + /// [`download_file_with_progress_using`] with the retry durations injected. /// See [`RetryTiming`] for why tests need this instead of paused Tokio time. /// @@ -460,19 +496,29 @@ async fn download_file_with_progress_timed( stalled += 1; } - if error.is_retryable() && stalled < MAX_STALLED_ATTEMPTS { + let exhausted = if stalled >= MAX_STALLED_ATTEMPTS { + Some(format!("{stalled} consecutive attempts made no progress")) + } else if attempt >= MAX_TOTAL_ATTEMPTS { + Some(format!( + "hit the {MAX_TOTAL_ATTEMPTS}-attempt ceiling while making only intermittent progress" + )) + } else { + None + }; + + if error.is_retryable() && exhausted.is_none() { wait_before_retry(url, stalled.max(1), &error, timing).await; continue; } let _ = tokio::fs::remove_file(&part_path).await; + let reason = exhausted.unwrap_or_else(|| "error is not retryable".to_string()); return Err(FbuildError::PackageError(format!( - "{} (gave up after {} attempts at byte offset {}; \ - {} consecutive attempts made no progress)", + "{} (gave up after {} attempts at byte offset {}; {})", error.into_fbuild_error(url), attempt, resume_from, - stalled + reason ))); } } @@ -531,8 +577,15 @@ async fn fetch_into_part( mut response, starts_at, total, + already_complete, } = opened; + // Nothing left to fetch. Return before touching the file: the `416` + // response body is an error document, not resource bytes. + if already_complete { + return Ok(()); + } + // The server declined the range and restarted the body. Anything already // written is now the wrong prefix, so drop it rather than append. let appending = starts_at == offset && offset > 0; diff --git a/crates/fbuild-packages-fetch/src/downloader_tests.rs b/crates/fbuild-packages-fetch/src/downloader_tests.rs index 5b2d2955..3e903d92 100644 --- a/crates/fbuild-packages-fetch/src/downloader_tests.rs +++ b/crates/fbuild-packages-fetch/src/downloader_tests.rs @@ -512,15 +512,6 @@ async fn streaming_download_gives_up_when_the_server_ignores_range() { ); } -#[test] -fn content_range_start_and_total_are_parsed() { - // Exercised through the public shape rather than a Response, which - // cannot be constructed here: the header grammar is the fragile part. - assert_eq!(parse_request_range("Range: bytes=1234-\r\n"), 1234); - assert_eq!(parse_request_range("range: bytes=0-\r\n"), 0); - assert_eq!(parse_request_range("GET / HTTP/1.1\r\n"), 0); -} - /// Retry timings short enough to run in real time. This test previously /// used `#[tokio::test(start_paused = true)]` against a real /// `TcpListener`, which flaked on loaded macOS runners: paused time @@ -617,3 +608,229 @@ fn format_download_progress_zero() { let msg = p.format_message(); assert!(msg.contains("0%"), "msg: {msg}"); } + +/// A server that always answers a ranged request with `416` **and a body**. +/// +/// S3, GCS and several CDNs do exactly this. The body is an error document, +/// not resource bytes, so appending it would corrupt a file that was already +/// complete — and because those bytes push the length past the expected +/// total, the short-body check cannot catch it either. +async fn run_range_not_satisfiable_server( + body: &'static [u8], + first_len: usize, + request_count: std::sync::Arc, +) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + const ERROR_DOC: &str = "InvalidRange"; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + let _ = ready_tx.send(()); + loop { + let (mut stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + let mut buf = [0u8; 2048]; + let read = stream.read(&mut buf).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buf[..read]).to_string(); + + if parse_request_range(&request) > 0 { + let head = format!( + "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Length: {}\r\n\r\n", + ERROR_DOC.len() + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(ERROR_DOC.as_bytes()).await; + } else { + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nAccept-Ranges: bytes\r\n\r\n", + first_len + ); + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(&body[..first_len]).await; + } + let _ = stream.shutdown().await; + } + }); + ready_rx.await.expect("416 test server should start"); + port +} + +/// A `416` answer must finalize the file, never append its error body. +#[tokio::test] +async fn streaming_download_treats_416_as_complete_without_appending_its_body() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + // The server announces exactly what it sends, so the first attempt is a + // complete download by its own account; a second, ranged request is only + // made if something retries. Force that by having the body be short of + // RESUME_BODY and letting the truncation check fire. + let port = + run_range_not_satisfiable_server(RESUME_BODY, RESUME_BODY.len(), request_count.clone()) + .await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_p: &DownloadProgress| {}; + + download_file_with_progress_using(&test_client(), &url, temp.path(), &mut progress) + .await + .expect("a complete first response should succeed"); + + let written = std::fs::read(temp.path().join("file")).unwrap(); + assert_eq!( + written, RESUME_BODY, + "the file must be exactly the resource, with no error document appended" + ); +} + +/// `416` reached through the resume path: the part file is already complete, +/// and the error body that comes with the status must not reach it. +#[tokio::test] +async fn a_416_response_is_not_written_to_the_part_file() { + let _guard = network_test_guard().await; + let temp = tempfile::TempDir::new().unwrap(); + let part = temp.path().join("file.part"); + std::fs::write(&part, RESUME_BODY).unwrap(); + + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = + run_range_not_satisfiable_server(RESUME_BODY, RESUME_BODY.len(), request_count.clone()) + .await; + let url = format!("http://127.0.0.1:{port}/file"); + let mut progress = |_p: &DownloadProgress| {}; + + fetch_into_part( + &test_client(), + &url, + &part, + RESUME_BODY.len() as u64, + "file", + &mut progress, + FAST_RETRY_TIMING, + ) + .await + .expect("416 means the resource is already complete, which is success"); + + assert_eq!( + std::fs::read(&part).unwrap(), + RESUME_BODY, + "the 416 error document must not be appended to the completed part file" + ); +} + +/// Progress alone must not license an unbounded loop. +/// +/// This server hands back one byte per attempt, so the stall budget never +/// fires — every attempt "makes progress". Only the absolute ceiling ends it, +/// and without that the install would hang rather than fail. +#[tokio::test] +async fn streaming_download_stops_at_the_absolute_attempt_ceiling() { + let _guard = network_test_guard().await; + let request_count = std::sync::Arc::new(AtomicUsize::new(0)); + let port = run_one_byte_at_a_time_server(request_count.clone()).await; + let url = format!("http://127.0.0.1:{port}/file"); + let temp = tempfile::TempDir::new().unwrap(); + let mut progress = |_p: &DownloadProgress| {}; + + let error = download_file_with_progress_timed( + &test_client(), + &url, + temp.path(), + &mut progress, + FAST_RETRY_TIMING, + ) + .await + .expect_err("a drip-feeding server must hit the ceiling, not run forever"); + + let message = error.to_string(); + assert!( + message.contains("ceiling"), + "the failure must say the absolute bound stopped it: {message}" + ); + assert_eq!( + request_count.load(Ordering::SeqCst), + MAX_TOTAL_ATTEMPTS as usize, + "should stop exactly at the ceiling" + ); + assert!(!temp.path().join("file").exists()); + assert!(!temp.path().join("file.part").exists()); +} + +/// Serves one byte per request, always claiming a much larger total. +async fn run_one_byte_at_a_time_server(request_count: std::sync::Arc) -> u16 { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + + tokio::spawn(async move { + let _ = ready_tx.send(()); + loop { + let (mut stream, _) = match listener.accept().await { + Ok(pair) => pair, + Err(_) => break, + }; + request_count.fetch_add(1, Ordering::SeqCst); + let mut buf = [0u8; 2048]; + let read = stream.read(&mut buf).await.unwrap_or(0); + let request = String::from_utf8_lossy(&buf[..read]).to_string(); + let start = parse_request_range(&request); + + // Always honor the range and always deliver exactly one byte, so + // the file advances forever and the stall budget never trips. + let total = 1_000_000usize; + let head = if start > 0 { + format!( + "HTTP/1.1 206 Partial Content\r\nContent-Length: {}\r\n\ + Content-Range: bytes {}-{}/{}\r\nAccept-Ranges: bytes\r\n\r\n", + total - start, + start, + total - 1, + total + ) + } else { + format!( + "HTTP/1.1 200 OK\r\nContent-Length: {total}\r\nAccept-Ranges: bytes\r\n\r\n" + ) + }; + let _ = stream.write_all(head.as_bytes()).await; + let _ = stream.write_all(b"x").await; + let _ = stream.shutdown().await; + } + }); + ready_rx.await.expect("drip server should start"); + port +} + +#[test] +fn content_range_values_are_parsed() { + assert_eq!( + parse_content_range_value("bytes 100-199/200"), + Some((100, Some(200))) + ); + // A `*` total is legal when the origin does not know the full size: the + // start still parses, the total is simply unknown. + assert_eq!( + parse_content_range_value("bytes 100-199/*"), + Some((100, None)) + ); + assert_eq!( + parse_content_range_value(" bytes 0-9/10 "), + Some((0, Some(10))) + ); + // Malformed inputs must yield None so the caller falls back to the + // offset it asked for, rather than trusting a garbage start. + assert_eq!(parse_content_range_value("items 100-199/200"), None); + assert_eq!(parse_content_range_value("bytes 100-199"), None); + assert_eq!(parse_content_range_value("bytes abc-199/200"), None); + assert_eq!(parse_content_range_value(""), None); +}