From 0ec7d3e42b8b74fd419d7997319fe724beb6c513 Mon Sep 17 00:00:00 2001 From: ethanstoner Date: Fri, 18 Sep 2026 10:56:59 -0700 Subject: [PATCH 1/3] Retry GCS 408, 502 and client timeouts like S3 gcs::is_retryable treated HTTP 408 and 502 as permanent and ignored the client's own timeout errors, so a bad gateway on a manifest CAS failed the push on GCS while the same response on S3 is retried (#18). Use the status set GCS documents as retryable (408, 429, 500, 502, 503, 504), leave 501 out, and share the same set with s3::is_transient_status. (cherry picked from commit 939348a2d40fe59af8dc91c6de050b78b5b9e4a0) Signed-off-by: Yordis Prieto --- crates/walgit-store/src/gcs.rs | 34 ++++++++++++++++++++++++++++++++-- crates/walgit-store/src/s3.rs | 6 +++--- 2 files changed, 35 insertions(+), 5 deletions(-) diff --git a/crates/walgit-store/src/gcs.rs b/crates/walgit-store/src/gcs.rs index fbe4a62c..4dd3b4f7 100644 --- a/crates/walgit-store/src/gcs.rs +++ b/crates/walgit-store/src/gcs.rs @@ -1201,9 +1201,13 @@ fn is_retryable(e: &google_cloud_storage::Error) -> bool { ); } if let Some(code) = e.http_status_code() { - return matches!(code, 503 | 504 | 429 | 500); + // The statuses GCS documents as retryable. The client's own policy + // retries all of 5xx; 501 is left out because an unimplemented + // request will not start working on a second try. Same set as + // `s3::is_transient_status`. + return matches!(code, 408 | 429 | 500 | 502 | 503 | 504); } - e.is_connect() || e.is_io() || e.is_transient_and_before_rpc() + e.is_connect() || e.is_io() || e.is_timeout() || e.is_transient_and_before_rpc() } fn map_error(key: &str, e: google_cloud_storage::Error) -> StoreError { @@ -1384,6 +1388,32 @@ mod tests { assert!(is_retryable(&e)); } + fn http_error(status: u16) -> google_cloud_storage::Error { + google_cloud_gax::error::Error::http(status, http::HeaderMap::new(), bytes::Bytes::new()) + } + + #[test] + fn transient_http_statuses_are_retryable() { + for status in [408, 429, 500, 502, 503, 504] { + assert!( + is_retryable(&http_error(status)), + "{status} should be retryable" + ); + } + for status in [400, 403, 404, 409, 412, 501] { + assert!( + !is_retryable(&http_error(status)), + "{status} should be permanent" + ); + } + } + + #[test] + fn is_retryable_timeout() { + let e = google_cloud_gax::error::Error::timeout("attempt timed out"); + assert!(is_retryable(&e)); + } + #[test] fn is_not_retryable_not_found() { let e = google_cloud_gax::error::Error::service( diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs index 08409d53..2138941b 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -261,7 +261,7 @@ fn is_transient_code(code: &str) -> bool { /// HTTP statuses that mean the same, for services whose error codes we do not /// recognise (rustfs and other S3-compatible stores do not all use AWS codes). fn is_transient_status(status: u16) -> bool { - matches!(status, 429 | 500 | 502 | 503 | 504) + matches!(status, 408 | 429 | 500 | 502 | 503 | 504) } /// Whether an SDK failure is worth another attempt at walgit's layer. @@ -1303,10 +1303,10 @@ mod tests { #[test] fn transient_statuses_are_recognised() { - for status in [429, 500, 502, 503, 504] { + for status in [408, 429, 500, 502, 503, 504] { assert!(is_transient_status(status), "{status} should be transient"); } - for status in [400, 403, 404, 409, 412] { + for status in [400, 403, 404, 409, 412, 501] { assert!(!is_transient_status(status), "{status} should be permanent"); } } From cdeb9c82c0cd9f27f6c0601eb95646bfe45e84a4 Mon Sep 17 00:00:00 2001 From: jilio <22717553+jilio@users.noreply.github.com> Date: Sun, 11 Oct 2026 23:18:04 +1100 Subject: [PATCH 2/3] Send S3 conditional headers with quoted entity-tags S3Store keeps ETags unquoted in Version (quotes are stripped on read) and put them on the wire as they are: `If-Match: abc` on the manifest CAS and on GETs, `If-None-Match: abc` on the conditional manifest GET. HTTP defines both headers over quoted entity-tags (RFC 9110 sections 8.8.3 and 13.1.1; the S3 API docs defer to RFC 7232), and the ETag header of every S3 response carries that form. AWS S3 (its CLI example sends a bare value) and rustfs accept the bare value too, but that is not documented for S3-compatible stores in general, and Cloudflare R2's docs are silent on it. A store that compares the tag strictly would refuse every manifest CAS (no push could ever commit) and never answer the conditional GET with 304. entity_tag() renders a Version as a quoted entity-tag (once; an already quoted value is left alone) and every conditional header uses it. Versions stay unquoted, so nothing stored or compared changes. Checked: the store contract suite passes against rustfs 1.0.1, which honours both forms; against Cloudflare R2, a quoted If-Match returned 200 on a match and 412 on a mismatch. A new unit test records the headers walgit sends to a fake S3; before this change it saw `If-Match: abc`. (cherry picked from commit 7143aad6f82624bd545847bc9bad6bed0abd820f) Signed-off-by: Yordis Prieto --- crates/walgit-store/src/s3.rs | 149 ++++++++++++++++++++++++++++++++-- 1 file changed, 140 insertions(+), 9 deletions(-) diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs index d788cc2e..08409d53 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -8,14 +8,15 @@ //! ## Version tokens //! //! S3 `ETags` are used as opaque `Version` strings. Quotes are stripped -//! consistently on read and never stored. For non-multipart uploads the -//! `ETag` is the MD5 of the content; for multipart uploads it is a compound -//! hash. Callers never parse the token — equality comparison suffices. +//! consistently on read and never stored; conditional headers put them back +//! ([`entity_tag`]). For non-multipart uploads the `ETag` is the MD5 of the +//! content; for multipart uploads it is a compound hash. Callers never parse +//! the token — equality comparison suffices. //! //! ## Conditional PUT //! -//! `PutMode::Create` → `If-None-Match: *` (object must not exist). -//! `PutMode::Update(v)` → `If-Match: ` (CAS on current `ETag`). +//! `PutMode::Create` → `If-None-Match: *` (object must not exist). +//! `PutMode::Update(v)` → `If-Match: ""` (CAS on current `ETag`). //! On failure the SDK returns a `PreconditionFailed` service error; we fill //! `current` via a follow-up HEAD when the SDK doesn't include it. //! @@ -124,10 +125,10 @@ impl S3Store { let mut builder = self.client.get_object().bucket(&self.bucket).key(key); if let Some(v) = &opts.if_none_match { - builder = builder.if_none_match(v.as_str()); + builder = builder.if_none_match(entity_tag(v)); } if let Some(v) = &opts.if_match { - builder = builder.if_match(v.as_str()); + builder = builder.if_match(entity_tag(v)); } if let Some(r) = &opts.range { builder = builder.range(Self::range_header(r)); @@ -393,7 +394,7 @@ impl ObjectStore for S3Store { builder = builder.if_none_match("*"); } PutMode::Update(v) => { - builder = builder.if_match(v.as_str()); + builder = builder.if_match(entity_tag(v)); } } @@ -977,6 +978,20 @@ impl S3Store { } } +/// A version token as the entity-tag a conditional header carries. `ETag`s are +/// stored unquoted, but `If-Match` / `If-None-Match` take quoted entity-tags +/// (RFC 9110 §8.8.3, §13.1.1), the form every response's `ETag` header has. +/// AWS S3 and rustfs also accept a bare value; other S3-compatible stores do +/// not document it (Cloudflare R2 documents neither form). +fn entity_tag(v: &Version) -> String { + let s = v.as_str(); + if s.len() >= 2 && s.starts_with('"') && s.ends_with('"') { + s.to_owned() + } else { + format!("\"{s}\"") + } +} + fn static_credentials( access_key: &str, secret_key: &str, @@ -997,13 +1012,16 @@ fn static_credentials( // 7. Multipart: CreateMultipartUpload + UploadPart + CompleteMultipartUpload // supported. No conditional headers on Create/Complete (same as real S3). // 8. ETags: quoted, MD5 for single-PUT, compound for multipart. Quotes -// stripped consistently in our Version. +// stripped consistently in our Version, restored on If-Match/If-None-Match +// (rustfs honours both forms). // 9. force_path_style: required for rustfs local dev. #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; + use aws_sdk_s3::error::SdkError; use aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Error; use aws_sdk_s3::operation::put_object::PutObjectError; @@ -1141,6 +1159,119 @@ mod tests { )); } + /// A fake S3 that records each request head and answers a PUT with 200 and + /// a GET with 304, both with a quoted `ETag` like real S3. + async fn recording_s3() -> (String, Arc>>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let heads: Arc>> = Arc::default(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let recorded = heads.clone(); + tokio::spawn(async move { + while let Ok((mut sock, _)) = listener.accept().await { + let recorded = recorded.clone(); + tokio::spawn(async move { + let mut buf = Vec::new(); + let mut chunk = [0u8; 8192]; + let end = loop { + let n = sock.read(&mut chunk).await.unwrap_or(0); + if n == 0 { + return; + } + buf.extend_from_slice(&chunk[..n]); + if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break i + 4; + } + }; + let head = String::from_utf8_lossy(&buf[..end]).into_owned(); + let body_len = head + .lines() + .filter_map(|l| l.split_once(':')) + .find(|(k, _)| k.eq_ignore_ascii_case("content-length")) + .and_then(|(_, v)| v.trim().parse::().ok()) + .unwrap_or(0); + while buf.len() < end + body_len { + let n = sock.read(&mut chunk).await.unwrap_or(0); + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + let resp = if head.starts_with("GET ") { + "HTTP/1.1 304 Not Modified\r\nETag: \"abc\"\r\nConnection: close\r\n\r\n" + } else { + "HTTP/1.1 200 OK\r\nETag: \"def\"\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + }; + recorded.lock().push(head); + let _ = sock.write_all(resp.as_bytes()).await; + }); + } + }); + (endpoint, heads) + } + + fn header<'a>(head: &'a str, name: &str) -> Option<&'a str> { + head.lines() + .filter_map(|l| l.split_once(':')) + .find(|(k, _)| k.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.trim()) + } + + #[test] + fn entity_tags_are_quoted_once() { + assert_eq!(entity_tag(&Version::new("abc")), "\"abc\""); + assert_eq!(entity_tag(&Version::new("\"abc\"")), "\"abc\""); + assert_eq!(entity_tag(&Version::new("abc-3")), "\"abc-3\""); + } + + /// Versions hold `ETag`s unquoted; on the wire a conditional header carries + /// the quoted entity-tag (RFC 9110), the form every `ETag` response has. + #[tokio::test] + async fn conditional_headers_carry_quoted_entity_tags() { + let (endpoint, heads) = recording_s3().await; + let store = S3Store { + client: client_for(&endpoint), + bucket: "b".into(), + http: reqwest::Client::new(), + multipart_threshold: u64::MAX, + multipart_part_size: 8 << 20, + }; + let put = store + .put( + "manifest.pb", + PutBody::Bytes(Bytes::from_static(b"x")), + PutMode::Update(Version::new("abc")).into(), + ) + .await + .unwrap(); + assert_eq!(put.version.as_str(), "def", "stored unquoted"); + let got = store + .get( + "manifest.pb", + GetOptions { + if_none_match: Some(Version::new("abc")), + ..Default::default() + }, + ) + .await + .unwrap(); + assert!(matches!(got, GetResult::NotModified { .. })); + let _ = store + .get( + "manifest.pb", + GetOptions { + if_match: Some(Version::new("abc")), + ..Default::default() + }, + ) + .await; + let heads = heads.lock().clone(); + assert_eq!(heads.len(), 3, "{heads:#?}"); + assert_eq!(header(&heads[0], "if-match"), Some("\"abc\"")); + assert_eq!(header(&heads[1], "if-none-match"), Some("\"abc\"")); + assert_eq!(header(&heads[2], "if-match"), Some("\"abc\"")); + } + #[test] fn transient_codes_are_recognised() { for code in [ From 0bf402da646d04b08bdc6d64208e2cfc31923482 Mon Sep 17 00:00:00 2001 From: Yad Konrad Date: Fri, 25 Sep 2026 10:25:21 -0400 Subject: [PATCH 3/3] Pass the pack lifecycle's sort budget in bytes GNU and BSD sort both read a bare --buffer-size number as KiB, so the lifecycle's byte budget (clamped to 64 MiB) reached sort as 64 GiB and never made it spill to the work directory. Measured on a 21 MB object list with a 1 MiB budget: GNU sort peaked at 46 MB RSS without the suffix and 7.2 MB with it; BSD sort at 87 MB and 3.6 MB. (cherry picked from commit 23320a1957f953272207dc2bfc30d51cc0b01eff) Signed-off-by: Yordis Prieto --- crates/walgit-git/src/pack_segments.rs | 23 +++++++++++--------- crates/walgit-git/src/pack_segments_tests.rs | 9 ++++++++ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/crates/walgit-git/src/pack_segments.rs b/crates/walgit-git/src/pack_segments.rs index 25dbf72d..d02ef05f 100644 --- a/crates/walgit-git/src/pack_segments.rs +++ b/crates/walgit-git/src/pack_segments.rs @@ -637,16 +637,19 @@ fn append_index( } fn sort(input: &Path, output: &Path, work: &Path, resources: Resources) -> Result<(), GitError> { - run( - Command::new("sort") - .args(["--stable", "--unique", "--key=1,1", "--parallel=1"]) - .arg(format!("--buffer-size={}", resources.sort_memory_bytes)) - .arg("--temporary-directory") - .arg(work), - input, - output, - work, - ) + run(&mut sort_command(work, resources), input, output, work) +} + +/// GNU and BSD `sort` both read a bare `--buffer-size` number as KiB; the `b` +/// suffix keeps the budget in bytes instead of 1024 times larger. +fn sort_command(work: &Path, resources: Resources) -> Command { + let mut command = Command::new("sort"); + command + .args(["--stable", "--unique", "--key=1,1", "--parallel=1"]) + .arg(format!("--buffer-size={}b", resources.sort_memory_bytes)) + .arg("--temporary-directory") + .arg(work); + command } fn difference(a: &Path, b: &Path, output: &Path, work: &Path) -> Result<(), GitError> { diff --git a/crates/walgit-git/src/pack_segments_tests.rs b/crates/walgit-git/src/pack_segments_tests.rs index 5a9b9037..ea9883d5 100644 --- a/crates/walgit-git/src/pack_segments_tests.rs +++ b/crates/walgit-git/src/pack_segments_tests.rs @@ -305,6 +305,15 @@ async fn empty_groups_conserve_everything_and_invalid_budgets_fail() { ); } +#[test] +fn sort_budget_is_passed_in_bytes() { + let command = sort_command(Path::new("work"), resources()); + assert!( + command.get_args().any(|a| a == "--buffer-size=1048576b"), + "{command:?}" + ); +} + #[test] fn type_split_keeps_hint_bytes_and_rejects_missing_objects() { let root = tempfile::tempdir().unwrap();