feat(jobs): a transient backend failure pauses at its cursor instead of failing

Step 3 of docs/plan/jobs-handling-recoverable-error.md, and it
deliberately does NOT add the retry loop the plan sketched. Reasoning
below.

`RunOutcome::from_domain_error(cursor, context, err)` routes a failed
operation to `PausedRetryable` when the error is transient and `Failed`
otherwise. Handlers call it instead of reaching for `Failed`, so an
outage stops a long scan at its cursor rather than discarding it —
`Failed` is terminal, and only `Paused` resumes.

Applied to `backend_consistency`'s enumeration failure first, because
that is the case with the most to lose: the job fails the whole run on
an enumeration error, so a brief 503 partway through a million-object
bucket used to throw away the entire audit.

## Why no bounded retry loop in the engine

The plan said "bounded exponential backoff, ~5 attempts" in
`run_or_resume`, and also warned "do not double-retry — the AWS SDK
already retries internally, so a second layer above it multiplies".
Checking before writing it, there are already TWO layers:

  * the AWS SDK retries internally;
  * `RetryBlobBackend` wraps every remote backend with exponential
    backoff — 3 retries, 100 ms initial, ×2, 10 s cap, all tunable via
    OXICLOUD_STORAGE_RETRY_*, and applied in di.rs for non-Local
    backends.

A third layer multiplies rather than adds: one logical operation could
span SDK × decorator × engine attempts, turning a brief outage into
minutes of held `migration_readonly` — the precise failure this plan
exists to stop.

Retrying here would also re-run a SCAN, not an operation. The retrying
belongs where it already is, per request; what was genuinely missing is
the conversion of an exhausted-retry failure into a resumable pause with
a reason, which is what this commit adds. If the attempt budget needs
tuning, `OXICLOUD_STORAGE_RETRY_MAX_RETRIES` is the knob, and it applies
to every backend call rather than only to jobs.

## Tests

`transient_failure_pauses_with_a_reason_and_keeps_the_cursor` asserts
the three things that matter: status Paused, cursor preserved,
`error_message` naming the cause. `permanent_failure_still_fails_terminally`
is the control — without it the classification could be inert and
everything would simply pause, which would look like success.

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

src/infrastructure/scheduler/recoverable.rs

@@ -223,6 +223,56 @@ impl RunOutcome {
             ),
         }
     }
+
+    /// Turn a failed operation into the right outcome:
+    /// [`RunOutcome::PausedRetryable`] when the error is transient,
+    /// [`RunOutcome::Failed`] otherwise.
+    ///
+    /// **This is where step 1's classification pays off.** Handlers
+    /// should route every backend error through here rather than
+    /// reaching for `Failed` directly, so "the provider is down" stops a
+    /// long scan at its cursor instead of discarding it.
+    ///
+    /// `cursor` is the resume position — normally the same value the
+    /// handler last checkpointed. Pass `None` only when nothing has been
+    /// settled yet; the run then resumes from the beginning.
+    ///
+    /// # Why the engine does not add its own retry loop
+    ///
+    /// The plan sketched bounded backoff *here*. Measuring first showed
+    /// two layers already exist below: the AWS SDK retries internally,
+    /// and `RetryBlobBackend` wraps every remote backend with its own
+    /// exponential backoff (defaults: 3 retries, 100 ms, ×2, 10 s cap —
+    /// all env-tunable). A third layer would multiply, not add: one
+    /// logical operation could span SDK × decorator × engine attempts,
+    /// turning a brief outage into minutes of held `migration_readonly`.
+    ///
+    /// The plan anticipated exactly this — "do not double-retry … the
+    /// AWS SDK already retries internally, so a second layer above it
+    /// multiplies" — so the retrying stays where it already is, at the
+    /// operation, and the engine supplies the part that was genuinely
+    /// missing: converting an exhausted-retry failure into a resumable
+    /// pause with a reason instead of a terminal `Failed`.
+    ///
+    /// Retrying at this level would also mean re-running a scan, not an
+    /// operation. Tuning attempts belongs in
+    /// `OXICLOUD_STORAGE_RETRY_*`, where it applies per request.
+    pub fn from_domain_error(
+        cursor: Option<&[u8]>,
+        context: &str,
+        err: &crate::domain::errors::DomainError,
+    ) -> Self {
+        if err.is_transient() {
+            RunOutcome::PausedRetryable {
+                cursor: cursor.map(<[u8]>::to_vec).unwrap_or_default(),
+                reason: format!("{context}: {err}"),
+            }
+        } else {
+            RunOutcome::Failed {
+                message: format!("{context}: {err}"),
+            }
+        }
+    }
 }
 
 /// Write `JobRunArgs` to `params` on a Fresh run, or read them back on a
@@ -1515,6 +1565,16 @@ mod tests {
                 .last()
                 .and_then(|s| s.state.lock().unwrap().cursor.clone())
         }
+
+        /// Test-only read — last-created run's `error_message`. What
+        /// separates an operator pause from a provider outage: both are
+        /// `Paused`, only one carries a reason.
+        fn last_error_message(&self) -> Option<String> {
+            let stores = self.stores.lock().unwrap();
+            stores
+                .last()
+                .and_then(|s| s.state.lock().unwrap().error_message.clone())
+        }
     }
 
     #[async_trait]
@@ -1739,6 +1799,51 @@ mod tests {
 
     // ─── Handlers ──────────────────────────────────────────────────────────
 
+    /// Hits a transient backend error partway through, exactly as a
+    /// remote backend does once its own retry decorator has given up.
+    struct TransientlyFailingHandler;
+    #[async_trait]
+    impl RecoverableJobHandler for TransientlyFailingHandler {
+        fn name(&self) -> &str {
+            "transient_failer"
+        }
+        async fn run_resumable(
+            &self,
+            store: &dyn JobStore,
+            _args: &JobRunArgs,
+            _resume_cursor: Option<Vec<u8>>,
+        ) -> RunOutcome {
+            store.checkpoint(vec![9, 9], 3).await.unwrap();
+            RunOutcome::from_domain_error(
+                Some(&[9, 9]),
+                "backend enumeration failed on s3",
+                &crate::domain::errors::DomainError::transient_backend("S3", "503 SlowDown"),
+            )
+        }
+    }
+
+    /// Same shape, but a permanent fault — the control that proves the
+    /// classification is doing the work rather than everything pausing.
+    struct PermanentlyFailingHandler;
+    #[async_trait]
+    impl RecoverableJobHandler for PermanentlyFailingHandler {
+        fn name(&self) -> &str {
+            "permanent_failer"
+        }
+        async fn run_resumable(
+            &self,
+            _store: &dyn JobStore,
+            _args: &JobRunArgs,
+            _resume_cursor: Option<Vec<u8>>,
+        ) -> RunOutcome {
+            RunOutcome::from_domain_error(
+                Some(&[9, 9]),
+                "backend enumeration failed on s3",
+                &crate::domain::errors::DomainError::internal_error("S3", "403 AccessDenied"),
+            )
+        }
+    }
+
     struct CompletingHandler;
     #[async_trait]
     impl RecoverableJobHandler for CompletingHandler {
@@ -1891,6 +1996,69 @@ mod tests {
         assert_eq!(provider.last_status(), Some(RunStatus::Completed));
     }
 
+    /// A transient backend failure must PAUSE with a reason, not fail.
+    ///
+    /// This is the whole point of the plan: `Failed` is terminal, so an
+    /// outage used to discard a partially-complete migration. The run has
+    /// to keep its cursor and stay resumable, and it has to say why it
+    /// stopped — a paused `backend_migration` still holds
+    /// `migration_readonly`, refusing writes application-wide, so
+    /// "someone paused this" and "the provider went down" cannot look
+    /// alike.
+    #[tokio::test]
+    async fn transient_failure_pauses_with_a_reason_and_keeps_the_cursor() {
+        let provider = Arc::new(MemProvider::new());
+        let provider_trait: Arc<dyn JobStoreProvider> = provider.clone();
+
+        let outcome = run_or_resume(
+            Arc::new(TransientlyFailingHandler),
+            provider_trait,
+            &JobRunArgs::default(),
+        )
+        .await;
+
+        // Reported Ok, not Err: the run did not fail, it stopped and can
+        // be resumed. A red job that a Resume click fixes reads as a bug
+        // rather than a decision waiting to be made.
+        assert!(outcome.is_ok(), "expected Ok, got {outcome:?}");
+        if let JobOutcome::Ok { extra, .. } = outcome {
+            assert_eq!(extra["paused"], true);
+            assert_eq!(extra["retryable"], true);
+            assert!(
+                extra["reason"].as_str().unwrap().contains("503"),
+                "the reason must reach the panel: {extra:?}"
+            );
+        }
+
+        assert_eq!(provider.last_status(), Some(RunStatus::Paused));
+        assert_eq!(
+            provider.last_cursor(),
+            Some(vec![9, 9]),
+            "resume position must survive, or the outage costs the whole scan"
+        );
+        let msg = provider.last_error_message().expect("reason recorded");
+        assert!(msg.contains("503"), "error_message names the cause: {msg}");
+    }
+
+    /// The control: a permanent fault still fails terminally. Without
+    /// this the classification could be doing nothing and everything
+    /// would simply pause, which looks like success in the test above.
+    #[tokio::test]
+    async fn permanent_failure_still_fails_terminally() {
+        let provider = Arc::new(MemProvider::new());
+        let provider_trait: Arc<dyn JobStoreProvider> = provider.clone();
+
+        let outcome = run_or_resume(
+            Arc::new(PermanentlyFailingHandler),
+            provider_trait,
+            &JobRunArgs::default(),
+        )
+        .await;
+
+        assert!(!outcome.is_ok(), "a 403 must not be retried forever");
+        assert_eq!(provider.last_status(), Some(RunStatus::Failed));
+    }
+
     #[tokio::test]
     async fn paused_run_persists_cursor_and_marks_status_paused() {
         let provider = Arc::new(MemProvider::new());

src/infrastructure/services/backend_consistency_service.rs

@@ -506,12 +506,18 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
                     // Whether the enumeration died on page 1 or page 900,
                     // the audit did not complete, and the operator needs to
                     // know that rather than read a green run.
-                    return RunOutcome::Failed {
-                        message: format!(
-                            "backend enumeration failed on {}: {e}",
-                            backend.backend_type()
-                        ),
-                    };
+                    // Transient (throttle, 5xx, connection reset) pauses at
+                    // the cursor so a resume continues the sweep;
+                    // everything else fails terminally. Losing a
+                    // half-finished audit of a million-object bucket to a
+                    // brief 503 is the case this distinction exists for —
+                    // the retry decorator has already given up by the time
+                    // the error arrives here.
+                    return RunOutcome::from_domain_error(
+                        cursor.as_ref().map(|s| s.as_bytes()),
+                        &format!("backend enumeration failed on {}", backend.backend_type()),
+                        &e,
+                    );
                 }
             };