Apply configured store retry budget

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,

crates/walgit-store/src/gcs.rs

@@ -37,16 +37,13 @@ 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).
 const READ_CHUNK_DEADLINE: std::time::Duration = std::time::Duration::from_mins(1);
 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<T, F, Fut>(
     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<google_cloud_auth::signer::Signer>,
     signing_service_account: Option<String>,
+    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=<pos>-<end>`, 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,15 +403,15 @@ 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 {
                                 return None;
                             }
                         }
                         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,19 +452,21 @@ 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(),
             creds: None,
             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::<Result<ObjectMeta>>(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();

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<GetResult> {

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

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"