diff --git a/CHANGELOG.md b/CHANGELOG.md index 3572a298..7b1aae73 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -236,6 +236,10 @@ limits, and required install commands. half-open connection. A connect now fails after 10 s, and a connection that sends nothing for 60 s fails as a network error. Downloads that keep streaming are not cut off (#570). +- Patch blob and diff downloads stream straight to the `.socket` cache instead + of being held in memory whole first, so a large patch artifact no longer + costs its full size in RAM during `apply`, `get`, `repair` or `rollback` + (#571). ### Maintenance diff --git a/crates/socket-patch-core/src/api/blob_fetcher.rs b/crates/socket-patch-core/src/api/blob_fetcher.rs index e1afdcf4..86a51e19 100644 --- a/crates/socket-patch-core/src/api/blob_fetcher.rs +++ b/crates/socket-patch-core/src/api/blob_fetcher.rs @@ -1,7 +1,8 @@ use std::collections::HashSet; use std::path::Path; -use crate::api::client::ApiClient; +use crate::api::client::{ApiClient, ApiError, BinaryBody}; +use crate::hash::git_sha256::compute_git_sha256_from_reader; use crate::manifest::operations::get_after_hash_blobs; use crate::manifest::schema::PatchManifest; use crate::patch::apply::PatchSources; @@ -113,10 +114,10 @@ pub async fn fetch_missing_blobs( } // `blobs_path` is created by the first successful write - // (`write_cache_entry_atomic`), never up front: a fetch that lands + // (`stream_cache_entry_atomic`), never up front: a fetch that lands // nothing leaves no `.socket/blobs/` husk behind. let hashes: Vec = missing.into_iter().collect(); - download_hashes(&hashes, blobs_path, client, on_progress).await + download_entries(&hashes, blobs_path, client, on_progress, Entry::Blob).await } /// Download specific blobs identified by their hashes. @@ -166,7 +167,8 @@ pub async fn fetch_blobs_by_hash( }; } - let download_result = download_hashes(&to_download, blobs_path, client, on_progress).await; + let download_result = + download_entries(&to_download, blobs_path, client, on_progress, Entry::Blob).await; results.extend(download_result.results); FetchMissingBlobsResult { @@ -239,66 +241,7 @@ async fn fetch_missing_diff_archives( // `archives_dir` is created by the first successful write, never up // front (see `fetch_missing_blobs`). let uuids: Vec = missing.into_iter().collect(); - let total = uuids.len(); - let mut downloaded = 0usize; - let mut failed = 0usize; - let mut results = Vec::with_capacity(total); - - for (i, uuid) in uuids.iter().enumerate() { - if let Some(ref cb) = on_progress { - cb(uuid, i + 1, total); - } - - let fetch_result = client.fetch_diff(uuid).await; - - match fetch_result { - Ok(Some(data)) => { - let archive_path = archives_dir.join(format!("{}.tar.gz", uuid)); - match write_cache_entry_atomic(&archive_path, &data).await { - Ok(()) => { - results.push(BlobFetchResult { - hash: uuid.clone(), - success: true, - error: None, - }); - downloaded += 1; - } - Err(e) => { - results.push(BlobFetchResult { - hash: uuid.clone(), - success: false, - error: Some(format!("Failed to write archive to disk: {}", e)), - }); - failed += 1; - } - } - } - Ok(None) => { - results.push(BlobFetchResult { - hash: uuid.clone(), - success: false, - error: Some("Diff archive not found on server".to_string()), - }); - failed += 1; - } - Err(e) => { - results.push(BlobFetchResult { - hash: uuid.clone(), - success: false, - error: Some(e.to_string()), - }); - failed += 1; - } - } - } - - FetchMissingBlobsResult { - total, - downloaded, - failed, - skipped: 0, - results, - } + download_entries(&uuids, archives_dir, client, on_progress, Entry::Diff).await } /// What kind of artifact a fetch or cleanup result counts, for human @@ -448,8 +391,10 @@ fn concise_fetch_error<'a>(err: &'a str, id: &str) -> std::borrow::Cow<'a, str> // ── Internal helpers ────────────────────────────────────────────────── -/// Write `bytes` to `dest` atomically: stage a temp file in the same -/// directory, then `rename(2)` it over `dest`. +/// Stream `body` to `dest` atomically: copy it chunk by chunk into a temp +/// file in the same directory, check it against `expected_hash` (a blob's +/// git-sha256 name) when given, then `rename(2)` it over `dest`. The body is +/// never held in memory whole (#571). /// /// The destinations here are *content-addressed* cache entries — /// `blobs/` and `archives/.tar.gz`. A plain `tokio::fs::write` @@ -468,127 +413,189 @@ fn concise_fetch_error<'a>(err: &'a str, id: &str) -> std::borrow::Cow<'a, str> /// content-addressed cache entries, not user-owned files — post-crash loss /// of a cache entry is harmless, so the extra durability isn't worth the /// I/O. Do not "consolidate" this into the hardened writer. -async fn write_cache_entry_atomic(dest: &Path, bytes: &[u8]) -> std::io::Result<()> { +async fn stream_cache_entry_atomic( + dest: &Path, + body: &mut BinaryBody, + expected_hash: Option<&str>, +) -> Result<(), EntryError> { let parent = dest.parent().ok_or_else(|| { - std::io::Error::new( + EntryError::Write(std::io::Error::new( std::io::ErrorKind::InvalidInput, "cache entry path has no parent directory", - ) + )) })?; let stem = dest .file_name() .map(|n| n.to_string_lossy().into_owned()) .unwrap_or_else(|| "blob".to_string()); // The cache directory (`.socket/blobs/`, `.socket/diffs/`) is created - // here, on the first verified download, and nowhere earlier: a fetch - // that lands nothing (all 404, offline, every hash mismatched) must not - // leave an empty directory behind for the user to commit. An - // uncreatable parent surfaces as this entry's write failure, like any - // other disk error. - tokio::fs::create_dir_all(parent).await?; + // here, by the first download, and nowhere earlier. A fetch that lands + // nothing (all 404, offline, every hash mismatched, every body cut + // short) must not leave an empty directory behind for the user to + // commit, so a failure removes the directories this call created, while + // they are still empty. An uncreatable parent surfaces as this entry's + // write failure, like any other disk error. + let mut created_dirs = Vec::new(); + for dir in parent.ancestors() { + if dir.as_os_str().is_empty() || tokio::fs::symlink_metadata(dir).await.is_ok() { + break; + } + created_dirs.push(dir); + } // Leading dot keeps the stage out of editor/glob views; the uuid suffix // keeps concurrent writers of the same entry from colliding. let stage = parent.join(format!(".socket-dl-{}-{}", stem, uuid::Uuid::new_v4())); - if let Err(e) = tokio::fs::write(&stage, bytes).await { + let result = async { + // Inside the cleanup scope: a `create_dir_all` that makes some + // ancestors and then fails must not leave them behind either. + tokio::fs::create_dir_all(parent) + .await + .map_err(EntryError::Write)?; + let size = stage_body(&stage, body).await?; + if let Some(expected) = expected_hash { + let file = tokio::fs::File::open(&stage) + .await + .map_err(EntryError::Write)?; + let actual = compute_git_sha256_from_reader(size, file) + .await + .map_err(EntryError::Write)?; + if !blob_hash_matches(expected, &actual) { + return Err(EntryError::HashMismatch(actual)); + } + } + tokio::fs::rename(&stage, dest) + .await + .map_err(EntryError::Write) + } + .await; + if result.is_err() { // A partial stage would otherwise leak as a `.socket-dl-*` turd. let _ = tokio::fs::remove_file(&stage).await; - return Err(e); + // Deepest first; `remove_dir` only succeeds while a dir is empty. + for dir in created_dirs { + let _ = tokio::fs::remove_dir(dir).await; + } } - if let Err(e) = tokio::fs::rename(&stage, dest).await { - let _ = tokio::fs::remove_file(&stage).await; - return Err(e); + result +} + +/// Copy `body` chunk by chunk into a new file at `stage`, returning the +/// byte count. Only one chunk is held in memory at a time. +async fn stage_body(stage: &Path, body: &mut BinaryBody) -> Result { + use tokio::io::AsyncWriteExt; + let mut file = tokio::fs::File::create(stage) + .await + .map_err(EntryError::Write)?; + let mut size: u64 = 0; + while let Some(chunk) = body.chunk().await.map_err(EntryError::Body)? { + let chunk = chunk.as_ref(); + file.write_all(chunk).await.map_err(EntryError::Write)?; + size += chunk.len() as u64; } - Ok(()) + file.flush().await.map_err(EntryError::Write)?; + Ok(size) +} + +/// Why one cache entry was not stored. +#[derive(Debug)] +enum EntryError { + /// The response body failed or stalled mid-read. + Body(ApiError), + /// A disk error while staging, hashing or renaming. + Write(std::io::Error), + /// The blob's content hashed to this, not to its name. + HashMismatch(String), } /// Compare an expected blob hash against the hash computed from the /// downloaded bytes. /// /// Git object hashes are hex, and hex is case-insensitive. The content -/// hasher ([`compute_git_sha256_from_bytes`]) always emits lowercase, but +/// hasher ([`compute_git_sha256_from_reader`]) always emits lowercase, but /// [`ApiClient::fetch_blob`]'s validator accepts uppercase hex too — so a /// manifest (or server) that uses uppercase would download byte-for-byte /// correct content and then be wrongly rejected by a case-sensitive /// comparison. Compare ignoring ASCII case to keep the two consistent. /// -/// [`compute_git_sha256_from_bytes`]: crate::hash::git_sha256::compute_git_sha256_from_bytes +/// [`compute_git_sha256_from_reader`]: crate::hash::git_sha256::compute_git_sha256_from_reader fn blob_hash_matches(expected: &str, actual: &str) -> bool { expected.eq_ignore_ascii_case(actual) } -/// Download a list of blob hashes sequentially, writing each to -/// `blobs_path/`. -async fn download_hashes( - hashes: &[String], - blobs_path: &Path, +/// The two kinds of cache entry [`download_entries`] stores. +#[derive(Debug, Clone, Copy)] +enum Entry { + /// `blobs/`, verified against its git-sha256 name. + Blob, + /// `diffs/.tar.gz`, stored as served. + Diff, +} + +/// Download `ids` sequentially, streaming each into its cache entry under +/// `dir` (see [`stream_cache_entry_atomic`]). The one download loop behind +/// [`fetch_missing_blobs`], [`fetch_blobs_by_hash`] and diff-mode +/// [`fetch_missing_sources`]. +async fn download_entries( + ids: &[String], + dir: &Path, client: &ApiClient, on_progress: Option<&OnProgress>, + entry: Entry, ) -> FetchMissingBlobsResult { - let total = hashes.len(); + let total = ids.len(); let mut downloaded: usize = 0; let mut failed: usize = 0; let mut results: Vec = Vec::with_capacity(total); - for (i, hash) in hashes.iter().enumerate() { + for (i, id) in ids.iter().enumerate() { if let Some(ref cb) = on_progress { - cb(hash, i + 1, total); + cb(id, i + 1, total); } - match client.fetch_blob(hash).await { - Ok(Some(data)) => { - // Verify content hash matches expected hash before writing - let actual_hash = crate::hash::git_sha256::compute_git_sha256_from_bytes(&data); - if !blob_hash_matches(hash, &actual_hash) { - results.push(BlobFetchResult { - hash: hash.clone(), - success: false, - error: Some(format!( - "Content hash mismatch: expected {}, got {}", - hash, actual_hash - )), - }); - failed += 1; - continue; - } - - let blob_path = blobs_path.join(hash); - match write_cache_entry_atomic(&blob_path, &data).await { - Ok(()) => { - results.push(BlobFetchResult { - hash: hash.clone(), - success: true, - error: None, - }); - downloaded += 1; - } - Err(e) => { - results.push(BlobFetchResult { - hash: hash.clone(), - success: false, - error: Some(format!("Failed to write blob to disk: {}", e)), - }); - failed += 1; + let (fetched, dest, expected_hash, noun, not_found) = match entry { + Entry::Blob => ( + client.fetch_blob(id).await, + dir.join(id), + Some(id.as_str()), + "blob", + "Blob not found on server", + ), + Entry::Diff => ( + client.fetch_diff(id).await, + dir.join(format!("{}.tar.gz", id)), + None, + "archive", + "Diff archive not found on server", + ), + }; + let error = match fetched { + Ok(Some(mut body)) => { + match stream_cache_entry_atomic(&dest, &mut body, expected_hash).await { + Ok(()) => None, + Err(EntryError::Body(e)) => Some(e.to_string()), + Err(EntryError::Write(e)) => { + Some(format!("Failed to write {} to disk: {}", noun, e)) } + Err(EntryError::HashMismatch(actual)) => Some(format!( + "Content hash mismatch: expected {}, got {}", + id, actual + )), } } - Ok(None) => { - results.push(BlobFetchResult { - hash: hash.clone(), - success: false, - error: Some("Blob not found on server".to_string()), - }); - failed += 1; - } - Err(e) => { - results.push(BlobFetchResult { - hash: hash.clone(), - success: false, - error: Some(e.to_string()), - }); - failed += 1; - } + Ok(None) => Some(not_found.to_string()), + Err(e) => Some(e.to_string()), + }; + if error.is_none() { + downloaded += 1; + } else { + failed += 1; } + results.push(BlobFetchResult { + hash: id.clone(), + success: error.is_none(), + error, + }); } FetchMissingBlobsResult { @@ -1036,6 +1043,34 @@ mod tests { // ── Atomic cache-entry write ───────────────────────────────────── + /// A [`BinaryBody`] streaming `bytes` from a local mock server. The + /// server is returned so it outlives the read. + async fn served_body(bytes: &[u8]) -> (wiremock::MockServer, BinaryBody) { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::any()) + .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(bytes.to_vec())) + .mount(&server) + .await; + let client = ApiClient::new(crate::api::client::ApiClientOptions { + api_url: server.uri(), + api_token: None, + use_public_proxy: true, + org_slug: None, + }); + let body = client + .fetch_diff("11111111-1111-4111-8111-111111111111") + .await + .unwrap() + .expect("200 serves a body"); + (server, body) + } + + /// [`stream_cache_entry_atomic`] with no hash check, over `bytes`. + async fn write_cache_entry_atomic(dest: &Path, bytes: &[u8]) -> Result<(), EntryError> { + let (_server, mut body) = served_body(bytes).await; + stream_cache_entry_atomic(dest, &mut body, None).await + } + #[tokio::test] async fn test_write_cache_entry_atomic_writes_exact_bytes_no_litter() { let dir = tempfile::tempdir().unwrap(); @@ -1128,7 +1163,10 @@ mod tests { let result = write_cache_entry_atomic(&ro.join("x"), b"bytes").await; let err = result.expect_err("stage write into a read-only dir must fail"); - assert_eq!(err.kind(), std::io::ErrorKind::PermissionDenied); + assert!( + matches!(&err, EntryError::Write(e) if e.kind() == std::io::ErrorKind::PermissionDenied), + "{err:?}" + ); // Restore before asserting/teardown so cleanup cannot mask failure. std::fs::set_permissions(&ro, std::fs::Permissions::from_mode(0o755)).unwrap(); diff --git a/crates/socket-patch-core/src/api/client.rs b/crates/socket-patch-core/src/api/client.rs index 7d862c05..2959a243 100644 --- a/crates/socket-patch-core/src/api/client.rs +++ b/crates/socket-patch-core/src/api/client.rs @@ -175,6 +175,34 @@ pub async fn hold_back_debug(fut: impl std::future::Future) -> He HeldBack { value, debug } } +/// The body of a 200 blob or diff response, read chunk by chunk. +/// +/// [`ApiClient::fetch_blob`] / [`ApiClient::fetch_diff`] return this instead +/// of the whole body so a large patch artifact streams to disk without being +/// held in memory (#571). The per-read idle bound of [`ApiTimeouts`] applies +/// to every chunk. +#[derive(Debug)] +pub struct BinaryBody { + resp: reqwest::Response, + kind: String, + identifier: String, +} + +impl BinaryBody { + /// The next chunk of the body, or `Ok(None)` once it is complete. A + /// failed or stalled read is an [`ApiError::Network`]. + pub async fn chunk(&mut self) -> Result>, ApiError> { + self.resp.chunk().await.map_err(|e| { + ApiError::Network(format!( + "Error reading {} body for {}: {}", + self.kind, + self.identifier, + network_error_detail(&e) + )) + }) + } +} + /// Options for constructing an [`ApiClient`]. #[derive(Debug, Clone)] pub struct ApiClientOptions { @@ -989,10 +1017,10 @@ impl ApiClient { /// Fetch a blob by its SHA-256 hash. /// - /// Returns the raw binary content, or `Ok(None)` if not found. - /// Uses the authenticated endpoint when token and org slug are - /// available, otherwise falls back to the public proxy. - pub async fn fetch_blob(&self, hash: &str) -> Result>, ApiError> { + /// Returns the response body as a [`BinaryBody`] stream, or `Ok(None)` + /// if not found. Uses the authenticated endpoint when token and org + /// slug are available, otherwise falls back to the public proxy. + pub async fn fetch_blob(&self, hash: &str) -> Result, ApiError> { // Validate hash format: SHA-256 = 64 hex characters if !is_valid_sha256_hex(hash) { return Err(ApiError::InvalidHash(format!( @@ -1005,10 +1033,11 @@ impl ApiClient { /// Fetch a per-file diff archive (tar.gz of bsdiff deltas) by patch UUID. /// - /// Returns the raw archive bytes, or `Ok(None)` if not found (404). The - /// public proxy serves these under `/patch/diff/`; the - /// authenticated API serves them under `/v0/orgs//patches/diff/`. - pub async fn fetch_diff(&self, uuid: &str) -> Result>, ApiError> { + /// Returns the archive body as a [`BinaryBody`] stream, or `Ok(None)` if + /// not found (404). The public proxy serves these under + /// `/patch/diff/`; the authenticated API serves them under + /// `/v0/orgs//patches/diff/`. + pub async fn fetch_diff(&self, uuid: &str) -> Result, ApiError> { if !is_valid_uuid(uuid) { return Err(ApiError::InvalidHash(format!( "Invalid patch UUID: {}", @@ -1062,12 +1091,13 @@ impl ApiClient { /// /// `kind` is the URL segment (`blob` / `diff`), doubling as the /// noun in log + error messages. `identifier` is the hash or UUID - /// interpolated into the URL. + /// interpolated into the URL. A 200 returns the unread body: callers + /// stream it to disk instead of buffering it whole. async fn fetch_binary( &self, kind: &str, identifier: &str, - ) -> Result>, ApiError> { + ) -> Result, ApiError> { let (url, use_auth) = self.binary_url(kind, identifier); debug_log(&format!("GET {} {}", kind, url)); @@ -1093,15 +1123,11 @@ impl ApiClient { let status = resp.status(); if status == StatusCode::OK { - let bytes = resp.bytes().await.map_err(|e| { - ApiError::Network(format!( - "Error reading {} body for {}: {}", - kind, - identifier, - network_error_detail(&e) - )) - })?; - return Ok(Some(bytes.to_vec())); + return Ok(Some(BinaryBody { + resp, + kind: kind.to_string(), + identifier: identifier.to_string(), + })); } if status == StatusCode::NOT_FOUND { return Ok(None); diff --git a/crates/socket-patch-core/tests/api_timeout_e2e.rs b/crates/socket-patch-core/tests/api_timeout_e2e.rs index 793f56df..54a436f9 100644 --- a/crates/socket-patch-core/tests/api_timeout_e2e.rs +++ b/crates/socket-patch-core/tests/api_timeout_e2e.rs @@ -11,7 +11,7 @@ use std::time::{Duration, Instant}; -use socket_patch_core::api::client::{ApiClient, ApiClientOptions, ApiError}; +use socket_patch_core::api::client::{ApiClient, ApiClientOptions, ApiError, BinaryBody}; use socket_patch_core::api::retry::{ ApiRetryPolicy, ApiTimeouts, RetryHooks, API_CONNECT_TIMEOUT, API_READ_TIMEOUT, }; @@ -228,6 +228,41 @@ async fn stalled_json_bodies_are_network_errors_on_both_clients() { ); } +/// Read a blob/diff body to the end. +async fn drain(mut body: BinaryBody) -> Result, ApiError> { + let mut bytes = Vec::new(); + while let Some(chunk) = body.chunk().await? { + bytes.extend_from_slice(chunk.as_ref()); + } + Ok(bytes) +} + +#[tokio::test] +async fn stalled_binary_bodies_are_network_errors_on_both_clients() { + // The headers arrive, then the body stalls: the idle bound fires on the + // streamed chunk read, not just on the wait for headers. + let uri = stalled_json_body_server().await; + for proxy in [false, true] { + let api = client(&uri, proxy); + for (what, body) in [ + ("fetch_blob", api.fetch_blob(HASH).await), + ("fetch_diff", api.fetch_diff(UUID).await), + ] { + let body = body + .unwrap_or_else(|e| panic!("{what} proxy={proxy}: headers arrived: {e:?}")) + .expect("200 serves a body"); + let error = tokio::time::timeout(GUARD, drain(body)) + .await + .unwrap_or_else(|_| panic!("{what} proxy={proxy}: stalled body unbounded")) + .expect_err("partial body must fail"); + assert!( + matches!(&error, ApiError::Network(message) if message.contains("timed out")), + "{what} proxy={proxy}: {error:?}" + ); + } + } +} + #[tokio::test] async fn completed_malformed_json_remains_a_parse_error() { let server = wiremock::MockServer::start().await; @@ -260,11 +295,13 @@ async fn a_body_that_keeps_streaming_past_the_bound_still_arrives() { proxy, ); let started = Instant::now(); - let body = tokio::time::timeout(GUARD, api.fetch_blob(HASH)) - .await - .expect("trickled blob still pending") - .expect("a streaming body must not time out") - .expect("200 is a blob"); + let body = tokio::time::timeout(GUARD, async { + let body = api.fetch_blob(HASH).await?.expect("200 is a blob"); + drain(body).await + }) + .await + .expect("trickled blob still pending") + .expect("a streaming body must not time out"); assert_eq!(body.len(), total, "proxy={proxy}"); assert!( started.elapsed() > READ * 3, diff --git a/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs b/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs index cc11c8dc..31ebb30f 100644 --- a/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs +++ b/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs @@ -341,7 +341,7 @@ async fn fetch_blobs_by_hash_mixes_skip_and_download_attempt() { // ── Content-hash verification (mock-server driven) ────────────────── // -// These drive the success and mismatch branches of `download_hashes`'s +// These drive the success and mismatch branches of `download_entries`'s // content verification, which the closed-port tests above can never reach // (they fail before any body is returned). The blob's name IS its // git-sha256, so the server must serve bytes that hash to the requested @@ -363,7 +363,7 @@ fn proxy_client(base: &str) -> ApiClient { } /// A blob whose content hashes to the requested name is written to disk -/// and counted as downloaded. Proves the happy path of `download_hashes`'s +/// and counted as downloaded. Proves the happy path of `download_entries`'s /// verify-then-write logic end to end. #[tokio::test] async fn fetch_missing_blobs_accepts_and_writes_matching_content() { @@ -732,3 +732,227 @@ async fn get_missing_blobs_reports_missing_afterhash() { "staged blob must not be reported missing" ); } + +// ── Streaming (#571) ───────────────────────────────────────────────── + +/// A one-response-per-connection server answering every request with +/// `200` and `Content-Length: head.len() + tail.len()`. It sends `head`, +/// then waits for `release` before sending `tail` (or, with `tail: None`, +/// closes the connection after `head`, cutting the body short). +async fn split_body_server( + head: Vec, + tail: Option>, + declared: usize, + release: std::sync::Arc, +) -> String { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let (head, tail, release) = (head.clone(), tail.clone(), release.clone()); + tokio::spawn(async move { + let mut request = Vec::new(); + let mut buf = [0u8; 4096]; + while !request.windows(4).any(|w| w == b"\r\n\r\n") { + match sock.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => request.extend_from_slice(&buf[..n]), + } + } + let headers = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/octet-stream\r\n\ + content-length: {declared}\r\nconnection: close\r\n\r\n" + ); + if sock.write_all(headers.as_bytes()).await.is_err() + || sock.write_all(&head).await.is_err() + || sock.flush().await.is_err() + { + return; + } + if let Some(tail) = tail { + release.notified().await; + let _ = sock.write_all(&tail).await; + } + let _ = sock.shutdown().await; + }); + } + }); + format!("http://{addr}") +} + +/// The size of the `.socket-dl-*` stage file in `dir`, if one exists. +/// +/// Sized through `std::fs::metadata(path)`, not `DirEntry::metadata`: on +/// Windows the latter reports the directory entry's cached size, which +/// NTFS does not update while the writer's handle is still open. +fn stage_len(dir: &Path) -> Option { + std::fs::read_dir(dir).ok()?.find_map(|e| { + let e = e.ok()?; + e.file_name() + .to_string_lossy() + .starts_with(".socket-dl-") + .then(|| std::fs::metadata(e.path()).ok().map(|m| m.len())) + .flatten() + }) +} + +/// Blob and diff downloads stream to disk: the first part of a body is +/// already in the stage file while the server is still holding back the +/// rest. Before #571 `fetch_binary` buffered the whole body in memory, so +/// nothing reached disk until the response completed. +#[tokio::test] +async fn blob_and_diff_bodies_reach_disk_before_the_response_completes() { + let head = vec![b'a'; 256 * 1024]; + let tail = vec![b'b'; 256 * 1024]; + let content = [head.clone(), tail.clone()].concat(); + let hash = compute_git_sha256_from_bytes(&content); + let uuid = "11111111-1111-4111-8111-111111111111"; + + for mode in [DownloadMode::File, DownloadMode::Diff] { + let release = std::sync::Arc::new(tokio::sync::Notify::new()); + let uri = split_body_server( + head.clone(), + Some(tail.clone()), + content.len(), + release.clone(), + ) + .await; + let tmp = tempfile::tempdir().unwrap(); + let blobs = tmp.path().join("blobs"); + let diffs = tmp.path().join("diffs"); + let (manifest, dir) = match mode { + DownloadMode::File => (manifest_with_after_hashes(&[&hash]), blobs.clone()), + DownloadMode::Diff => (manifest_with_uuids(&[uuid]), diffs.clone()), + }; + let sources = PatchSources { + blobs_path: &blobs, + diffs_path: Some(&diffs), + mem_blobs: None, + }; + let client = proxy_client(&uri); + + let watch = async { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + let mut streamed = false; + while std::time::Instant::now() < deadline { + if stage_len(&dir) == Some(head.len() as u64) { + streamed = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + // Release the tail either way so the download can finish. + release.notify_one(); + streamed + }; + let (result, streamed) = tokio::join!( + fetch_missing_sources(&manifest, &sources, mode, &client, None), + watch + ); + assert!( + streamed, + "{mode:?}: the first {} bytes must be on disk while the rest is held back", + head.len() + ); + assert_eq!(result.downloaded, 1, "{mode:?}: {:?}", result.results); + let entry = match mode { + DownloadMode::File => blobs.join(&hash), + DownloadMode::Diff => diffs.join(format!("{uuid}.tar.gz")), + }; + assert_eq!(std::fs::read(&entry).unwrap(), content, "{mode:?}"); + assert_eq!(dir_entry_count(&dir), 1, "{mode:?}: no stage litter"); + } +} + +/// A body cut short mid-stream fails that entry with the body-read error +/// and leaves nothing behind: no entry, no stage, and no cache directory +/// the download created. Same for a blob whose streamed content does not +/// hash to its name. +#[tokio::test] +async fn failed_streams_leave_no_stage_and_no_created_cache_dir() { + let content = vec![b'c'; 64 * 1024]; + let hash = compute_git_sha256_from_bytes(&content); + let uuid = "11111111-1111-4111-8111-111111111111"; + let release = std::sync::Arc::new(tokio::sync::Notify::new()); + + // Cut short: the server declares twice what it sends, then closes. + let uri = split_body_server(content.clone(), None, content.len() * 2, release.clone()).await; + for mode in [DownloadMode::File, DownloadMode::Diff] { + let tmp = tempfile::tempdir().unwrap(); + let blobs = tmp.path().join("blobs"); + let diffs = tmp.path().join("diffs"); + let manifest = match mode { + DownloadMode::File => manifest_with_after_hashes(&[&hash]), + DownloadMode::Diff => manifest_with_uuids(&[uuid]), + }; + let sources = PatchSources { + blobs_path: &blobs, + diffs_path: Some(&diffs), + mem_blobs: None, + }; + let result = + fetch_missing_sources(&manifest, &sources, mode, &proxy_client(&uri), None).await; + assert_eq!(result.failed, 1, "{mode:?}"); + let error = result.results[0].error.as_deref().unwrap(); + assert!(error.contains("Error reading"), "{mode:?}: {error}"); + assert!(!blobs.exists() && !diffs.exists(), "{mode:?}: no cache dir"); + } + + // Mismatch: the full body arrives but hashes to something else. + let wrong = compute_git_sha256_from_bytes(b"something else"); + let uri = split_body_server(content.clone(), None, content.len(), release).await; + let tmp = tempfile::tempdir().unwrap(); + let blobs = tmp.path().join(".socket").join("blobs"); + let result = fetch_missing_blobs( + &manifest_with_after_hashes(&[&wrong]), + &blobs, + &proxy_client(&uri), + None, + ) + .await; + assert_eq!(result.failed, 1); + let error = result.results[0].error.as_deref().unwrap(); + assert_eq!( + error, + format!("Content hash mismatch: expected {wrong}, got {hash}") + ); + assert_eq!( + dir_entry_count(tmp.path()), + 0, + "a rejected blob must leave neither .socket/ nor .socket/blobs/ behind" + ); +} + +/// A cache directory `create_dir_all` only partly creates (here a new +/// `.socket/` under which the 300-byte leaf name is too long for the +/// filesystem) is removed again, like any other failed download's. +#[tokio::test] +async fn partly_created_cache_dirs_are_removed_on_failure() { + let content = b"the genuine patched file body"; + let hash = compute_git_sha256_from_bytes(content); + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_matcher(format!("/patch/blob/{hash}"))) + .respond_with(ResponseTemplate::new(200).set_body_bytes(content.to_vec())) + .mount(&server) + .await; + + let tmp = tempfile::tempdir().unwrap(); + let blobs = tmp.path().join(".socket").join("x".repeat(300)); + let result = fetch_missing_blobs( + &manifest_with_after_hashes(&[&hash]), + &blobs, + &proxy_client(&server.uri()), + None, + ) + .await; + assert_eq!(result.failed, 1, "{:?}", result.results); + let error = result.results[0].error.as_deref().unwrap(); + assert!(error.starts_with("Failed to write blob to disk"), "{error}"); + assert_eq!( + dir_entry_count(tmp.path()), + 0, + "the .socket/ that create_dir_all made before failing must be removed" + ); +} diff --git a/crates/socket-patch-core/tests/covgap_api_blob_fetcher.rs b/crates/socket-patch-core/tests/covgap_api_blob_fetcher.rs index 1e0bc614..041c323a 100644 --- a/crates/socket-patch-core/tests/covgap_api_blob_fetcher.rs +++ b/crates/socket-patch-core/tests/covgap_api_blob_fetcher.rs @@ -2,7 +2,7 @@ //! branches: //! //! * an uncreatable cache directory — the dir is created by the first -//! verified download inside `write_cache_entry_atomic`, never up front, +//! download inside `stream_cache_entry_atomic`, never up front, //! so the failure is a per-entry "Failed to write ... to disk" and a //! fetch that lands nothing leaves no `.socket/blobs/` husk — driven //! cross-platform via ENOTDIR (the target directory is routed *through @@ -11,7 +11,7 @@ //! twin is tested in `blob_fetcher_edges_e2e.rs`, this one never ran; //! * the "Failed to write blob/archive to disk" arms where the download //! succeeded but the atomic cache write failed (unix-only, read-only directory); -//! * mixed-outcome aggregation across `download_hashes` arms — each arm +//! * mixed-outcome aggregation across `download_entries` arms — each arm //! is individually covered by the sibling suite, but a single run //! combining success + 404 + hash-mismatch never was. //! @@ -497,7 +497,7 @@ async fn fetch_missing_sources_diff_disk_write_failure_is_per_archive_failure() // ── Mixed-outcome aggregation ──────────────────────────────────────── -/// One run combining all three `download_hashes` arms: a good blob, a +/// One run combining all three `download_entries` arms: a good blob, a /// 404, and a hash-mismatch. Counters aggregate per-arm, each entry /// carries its own arm's error, only the good blob lands on disk, and /// `format_fetch_result` renders the genuinely mixed result.