From a3b1309ce3fbe46cd90b7d3cd1f13541072aef28 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 21:00:17 +0000 Subject: [PATCH 1/5] Start refactor for #571 Assisted-by: Claude Code:claude-opus-5-5 From ae5d5286b94d101598174ca09905c08da64bc6d4 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 21:16:43 +0000 Subject: [PATCH 2/5] Stream patch blob and diff downloads to disk apply, get, repair and rollback used to hold every patch blob and diff archive fully in memory before writing it to the .socket cache. Each body now streams chunk by chunk into its .socket-dl-* stage file. A blob is hashed from the staged file, then renamed into place. ApiClient::fetch_blob and fetch_diff return a BinaryBody stream. The two near-identical blob and diff download loops in blob_fetcher are now one. Error messages, counts and the no-litter guarantees are unchanged. A failed download still leaves no stage file and no cache directory it created. Fixes #571. Assisted-by: Claude Code:claude-opus-5-5 --- .../socket-patch-core/src/api/blob_fetcher.rs | 320 ++++++++++-------- crates/socket-patch-core/src/api/client.rs | 64 ++-- .../tests/api_timeout_e2e.rs | 49 ++- .../tests/blob_fetcher_edges_e2e.rs | 187 ++++++++++ 4 files changed, 453 insertions(+), 167 deletions(-) diff --git a/crates/socket-patch-core/src/api/blob_fetcher.rs b/crates/socket-patch-core/src/api/blob_fetcher.rs index e1afdcf47..520591327 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,187 @@ 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); + } + tokio::fs::create_dir_all(parent) + .await + .map_err(EntryError::Write)?; // 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 { + 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 +1041,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 +1161,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 7d862c050..2959a243c 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 793f56dfb..54a436f9a 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 cc11c8dc3..215870085 100644 --- a/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs +++ b/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs @@ -732,3 +732,190 @@ 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. +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(|| e.metadata().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" + ); +} From 0b4de631a8610323894424026028247339d0f617 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 21:19:08 +0000 Subject: [PATCH 3/5] Note streamed blob downloads in the changelog Assisted-by: Claude Code:claude-opus-5-5 --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3572a298c..7b1aae734 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 From 3362893397851fd76a46e46934018096289af941 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 21:43:18 +0000 Subject: [PATCH 4/5] Clean up cache dirs a failed mkdir made create_dir_all can create some ancestors of the cache directory and then fail. Those empty directories were left behind, because the creation ran before the failure cleanup. It now runs inside that cleanup, so a download that lands nothing never leaves a .socket/ or .socket/blobs/ husk. Assisted-by: Claude Code:claude-opus-5-5 --- .../socket-patch-core/src/api/blob_fetcher.rs | 8 ++-- .../tests/blob_fetcher_edges_e2e.rs | 37 ++++++++++++++++++- .../tests/covgap_api_blob_fetcher.rs | 6 +-- 3 files changed, 43 insertions(+), 8 deletions(-) diff --git a/crates/socket-patch-core/src/api/blob_fetcher.rs b/crates/socket-patch-core/src/api/blob_fetcher.rs index 520591327..86a51e19d 100644 --- a/crates/socket-patch-core/src/api/blob_fetcher.rs +++ b/crates/socket-patch-core/src/api/blob_fetcher.rs @@ -442,14 +442,16 @@ async fn stream_cache_entry_atomic( } created_dirs.push(dir); } - tokio::fs::create_dir_all(parent) - .await - .map_err(EntryError::Write)?; // 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())); 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) 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 215870085..035dda349 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() { @@ -919,3 +919,36 @@ async fn failed_streams_leave_no_stage_and_no_created_cache_dir() { "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 1e0bc6149..041c323ad 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. From ae928ae91c2b1c57d9e5367d1d0c72bd66424651 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 22:07:53 +0000 Subject: [PATCH 5/5] Size the stage file through its path in tests The streaming test read the stage size from DirEntry::metadata. On Windows that is the directory entry's cached size, which NTFS leaves at 0 while the downloader's handle is open, so the test saw nothing streamed. std::fs::metadata asks the file itself. Assisted-by: Claude Code:claude-opus-5-5 --- crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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 035dda349..31ebb30fd 100644 --- a/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs +++ b/crates/socket-patch-core/tests/blob_fetcher_edges_e2e.rs @@ -782,13 +782,17 @@ async fn split_body_server( } /// 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(|| e.metadata().ok().map(|m| m.len())) + .then(|| std::fs::metadata(e.path()).ok().map(|m| m.len())) .flatten() }) }