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"