diff --git a/crates/walgit-config/src/lib.rs b/crates/walgit-config/src/lib.rs index e3cdaf15..dd9a5f96 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..96e4a1e0 100644 --- a/crates/walgit-store/src/gcs.rs +++ b/crates/walgit-store/src/gcs.rs @@ -12,6 +12,7 @@ use futures::StreamExt; use google_cloud_auth::credentials::Builder as AuthBuilder; use google_cloud_gax::error::rpc::Code; +use google_cloud_gax::retry_policy::{NeverRetry, RetryPolicyExt}; use google_cloud_storage::builder::storage::SignedUrlBuilder; use google_cloud_storage::client::{Storage, StorageControl}; use google_cloud_storage::model_ext::ReadRange; @@ -37,7 +38,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 +45,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 +64,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 +81,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 +96,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 +158,7 @@ pub struct GcsStore { bucket_resource: String, // "projects/_/buckets/{bucket}" signing_signer: Option, signing_service_account: Option, + max_retries: u32, } impl GcsStore { @@ -186,6 +186,7 @@ impl GcsStore { let storage = Storage::builder() .with_endpoint(endpoint.clone()) + .with_retry_policy(NeverRetry) .build() .await?; let mut bulk = Vec::new(); @@ -194,6 +195,7 @@ impl GcsStore { bulk.push( Storage::builder() .with_endpoint(endpoint.clone()) + .with_retry_policy(NeverRetry) .build() .await?, ); @@ -211,6 +213,7 @@ impl GcsStore { let control = StorageControl::builder() .with_endpoint(endpoint) + .with_retry_policy(NeverRetry) .build() .await?; @@ -234,6 +237,7 @@ impl GcsStore { bucket_resource, signing_signer, signing_service_account: cfg.gcs.signing_service_account.clone(), + max_retries: cfg.max_retries, }) } @@ -266,6 +270,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 +340,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 +357,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 +407,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 +415,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 +430,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 +456,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 +470,7 @@ impl BulkHttp { bucket, permits: std::sync::Arc::new(tokio::sync::Semaphore::new(8)), endpoint, + max_retries, } } @@ -687,7 +694,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 @@ -738,7 +745,12 @@ impl ObjectStore for GcsStore { return Ok(GetResult::Object { meta, body }); } let (client, permit) = self.data_client(key, opts.range.is_some()).await; - let mut builder = client.read_object(self.bucket_resource.clone(), key.to_owned()); + let mut builder = client + .read_object(self.bucket_resource.clone(), key.to_owned()) + .with_retry_policy( + google_cloud_storage::retry_policy::RetryableErrors + .with_attempt_limit(self.max_retries.saturating_add(1)), + ); if let Some(v) = &opts.if_match { if let Some(generation) = parse_generation(v) { @@ -777,7 +789,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 +926,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 +951,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 +1004,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 @@ -1227,6 +1240,63 @@ fn map_error(key: &str, e: google_cloud_storage::Error) -> StoreError { mod tests { use super::*; + #[tokio::test] + async fn metadata_read_retry_budget_is_extra_attempts() { + let attempts = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = attempts.clone(); + let result = call( + "head", + "k", + std::time::Duration::from_secs(1), + 2, + move || { + let seen = seen.clone(); + async move { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err::<(), _>(google_cloud_gax::error::Error::service( + google_cloud_gax::error::rpc::Status::default() + .set_code(Code::Unavailable) + .set_message("503"), + )) + } + }, + ) + .await; + assert!(result.is_err()); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 3, + "initial read plus two configured retries" + ); + + attempts.store(0, std::sync::atomic::Ordering::SeqCst); + let seen = attempts.clone(); + let result = call( + "delete", + "k", + std::time::Duration::from_secs(1), + 0, + move || { + let seen = seen.clone(); + async move { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err::<(), _>(google_cloud_gax::error::Error::service( + google_cloud_gax::error::rpc::Status::default() + .set_code(Code::Unavailable) + .set_message("503"), + )) + } + }, + ) + .await; + assert!(result.is_err()); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::SeqCst), + 1, + "a zero retry budget makes exactly one attempt" + ); + } + #[test] fn gen_version_formats_decimal() { assert_eq!(gen_version(1_234_567_890).as_str(), "1234567890"); @@ -1582,7 +1652,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..d1f00dd0 100644 --- a/crates/walgit-store/src/s3.rs +++ b/crates/walgit-store/src/s3.rs @@ -58,12 +58,17 @@ use crate::{ /// S3-compatible object store. pub struct S3Store { + /// Mutations are deliberately single-attempt: retrying a conditional write + /// can turn a landed CAS with a lost response into a false 412. client: S3Client, + /// Idempotent metadata reads use the configured retry budget. + read_client: S3Client, bucket: String, /// reqwest client for streaming GETs via presigned URLs. http: reqwest::Client, multipart_threshold: u64, multipart_part_size: u64, + max_retries: u32, } impl S3Store { @@ -88,25 +93,37 @@ impl S3Store { ); let region = aws_sdk_s3::config::Region::new(cfg.s3.region.clone()); - let mut s3_config = aws_sdk_s3::Config::builder() - .region(region) - .credentials_provider(creds) - .force_path_style(cfg.s3.force_path_style) - .behavior_version_latest(); + let build_config = |retry_config| { + let mut s3_config = aws_sdk_s3::Config::builder() + .region(region.clone()) + .credentials_provider(creds.clone()) + .force_path_style(cfg.s3.force_path_style) + .retry_config(retry_config) + .behavior_version_latest(); - if !cfg.s3.endpoint.is_empty() { - s3_config = s3_config.endpoint_url(&cfg.s3.endpoint); - } + if !cfg.s3.endpoint.is_empty() { + s3_config = s3_config.endpoint_url(&cfg.s3.endpoint); + } + s3_config.build() + }; - let client = S3Client::from_conf(s3_config.build()); + let client = S3Client::from_conf(build_config( + aws_sdk_s3::config::retry::RetryConfig::disabled(), + )); + let read_client = S3Client::from_conf(build_config( + aws_sdk_s3::config::retry::RetryConfig::standard() + .with_max_attempts(cfg.max_retries.saturating_add(1)), + )); let http = reqwest::Client::builder().build()?; Ok(S3Store { client, + read_client, bucket: cfg.bucket.clone(), http, multipart_threshold: cfg.multipart_threshold.as_u64(), multipart_part_size: cfg.multipart_part_size.as_u64(), + max_retries: cfg.max_retries, }) } @@ -138,14 +155,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 { @@ -337,7 +382,7 @@ impl ObjectStore for S3Store { async fn head(&self, key: &str) -> Result> { let resp = self - .client + .read_client .head_object() .bucket(&self.bucket) .key(key) @@ -476,7 +521,7 @@ impl ObjectStore for S3Store { prefix: &str, start_after: Option<&str>, ) -> BoxStream<'static, Result> { - let client = self.client.clone(); + let client = self.read_client.clone(); let bucket = self.bucket.clone(); let prefix = prefix.to_owned(); let start_after = start_after.map(std::borrow::ToOwned::to_owned); @@ -556,7 +601,7 @@ impl ObjectStore for S3Store { let mut continuation_token: Option = None; loop { let mut builder = self - .client + .read_client .list_objects_v2() .bucket(&self.bucket) .prefix(prefix) @@ -1007,18 +1052,29 @@ mod tests { use aws_sdk_s3::error::SdkError; use aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Error; use aws_sdk_s3::operation::put_object::PutObjectError; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; /// A fake S3 that answers every request with one status and error code. /// Bound on an ephemeral port; the accept loop dies with the test runtime. async fn fake_s3(status: u16, code: &'static str) -> S3Client { + let (endpoint, _) = counting_fake_s3(status, code).await; + client_for(&endpoint) + } + + async fn counting_fake_s3(status: u16, code: &'static str) -> (String, Arc) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); + let attempts = Arc::new(AtomicUsize::new(0)); + let seen = attempts.clone(); tokio::spawn(async move { while let Ok((mut sock, _)) = listener.accept().await { + let seen = seen.clone(); tokio::spawn(async move { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let mut buf = [0u8; 8192]; let _ = sock.read(&mut buf).await; + seen.fetch_add(1, Ordering::SeqCst); let body = format!( "{code}fake" ); @@ -1031,23 +1087,73 @@ mod tests { }); } }); - client_for(&format!("http://127.0.0.1:{port}")) + (format!("http://127.0.0.1:{port}"), attempts) } /// SDK retries are disabled so each test observes exactly the error the /// service produced; walgit's own retry layer is what these tests cover. fn client_for(endpoint: &str) -> S3Client { + client_for_attempts(endpoint, 1) + } + + fn client_for_attempts(endpoint: &str, max_attempts: u32) -> S3Client { let conf = aws_sdk_s3::config::Config::builder() .region(aws_sdk_s3::config::Region::new("us-east-1")) .credentials_provider(static_credentials("test", "test", None)) .endpoint_url(endpoint) .force_path_style(true) - .retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled()) + .retry_config( + aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(max_attempts), + ) .behavior_version_latest() .build(); S3Client::from_conf(conf) } + fn store_for_attempts(endpoint: &str, max_retries: u32) -> S3Store { + S3Store { + client: client_for_attempts(endpoint, 1), + read_client: client_for_attempts(endpoint, max_retries.saturating_add(1)), + bucket: "b".into(), + http: reqwest::Client::new(), + multipart_threshold: u64::MAX, + multipart_part_size: 8 * 1024 * 1024, + max_retries, + } + } + + #[tokio::test] + async fn reads_retry_but_mutations_make_one_attempt() { + let (endpoint, attempts) = counting_fake_s3(503, "SlowDown").await; + let store = store_for_attempts(&endpoint, 2); + + assert!(store.head("k").await.unwrap_err().is_retryable()); + assert_eq!(attempts.load(Ordering::SeqCst), 3); + + attempts.store(0, Ordering::SeqCst); + assert!( + store + .put( + "k", + PutBody::Bytes(bytes::Bytes::from_static(b"x")), + PutMode::Create.into(), + ) + .await + .unwrap_err() + .is_retryable() + ); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + + attempts.store(0, Ordering::SeqCst); + assert!(store.delete("k", None).await.unwrap_err().is_retryable()); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + + let zero = store_for_attempts(&endpoint, 0); + attempts.store(0, Ordering::SeqCst); + assert!(zero.head("k").await.unwrap_err().is_retryable()); + assert_eq!(attempts.load(Ordering::SeqCst), 1); + } + async fn put_error(client: &S3Client) -> SdkError { client .put_object() diff --git a/docs/ROUNDTRIPS.md b/docs/ROUNDTRIPS.md index 447af901..bbca5b31 100644 --- a/docs/ROUNDTRIPS.md +++ b/docs/ROUNDTRIPS.md @@ -59,6 +59,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/S3 reads and interrupted bulk reads retry with jitter; writes and deletes are always single-attempt | up to `store.max_retries` extra read 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 79584aa9..40e19df8 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 reads only (jittered backoff; writes/deletes are single-attempt; 0 disables) multipart_threshold = "64MiB" # PUTs above this use multipart/resumable uploads multipart_part_size = "32MiB"