for download retry on rate limit reached from S3 gateway

ehsan6sha · ehsan6sha · commit 2f751599c487 · 2026-04-23T09:45:00.000-04:00
diff --git a/crates/fula-client/src/encryption.rs b/crates/fula-client/src/encryption.rs
@@ -88,6 +88,50 @@ fn flush_backoff_delay(attempt: usize) -> std::time::Duration {
 static FLUSH_BACKOFF_COUNT: std::sync::atomic::AtomicU64 =
     std::sync::atomic::AtomicU64::new(0);
 
+/// Fixed delay base (ms) between transient-error retries on HAMT blob-backend
+/// GET/PUT. Chosen so a fully drained nginx `limit_req` burst (token bucket
+/// with sub-second refill at the gateway's configured rate) has time to refill
+/// before the next attempt. Not exponential: the rate-limit condition resets
+/// per second, so later attempts don't benefit from longer waits.
+#[cfg(not(target_arch = "wasm32"))]
+const BLOB_BACKEND_RETRY_BASE_MS: u64 = 300;
+/// Maximum added jitter for blob-backend retries. De-synchronises fan-out
+/// retries so many concurrent walkers don't all wake at the same instant.
+#[cfg(not(target_arch = "wasm32"))]
+const BLOB_BACKEND_RETRY_JITTER_MS: u64 = 100;
+/// Total attempts (including the first try) for blob-backend GET/PUT.
+#[cfg(not(target_arch = "wasm32"))]
+const BLOB_BACKEND_MAX_ATTEMPTS: u32 = 4;
+
+/// Compute the fixed-plus-jitter sleep duration for a blob-backend retry.
+///
+/// Same jitter source (`SystemTime::subsec_nanos()`) as `flush_backoff_delay`;
+/// no `rand` dependency pulled in just for this.
+#[cfg(not(target_arch = "wasm32"))]
+fn blob_backend_retry_delay() -> std::time::Duration {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    let jitter = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| u64::from(d.subsec_nanos()))
+        .unwrap_or(0)
+        % (BLOB_BACKEND_RETRY_JITTER_MS + 1);
+    std::time::Duration::from_millis(BLOB_BACKEND_RETRY_BASE_MS + jitter)
+}
+
+/// Process-wide counter bumped each time `S3BlobBackend::{get,put}` sleeps on
+/// a transient-5xx retry. Observable via [`blob_backend_retry_count`] so the
+/// fault-injection integration test can assert the retry path actually ran.
+#[cfg(not(target_arch = "wasm32"))]
+static BLOB_BACKEND_RETRY_COUNT: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+/// Read the total number of blob-backend transient-5xx retries since process
+/// start. Native-only.
+#[cfg(not(target_arch = "wasm32"))]
+pub fn blob_backend_retry_count() -> u64 {
+    BLOB_BACKEND_RETRY_COUNT.load(std::sync::atomic::Ordering::Relaxed)
+}
+
 /// Read the total number of flush-forest backoff sleeps observed since
 /// process start. Native-only.
 #[cfg(not(target_arch = "wasm32"))]
@@ -294,21 +338,70 @@ fn client_err_to_crypto(err: ClientError) -> CryptoError {
 #[cfg(not(target_arch = "wasm32"))]
 #[async_trait::async_trait]
 impl BlobBackend for S3BlobBackend {
+    /// Retries transient 5xx responses (nginx `limit_req` 503, upstream
+    /// 429/500/502/503/504, S3 `SlowDown`/`InternalError`/`ServiceUnavailable`)
+    /// with a fixed 300 ms + 0-100 ms jitter delay, up to 4 attempts total.
+    /// Non-transient errors (auth failure, NotFound, etc.) short-circuit.
     async fn get(&self, path: &str) -> fula_crypto::Result<Vec<u8>> {
-        let bytes = self
-            .inner
-            .get_object(&self.bucket, path)
-            .await
-            .map_err(client_err_to_crypto)?;
-        Ok(bytes.to_vec())
+        let mut attempt: u32 = 0;
+        loop {
+            attempt += 1;
+            match self.inner.get_object(&self.bucket, path).await {
+                Ok(bytes) => return Ok(bytes.to_vec()),
+                Err(e)
+                    if attempt < BLOB_BACKEND_MAX_ATTEMPTS
+                        && crate::multipart::is_transient(&e) =>
+                {
+                    tracing::debug!(
+                        bucket = %self.bucket,
+                        path = %path,
+                        attempt,
+                        error = %e,
+                        "S3BlobBackend::get retrying transient 5xx"
+                    );
+                    BLOB_BACKEND_RETRY_COUNT
+                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                    tokio::time::sleep(blob_backend_retry_delay()).await;
+                    continue;
+                }
+                Err(e) => return Err(client_err_to_crypto(e)),
+            }
+        }
     }
 
+    /// Same retry policy as `get`. `put_object` is idempotent on v7 HAMT
+    /// node keys — they are content-addressed (blake3 over the plaintext
+    /// node), so re-uploading the same bytes at the same path is safe.
     async fn put(&self, path: &str, bytes: Vec<u8>) -> fula_crypto::Result<()> {
-        self.inner
-            .put_object(&self.bucket, path, bytes)
-            .await
-            .map(|_| ())
-            .map_err(client_err_to_crypto)
+        let mut attempt: u32 = 0;
+        loop {
+            attempt += 1;
+            // Clone the body each attempt: reqwest consumes the body, and we
+            // want to re-send the same bytes on retry. The retry path is cold
+            // and HAMT node blobs are small (sub-KB typical), so this is
+            // negligible on the happy path too.
+            let body = bytes.clone();
+            match self.inner.put_object(&self.bucket, path, body).await {
+                Ok(_) => return Ok(()),
+                Err(e)
+                    if attempt < BLOB_BACKEND_MAX_ATTEMPTS
+                        && crate::multipart::is_transient(&e) =>
+                {
+                    tracing::debug!(
+                        bucket = %self.bucket,
+                        path = %path,
+                        attempt,
+                        error = %e,
+                        "S3BlobBackend::put retrying transient 5xx"
+                    );
+                    BLOB_BACKEND_RETRY_COUNT
+                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                    tokio::time::sleep(blob_backend_retry_delay()).await;
+                    continue;
+                }
+                Err(e) => return Err(client_err_to_crypto(e)),
+            }
+        }
     }
 }
 
diff --git a/crates/fula-client/src/lib.rs b/crates/fula-client/src/lib.rs
@@ -57,7 +57,7 @@ pub(crate) static TEST_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()
 
 pub use client::FulaClient;
 pub use config::Config;
-pub use encryption::{EncryptedClient, EncryptionConfig, DecryptedObjectInfo, FileMetadata, DirectoryListing, PinningCredentials};
+pub use encryption::{EncryptedClient, EncryptionConfig, DecryptedObjectInfo, FileMetadata, DirectoryListing, PinningCredentials, S3BlobBackend};
 
 /// Crash-injection atomics used by the workspace integration tests to
 /// simulate client-side aborts between phases of `migrate_v1_to_v7_internal`.
@@ -102,6 +102,16 @@ pub fn flush_backoff_count() -> u64 {
     encryption::flush_backoff_count()
 }
 
+/// Total `S3BlobBackend::{get, put}` retries triggered by transient 5xx
+/// responses (typically nginx `limit_req` 503 on bursty HAMT walks) since
+/// process start. Monotonic process-wide counter.
+///
+/// Native-only; the wasm32 `BlobBackend` impl is single-attempt.
+#[cfg(not(target_arch = "wasm32"))]
+pub fn blob_backend_retry_count() -> u64 {
+    encryption::blob_backend_retry_count()
+}
+
 /// Count of WAL groups discarded on load due to partial-group truncation
 /// (M-4). A transactional multi-entry op written via the internal
 /// `append_group` path is applied all-or-none on replay: if any member is
diff --git a/crates/fula-client/src/multipart.rs b/crates/fula-client/src/multipart.rs
@@ -458,8 +458,10 @@ where
 ///
 /// Native-only: `retry_idempotent` short-circuits to a single attempt on
 /// wasm32 (no sleep primitive), so the classifier has no caller there.
+/// Shared with `S3BlobBackend::{get, put}` in `encryption.rs` so the
+/// blob-backend retry loop matches the same transient set.
 #[cfg(not(target_arch = "wasm32"))]
-fn is_transient(err: &ClientError) -> bool {
+pub(crate) fn is_transient(err: &ClientError) -> bool {
     match err {
         ClientError::Http(e) => {
             // `is_connect` exists only on native reqwest; the wasm build
diff --git a/crates/fula-client/tests/blob_backend_retries_transient.rs b/crates/fula-client/tests/blob_backend_retries_transient.rs
@@ -0,0 +1,204 @@
+//! Regression test for the v7 LIST-failure fix.
+//!
+//! Background: after the v1→v7 migration of a large bucket, LIST walks the
+//! sharded HAMT via `list_all_files`, which fans out to many concurrent
+//! `S3BlobBackend::get` calls on `/bucket/__fula_forest_v7_nodes/<key>`. When
+//! the gateway is fronted by an nginx `limit_req` with a `burst` smaller than
+//! the in-flight count, some requests come back as HTTP 503 with an empty
+//! body — a single one of those aborted the whole LIST.
+//!
+//! The fix adds bounded fixed-delay retries (300 ms base + 0-100 ms jitter,
+//! up to 4 attempts) in `S3BlobBackend::{get, put}` for the transient class
+//! HTTP 429/500/502/503/504 / `SlowDown` / `InternalError` /
+//! `ServiceUnavailable`. This test exercises that retry path end-to-end
+//! against a `wiremock::MockServer` that emits 503 → 503 → 200 and asserts:
+//!
+//! 1. The outer call returns `Ok`.
+//! 2. `blob_backend_retry_count()` observes exactly the number of retry
+//!    sleeps we expect.
+//! 3. Non-transient responses (e.g. 404) are NOT retried.
+
+#![cfg(not(target_arch = "wasm32"))]
+
+use fula_client::{Config, FulaClient, S3BlobBackend, blob_backend_retry_count};
+use fula_crypto::BlobBackend;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use wiremock::matchers::{method, path};
+use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
+
+/// The retry counter is process-wide; tests in this file all mutate it.
+/// Serialize counter-sensitive tests under a shared lock so the `before →
+/// after` delta measured by each test is actually its own increments.
+static COUNTER_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// Responder that emits the Nth response from a list; cycles if called more
+/// than `responses.len()` times. Lets us script "503, 503, 200" against a
+/// single path without wiremock's expectation counters fighting us.
+struct Scripted {
+    calls: Arc<AtomicUsize>,
+    responses: Vec<ResponseTemplate>,
+}
+
+impl Respond for Scripted {
+    fn respond(&self, _req: &Request) -> ResponseTemplate {
+        let idx = self.calls.fetch_add(1, Ordering::SeqCst);
+        let slot = idx.min(self.responses.len() - 1);
+        self.responses[slot].clone()
+    }
+}
+
+fn mk_client(endpoint: &str) -> FulaClient {
+    // 10 s per-request timeout is ample for mock responses; the test doesn't
+    // need a real connect timeout.
+    let cfg = Config::new(endpoint);
+    FulaClient::new(cfg).expect("build FulaClient")
+}
+
+#[tokio::test]
+async fn get_retries_through_two_503s_then_succeeds() {
+    let _guard = COUNTER_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+    let server = MockServer::start().await;
+
+    let calls = Arc::new(AtomicUsize::new(0));
+    let responder = Scripted {
+        calls: calls.clone(),
+        responses: vec![
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(200).set_body_bytes(b"hamt-node-bytes".to_vec()),
+        ],
+    };
+    Mock::given(method("GET"))
+        .and(path("/images/__fula_forest_v7_nodes/deadbeef"))
+        .respond_with(responder)
+        .mount(&server)
+        .await;
+
+    let before = blob_backend_retry_count();
+    let client = mk_client(&server.uri());
+    let backend = S3BlobBackend::new(client, "images".to_string());
+
+    let got = backend
+        .get("__fula_forest_v7_nodes/deadbeef")
+        .await
+        .expect("retry should absorb two 503s");
+
+    assert_eq!(got, b"hamt-node-bytes");
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        3,
+        "mock should have been hit once per attempt"
+    );
+    let retries = blob_backend_retry_count() - before;
+    assert_eq!(retries, 2, "two 503s → two retry sleeps, then success");
+}
+
+#[tokio::test]
+async fn put_retries_through_one_503_then_succeeds() {
+    let _guard = COUNTER_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+    let server = MockServer::start().await;
+
+    let calls = Arc::new(AtomicUsize::new(0));
+    let responder = Scripted {
+        calls: calls.clone(),
+        responses: vec![
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(200).insert_header("ETag", "\"abc123\""),
+        ],
+    };
+    Mock::given(method("PUT"))
+        .and(path("/images/__fula_forest_v7_nodes/cafebabe"))
+        .respond_with(responder)
+        .mount(&server)
+        .await;
+
+    let before = blob_backend_retry_count();
+    let client = mk_client(&server.uri());
+    let backend = S3BlobBackend::new(client, "images".to_string());
+
+    backend
+        .put("__fula_forest_v7_nodes/cafebabe", b"encrypted-node-blob".to_vec())
+        .await
+        .expect("retry should absorb the 503");
+
+    assert_eq!(calls.load(Ordering::SeqCst), 2);
+    let retries = blob_backend_retry_count() - before;
+    assert_eq!(retries, 1, "one 503 → one retry sleep");
+}
+
+#[tokio::test]
+async fn get_gives_up_after_max_attempts_on_persistent_503() {
+    let _guard = COUNTER_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+    let server = MockServer::start().await;
+
+    let calls = Arc::new(AtomicUsize::new(0));
+    let responder = Scripted {
+        calls: calls.clone(),
+        // Always 503 — more entries than attempts so Scripted never wraps.
+        responses: vec![
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(503),
+            ResponseTemplate::new(503),
+        ],
+    };
+    Mock::given(method("GET"))
+        .and(path("/images/__fula_forest_v7_nodes/persistently-broken"))
+        .respond_with(responder)
+        .mount(&server)
+        .await;
+
+    let client = mk_client(&server.uri());
+    let backend = S3BlobBackend::new(client, "images".to_string());
+
+    let err = backend
+        .get("__fula_forest_v7_nodes/persistently-broken")
+        .await
+        .expect_err("persistent 503 must eventually surface");
+
+    // Four total attempts = `BLOB_BACKEND_MAX_ATTEMPTS`.
+    assert_eq!(calls.load(Ordering::SeqCst), 4, "attempts capped at 4");
+    // Error message should mention the storage-backend failure; we don't
+    // pin the exact string (ClientError::to_string is wrapped via
+    // client_err_to_crypto → CryptoError::Storage).
+    let msg = err.to_string();
+    assert!(
+        msg.to_lowercase().contains("storage") || msg.contains("503"),
+        "unexpected error message: {msg}"
+    );
+}
+
+#[tokio::test]
+async fn get_does_not_retry_on_404() {
+    let _guard = COUNTER_LOCK.lock().unwrap_or_else(|p| p.into_inner());
+    let server = MockServer::start().await;
+
+    let calls = Arc::new(AtomicUsize::new(0));
+    let responder = Scripted {
+        calls: calls.clone(),
+        responses: vec![ResponseTemplate::new(404)],
+    };
+    Mock::given(method("GET"))
+        .and(path("/images/__fula_forest_v7_nodes/not-there"))
+        .respond_with(responder)
+        .mount(&server)
+        .await;
+
+    let before = blob_backend_retry_count();
+    let client = mk_client(&server.uri());
+    let backend = S3BlobBackend::new(client, "images".to_string());
+
+    let _err = backend
+        .get("__fula_forest_v7_nodes/not-there")
+        .await
+        .expect_err("404 must not be retried");
+
+    assert_eq!(calls.load(Ordering::SeqCst), 1, "404 is terminal");
+    assert_eq!(
+        blob_backend_retry_count(),
+        before,
+        "non-transient error must not bump retry counter"
+    );
+}
