feat(errors): classify transient failures on the type, not by string-matching

Step 1 of docs/plan/jobs-handling-recoverable-error.md, and the blocker
for the rest of it: the engine cannot retry-then-pause until it can tell
"the provider is down" from "this data is wrong". Both arrived as
`ErrorKind::InternalError`, so the distinction survived only inside a
formatted message.

`RetryBlobBackend` was reading that message. Literally:

    let msg = err.to_string().to_lowercase();
    msg.contains("timeout") || msg.contains("503") || msg.contains("reset by peer")

Fragile in a specific way — an SDK reformatting its `Display` turns
retrying off with nothing failing to say so — and blind to any status
code that never made it into the text.

Adds `ErrorKind::TransientBackend` and `DomainError::is_transient()`.
One predicate, so the retry decorator and the job engine cannot classify
the same failure differently. `Timeout` counts (transient by
construction); everything else must say so explicitly. The default is
"not retryable" because that fails visibly, whereas retrying a permanent
fault burns attempts and — once the engine wires this up — holds
`migration_readonly` while it does.

A kind rather than a `transient: bool` field: 21 struct-literal sites
construct `DomainError` directly and would all have needed touching for
a change that is conceptually about classification. The plan allowed
either.

`s3_domain_error` does the classification where the status is still in
hand. Transient: 5xx, 429, and the SlowDown / RequestTimeout /
ThrottlingException codes that arrive as 400 (status alone is not
enough), plus dispatch-level I/O and timeouts. Permanent: other 4xx —
credentials, missing bucket, malformed request — and construction
failures. `ResponseError` counts as transient since truncation on the
wire is the usual cause and the attempt cap bounds being wrong.

Applied at the five S3 sites that wrap an SDK error, including
`ListObjectsV2` — `backend_consistency` fails the whole run on an
enumeration error, so a throttle midway through a large bucket should be
retryable rather than discarding the sweep.

The exhaustive `ErrorKind` match in `interfaces/errors.rs` forced the
HTTP decision, which is the right friction: 503, not 500. The request
was fine and may succeed shortly, which is what a caller needs to decide
whether to retry and what a proxy keys off to avoid caching the failure.

The substring matcher stays for now, behind the typed check, with the
deletion condition written down: it goes when every backend wrapping a
remote SDK error classifies at the point of wrapping. Removing it before
then would silently reduce retrying on the unconverted backends, which
is the worse direction. Azure is the one left, and it is queued for the
official-SDK migration anyway.

Not yet wired: `RunOutcome::PausedRetryable` (step 2) and the engine's
bounded backoff (step 3). This commit only makes the distinction
representable.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/domain/errors.rs

@@ -23,6 +23,30 @@ pub enum ErrorKind {
     AccessDenied,
     /// Timeout expired
     Timeout,
+    /// A dependency failed in a way that may clear on its own — an HTTP
+    /// 5xx or 429 from object storage, a connection reset, a DNS
+    /// failure.
+    ///
+    /// Distinct from [`ErrorKind::InternalError`] because the engine has
+    /// to tell "the provider is down" from "this data is wrong": the
+    /// first is worth retrying and then pausing so an operator can
+    /// resume, the second is terminal. Flattening both into
+    /// `InternalError` is what forced `RetryBlobBackend` to classify by
+    /// string-matching `Display` output — fragile in exactly the way
+    /// that turns an SDK's cosmetic reformat into a silent behaviour
+    /// change.
+    ///
+    /// **Set it deliberately, at the point where the status code is
+    /// still visible** — the port wrapping the SDK error. By the time an
+    /// error reaches the engine, the code survives only inside a
+    /// formatted string.
+    ///
+    /// Not a promise that a retry succeeds. A deterministic 500 (Azurite
+    /// answering the CRC64 ranged GET) is a permanent fault wearing a
+    /// retryable status code, which no status-based taxonomy can get
+    /// right — the bounded attempt cap is the safety net for exactly
+    /// that. See `docs/plan/jobs-handling-recoverable-error.md`.
+    TransientBackend,
     /// Internal system error
     InternalError,
     /// Functionality not implemented
@@ -58,6 +82,10 @@ impl ErrorKind {
             ErrorKind::InvalidInput => "Invalid Input",
             ErrorKind::AccessDenied => "Access Denied",
             ErrorKind::Timeout => "Timeout",
+            // Wire value — the SPA switches on `error_type`, so this
+            // string is a contract. Additive here; nothing keys off it
+            // yet.
+            ErrorKind::TransientBackend => "Transient Backend",
             ErrorKind::InternalError => "Internal Error",
             ErrorKind::NotImplemented => "Not Implemented",
             ErrorKind::UnsupportedOperation => "Unsupported Operation",
@@ -148,6 +176,36 @@ impl DomainError {
         }
     }
 
+    /// A dependency failed in a way that may clear on its own. See
+    /// [`ErrorKind::TransientBackend`] for what qualifies and why the
+    /// classification belongs at the port rather than downstream.
+    pub fn transient_backend<S: Into<String>>(entity_type: &'static str, message: S) -> Self {
+        Self {
+            kind: ErrorKind::TransientBackend,
+            entity_type,
+            entity_id: None,
+            message: message.into(),
+            source: None,
+        }
+    }
+
+    /// Whether retrying this operation could plausibly succeed.
+    ///
+    /// The single place that answers the question, so a retry decorator
+    /// and the job engine cannot disagree about the same error — they
+    /// did while the answer was `Display` string-matching in one of
+    /// them and nothing in the other.
+    ///
+    /// `Timeout` is included because it is transient by construction;
+    /// everything else must say so explicitly via
+    /// [`ErrorKind::TransientBackend`]. Defaulting to "not retryable" is
+    /// the safe direction: a missed retry surfaces as a visible failure,
+    /// whereas retrying a permanent fault burns attempts and, in the
+    /// job engine, holds `migration_readonly` while it does.
+    pub fn is_transient(&self) -> bool {
+        matches!(self.kind, ErrorKind::Timeout | ErrorKind::TransientBackend)
+    }
+
     /// Creates an internal error
     pub fn internal_error<S: Into<String>>(entity_type: &'static str, message: S) -> Self {
         Self {
@@ -317,3 +375,41 @@ impl From<uuid::Error> for DomainError {
         }
     }
 }
+
+#[cfg(test)]
+mod transient_tests {
+    use super::*;
+
+    /// The retry decorator and the job engine both branch on this, so
+    /// the set has to be deliberate rather than incidental.
+    #[test]
+    fn only_timeout_and_transient_backend_are_retryable() {
+        assert!(DomainError::transient_backend("S3", "503").is_transient());
+        assert!(DomainError::timeout("S3", "read timed out").is_transient());
+
+        // Everything else defaults to permanent. Retrying a genuine
+        // fault burns attempts and, in the job engine, holds
+        // `migration_readonly` while it does — so the default has to be
+        // "no".
+        for e in [
+            DomainError::internal_error("S3", "decode failed"),
+            DomainError::new(ErrorKind::NotFound, "Blob", "missing"),
+            DomainError::new(ErrorKind::AccessDenied, "S3", "bad credentials"),
+            DomainError::new(ErrorKind::InvalidInput, "S3", "malformed key"),
+            DomainError::new(ErrorKind::UnsupportedOperation, "S3", "no enumeration"),
+        ] {
+            assert!(
+                !e.is_transient(),
+                "{:?} must not be retryable by default",
+                e.kind
+            );
+        }
+    }
+
+    /// `error_type` is a wire contract the SPA switches on, so this
+    /// string is not free to churn.
+    #[test]
+    fn transient_backend_has_a_stable_wire_name() {
+        assert_eq!(ErrorKind::TransientBackend.as_str(), "Transient Backend");
+    }
+}

src/infrastructure/services/retry_blob_backend.rs

@@ -99,7 +99,28 @@ where
 }
 
 /// Determine if an error is likely transient (network timeout, 5xx, etc.).
+///
+/// Asks the error first. `DomainError::is_transient` is the single
+/// answer to that question, so this decorator and the job engine cannot
+/// classify the same failure differently.
+///
+/// **The substring arm is transitional.** It is what this function used
+/// to be, in full: a `to_lowercase()` scan of `Display` output for
+/// "timeout", "503", "reset by peer" and friends. That is fragile in a
+/// specific way — an SDK reformatting its error text silently turns
+/// retries off, with nothing failing to say so — and it cannot see a
+/// status code that never made it into the message.
+///
+/// It stays only until every backend classifies at the point of
+/// wrapping, where the status is still in hand. Deleting it before then
+/// would silently REDUCE retrying on the backends not yet converted,
+/// which is the worse direction to be wrong in. Delete it once
+/// `grep -rn "transient_backend" src/infrastructure/services/` covers
+/// every backend that wraps a remote SDK error.
 fn is_retryable(err: &DomainError) -> bool {
+    if err.is_transient() {
+        return true;
+    }
     let msg = err.to_string().to_lowercase();
     msg.contains("timeout")
         || msg.contains("connection")

src/infrastructure/services/s3_blob_backend.rs

@@ -170,12 +170,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(body)
                 .send()
                 .await
-                .map_err(|e| {
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to upload blob {}: {}", hash, e),
-                    )
-                })?;
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
 
             // Clean up local source after successful upload
             let _ = fs::remove_file(&source_path).await;
@@ -215,12 +210,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(body)
                 .send()
                 .await
-                .map_err(|e| {
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to upload blob {}: {}", hash, e),
-                    )
-                })?;
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
 
             Ok(size)
         })
@@ -253,12 +243,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(ByteStream::from(data))
                 .send()
                 .await
-                .map_err(|e| {
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to upload blob {}: {}", hash, e),
-                    )
-                })?;
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
             Ok(size)
         })
     }
@@ -363,12 +348,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .key(&key)
                 .send()
                 .await
-                .map_err(|e| {
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to delete blob {}: {}", hash, e),
-                    )
-                })?;
+                .map_err(|e| s3_domain_error("S3", format!("Failed to delete blob {hash}"), &e))?;
 
             Ok(())
         })
@@ -565,11 +545,11 @@ impl BlobStorageBackend for S3BlobBackend {
                 }
 
                 let resp = req.send().await.map_err(|e| {
-                    DomainError::new(
-                        ErrorKind::InternalError,
-                        "Blob",
-                        format!("S3 ListObjectsV2 failed: {e}"),
-                    )
+                    // Classified, because `backend_consistency` fails the
+                    // whole run on an enumeration error — a throttle
+                    // midway through a million-object bucket should be
+                    // retryable rather than throwing the sweep away.
+                    s3_domain_error("Blob", "S3 ListObjectsV2 failed".to_string(), &e)
                 })?;
 
                 requests += 1;
@@ -638,6 +618,60 @@ impl BlobStorageBackend for S3BlobBackend {
     }
 }
 
+/// Wrap an SDK error as a `DomainError` that says whether retrying it
+/// could help.
+///
+/// The classification has to happen HERE. One layer up the status code
+/// survives only inside a formatted string, which is what forced
+/// `RetryBlobBackend` to grep its own error text for "503" — a check
+/// that silently stops working when an SDK reformats `Display`.
+///
+/// Transient: 5xx and 429 from the service, plus dispatch-level I/O and
+/// timeouts (DNS, TLS, connection refused, TCP reset). Permanent:
+/// everything 4xx except 429 — credentials, a missing bucket, a
+/// malformed request — and client-side construction failures, none of
+/// which a second attempt changes.
+///
+/// `ResponseError` (a reply the SDK could not parse) counts as
+/// transient: truncation on the wire is the usual cause, and the
+/// attempt cap bounds the cost of being wrong.
+pub(crate) fn s3_domain_error<E>(
+    entity: &'static str,
+    context: String,
+    err: &aws_sdk_s3::error::SdkError<E>,
+) -> DomainError
+where
+    E: aws_sdk_s3::error::ProvideErrorMetadata + std::fmt::Debug,
+{
+    use aws_sdk_s3::error::SdkError;
+
+    let transient = match err {
+        SdkError::ServiceError(svc) => {
+            let status = svc.raw().status().as_u16();
+            let code = svc.err().meta().code().unwrap_or_default();
+            status >= 500
+                || status == 429
+                // Throttling can arrive as 400 with a code rather than
+                // 429, so the status alone is not enough.
+                || code.eq_ignore_ascii_case("SlowDown")
+                || code.eq_ignore_ascii_case("RequestTimeout")
+                || code.eq_ignore_ascii_case("ThrottlingException")
+        }
+        SdkError::DispatchFailure(d) => d.is_io() || d.is_timeout(),
+        SdkError::TimeoutError(_) => true,
+        SdkError::ResponseError(_) => true,
+        SdkError::ConstructionFailure(_) => false,
+        _ => false,
+    };
+
+    let message = format!("{context}: {}", format_s3_error(err));
+    if transient {
+        DomainError::transient_backend(entity, message)
+    } else {
+        DomainError::internal_error(entity, message)
+    }
+}
+
 /// Extract an actionable error string from an aws-sdk-s3 error.
 ///
 /// `SdkError::Display` renders literally `"service error"` when the
@@ -656,6 +690,10 @@ impl BlobStorageBackend for S3BlobBackend {
 /// - `unknown SDK error: <debug>` — anything else, with the full
 ///   `Debug` output so the operator + audit stream see the real cause
 ///   instead of `"service error"`.
+///
+/// Formatting only. Whether the error is worth retrying is
+/// [`s3_domain_error`]'s job, from the structured variant rather than
+/// from this string.
 fn format_s3_error<E>(err: &aws_sdk_s3::error::SdkError<E>) -> String
 where
     E: aws_sdk_s3::error::ProvideErrorMetadata + std::fmt::Debug,

src/interfaces/errors.rs

@@ -132,6 +132,11 @@ impl From<DomainError> for AppError {
             ErrorKind::QuotaExceeded => StatusCode::INSUFFICIENT_STORAGE,
             ErrorKind::Conflict => StatusCode::CONFLICT,
             ErrorKind::PreconditionFailed => StatusCode::PRECONDITION_FAILED,
+            // 503, not 500: the request was fine and the same request
+            // may well succeed shortly. That is what a caller needs to
+            // decide whether to retry, and it is what a reverse proxy
+            // keys off to avoid caching the failure.
+            ErrorKind::TransientBackend => StatusCode::SERVICE_UNAVAILABLE,
         };
 
         Self {