blob-client: a rate-limited PUT backs off for seconds, not 300 ms

crates/engram-blob-client/Cargo.toml

@@ -25,7 +25,7 @@ workspace-hack = { version = "0.1", path = "../../workspace-hack" }
 engram-testkit.workspace = true
 futures.workspace = true
 tempfile = "3"
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "test-util"] }
 # blobbench (examples/blobbench.rs) only — the manual before/after
 # throughput harness. Not a CI lane; compiled by --all-targets so it
 # can't rot.

crates/engram-chunk-store/src/admission.rs

@@ -102,11 +102,11 @@ impl Permit {
 }
 
 /// Whether a store error can mean the link or the store is overloaded:
-/// an SDK, I/O or protocol error (a timeout arrives as an SDK error). A
-/// missing object or a bad configuration cannot.
+/// a throttle, an SDK, I/O or protocol error (a timeout arrives as an SDK
+/// error). A missing object or a bad configuration cannot.
 pub(crate) fn is_overload(e: &engram_core::error::BlobError) -> bool {
     use engram_core::error::BlobError as B;
-    matches!(e, B::Sdk(_) | B::Io(_) | B::Protocol(_))
+    matches!(e, B::Throttled(_) | B::Sdk(_) | B::Io(_) | B::Protocol(_))
 }
 
 impl Admission {

crates/engram-chunk-store/src/error.rs

@@ -115,6 +115,7 @@ impl ChunkStoreError {
             Self::Blob(b) => Self::Blob(match b {
                 BlobError::NotFound => BlobError::NotFound,
                 BlobError::Sdk(e) => BlobError::Sdk(e.to_string().into()),
+                BlobError::Throttled(e) => BlobError::Throttled(e.to_string().into()),
                 BlobError::Config(m) => BlobError::Config(m.clone()),
                 BlobError::Protocol(m) => BlobError::Protocol(m.clone()),
                 BlobError::Io(e) => BlobError::Io(io(e)),

crates/engram-core/src/error.rs

@@ -252,6 +252,9 @@ pub enum BlobError {
     NotFound,
     /// Underlying SDK or HTTP error.
     Sdk(BoxError),
+    /// The store refused the request because of load (HTTP 429, or 503
+    /// and S3 `SlowDown`). The request can succeed after a backoff.
+    Throttled(BoxError),
     /// Configuration was invalid or missing (no creds, bad endpoint).
     Config(String),
     /// Backend returned an unexpected response shape.
@@ -269,6 +272,7 @@ impl fmt::Display for BlobError {
         match self {
             Self::NotFound => write!(f, "blob not found"),
             Self::Sdk(e) => write!(f, "blob sdk error: {e}"),
+            Self::Throttled(e) => write!(f, "blob store throttled the request: {e}"),
             Self::Config(msg) => write!(f, "blob config error: {msg}"),
             Self::Protocol(msg) => write!(f, "blob protocol error: {msg}"),
             Self::Io(e) => write!(f, "blob io error: {e}"),
@@ -282,7 +286,7 @@ impl fmt::Display for BlobError {
 impl StdError for BlobError {
     fn source(&self) -> Option<&(dyn StdError + 'static)> {
         match self {
-            Self::Sdk(e) => Some(&**e),
+            Self::Sdk(e) | Self::Throttled(e) => Some(&**e),
             Self::Io(e) => Some(e),
             _ => None,
         }

crates/engram-core/src/traits/storage.rs

@@ -123,6 +123,8 @@ pub struct BlobObjectMeta {
 /// - Streaming both directions; never materialize bodies in RAM.
 /// - Mapping backend NotFound (404 / NoSuchKey) into
 ///   `BlobError::NotFound`; idempotent delete on missing keys.
+/// - Mapping a refusal for load (HTTP 429 / 503) into
+///   `BlobError::Throttled`.
 /// - All other errors surface as `BlobError::Sdk` with the underlying
 ///   error in the source chain.
 #[async_trait]

crates/engram-storage-gcs/src/lib.rs

@@ -164,17 +164,12 @@ fn map_http_err(e: GcsHttpError) -> BlobError {
     // Both shapes mean the same thing: object doesn't exist.
     // Collapse them so callers downstream of `get_streaming` get
     // the same `BlobError::NotFound` that callers of `head` do.
-    let is_404 = match &e {
-        GcsHttpError::Response(resp) => resp.code == 404,
-        GcsHttpError::HttpClient(rq) | GcsHttpError::RawResponse(rq, _) => {
-            rq.status().is_some_and(|s| s.as_u16() == 404)
-        }
-        _ => false,
-    };
-    if is_404 {
-        BlobError::NotFound
-    } else {
-        BlobError::Sdk(Box::new(e))
+    // GCS answers 429 when a bucket's write rate grows faster than it
+    // redistributes load, and 503 when a backend is busy.
+    match http_status(&e) {
+        Some(404) => BlobError::NotFound,
+        Some(429 | 503) => BlobError::Throttled(Box::new(e)),
+        _ => BlobError::Sdk(Box::new(e)),
     }
 }
 
@@ -402,6 +397,42 @@ mod tests {
 
     use engram_testkit::blob_conformance;
 
+    fn json_error(code: u16, message: &str) -> GcsHttpError {
+        GcsHttpError::Response(gcloud_storage::http::error::ErrorResponse {
+            code,
+            errors: Vec::new(),
+            message: message.to_string(),
+        })
+    }
+
+    #[test]
+    fn status_codes_map_to_blob_errors() {
+        let rate_limited = json_error(
+            429,
+            "Your request distribution is too uneven across the key-ranges in your bucket.",
+        );
+        assert!(matches!(
+            map_http_err(rate_limited),
+            BlobError::Throttled(_)
+        ));
+        assert!(matches!(
+            map_http_err(json_error(503, "backendError")),
+            BlobError::Throttled(_)
+        ));
+        assert!(matches!(
+            map_http_err(json_error(404, "No such object")),
+            BlobError::NotFound
+        ));
+        assert!(matches!(
+            map_http_err(json_error(500, "internal")),
+            BlobError::Sdk(_)
+        ));
+        assert!(matches!(
+            map_http_err(json_error(403, "forbidden")),
+            BlobError::Sdk(_)
+        ));
+    }
+
     /// Pins the rustls CryptoProvider contract for this crate's clients.
     ///
     /// `connect()` reaches TLS twice — the tuned client it builds, and the

crates/engram-storage-s3/src/lib.rs

@@ -204,14 +204,20 @@ impl PartAccumulator {
     }
 }
 
-/// Wrap any SDK error as [`BlobError::Sdk`], keeping the full context
-/// chain (`DisplayErrorContext` renders service code + message).
-fn sdk_err<E, R>(err: SdkError<E, R>) -> BlobError
+/// Wrap an SDK error as [`BlobError::Throttled`] when S3 refused it for
+/// load (503 `SlowDown`, or 429), else as [`BlobError::Sdk`], keeping the
+/// full context chain (`DisplayErrorContext` renders service code +
+/// message).
+fn sdk_err<E>(err: SdkError<E>) -> BlobError
 where
     E: std::error::Error + Send + Sync + 'static,
-    R: std::fmt::Debug + Send + Sync + 'static,
 {
-    BlobError::Sdk(err.into_service_error_or_self())
+    let status = err.raw_response().map(|r| r.status().as_u16());
+    let inner = err.into_service_error_or_self();
+    match status {
+        Some(429 | 503) => BlobError::Throttled(inner),
+        _ => BlobError::Sdk(inner),
+    }
 }
 
 /// The SDK's error Display shows only the outermost layer; this

crates/engram-testkit/src/storage.rs

@@ -104,6 +104,8 @@ pub enum InjectedError {
     Sdk(String),
     /// `BlobError::Protocol(_)` — an unexpected response shape.
     Protocol(String),
+    /// `BlobError::Throttled(_)` — the store refused the request for load.
+    Throttled(String),
 }
 
 impl InjectedError {
@@ -117,6 +119,7 @@ impl InjectedError {
                 msg.clone(),
             )),
             Self::Protocol(msg) => BlobError::Protocol(msg.clone()),
+            Self::Throttled(msg) => BlobError::Throttled(msg.clone().into()),
         }
     }
 }