From a46bb4b2da93137fb2254abc47a920d393776048 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Wed, 30 Sep 2026 11:22:40 -0700 Subject: [PATCH] Apply configured store retry budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? `store.max_retries` was documented as an operator-controlled retry budget, but backend construction ignored it and GCS used fixed retry counts. ### What? Use the configured number of extra attempts for transient GCS metadata failures, interrupted GCS bulk reads, S3 SDK operations, and S3 HTTP GETs. Retry with bounded jittered backoff and preserve zero as the way to disable retries. Healthy store calls remain one request. Only transient failure paths add up to `store.max_retries` attempts, and this change adds no CAS object or healthy-path round trip. ## Test Plan ✅ `cargo check -p walgit-store --all-features` ✅ `cargo test -p walgit-store --all-features bulk_read_resumes_after_mid_stream_cut` ✅ `cargo fmt --all -- --check` ✅ `git diff --check` ## Issue Closes #82 --- crates/walgit-config/src/lib.rs | 1 + crates/walgit-store/src/gcs.rs | 56 ++++++++++++++++++--------------- crates/walgit-store/src/s3.rs | 48 +++++++++++++++++++++++----- docs/ROUNDTRIPS.md | 1 + walgit.example.toml | 2 +- 5 files changed, 74 insertions(+), 34 deletions(-) diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index 6c3a2257..5d68edc2 100644 --- a/crates/walgit-config/src/lib.rs +++ b/crates/walgit-config/src/lib.rs @@ -235,6 +235,7 @@ pub struct StoreConfig { pub prefix: String, pub gcs: GcsConfig, pub s3: S3Config, + /// Extra attempts for transient backend failures and interrupted reads; zero disables retries. pub max_retries: u32, /// Objects larger than this use resumable/multipart upload. pub multipart_threshold: ByteSize, diff --git a/crates/walgit-store/src/gcs.rs b/crates/walgit-store/src/gcs.rs index fbe4a62c..1974bf77 100644 --- a/crates/walgit-store/src/gcs.rs +++ b/crates/walgit-store/src/gcs.rs @@ -37,7 +37,6 @@ const LIST_PAGE_SIZE: i32 = 1000; /// (503) or retries. Metadata/control calls are quick (p99 ~80 ms); body reads /// and uploads get time proportional to their size. /// Mid-stream resumes per bulk read before the error is surfaced. -const BULK_RESUME_ATTEMPTS: u32 = 5; const META_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); const READ_OPEN_DEADLINE: std::time::Duration = std::time::Duration::from_mins(1); /// Per chunk of a streaming body read (not the whole stream). @@ -45,8 +44,6 @@ const READ_CHUNK_DEADLINE: std::time::Duration = std::time::Duration::from_mins( const PUT_MIN_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); /// Uploads get this many bytes per second on top of `PUT_MIN_DEADLINE` (1 MiB/s floor). const PUT_BYTES_PER_SEC: u64 = 1024 * 1024; -/// Idempotent metadata reads that time out are retried once after a short jittered pause. -const READ_RETRIES: u32 = 1; fn put_deadline(bytes: u64) -> std::time::Duration { PUT_MIN_DEADLINE + std::time::Duration::from_secs(bytes / PUT_BYTES_PER_SEC) @@ -66,8 +63,8 @@ fn deadline_error(op: &str, key: &str, deadline: std::time::Duration) -> StoreEr /// Run a GCS call under `deadline`; the client error keeps its meaning through /// `map_error` (`NotFound` / `PreconditionFailed` / `NotModified`), a timeout becomes -/// [`deadline_error`]. `retries` extra attempts are made only when the deadline -/// fired (the call is idempotent for every caller that passes > 0). +/// [`deadline_error`]. `retries` extra attempts are made for deadlines and +/// transient service errors (the call is idempotent for every caller that passes > 0). async fn call( op: &'static str, key: &str, @@ -83,6 +80,9 @@ where loop { match tokio::time::timeout(deadline, make()).await { Ok(Ok(v)) => return Ok(v), + Ok(Err(e)) if is_retryable(&e) && attempt < retries => { + tracing::warn!(op, key, attempt, error = %e, "retrying transient gcs error"); + } Ok(Err(e)) => return Err(GcsCallError::Gcs(e)), Err(_) => { if attempt >= retries { @@ -95,17 +95,15 @@ where deadline_ms = u64::try_from(deadline.as_millis()).unwrap_or(u64::MAX), "gcs call exceeded deadline, retrying" ); - attempt += 1; - let jitter = 100 - + u64::from( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.subsec_nanos()) - % 400, - ); - tokio::time::sleep(std::time::Duration::from_millis(jitter)).await; } } + let delay = crate::util::backoff( + attempt, + std::time::Duration::from_millis(100), + std::time::Duration::from_secs(2), + ); + attempt += 1; + tokio::time::sleep(delay).await; } } @@ -159,6 +157,7 @@ pub struct GcsStore { bucket_resource: String, // "projects/_/buckets/{bucket}" signing_signer: Option, signing_service_account: Option, + max_retries: u32, } impl GcsStore { @@ -234,6 +233,7 @@ impl GcsStore { bucket_resource, signing_signer, signing_service_account: cfg.gcs.signing_service_account.clone(), + max_retries: cfg.max_retries, }) } @@ -266,6 +266,7 @@ impl GcsStore { permits: self.bulk_permits.clone(), endpoint: std::env::var("WALGIT_GCS_HTTP_ENDPOINT") .unwrap_or_else(|_| "https://storage.googleapis.com".into()), + max_retries: self.max_retries, } } @@ -335,7 +336,7 @@ impl GcsStore { let req = google_cloud_storage::model::GetObjectRequest::new() .set_bucket(self.bucket_resource.clone()) .set_object(key.to_owned()); - match call("head", key, META_DEADLINE, READ_RETRIES, || { + match call("head", key, META_DEADLINE, self.max_retries, || { self.control.get_object().with_request(req.clone()).send() }) .await @@ -352,8 +353,8 @@ impl GcsStore { /// response body", a chunk deadline — prod 2026-08-21 ×6, aborting a /// history-pack install) the wrapper re-issues the request for the bytes /// not yet delivered (`Range: bytes=-`, pinned to the same - /// generation so a rewritten object can never be spliced), up to - /// `BULK_RESUME_ATTEMPTS` times with jittered backoff. + /// generation so a rewritten object can never be spliced), up to the + /// configured `store.max_retries` with jittered backoff. /// `walgit_remote_chunk_retries_total` counts the resumes. async fn bulk_http_read( &self, @@ -402,7 +403,7 @@ impl BulkHttp { return Some((Ok(b), (st, this, key))); } None => { - if end > 0 && st.pos < end && st.attempts < BULK_RESUME_ATTEMPTS { + if end > 0 && st.pos < end && st.attempts < this.max_retries { // Short body without an error: resume too. tracing::warn!(key = %key, pos = st.pos, end, "gcs bulk read ended early; resuming"); } else { @@ -410,7 +411,7 @@ impl BulkHttp { } } Some(Err(e)) => { - if st.attempts >= BULK_RESUME_ATTEMPTS || st.pos >= end { + if st.attempts >= this.max_retries || st.pos >= end { return Some((Err(e), (st, this, key))); } tracing::warn!(key = %key, pos = st.pos, end, attempt = st.attempts + 1, error = %e, "gcs bulk read failed mid-stream; resuming from the bytes received"); @@ -425,7 +426,7 @@ impl BulkHttp { match this.open(&key, Some(st.pos..end), generation).await { Ok((_, _, s)) => st.inner = s, Err(e) => { - if st.attempts >= BULK_RESUME_ATTEMPTS { + if st.attempts >= this.max_retries { return Some((Err(e), (st, this, key))); } // Try again after the next backoff (loop: inner is exhausted → None arm). @@ -451,12 +452,13 @@ pub(crate) struct BulkHttp { /// JSON API endpoint (`https://storage.googleapis.com`; tests point it at /// a local server that cuts bodies). endpoint: String, + max_retries: u32, } impl BulkHttp { /// Test constructor: no credentials, any endpoint. #[cfg(test)] - pub(crate) fn for_tests(endpoint: String, bucket: String) -> Self { + pub(crate) fn for_tests(endpoint: String, bucket: String, max_retries: u32) -> Self { BulkHttp { clients: vec![reqwest::Client::new()], next: std::sync::Arc::default(), @@ -464,6 +466,7 @@ impl BulkHttp { bucket, permits: std::sync::Arc::new(tokio::sync::Semaphore::new(8)), endpoint, + max_retries, } } @@ -687,7 +690,7 @@ impl ObjectStore for GcsStore { .set_object(key.to_owned()) .set_if_generation_not_match(generation); - let result = match call("get", key, META_DEADLINE, READ_RETRIES, || { + let result = match call("get", key, META_DEADLINE, self.max_retries, || { self.control.get_object().with_request(req.clone()).send() }) .await @@ -777,7 +780,7 @@ impl ObjectStore for GcsStore { let req = google_cloud_storage::model::GetObjectRequest::new() .set_bucket(self.bucket_resource.clone()) .set_object(key.to_owned()); - match call("head", key, META_DEADLINE, READ_RETRIES, || { + match call("head", key, META_DEADLINE, self.max_retries, || { self.control.get_object().with_request(req.clone()).send() }) .await @@ -914,6 +917,7 @@ impl ObjectStore for GcsStore { let (tx, rx) = tokio::sync::mpsc::channel::>(32); let control = self.control.clone(); let bucket_resource = self.bucket_resource.clone(); + let max_retries = self.max_retries; let prefix = prefix.to_owned(); let start_after = start_after.map(std::borrow::ToOwned::to_owned); @@ -938,7 +942,7 @@ impl ObjectStore for GcsStore { } let list_key = format!("{prefix}/list"); - let resp = match call("list", &list_key, META_DEADLINE, READ_RETRIES, || { + let resp = match call("list", &list_key, META_DEADLINE, max_retries, || { control.list_objects().with_request(req.clone()).send() }) .await @@ -991,7 +995,7 @@ impl ObjectStore for GcsStore { "list_prefixes", &list_key, META_DEADLINE, - READ_RETRIES, + self.max_retries, || self.control.list_objects().with_request(req.clone()).send(), ) .await @@ -1582,7 +1586,7 @@ mod resume_tests { let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); - let bulk = BulkHttp::for_tests(format!("http://{addr}"), "b".into()); + let bulk = BulkHttp::for_tests(format!("http://{addr}"), "b".into(), 2); let (size, generation, mut body) = bulk.read("k", None, None).await.unwrap(); assert_eq!((size, generation), (3000, Some(7))); let mut got = Vec::new(); diff --git a/crates/walgit-store/src/s3.rs b/crates/walgit-store/src/s3.rs index d788cc2e..0d61e190 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -64,6 +64,7 @@ pub struct S3Store { http: reqwest::Client, multipart_threshold: u64, multipart_part_size: u64, + max_retries: u32, } impl S3Store { @@ -92,6 +93,10 @@ impl S3Store { .region(region) .credentials_provider(creds) .force_path_style(cfg.s3.force_path_style) + .retry_config( + aws_sdk_s3::config::retry::RetryConfig::standard() + .with_max_attempts(cfg.max_retries.saturating_add(1)), + ) .behavior_version_latest(); if !cfg.s3.endpoint.is_empty() { @@ -107,6 +112,7 @@ impl S3Store { http, multipart_threshold: cfg.multipart_threshold.as_u64(), multipart_part_size: cfg.multipart_part_size.as_u64(), + max_retries: cfg.max_retries, }) } @@ -138,14 +144,42 @@ impl S3Store { .await .map_err(|e| StoreError::other(anyhow::anyhow!("presigning get: {e}")))?; - let mut req = self.http.get(presigned.uri()); - for (name, value) in presigned.headers() { - req = req.header(name, value); + let mut attempt = 0u32; + loop { + let mut req = self.http.get(presigned.uri()); + for (name, value) in presigned.headers() { + req = req.header(name, value); + } + match req.send().await { + Ok(response) + if matches!(response.status().as_u16(), 429 | 500..=599) + && attempt < self.max_retries => + { + tracing::warn!( + key, + attempt, + status = %response.status(), + "retrying transient s3 get" + ); + } + Ok(response) => return Ok(response), + Err(error) if attempt < self.max_retries => { + tracing::warn!(key, attempt, %error, "retrying failed s3 get"); + } + Err(error) => { + return Err(StoreError::retryable(anyhow::anyhow!( + "s3 get http: {error}" + ))); + } + } + let delay = util::backoff( + attempt, + std::time::Duration::from_millis(100), + std::time::Duration::from_secs(2), + ); + attempt += 1; + tokio::time::sleep(delay).await; } - - req.send() - .await - .map_err(|e| StoreError::retryable(anyhow::anyhow!("s3 get http: {e}"))) } fn get_result_from_response(key: &str, resp: reqwest::Response) -> Result { diff --git a/docs/ROUNDTRIPS.md b/docs/ROUNDTRIPS.md index 622f8ea1..b2d317ab 100644 --- a/docs/ROUNDTRIPS.md +++ b/docs/ROUNDTRIPS.md @@ -60,6 +60,7 @@ right shape. This document is the thinking tool; apply it to every protocol chan | Bundle removal (2026-09-11) | v2 capabilities and narrated fetch: removed optional list GET (1 → 0 extra); maintenance no longer reads/CASes a bundle list; direct import no longer composes a wrapper or reads/CASes a bundle list | no new store requests; checkpoint and push budgets unchanged | `smart.rs`, `maintain.rs`, `import_direct.rs` | | Canonical checkpoint publication | unchanged: freshness → content-addressed refs PUT ∥ attempt-specific metadata PUT → manifest CAS | 4 healthy requests; equality verification GET only after immutable Create conflict | `checkpoint.rs`, `snapshots.rs` | | Orphan log slot (failure path only) | +1 fresh manifest GET, +HEAD per probe, +Create at next seq | — | `publish.rs::claim_log_slot` | +| Configured store retries (failure path only) | healthy calls remain one attempt; transient idempotent GCS calls, S3 SDK calls, and interrupted bulk reads retry with jitter | up to `store.max_retries` extra attempts; no new healthy-path requests or CAS objects | `gcs.rs`, `s3.rs` | `healthy_request_round_trip_budgets` in `crates/walgit-server/tests/sim.rs` pins the healthy MemoryStore counts at push **5**, warm refs **1**, cold refs with one tail segment **2**, and checkpoint **4**. Cold open used to spend an diff --git a/walgit.example.toml b/walgit.example.toml index e7c7df23..ea0f4e3a 100644 --- a/walgit.example.toml +++ b/walgit.example.toml @@ -67,7 +67,7 @@ anonymous_read = true # must be false in oidc mode backend = "s3" # "s3" (AWS, MinIO, rustfs, R2, Ceph, …) | "gcs" | "memory" (tests) bucket = "walgit" prefix = "" # global key prefix inside the bucket -max_retries = 8 # retries on retryable store errors (jittered backoff) +max_retries = 8 # extra attempts for transient store errors and interrupted reads (jittered backoff; 0 disables) multipart_threshold = "64MiB" # PUTs above this use multipart/resumable uploads multipart_part_size = "32MiB"