Merge pull request #716 from EdouardVanbelle/feat/jobs-with-recoverable-error

docs/plan/derived-blobs.md

@@ -956,8 +956,8 @@ Findings each job reports today, and where the new tables land:
 | 6 | `storage.blobs.ref_count` | recompute | `refcount_mismatch` | ✓ chunk level only |
 | 7 | `chunk_manifests.ref_count` | recompute | `refcount_mismatch` (manifests_consistency) | ✓ manifest level |
 | 8 | manifest orphan reaping | GC predicate | registry `NOT EXISTS` union, no `ref_count` | ✓ |
-| 9 | derived/attached → Blob | dangling | — | ✗ new check needed |
+| 9 | derived/attached → Blob | dangling | `derived_dangling_blob`, `attached_dangling_blob` (satellites_consistency) | ✓ |
-| 10 | `content_derived_blobs.source_hash` → Blob | orphan mapping | — | ✗ new check needed |
+| 10 | `content_derived_blobs.source_hash` → Blob | orphan mapping | `derived_orphan_mapping` (satellites_consistency) | ✓ |
 | 11 | chunk at `ref_count = 0` past grace, still present | GC lag | — | ✗ a stalled GC is silent |
 | 12 | `blob_extracted_text`, `faces.faces` orphans | dependents | search worker self-janitors; `faces` unverified | ~ verify |
 | 13 | `file_attached_blobs.file_id` → `files` | dangling | FK `ON DELETE CASCADE` | ✓ DB-enforced |

docs/plan/jobs-handling-recoverable-error.md

@@ -1,7 +1,60 @@
 # Recoverable errors in jobs — retry, then pause
 
-**Status: not started.** Design settled 2026-08-31, from a live
+**Status: implemented 2026-09-08**, except the online-migration
-diagnosis (see [Motivating incident](#motivating-incident)).
+follow-up in [Where this should end up](#where-this-should-end-up),
+which remains deliberately out of scope. Design settled 2026-08-31,
+from a live diagnosis (see
+[Motivating incident](#motivating-incident)).
+
+Steps 1–4 are in, and §Testing is enforced by
+`tests/api/backend_migration_blackhole.hurl`. Validated by hand against
+a real S3 endpoint in both directions, with the network blackholed
+mid-run: source-probe failure, target-upload failure and target-init
+failure all reach `Paused` with the cause recorded and the cursor
+positioned so nothing is skipped.
+
+Two things the design did not anticipate, both found during
+implementation and worth reading before touching this area:
+
+* **Classification was not enough on its own.** A backend that *fails*
+  was the case this plan modelled. A backend that never *answers* has
+  no error to classify, so no amount of retry policy sees it. The S3
+  client was built from a bare `config::Builder::new()` and carried no
+  `TimeoutConfig` at all, which also meant `SdkError::TimeoutError` —
+  an arm the classifier already handled — was unreachable in
+  production. Fixed with `TimeoutBlobBackend` (innermost, below retry)
+  plus the SDK's own connect/read timeouts and stalled-stream
+  protection. See that module's docs for why the bound lives in the
+  chain rather than being configured per SDK.
+
+* **`NotFound` was returned for every read failure.** Nine sites across
+  S3, Azure and local. `blob_exists` was among them, and it is the
+  migration's FIRST probe of the source — so a transient outage read as
+  "blob absent", took the permanent branch, advanced the cursor past
+  the row and could reach `finish_completed` with the pointer flipped
+  to a target missing everything the outage covered. A migration
+  reporting success having silently dropped whatever was unreachable.
+  That path is why the retry-then-pause policy alone would not have
+  been sufficient.
+
+Reporting bugs surfaced by the same testing, all fixed: a paused run
+logged `outcome="ok"`, a resumed run inherited the previous attempt's
+`error_message` through to `Completed`, and four of five migration
+counters reset per segment while `scanned_count` alone was cumulative.
+
+Known remaining, none of them blocking:
+
+* The online-migration shape below (the gate is still held for the
+  whole copy, so a paused migration freezes writes until an operator
+  resumes or cancels).
+* Per-backend SDK retry tuning — see [Scope](#scope-azure-and-s3-both).
+  Two retry layers currently stack multiplicatively, so the configured
+  retry count is not the effective one and detection takes ~2min rather
+  than ~30s.
+* `scanned_count` over-reported once (2522 against 2022 rows) on a run
+  with several pause/resume cycles. Not reproduced in five runs since;
+  now observable rather than inferable, because counters are persisted
+  per batch.
 
 A job that hits a failing backend today has two possible endings, and
 neither is right for an outage: it fails the run (throwing away a
@@ -90,7 +143,7 @@ implementation possible, and it is step 1.
 
 ---
 
-## Step 1 — errors say whether they are retryable
+## Step 1 — errors say whether they are retryable — **DONE**
 
 Today both backends wrap SDK errors into
 `DomainError::internal_error("Azure", format!("…{e}"))`, so the status
@@ -120,7 +173,7 @@ own backoff, so a second layer above it multiplies. Check what the S3
 backend inherits before adding anything, and consider making the
 engine's cap the *outer* bound with SDK retries reduced or disabled.
 
-## Step 2 — an outcome the engine can act on
+## Step 2 — an outcome the engine can act on — **DONE**
 
 `RunOutcome` grows a variant meaning "the environment failed, this is
 worth trying again later":
@@ -143,7 +196,7 @@ differs is `error_message`, which must let the panel — and an operator —
 tell "I paused this" from "the provider went down". Without that
 distinction a paused run is an unexplained one.
 
-## Step 3 — the engine implements the policy
+## Step 3 — the engine implements the policy — **DONE**
 
 In `run_or_resume`:
 
@@ -159,7 +212,7 @@ Handlers then return the retryable outcome and get the policy for free.
 
 ---
 
-## Step 4 — `migration_readonly`, the sharp edge
+## Step 4 — `migration_readonly`, the sharp edge — **DONE** (conservative option; gate still held while paused, confirmed as intended)
 
 `backend_migration` holds a gate that refuses writes **application-wide**
 until cutover. What happens to it on pause is a correctness question,
@@ -238,7 +291,36 @@ SDK retry tuning stays a separate, optional refinement — and for Azure
 specifically it should wait for the official SDK, since `azure_core`
 0.21 is archived and queued for replacement.
 
-## Testing
+## Testing — **DONE**
+
+Implemented as `tests/api/backend_migration_blackhole.hurl`, though not
+the way this section anticipated. Two departures worth recording:
+
+**The fixture is an unreachable address, not Azurite.** Azurite's
+deterministic CRC64 500 is a *failure*, and failures were never the
+hard case — they surface, get classified and retry. The case that hung
+is a peer that never answers, so the entry points at `192.0.2.1`
+(TEST-NET-1, RFC 5737, guaranteed unrouted): a SYN goes unanswered,
+with no RST and no ICMP. Note the existing `s3_stub` entry
+(`127.0.0.1:9999`) is NOT usable for this — nothing listens, so the
+connection is refused instantly, and a test built on it would pass with
+no timeout configured anywhere.
+
+**The bound is a polling budget, not a request duration.**
+`backend_migration` is a detached job: the trigger returns 202 in
+milliseconds regardless of how long the backend hangs, so timing the
+trigger proves nothing. The file polls `/runs` with `retry: 60,
+retry-interval: 2000` — 120s, three orders of magnitude below the
+unbounded socket's ~15min. That budget is the assertion; every other
+assert in the file would eventually pass even unbounded.
+
+It targets the **target-init** failure deliberately, because that path
+pauses BEFORE `migration_readonly` is engaged and so cannot leave the
+shared suite's server read-only. It also cancels its own run: a Paused
+row left behind would be resumed by the next `backend_migration`
+trigger in the suite.
+
+The original notes follow.
 
 The `backend_consistency_azure.hurl` scenario and its Azurite service
 are already wired (`tests/common/docker-compose.test.yml`,

frontend/src/lib/api/types.ts

@@ -597,9 +597,33 @@ export interface FolderAncestorsResponse {
  * discriminant is the `outcome` field, not the object key.
  */
 export type JobOutcome =
-	| { outcome: 'ok'; count: number; extra?: unknown }
+	| { outcome: 'ok'; count: number; extra?: JobOutcomeExtra }
 	| { outcome: 'err'; message: string };
 
+/**
+ * The parts of a job outcome's free-form `extra` the panel reads.
+ *
+ * Deliberately narrow — most keys are per-job counters nothing generic
+ * should switch on. These three describe the RUN's shape rather than
+ * its work, and the panel has to render them:
+ *
+ * A run that stopped because the backend was unreachable reports
+ * `outcome: 'ok'` — it did not fail, it paused and can be resumed. Read
+ * alone that renders as a green "ok" pill, which is exactly wrong: a
+ * paused `backend_migration` still holds `migration_readonly` and is
+ * refusing writes application-wide. `retryable` is what lets the row
+ * say so.
+ */
+export interface JobOutcomeExtra {
+	/** The run stopped at its cursor and can be resumed. */
+	paused?: boolean;
+	/** It stopped because the ENVIRONMENT failed, not because an
+	 *  operator asked — `reason` says what. */
+	retryable?: boolean;
+	reason?: string;
+	[key: string]: unknown;
+}
+
 /**
  * `JobSummary` — one row per registered job in `GET /api/admin/jobs`.
  * Cadence + last-run bookkeeping. `interval_ms` / `next_run_at` are
@@ -694,6 +718,14 @@ export interface JobSummary {
 	 *  for this job. Distinct from `running` — a paused run is
 	 *  resumable via the same trigger endpoint. */
 	paused_run?: PausedRunBrief;
+	/** Status of this job's most recent run row (recoverable jobs only).
+	 *
+	 *  **Prefer this over `last_outcome` wherever they could disagree.**
+	 *  `last_outcome` is the backend's in-memory record of the last
+	 *  dispatch, so anything that changes a run row without running the
+	 *  handler leaves it stale — cancelling a Paused run is a direct SQL
+	 *  flip, and the panel went on rendering the pause it replaced. */
+	last_run_status?: RunStatus;
 	/** Present iff `OXICLOUD_STARTUP_JOBS` names this job — the flags it
 	 *  is dispatched with at every boot. Worth showing: a job configured
 	 *  with `repair: true` deletes on every restart, and the row would

frontend/src/lib/components/AdminJobsPanel.svelte

@@ -516,8 +516,58 @@
 		}
 	}
 
+	// A run that stopped because the backend was unreachable, rather than
+	// because an operator asked it to stop.
+	//
+	// It reports `outcome: 'ok'` on the wire — correctly, since it did
+	// not fail and a Resume continues it — but rendering that as a plain
+	// green "ok" hides the thing worth acting on. A paused
+	// `backend_migration` is still holding `migration_readonly` and
+	// refusing writes across the whole app; the row has to say so.
+	/// Lifecycle label for the State column. Lower-cased to match the
+	/// existing `running` pill rather than shouting the DB's PascalCase.
+	function runStatusLabel(status: RunStatus): string {
+		switch (status) {
+			case 'Running':
+				return t('admin.jobs.state_running', 'running');
+			case 'Paused':
+				return t('admin.jobs.state_paused', 'paused');
+			case 'CancelRequested':
+				return t('admin.jobs.state_cancelling', 'cancelling');
+			case 'Cancelled':
+				return t('admin.jobs.state_cancelled', 'cancelled');
+			case 'Completed':
+				return t('admin.jobs.state_completed', 'completed');
+			case 'Failed':
+				return t('admin.jobs.state_failed', 'failed');
+		}
+	}
+
+	function backendFailureReason(job: JobSummary): string | undefined {
+		if (job.last_outcome?.outcome !== 'ok') return undefined;
+		const reason = job.last_outcome.extra?.reason;
+		return typeof reason === 'string' ? reason : undefined;
+	}
+
+	/// How the last dispatch turned out. NOT where the run is in its
+	/// lifecycle — that is the State column, driven by
+	/// `last_run_status`. A paused run legitimately has no outcome yet,
+	/// and saying so is the honest answer.
 	function outcomeLabel(job: JobSummary): string {
-		if (!job.last_outcome) return t('admin.jobs.never', 'never');
+		if (!job.last_outcome) {
+			// "never" means never ran. A job with a run row DID run — the
+			// outcome simply is not in memory, because `last_outcome` is
+			// populated per dispatch and a restart empties it. Saying
+			// "never" there is a lie the run history immediately
+			// contradicts: Ed saw it on a job whose last run was 8h ago.
+			//
+			// "—" is the honest answer: no outcome recorded. The State
+			// column still shows what the run did, and the drawer has
+			// the history.
+			return job.last_run_status
+				? t('admin.jobs.outcome_unknown', '—')
+				: t('admin.jobs.never', 'never');
+		}
 		if (job.last_outcome.outcome === 'ok') {
 			// `ok` on the wire = dispatch completed. If any actionable
 			// findings surfaced, we flip to "issues" (amber). If only
@@ -939,11 +989,26 @@
 								{/if}
 							</div>
 						</td>
+						<!-- STATE = where the run is in its lifecycle. Distinct
+						     from Outcome, which is how the work turned out.
+						     They are orthogonal: a Paused run has no outcome
+						     yet, and a Completed run's outcome may still be
+						     "issues". Conflating them is what made a paused
+						     migration render as a green "ok". -->
 						<td>
 							{#if isRunning(job)}
 								<span class="jobs-panel__pill jobs-panel__pill--running">
 									{t('admin.jobs.state_running', 'running')}
 								</span>
+							{:else if job.last_run_status}
+								<!-- From the run ROW, not from memory: the
+								     in-memory outcome is empty after a restart and
+								     stale after a cancel, both of which the row
+								     gets right. Reason on hover when the run
+								     stopped on a backend failure. -->
+								<span class={statusClass(job.last_run_status)} title={backendFailureReason(job)}>
+									{runStatusLabel(job.last_run_status)}
+								</span>
 							{:else}
 								<span class="jobs-panel__muted">—</span>
 							{/if}

src/common/config.rs

@@ -2,6 +2,8 @@ use std::env;
 use std::path::PathBuf;
 use std::time::Duration;
 
+use crate::infrastructure::services::timeout_blob_backend::TimeoutPolicy;
+
 /// Cache configuration
 #[derive(Debug, Clone)]
 pub struct CacheConfig {
@@ -268,6 +270,9 @@ pub struct StorageConfig {
     pub encryption: EncryptionConfig,
     /// Retry policy for remote backends.
     pub retry: RetryConfig,
+    /// Wall-clock bounds on backend calls, so a stalled endpoint
+    /// surfaces as a transient error instead of hanging indefinitely.
+    pub timeout: TimeoutPolicy,
 }
 
 /// Which blob storage backend to use.
@@ -1295,6 +1300,7 @@ impl Default for StorageConfig {
             cache: BlobCacheConfig::default(),
             encryption: EncryptionConfig::default(),
             retry: RetryConfig::default(),
+            timeout: TimeoutPolicy::default(),
         }
     }
 }
@@ -3722,6 +3728,29 @@ impl AppConfig {
             config.storage.retry.backoff_multiplier = n;
         }
 
+        // Backend call timeouts. `0` means "unbounded" for that class,
+        // which is the default for writes — see `TimeoutPolicy`.
+        for (var, slot) in [
+            (
+                "OXICLOUD_STORAGE_TIMEOUT_METADATA_MS",
+                &mut config.storage.timeout.metadata,
+            ),
+            (
+                "OXICLOUD_STORAGE_TIMEOUT_OPEN_MS",
+                &mut config.storage.timeout.open,
+            ),
+            (
+                "OXICLOUD_STORAGE_TIMEOUT_WRITE_MS",
+                &mut config.storage.timeout.write,
+            ),
+        ] {
+            if let Ok(v) = env::var(var)
+                && let Ok(n) = v.parse::<u64>()
+            {
+                *slot = (n > 0).then(|| std::time::Duration::from_millis(n));
+            }
+        }
+
         // OIDC configuration
         if let Ok(v) = env::var("OXICLOUD_OIDC_ENABLED") {
             config.oidc.enabled = v.parse::<bool>().unwrap_or(false);

src/common/di.rs

@@ -329,7 +329,8 @@ impl AppServiceFactory {
         // just before the struct init.
         let active_backend_name = Arc::new(std::sync::RwLock::new(active_backend_name));
 
-        // Stack decorators: retry → encryption → cache (inner-to-outer).
+        // Stack decorators: timeout → retry → encryption → cache
+        // (inner-to-outer).
         //
         // Encryption is applied INSIDE build_entry_backend (per-entry
         // key), so it's already on the base returned above when the
@@ -343,6 +344,32 @@ impl AppServiceFactory {
         // gates the "remote-only" decorators the same as before.
         let mut blob_backend: Arc<dyn BlobStorageBackend> = base_backend;
 
+        // Timeout decorator — INNERMOST, and applied to every backend
+        // kind including Local.
+        //
+        // It has to sit below retry: a call that never returns produces
+        // no error, so retry has nothing to react to and the job never
+        // pauses. Converting the hang into a transient error first is
+        // what gives every layer above it something to act on.
+        //
+        // Unconditional by design. Retry is gated on "not Local" because
+        // the kernel already retries local I/O, but a bound that never
+        // fires is free, and keeping the chain uniform avoids a class of
+        // backend-specific surprise.
+        {
+            use crate::infrastructure::services::timeout_blob_backend::TimeoutBlobBackend;
+            let policy = self.config.storage.timeout.clone();
+            if policy.is_enabled() {
+                blob_backend = Arc::new(TimeoutBlobBackend::new(blob_backend, policy.clone()));
+                tracing::info!(
+                    metadata_ms = policy.metadata.map(|d| d.as_millis() as u64),
+                    open_ms = policy.open.map(|d| d.as_millis() as u64),
+                    write_ms = policy.write.map(|d| d.as_millis() as u64),
+                    "Blob storage timeout decorator enabled"
+                );
+            }
+        }
+
         // Retry decorator (for remote backends)
         if self.config.storage.retry.enabled && active_backend_kind != StorageBackendType::Local {
             use crate::infrastructure::services::retry_blob_backend::{

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/scheduler/engine.rs

@@ -275,6 +275,30 @@ fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option<ErrCause>, elapse
     // structured log renderer to project the `elapsed_ms` field.
     let elapsed = format_elapsed(elapsed_ms);
     match outcome {
+        // A paused run is carried as `Ok` — the handler did its job and
+        // stopped cleanly at a checkpoint — but logging it as `ok` says
+        // the opposite of what an operator needs to know: the migration
+        // is blocked and will not progress until the backend returns.
+        // Same distinction the admin panel draws between a run's STATE
+        // and its OUTCOME; this line only ever showed the outcome.
+        JobOutcome::Ok { count, extra }
+            if extra.get("paused") == Some(&serde_json::Value::Bool(true)) =>
+        {
+            tracing::warn!(
+                target: "oxicloud::scheduler",
+                event = "job.run",
+                job = %name,
+                outcome = "paused",
+                retryable = extra.get("retryable") == Some(&serde_json::Value::Bool(true)),
+                count = *count,
+                elapsed_ms = elapsed_ms,
+                extra = %extra,
+                "job {} PAUSED after {} — count={} (resume when the cause clears)",
+                name,
+                elapsed,
+                count,
+            );
+        }
         JobOutcome::Ok { count, extra } => {
             tracing::info!(
                 target: "oxicloud::scheduler",

src/infrastructure/scheduler/mod.rs

@@ -33,9 +33,9 @@ pub use engine::SchedulerEngine;
 pub use handler::JobHandler;
 pub use pg_job_store::{PgJobStore, PgJobStoreProvider};
 pub use recoverable::{
-    Finding, JobStore, JobStoreProvider, OpenedRun, ProgressKind, RecoverableAdapter,
+    CANCEL_INTENT_PARAM, CANCEL_INTENT_TERMINATE, Finding, JobStore, JobStoreProvider, OpenedRun,
-    RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, RunSummary, derive_progress,
+    ProgressKind, RecoverableAdapter, RecoverableJobHandler, RunOutcome, RunProgress, RunStatus,
-    record_or_log, run_or_resume,
+    RunSummary, derive_progress, record_or_log, run_or_resume,
 };
 pub use registry::{
     JobEntry, JobRegistry, JobSummary, PausedRunBrief, RegisterError, StartupTrigger,

src/infrastructure/scheduler/pg_job_store.rs

@@ -229,19 +229,22 @@ impl JobStore for PgJobStore {
         Ok(row.and_then(|(v,)| v))
     }
 
-    async fn scanned_count(&self) -> Result<u64, DomainError> {
+    // `(stats ->> $2)::BIGINT` — text extraction then cast, rather
-        // `(stats->>'scanned_count')::BIGINT` — text cast rather than
+    // than `->` numeric extraction, because the stored value has been
-        // `->` numeric extraction because the stored value has been
+    // written via `((...)::text)::jsonb` in `checkpoint` and may
-        // written via `((...)::text)::jsonb` in `checkpoint`, which
+    // present as either a JSON number or a JSON string depending on
-        // may present as either a JSON number or a JSON string
+    // prior versions. `::BIGINT` handles both.
-        // depending on prior versions. `::BIGINT` handles both.
+    //
+    // `scanned_count()` is the trait's default wrapper around this.
+    async fn stat_u64(&self, key: &str) -> Result<u64, DomainError> {
         let row: Option<(Option<i64>,)> = sqlx::query_as(
-            "SELECT (stats ->> 'scanned_count')::BIGINT FROM jobs.recoverable_runs WHERE id = $1",
+            "SELECT (stats ->> $2)::BIGINT FROM jobs.recoverable_runs WHERE id = $1",
         )
         .bind(self.run_id)
+        .bind(key)
         .fetch_optional(self.pool.as_ref())
         .await
-        .map_err(|e| map_sqlx_err("scanned_count", e))?;
+        .map_err(|e| map_sqlx_err("stat_u64", e))?;
         Ok(row.and_then(|(v,)| v).unwrap_or(0).max(0) as u64)
     }
 
@@ -327,6 +330,56 @@ impl JobStore for PgJobStore {
         Ok(())
     }
 
+    async fn mark_paused_retryable(
+        &self,
+        cursor: Option<Vec<u8>>,
+        reason: &str,
+    ) -> Result<(), DomainError> {
+        // `status = 'Paused'`, so resume is the same operation an
+        // operator pause produces — the only difference is that
+        // `error_message` is populated, which is what lets the panel say
+        // WHY it stopped. `completed_at` stays NULL: the run is not
+        // over.
+        //
+        // One statement per cursor shape, matching `mark_paused`: a
+        // COALESCE would overwrite a real cursor with NULL when the
+        // handler had not advanced since the last checkpoint.
+        if let Some(c) = cursor {
+            sqlx::query(
+                r#"
+                UPDATE jobs.recoverable_runs
+                   SET status           = 'Paused',
+                       cursor           = $2,
+                       error_message    = $3,
+                       last_progress_at = NOW()
+                 WHERE id = $1
+                "#,
+            )
+            .bind(self.run_id)
+            .bind(&c[..])
+            .bind(reason)
+            .execute(self.pool.as_ref())
+            .await
+            .map_err(|e| map_sqlx_err("mark_paused_retryable", e))?;
+        } else {
+            sqlx::query(
+                r#"
+                UPDATE jobs.recoverable_runs
+                   SET status           = 'Paused',
+                       error_message    = $2,
+                       last_progress_at = NOW()
+                 WHERE id = $1
+                "#,
+            )
+            .bind(self.run_id)
+            .bind(reason)
+            .execute(self.pool.as_ref())
+            .await
+            .map_err(|e| map_sqlx_err("mark_paused_retryable", e))?;
+        }
+        Ok(())
+    }
+
     async fn mark_failed(&self, message: &str) -> Result<(), DomainError> {
         sqlx::query(
             r#"
@@ -824,10 +877,18 @@ impl PgJobStoreProvider {
                         // In practice this is a rare edge case that
                         // ONLY hits if two admin triggers land in
                         // the same microsecond.
+                        //
+                        // `error_message` is cleared here: it records why
+                        // the LAST attempt stopped, so carrying it past a
+                        // resume leaves a Completed run still displaying a
+                        // transient error it recovered from — a failure
+                        // that did not happen. Same stale-state shape as
+                        // the read-only banner outliving its migration.
                         let row: Option<(DateTime<Utc>, Option<Vec<u8>>)> = sqlx::query_as(
                             r#"
                             UPDATE jobs.recoverable_runs
                                SET status           = 'Running',
+                                   error_message    = NULL,
                                    last_progress_at = NOW()
                              WHERE id = $1
                             RETURNING started_at, cursor

src/infrastructure/scheduler/recoverable.rs

@@ -157,6 +157,32 @@ pub enum RunOutcome {
     Paused {
         cursor: Vec<u8>,
     },
+    /// The ENVIRONMENT failed after a bounded number of attempts, and
+    /// this is worth trying again later.
+    ///
+    /// Lands as `Paused` in the row, so resume works unchanged. What
+    /// differs is `error_message`: an operator has to be able to tell "I
+    /// paused this" from "the provider went down", and a paused run with
+    /// no explanation is an unexplained one.
+    ///
+    /// Distinct from both neighbours, and the distinction is the point:
+    ///
+    /// | outcome | meaning | resumes? |
+    /// |---|---|---|
+    /// | `Failed` | the data or the request is wrong | no — terminal |
+    /// | `Paused` | an operator asked it to stop | yes |
+    /// | `PausedRetryable` | the environment failed | yes, and says why |
+    ///
+    /// Reached only after the handler has already retried — see
+    /// `retry_transient` — because a single transient error is not news.
+    /// The cap exists because no status-based taxonomy can tell a
+    /// deterministic 5xx from a passing one (Azurite answers 500 to a
+    /// CRC64 ranged GET, every time), so the policy is deliberately
+    /// "retry as if transient, then hand the decision to a human".
+    PausedRetryable {
+        cursor: Vec<u8>,
+        reason: String,
+    },
     Failed {
         message: String,
     },
@@ -197,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
@@ -484,7 +560,44 @@ pub trait JobStore: Send + Sync {
     /// Returns `0` if the key is absent (fresh row) or not a
     /// number. Callers on a Fresh run can safely skip this — the
     /// answer is trivially 0 and the write path starts fresh.
-    async fn scanned_count(&self) -> Result<u64, DomainError>;
+    async fn scanned_count(&self) -> Result<u64, DomainError> {
+        self.stat_u64("scanned_count").await
+    }
+
+    /// Read any numeric key out of the run's `stats` JSONB.
+    ///
+    /// The generalisation of [`Self::scanned_count`], which is now
+    /// one caller of it. Handlers use this on a Resume path to
+    /// restore their own cumulative counters — see
+    /// [`Self::checkpoint_counters`].
+    ///
+    /// Returns `0` when the key is absent or not a number, so a
+    /// fresh run and a run that never wrote the key are the same
+    /// answer.
+    async fn stat_u64(&self, key: &str) -> Result<u64, DomainError>;
+
+    /// Handler-callable. Merge the handler's OWN cumulative counters
+    /// into `stats` mid-run.
+    ///
+    /// Distinct from [`Self::merge_stats`], which stays engine-only and
+    /// runs once at `Completed`. That timing is the problem this
+    /// solves: counters written only at the end are lost by a pause,
+    /// so every resumed segment restarts them at zero and the final
+    /// row reports the LAST segment rather than the run. `backend_
+    /// migration` showed this as `copied: 0` on a migration that had
+    /// copied plenty, next to a `scanned_count` that was cumulative
+    /// because `checkpoint` had been persisting it all along.
+    ///
+    /// Pass ABSOLUTE values, not deltas — the merge is
+    /// `stats = stats || $1`, so each write displaces the last. Keys
+    /// are the handler's own; do not write engine-owned
+    /// `scanned_count` / `finding_count` through here.
+    async fn checkpoint_counters(
+        &self,
+        counters: &serde_json::Map<String, serde_json::Value>,
+    ) -> Result<(), DomainError> {
+        self.merge_stats(counters).await
+    }
 
     /// Persist one finding to `jobs.run_findings` and bump
     /// `stats.finding_count` on the parent run. Consistency handlers
@@ -554,6 +667,27 @@ pub trait JobStore: Send + Sync {
     /// returned. Handler code MUST NOT call this.
     async fn mark_paused(&self, cursor: Option<Vec<u8>>) -> Result<(), DomainError>;
 
+    /// Engine-only. Called by [`run_or_resume`] on
+    /// [`RunOutcome::PausedRetryable`]. Handler code MUST NOT call this.
+    ///
+    /// Writes `status = Paused` — so resume is the same operation — plus
+    /// `error_message = reason`. The reason is the whole point: without
+    /// it the panel cannot distinguish an operator pause from a provider
+    /// outage, and a paused migration holding `migration_readonly` looks
+    /// like someone forgot about it.
+    ///
+    /// Separate method rather than an extra argument on
+    /// [`Self::mark_paused`] because the two carry different meaning and
+    /// only one of them writes `error_message`. A `reason: Option<&str>`
+    /// parameter would let a caller write a Paused row with an
+    /// error message and no error, which is the state this exists to
+    /// distinguish from.
+    async fn mark_paused_retryable(
+        &self,
+        cursor: Option<Vec<u8>>,
+        reason: &str,
+    ) -> Result<(), DomainError>;
+
     /// Engine-only. Called by [`run_or_resume`] on
     /// [`RunOutcome::Failed`]. Handler code MUST NOT call this.
     async fn mark_failed(&self, message: &str) -> Result<(), DomainError>;
@@ -1018,6 +1152,47 @@ pub async fn run_or_resume(
                 )
             }
         }
+        RunOutcome::PausedRetryable { cursor, reason } => {
+            let cursor_hex = hex::encode(&cursor);
+            log_terminal_write_err(
+                "mark_paused_retryable",
+                run_id,
+                store.mark_paused_retryable(Some(cursor), &reason).await,
+            );
+            // Audited, not merely logged. Writes are refused app-wide
+            // while `backend_migration` holds `migration_readonly`, so a
+            // run that stopped on a provider outage is an operational
+            // event someone has to act on — and "why is the app
+            // read-only" must be answerable afterwards.
+            tracing::info!(
+                target: "audit",
+                event = "job.paused_retryable",
+                reason = "backend_unavailable",
+                job = %job.name(),
+                run_id = %run_id,
+                cursor_hex = %cursor_hex,
+                detail = %reason,
+                "👮🏻‍♂️ `{}` paused after exhausting retries: {reason}",
+                job.name(),
+            );
+            // `ok`, not `err`: the run did not fail, it stopped and can
+            // be resumed. Reporting it as an error would put a red job
+            // in the panel that a Resume click fixes, which reads as a
+            // bug rather than as a decision waiting to be made.
+            JobOutcome::ok_with(
+                stats.finding_count,
+                serde_json::json!({
+                    "paused":            true,
+                    "retryable":         true,
+                    "reason":            reason,
+                    "run_id":            run_id.to_string(),
+                    "cursor_hex":        cursor_hex,
+                    "finding_count":     stats.finding_count,
+                    "scanned_count":     stats.scanned_count,
+                    "severity_counts":   stats.by_severity,
+                }),
+            )
+        }
         RunOutcome::Failed { message } => {
             log_terminal_write_err("mark_failed", run_id, store.mark_failed(&message).await);
             JobOutcome::err(format!("{message} (run_id={run_id})"))
@@ -1315,8 +1490,14 @@ mod tests {
         async fn get_string_param(&self, key: &str) -> Result<Option<String>, DomainError> {
             Ok(self.state.lock().unwrap().string_params.get(key).cloned())
         }
-        async fn scanned_count(&self) -> Result<u64, DomainError> {
+        /// Mirrors the PG row: `scanned_count` is its own column-like
-            Ok(self.state.lock().unwrap().scanned_count)
+        /// field, every other counter lives in the merged stats map.
+        async fn stat_u64(&self, key: &str) -> Result<u64, DomainError> {
+            let s = self.state.lock().unwrap();
+            if key == "scanned_count" {
+                return Ok(s.scanned_count);
+            }
+            Ok(s.extra_stats.get(key).and_then(|v| v.as_u64()).unwrap_or(0))
         }
         async fn merge_stats(
             &self,
@@ -1340,6 +1521,23 @@ mod tests {
             }
             Ok(())
         }
+        async fn mark_paused_retryable(
+            &self,
+            cursor: Option<Vec<u8>>,
+            reason: &str,
+        ) -> Result<(), DomainError> {
+            let mut s = self.state.lock().unwrap();
+            s.status = RunStatus::Paused;
+            // Both, deliberately: Paused so resume works, `error_message`
+            // so a test can assert the two pause shapes are
+            // distinguishable — which is the whole reason the variant
+            // exists.
+            s.error_message = Some(reason.to_string());
+            if let Some(c) = cursor {
+                s.cursor = Some(c);
+            }
+            Ok(())
+        }
         async fn mark_failed(&self, message: &str) -> Result<(), DomainError> {
             let mut s = self.state.lock().unwrap();
             s.status = RunStatus::Failed;
@@ -1410,6 +1608,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]
@@ -1634,6 +1842,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 {
@@ -1786,6 +2039,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());
@@ -1853,6 +2169,40 @@ mod tests {
         assert_eq!(*seen.lock().unwrap(), Some(b"halfway".to_vec()));
     }
 
+    /// Counters written mid-run must survive to be read back, because
+    /// that round-trip is the whole mechanism by which a resumed
+    /// segment continues its totals instead of restarting them at zero.
+    /// `backend_migration` reported `copied: 0` on a migration that had
+    /// copied thousands precisely because nothing persisted them until
+    /// `Completed`, which a paused run never reaches.
+    #[tokio::test]
+    async fn checkpoint_counters_round_trip_through_stats() {
+        let provider = Arc::new(MemProvider::new());
+        let store = provider.open_or_start("counter_job").await.unwrap();
+        let store: Arc<dyn JobStore> = match store {
+            OpenedRun::Fresh { store: s } | OpenedRun::Resumed { store: s, .. } => s,
+            OpenedRun::AlreadyActive { .. } => panic!("fresh provider cannot be active"),
+        };
+
+        // Absent keys read as 0, so a fresh run needs no special case.
+        assert_eq!(store.stat_u64("copied").await.unwrap(), 0);
+
+        let mut counters = serde_json::Map::new();
+        counters.insert("copied".into(), serde_json::json!(120u64));
+        counters.insert("skipped".into(), serde_json::json!(7u64));
+        store.checkpoint_counters(&counters).await.unwrap();
+
+        assert_eq!(store.stat_u64("copied").await.unwrap(), 120);
+        assert_eq!(store.stat_u64("skipped").await.unwrap(), 7);
+
+        // Absolute, not additive: a later batch's write displaces the
+        // earlier one rather than summing with it. The handler owns the
+        // running total; the store only records it.
+        counters.insert("copied".into(), serde_json::json!(300u64));
+        store.checkpoint_counters(&counters).await.unwrap();
+        assert_eq!(store.stat_u64("copied").await.unwrap(), 300);
+    }
+
     #[tokio::test]
     async fn concurrent_trigger_hits_already_active() {
         let provider = Arc::new(MemProvider::new());

src/infrastructure/scheduler/registry.rs

@@ -249,11 +249,12 @@ impl JobRegistry {
                     last_outcome,
                     running: state.current_run_start.is_some(),
                     recoverable: entry.handler.is_recoverable(),
-                    // Both populated in the `list_jobs` handler — one
+                    // All populated in the `list_jobs` handler — two
                     // from a DB round-trip, one from AppConfig. Kept
                     // out of the registry snapshot so the in-memory
                     // scheduler state pulls in neither dependency.
                     paused_run: None,
+                    last_run_status: None,
                     startup: None,
                 }
             })
@@ -349,6 +350,18 @@ pub struct JobSummary {
     /// job, most of which the job ignored with no way to tell.
     #[serde(skip_serializing_if = "<[_]>::is_empty")]
     pub parameters: &'static [JobParam],
+    /// Status of this job's most recent run row, for recoverable jobs.
+    ///
+    /// Populated by the `list_jobs` handler from the DB, and it exists
+    /// because [`Self::last_outcome`] cannot answer this: that field is
+    /// in-memory, written when a dispatch completes through the engine,
+    /// so anything changing a run row without running the handler leaves
+    /// it stale. Cancelling a Paused run is exactly that — a direct SQL
+    /// flip — and the panel went on showing the pause's outcome.
+    ///
+    /// Prefer this over `last_outcome` wherever the two could disagree.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub last_run_status: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub interval_ms: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]

src/infrastructure/services/azure_blob_backend.rs

@@ -143,9 +143,10 @@ impl BlobStorageBackend for AzureBlobBackend {
             })?;
             let file_size = data.len() as u64;
 
-            client.put_block_blob(data).await.map_err(|e| {
+            client
-                DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}"))
+                .put_block_blob(data)
-            })?;
+                .await
+                .map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
 
             let _ = fs::remove_file(&source_path).await;
             Ok(file_size)
@@ -169,9 +170,10 @@ impl BlobStorageBackend for AzureBlobBackend {
 
             // `Bytes` converts into `azure_core::Body` by reference count —
             // the old `data.to_vec()` copied every chunk once more.
-            client.put_block_blob(data).await.map_err(|e| {
+            client
-                DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}"))
+                .put_block_blob(data)
-            })?;
+                .await
+                .map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
 
             Ok(size)
         })
@@ -190,9 +192,10 @@ impl BlobStorageBackend for AzureBlobBackend {
         Box::pin(async move {
             let client = self.blob_client(&hash);
             let size = data.len() as u64;
-            client.put_block_blob(data).await.map_err(|e| {
+            client
-                DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}"))
+                .put_block_blob(data)
-            })?;
+                .await
+                .map_err(|e| azure_domain_error(format!("Failed to upload blob {hash}"), &e))?;
             Ok(size)
         })
     }
@@ -236,11 +239,7 @@ impl BlobStorageBackend for AzureBlobBackend {
             let first = match pages.next().await {
                 Some(Ok(response)) => response,
                 Some(Err(e)) => {
-                    return Err(DomainError::new(
+                    return Err(azure_read_error(format!("Failed to get blob {hash}"), &e));
-                        ErrorKind::NotFound,
-                        "Azure",
-                        format!("Failed to get blob {hash}: {e}"),
-                    ));
                 }
                 None => {
                     let empty: BlobStream =
@@ -321,10 +320,9 @@ impl BlobStorageBackend for AzureBlobBackend {
             let first = match pages.next().await {
                 Some(Ok(response)) => response,
                 Some(Err(e)) => {
-                    return Err(DomainError::new(
+                    return Err(azure_read_error(
-                        ErrorKind::NotFound,
+                        format!("Failed to get blob range {hash}"),
-                        "Azure",
+                        &e,
-                        format!("Failed to get blob range {hash}: {e}"),
                     ));
                 }
                 None => {
@@ -371,9 +369,9 @@ impl BlobStorageBackend for AzureBlobBackend {
                     if status == Some(azure_core::StatusCode::NotFound) {
                         Ok(())
                     } else {
-                        Err(DomainError::internal_error(
+                        Err(azure_domain_error(
-                            "Azure",
+                            format!("Failed to delete blob {hash}"),
-                            format!("Failed to delete blob {hash}: {e}"),
+                            &e,
                         ))
                     }
                 }
@@ -395,9 +393,13 @@ impl BlobStorageBackend for AzureBlobBackend {
                     if status == Some(azure_core::StatusCode::NotFound) {
                         Ok(false)
                     } else {
-                        Err(DomainError::internal_error(
+                        // Only the 404 means "absent"; everything else keeps
-                            "Azure",
+                        // its transient/permanent class so the migration's
-                            format!("Failed to check blob {hash}: {e}"),
+                        // source probe can pause on an outage instead of
+                        // recording a permanent finding.
+                        Err(azure_domain_error(
+                            format!("Failed to check blob {hash}"),
+                            &e,
                         ))
                     }
                 }
@@ -412,13 +414,10 @@ impl BlobStorageBackend for AzureBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let client = self.blob_client(&hash);
-            let props = client.get_properties().await.map_err(|e| {
+            let props = client
-                DomainError::new(
+                .get_properties()
-                    ErrorKind::NotFound,
+                .await
-                    "Azure",
+                .map_err(|e| azure_read_error(format!("Failed to stat blob {hash}"), &e))?;
-                    format!("Failed to stat blob {hash}: {e}"),
-                )
-            })?;
             Ok(props.blob.properties.content_length)
         })
     }
@@ -558,12 +557,16 @@ impl BlobStorageBackend for AzureBlobBackend {
 
                 while let Some(page) = pages.next().await {
                     let page = page.map_err(|e| {
-                        DomainError::internal_error(
+                        // Classified: `backend_consistency` fails the whole
-                            "Blob",
+                        // run on an enumeration error, so a throttle
+                        // partway through the 256-shard walk should be
+                        // retryable rather than discarding the sweep.
+                        azure_domain_error(
                             format!(
-                                "Azure ListBlobs failed on shard {shard:02x} of container '{}': {e}",
+                                "Azure ListBlobs failed on shard {shard:02x} of container '{}'",
                                 self.container_name
                             ),
+                            &e,
                         )
                     })?;
 
@@ -654,6 +657,77 @@ impl BlobStorageBackend for AzureBlobBackend {
     }
 }
 
+/// Wrap an `azure_core` error as a `DomainError` that says whether
+/// retrying it could help. Azure counterpart of `s3_domain_error`.
+///
+/// `azure_core::error::ErrorKind::HttpResponse` carries the status, so
+/// this works on the archived 0.21 SDK — no need to wait for the
+/// official-crate migration. That matters because Azure is the backend
+/// the retry-then-pause plan was written for: a ranged GET carrying
+/// `x-ms-range-get-content-crc64` that Azurite answers 500 to, retried
+/// forever by `azure_core` while `migration_readonly` refused writes
+/// application-wide.
+///
+/// Transient: 5xx, 429, 408. Also `Io` — connection resets, DNS, TLS.
+/// Permanent: other 4xx (credentials, missing container, malformed
+/// request), `DataConversion`, `Credential`.
+///
+/// **A deterministic 500 still classifies as transient**, and that is
+/// deliberate rather than an oversight. Nothing at this layer can tell
+/// "this provider is briefly unwell" from "this provider will answer
+/// 500 to this exact request forever" — the Azurite CRC64 case is the
+/// second wearing the clothes of the first. So the policy is to retry
+/// as if transient and let the bounded attempt cap turn the difference
+/// into a Paused run an operator can act on.
+/// Read-path variant of [`azure_domain_error`]: only a real 404 is
+/// `NotFound`.
+///
+/// Every Azure read used to label EVERY failure `NotFound` — a refused
+/// connection, a 503, an expired SAS token all reported as "blob
+/// missing". That is the most dangerous wrong answer available on a read
+/// path, because callers ACT on NotFound by concluding the bytes are
+/// gone: a migration reading its source would treat an outage as "the
+/// source does not have this blob" and move past it.
+///
+/// Everything that is not a 404 goes through the normal classifier, so a
+/// 403 stays permanent instead of being retried.
+pub(crate) fn azure_read_error(context: String, err: &azure_core::Error) -> DomainError {
+    use azure_core::error::ErrorKind as AzKind;
+
+    if let AzKind::HttpResponse { status, .. } = err.kind()
+        && u16::from(*status) == 404
+    {
+        return DomainError::new(
+            ErrorKind::NotFound,
+            "Azure",
+            format!("{context}: not found"),
+        );
+    }
+    azure_domain_error(context, err)
+}
+
+pub(crate) fn azure_domain_error(context: String, err: &azure_core::Error) -> DomainError {
+    use azure_core::error::ErrorKind as AzKind;
+
+    let transient = match err.kind() {
+        AzKind::HttpResponse { status, .. } => {
+            let code = u16::from(*status);
+            code >= 500 || code == 429 || code == 408
+        }
+        AzKind::Io => true,
+        AzKind::DataConversion | AzKind::Credential | AzKind::MockFramework | AzKind::Other => {
+            false
+        }
+    };
+
+    let message = format!("{context}: {err}");
+    if transient {
+        DomainError::transient_backend("Azure", message)
+    } else {
+        DomainError::internal_error("Azure", message)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/infrastructure/services/backend_consistency_service.rs

@@ -378,9 +378,11 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
             },
         };
         if let Err(e) = backend.initialize().await {
-            return RunOutcome::Failed {
+            // Nothing has been scanned yet, so there is no cursor to keep
-                message: format!("probed backend init: {e}"),
+            // — but the distinction still matters: an unreachable endpoint
-            };
+            // pauses and can be resumed once it is back, while a wrong
+            // bucket or bad credentials stays terminal.
+            return RunOutcome::from_domain_error(None, "probed backend init", &e);
         }
         if let Some(name) = &probed_storage {
             tracing::info!(
@@ -506,12 +508,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 {
+                    // Transient (throttle, 5xx, connection reset) pauses at
-                        message: format!(
+                    // the cursor so a resume continues the sweep;
-                            "backend enumeration failed on {}: {e}",
+                    // everything else fails terminally. Losing a
-                            backend.backend_type()
+                    // 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,
+                    );
                 }
             };
 

src/infrastructure/services/backend_migration_service.rs

@@ -452,9 +452,11 @@ impl RecoverableJobHandler for BackendMigrationService {
         // for the swap-hot-swap call in `finish_completed`.
         let target = build_entry_backend_typed(target_entry, &self.storage_path_fallback);
         if let Err(e) = target.initialize().await {
-            return RunOutcome::Failed {
+            // Runs BEFORE `migration_readonly` is engaged, so pausing
-                message: format!("target backend init: {e}"),
+            // here holds no write freeze — an operator can leave it
-            };
+            // paused indefinitely and resume when the target comes back.
+            // A wrong bucket or bad credentials still fails terminally.
+            return RunOutcome::from_domain_error(None, "target backend init", &e);
         }
 
         // All guards passed. Engage server-wide read-only mode for
@@ -555,16 +557,33 @@ impl RecoverableJobHandler for BackendMigrationService {
             },
         };
 
-        let mut copied_count = 0u64;
+        // Restored on Resume, exactly like `already_scanned` above.
+        //
+        // These used to start at zero on every segment while
+        // `scanned_count` was restored, so one counter described the
+        // migration and the other four described the current segment.
+        // A run that paused and resumed then reported `copied: 0`
+        // beside a `scanned_count` in the thousands — the numbers were
+        // measuring different things and only one of them said so.
+        // `checkpoint_counters` below persists them per batch so a
+        // pause cannot discard them.
+        let restore = |key: &'static str| async move {
+            if is_fresh {
+                0
+            } else {
+                store.stat_u64(key).await.unwrap_or(0)
+            }
+        };
+        let mut copied_count = restore("copied").await;
         // Populated by the smart-skip probe below: target blob
         // already exists at the current head format+key, so a
         // rewrite would be identical bytes. Cheap (15-byte range
         // read via `is_at_head_format`), massive latency win on
         // resume + on backends where the source was rotated to the
         // same key as the target already had.
-        let mut skipped_count: u64 = 0;
+        let mut skipped_count: u64 = restore("skipped").await;
-        let mut failed_count = 0u64;
+        let mut failed_count = restore("failed").await;
-        let mut source_missing_count = 0u64;
+        let mut source_missing_count = restore("source_missing").await;
 
         loop {
             // Cooperative cancel poll between batches.
@@ -580,6 +599,23 @@ impl RecoverableJobHandler for BackendMigrationService {
                         source_missing = source_missing_count,
                         "backend_migration cancelled cooperatively, pausing"
                     );
+                    // A TERMINAL cancel must give writes back.
+                    //
+                    // Cancel ends the run with no swap, so the source
+                    // stays the active backend and there is nothing left
+                    // to protect. Leaving the gate set stranded the whole
+                    // application read-only with no way out: the flag is
+                    // persisted, so a restart reloaded it rather than
+                    // clearing it, and the only escape was editing
+                    // `admin_settings` by hand.
+                    //
+                    // A plain PAUSE deliberately keeps the gate. The
+                    // cursor stays valid only while nothing writes, so
+                    // resuming after allowing writes could miss a blob
+                    // written below the cursor — see the plan's
+                    // "Why NOT to release the gate on pause". Cancel is
+                    // the escape hatch, and it is the operator's call.
+                    self.release_readonly_on_terminal_cancel(store).await;
                     return RunOutcome::Paused {
                         cursor: cursor
                             .as_ref()
@@ -671,21 +707,72 @@ impl RecoverableJobHandler for BackendMigrationService {
                         .await;
                         continue;
                     }
+                    Err(e) if e.is_transient() => {
+                        // PAUSE. Skipping here was a data-loss path.
+                        //
+                        // The old comment called this "a network blip"
+                        // and `continue`d, reasoning that a re-run would
+                        // re-probe. It would not: the cursor advances to
+                        // the batch's last hash regardless, so a skipped
+                        // row is never revisited by THIS run — and unlike
+                        // a copy failure it recorded no finding, so
+                        // `failed` stayed 0, the run reached
+                        // `finish_completed`, and the pointer flipped to
+                        // a target missing every blob the outage
+                        // covered.
+                        //
+                        // That is the worst shape available: a migration
+                        // reporting success while having silently
+                        // dropped whatever was unreachable at the time.
+                        tracing::warn!(
+                            target: "oxicloud::migration",
+                            event = "backend_migration.source_unreachable",
+                            run_id = %store.run_id(),
+                            hash = %hash,
+                            copied = copied_count,
+                            error = %e,
+                            "source unreachable while probing; pausing at the last checkpoint"
+                        );
+                        return RunOutcome::from_domain_error(
+                            cursor.as_ref().map(|s| s.as_bytes()),
+                            &format!(
+                                "source unreachable while probing ({copied_count} blob(s) \
+                                 copied so far)"
+                            ),
+                            &e,
+                        );
+                    }
                     Err(e) => {
-                        // Transient probe failure on source is NOT a
+                        // Permanent probe failure. Still skipped rather
-                        // finding — treat like a network blip.
+                        // than fatal — one unprobeable blob must not
-                        // Skipping this row on this run; a re-run
+                        // abort the migration — but it now records a
-                        // will re-probe. If the failure is
+                        // finding, so the run cannot report clean while
-                        // persistent, `blobs_consistency` catches
+                        // having skipped rows, and `blobs_consistency`
-                        // it.
+                        // is not the only thing that would ever notice.
                         tracing::warn!(
                             target: "oxicloud::migration",
                             event = "backend_migration.source_probe_error",
                             run_id = %store.run_id(),
                             hash = %hash,
                             error = %e,
-                            "source blob_exists probe failed; skipping this row"
+                            "source blob_exists probe failed; recording finding, skipping row"
                         );
+                        failed_count += 1;
+                        record_or_log(
+                            store,
+                            BACKEND_MIGRATION_JOB_NAME,
+                            "migration_failed",
+                            "data_loss",
+                            None,
+                            serde_json::json!({
+                                "hash":   hash,
+                                "size":   size,
+                                "source": source_kind,
+                                "target": target_kind,
+                                "error":  format!("source probe failed: {e}"),
+                            }),
+                        )
+                        .await;
                         continue;
                     }
                 }
@@ -757,6 +844,58 @@ impl RecoverableJobHandler for BackendMigrationService {
                         }
                     }
                     Err(e) => {
+                        // A transient failure pauses IMMEDIATELY. Not
+                        // after a threshold — on the first one.
+                        //
+                        // The cursor advances to the batch's LAST hash,
+                        // after this loop. So continuing past a transient
+                        // failure lets the batch finish and the cursor
+                        // move BEYOND the blob that failed, and nothing
+                        // revisits it: the run would carry a `data_loss`
+                        // finding for a blob that was never damaged, only
+                        // briefly unreachable. Ed caught this in review of
+                        // a "tolerate N consecutive" version — that
+                        // version skipped up to N blobs per batch for
+                        // exactly this reason.
+                        //
+                        // Pausing here keeps the cursor at the PREVIOUS
+                        // batch's end, so a resume re-walks this batch
+                        // and retries the blob. Re-copying a few
+                        // already-present blobs is free — the walk
+                        // short-circuits on them.
+                        //
+                        // Tolerate-and-continue still applies to
+                        // PERMANENT failures, which is what it was built
+                        // for: one corrupt or unreadable blob must not
+                        // abort a migration of millions, and retrying it
+                        // would fail identically.
+                        //
+                        // The copy has already been retried beneath this
+                        // (RetryBlobBackend: 3 attempts with backoff), so
+                        // arriving here means 4 attempts failed.
+                        if e.is_transient() {
+                            tracing::warn!(
+                                target: "oxicloud::migration",
+                                event = "backend_migration.backend_unreachable",
+                                run_id = %store.run_id(),
+                                hash = %hash,
+                                copied = copied_count,
+                                error = %e,
+                                "backend unreachable; pausing at the last checkpoint so this \
+                                 blob is retried on resume"
+                            );
+                            // `migration_readonly` stays engaged — only
+                            // Cancel releases it. See
+                            // `release_readonly_on_terminal_cancel`.
+                            return RunOutcome::from_domain_error(
+                                cursor.as_ref().map(|s| s.as_bytes()),
+                                &format!(
+                                    "backend unreachable while copying ({copied_count} blob(s) \
+                                     copied so far)"
+                                ),
+                                &e,
+                            );
+                        }
                         failed_count += 1;
                         tracing::warn!(
                             target: "oxicloud::migration",
@@ -801,6 +940,29 @@ impl RecoverableJobHandler for BackendMigrationService {
                     message: format!("checkpoint: {e}"),
                 };
             }
+            // Persist the counters alongside the cursor. Absolute
+            // values, not deltas — the merge is last-write-wins, and
+            // the checkpoint above already made this batch's work part
+            // of the durable position. A failure here is logged but
+            // does NOT fail the run: the cursor is the correctness-
+            // critical write, these are reporting.
+            let counters: serde_json::Map<String, serde_json::Value> = serde_json::json!({
+                "copied":         copied_count,
+                "skipped":        skipped_count,
+                "failed":         failed_count,
+                "source_missing": source_missing_count,
+            })
+            .as_object()
+            .cloned()
+            .unwrap_or_default();
+            if let Err(e) = store.checkpoint_counters(&counters).await {
+                tracing::warn!(
+                    target: "oxicloud::migration",
+                    event = "backend_migration.counter_persist_failed",
+                    error = %e,
+                    "could not persist per-batch counters; totals may under-report after a resume"
+                );
+            }
             // Bump the shared progress snapshot so the server-status
             // header middleware surfaces fresh numbers on every
             // user's next API call. Guard is held only for a struct
@@ -834,6 +996,66 @@ impl RecoverableJobHandler for BackendMigrationService {
 }
 
 impl BackendMigrationService {
+    /// Clear `migration_readonly` when the cancel was TERMINAL.
+    ///
+    /// Cancel ends the run with no swap: the source is still the active
+    /// backend, so there is nothing left for the write freeze to
+    /// protect, and leaving it set locks the whole application out of
+    /// writes. The flag is persisted, so that state survived restarts —
+    /// the only escape was hand-editing `admin_settings`.
+    ///
+    /// **Pause is deliberately not this.** The cursor is a position in a
+    /// hash-ordered walk, and it stays valid only while nothing writes.
+    /// Release the gate on pause and a blob written afterwards whose
+    /// hash sorts BELOW the cursor is never visited, so the run
+    /// completes, flips the pointer, and reads for that hash 404 against
+    /// a target that never received it. Cancel is safe precisely because
+    /// it ENDS the run: a later retry starts fresh and rescans
+    /// everything.
+    ///
+    /// Distinguished by the same `cancel_intent` param the engine reads
+    /// to decide `Cancelled` vs `Paused`, so the two cannot disagree
+    /// about which kind of stop this was.
+    ///
+    /// Best effort, and deliberately so: a run that has already been
+    /// cancelled should not be turned into a hard failure by a DB blip
+    /// while releasing a flag. The in-memory store still happens, so
+    /// writes resume in THIS process even if the persist fails; the loud
+    /// warning is what tells an operator the DB copy needs attention.
+    async fn release_readonly_on_terminal_cancel(&self, store: &dyn JobStore) {
+        let terminal = store
+            .get_string_param(crate::infrastructure::scheduler::CANCEL_INTENT_PARAM)
+            .await
+            .ok()
+            .flatten()
+            .as_deref()
+            == Some(crate::infrastructure::scheduler::CANCEL_INTENT_TERMINATE);
+        if !terminal {
+            return;
+        }
+
+        if let Err(e) = persist_migration_readonly(self.pool.as_ref(), false).await {
+            tracing::warn!(
+                target: "oxicloud::migration",
+                event = "storage.migration_readonly.release_persist_failed",
+                run_id = %store.run_id(),
+                error = %e,
+                "could not persist migration_readonly=false after a terminal cancel; writes \
+                 resume in this process but a restart will come up read-only until \
+                 admin_settings is corrected"
+            );
+        }
+        self.migration_readonly.store(false, Ordering::Relaxed);
+        tracing::info!(
+            target: "audit",
+            event = "storage.migration_readonly.released",
+            reason = "migration_cancelled",
+            run_id = %store.run_id(),
+            "🚧 migration_readonly released after terminal cancel — writes resume, active \
+             backend unchanged"
+        );
+    }
+
     /// Terminal successful path — reached from both Completed sites
     /// in the batch loop (empty-first-batch and short-batch).
     ///

src/infrastructure/services/local_blob_backend.rs

@@ -615,13 +615,11 @@ impl BlobStorageBackend for LocalBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let blob_path = self.blob_path(&hash);
-            let file = File::open(&blob_path).await.map_err(|e| {
+            // Was unconditional NotFound: a stale NFS handle or an
-                DomainError::new(
+            // unmounted iSCSI target reported the blob as missing.
-                    ErrorKind::NotFound,
+            let file = File::open(&blob_path)
-                    "Blob",
+                .await
-                    format!("Failed to open blob {}: {}", hash, e),
+                .map_err(|e| local_io_error("Blob", format!("Failed to open blob {hash}"), &e))?;
-                )
-            })?;
             Ok(Box::pin(ReaderStream::with_capacity(file, STREAM_CHUNK_SIZE)) as BlobStream)
         })
     }
@@ -636,13 +634,9 @@ impl BlobStorageBackend for LocalBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let blob_path = self.blob_path(&hash);
-            let mut file = File::open(&blob_path).await.map_err(|e| {
+            let mut file = File::open(&blob_path)
-                DomainError::new(
+                .await
-                    ErrorKind::NotFound,
+                .map_err(|e| local_io_error("Blob", format!("Failed to open blob {hash}"), &e))?;
-                    "Blob",
-                    format!("Failed to open blob {}: {}", hash, e),
-                )
-            })?;
 
             file.seek(std::io::SeekFrom::Start(start))
                 .await
@@ -697,13 +691,9 @@ impl BlobStorageBackend for LocalBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let blob_path = self.blob_path(&hash);
-            let meta = fs::metadata(&blob_path).await.map_err(|e| {
+            let meta = fs::metadata(&blob_path)
-                DomainError::new(
+                .await
-                    ErrorKind::NotFound,
+                .map_err(|e| local_io_error("Blob", format!("Failed to stat blob {hash}"), &e))?;
-                    "Blob",
-                    format!("Failed to stat blob {}: {}", hash, e),
-                )
-            })?;
             Ok(meta.len())
         })
     }
@@ -918,6 +908,60 @@ impl BlobStorageBackend for LocalBlobBackend {
     }
 }
 
+/// Classify a filesystem error, because "local" does not mean
+/// "reliable".
+///
+/// A local backend is a PATH, and that path may be an iSCSI or NVMe-oF
+/// LUN, an NFS mount, or a disk with a failing sector. Those produce
+/// errors that clear on their own exactly like a remote 503 does, and
+/// treating every one as permanent means a migration off a briefly
+/// unreachable mount records data-loss findings for blobs that are
+/// perfectly intact.
+///
+/// It matters more here than for a remote backend, because
+/// `RetryBlobBackend` is only applied when the active backend is NOT
+/// Local (`di.rs`) — so nothing below this retries, and this
+/// classification is the only thing standing between a flaky mount and
+/// a run that concludes the data is gone.
+///
+/// **`NotFound` stays `NotFound`, and nothing else becomes it.** Callers
+/// act on that variant by concluding the bytes do not exist.
+///
+/// Transient: the network-mount family (timeouts, unreachable, reset,
+/// stale handle) plus `Interrupted` (EINTR) and `ResourceBusy` (EBUSY).
+///
+/// Permanent, deliberately: `PermissionDenied` and
+/// `ReadOnlyFilesystem` need an operator, retrying changes nothing.
+/// `StorageFull` likewise. `InvalidData` is corruption, which is a
+/// finding worth keeping. A bad sector surfaces as an uncategorised EIO
+/// and therefore lands here too — right, because the useful outcome is
+/// a `blob_corrupted`-style finding naming the blob, not a run that
+/// pauses forever waiting for a disk to heal.
+pub(crate) fn local_io_error(
+    entity: &'static str,
+    context: String,
+    err: &std::io::Error,
+) -> DomainError {
+    use std::io::ErrorKind as Io;
+
+    let message = format!("{context}: {err}");
+    match err.kind() {
+        Io::NotFound => DomainError::new(ErrorKind::NotFound, entity, message),
+        Io::TimedOut
+        | Io::HostUnreachable
+        | Io::NetworkUnreachable
+        | Io::NetworkDown
+        | Io::ConnectionReset
+        | Io::ConnectionAborted
+        | Io::NotConnected
+        | Io::BrokenPipe
+        | Io::StaleNetworkFileHandle
+        | Io::Interrupted
+        | Io::ResourceBusy => DomainError::transient_backend(entity, message),
+        _ => DomainError::internal_error(entity, message),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1085,4 +1129,56 @@ mod tests {
             "resume must start STRICTLY after the given hash"
         );
     }
+
+    /// "Local" does not mean reliable — the path can be an iSCSI LUN or
+    /// an NFS mount. The two directions this must never confuse:
+    ///
+    /// * a genuinely absent file must stay `NotFound`, because callers
+    ///   act on that by concluding the bytes do not exist;
+    /// * an unreachable mount must NOT become `NotFound`, which is what
+    ///   every one of these sites used to return unconditionally.
+    #[test]
+    fn local_io_errors_are_classified_not_all_notfound() {
+        use std::io::{Error, ErrorKind as Io};
+
+        let missing = local_io_error("Blob", "open".into(), &Error::from(Io::NotFound));
+        assert_eq!(missing.kind, ErrorKind::NotFound);
+        assert!(!missing.is_transient());
+
+        // Network-backed mounts and interrupted syscalls: retry helps.
+        for kind in [
+            Io::TimedOut,
+            Io::HostUnreachable,
+            Io::NetworkDown,
+            Io::ConnectionReset,
+            Io::StaleNetworkFileHandle,
+            Io::Interrupted,
+            Io::ResourceBusy,
+        ] {
+            let e = local_io_error("Blob", "open".into(), &Error::from(kind));
+            assert!(e.is_transient(), "{kind:?} should be retryable");
+            assert_ne!(
+                e.kind,
+                ErrorKind::NotFound,
+                "{kind:?} must never read as a missing blob"
+            );
+        }
+
+        // Operator-action or corruption: retrying changes nothing, and a
+        // finding naming the blob is the useful outcome.
+        for kind in [
+            Io::PermissionDenied,
+            Io::ReadOnlyFilesystem,
+            Io::StorageFull,
+            Io::InvalidData,
+        ] {
+            let e = local_io_error("Blob", "open".into(), &Error::from(kind));
+            assert!(!e.is_transient(), "{kind:?} should not be retryable");
+            assert_ne!(
+                e.kind,
+                ErrorKind::NotFound,
+                "{kind:?} is not a missing blob"
+            );
+        }
+    }
 }

src/infrastructure/services/mod.rs

@@ -64,6 +64,7 @@ pub mod thumb_derived_import_service;
 pub mod thumbnail_service;
 #[cfg(test)]
 mod thumbnail_service_test;
+pub mod timeout_blob_backend;
 pub mod transcode_import_service;
 pub mod trash_cleanup_service;
 pub mod tree_etag_flush_service;

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

@@ -7,6 +7,7 @@ use aws_sdk_s3::primitives::ByteStream;
 use bytes::Bytes;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
+use std::time::Duration;
 use tokio::fs;
 use tokio_util::io::ReaderStream;
 
@@ -39,9 +40,36 @@ impl S3BlobBackend {
             "oxicloud",
         );
 
+        // `Builder::new()` starts from nothing — in particular with no
+        // `TimeoutConfig` at all, which meant a lost network on an
+        // established connection produced no error until the OS gave up
+        // on TCP retransmission (~15 minutes). For that whole window a
+        // migration looked merely slow: no error, so no retry, no log
+        // and no pause. It also made the `SdkError::TimeoutError` arm of
+        // `s3_domain_error` unreachable.
+        //
+        // These bounds are deliberately not the ones in `TimeoutPolicy`:
+        // that decorator provides the configurable outer bound for every
+        // backend, while these are the SDK's finer, per-attempt
+        // instruments underneath it.
+        let timeouts = aws_sdk_s3::config::timeout::TimeoutConfig::builder()
+            .connect_timeout(Duration::from_secs(10))
+            // Time to first byte, not transfer duration — a large object
+            // is never punished for being large.
+            .read_timeout(Duration::from_secs(30))
+            .build();
+
         let mut builder = aws_sdk_s3::config::Builder::new()
             .region(aws_sdk_s3::config::Region::new(config.region.clone()))
             .credentials_provider(credentials)
+            .timeout_config(timeouts)
+            // The right tool for a network pulled mid-transfer: it
+            // measures throughput rather than elapsed time, so it can
+            // bound a streaming upload without capping how long a
+            // legitimately large one may take.
+            .stalled_stream_protection(
+                aws_sdk_s3::config::StalledStreamProtectionConfig::enabled().build(),
+            )
             .behavior_version_latest();
 
         if let Some(ref endpoint) = config.endpoint_url {
@@ -105,10 +133,14 @@ impl BlobStorageBackend for S3BlobBackend {
                 .send()
                 .await
                 .map_err(|e| {
-                    DomainError::internal_error(
+                    // Classified like every other SDK call. A refused
-                        "S3",
+                    // connection or a 5xx here is the endpoint being
-                        format!("Cannot access bucket '{}': {}", self.bucket, e),
+                    // down, not the configuration being wrong, and the
-                    )
+                    // jobs that call `initialize()` should pause rather
+                    // than fail on it. A genuine misconfiguration —
+                    // wrong bucket, bad credentials — still lands as 4xx
+                    // and stays terminal.
+                    s3_domain_error("S3", format!("Cannot access bucket '{}'", self.bucket), &e)
                 })?;
 
             tracing::info!("S3 blob backend initialized: bucket={}", self.bucket);
@@ -170,12 +202,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(body)
                 .send()
                 .await
-                .map_err(|e| {
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
-                    DomainError::internal_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 +242,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(body)
                 .send()
                 .await
-                .map_err(|e| {
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to upload blob {}: {}", hash, e),
-                    )
-                })?;
 
             Ok(size)
         })
@@ -253,12 +275,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .body(ByteStream::from(data))
                 .send()
                 .await
-                .map_err(|e| {
+                .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?;
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to upload blob {}: {}", hash, e),
-                    )
-                })?;
             Ok(size)
         })
     }
@@ -297,11 +314,30 @@ impl BlobStorageBackend for S3BlobBackend {
                 .send()
                 .await
                 .map_err(|e| {
-                    DomainError::new(
+                    // Only a real NoSuchKey is NotFound. This used to
-                        ErrorKind::NotFound,
+                    // label EVERY read failure that way — a refused
-                        "S3",
+                    // connection, a 503, an expired credential all
-                        format!("Failed to get blob {}: {}", hash, e),
+                    // reported as "blob missing".
-                    )
+                    //
+                    // That is the most dangerous wrong answer available
+                    // here, because callers ACT on NotFound by concluding
+                    // the bytes are gone. A migration reading its source
+                    // through this would treat an outage as "the source
+                    // does not have this blob" and move on.
+                    //
+                    // Everything else goes through the normal classifier,
+                    // so a 403 stays permanent rather than being retried
+                    // forever.
+                    if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e
+                        && svc.err().is_no_such_key()
+                    {
+                        return DomainError::new(
+                            ErrorKind::NotFound,
+                            "S3",
+                            format!("Failed to get blob {hash}: no such key"),
+                        );
+                    }
+                    s3_domain_error("S3", format!("Failed to get blob {hash}"), &e)
                 })?;
 
             // Convert S3 ByteStream into a Stream<Item = Result<Bytes, io::Error>>
@@ -336,11 +372,21 @@ impl BlobStorageBackend for S3BlobBackend {
                 .send()
                 .await
                 .map_err(|e| {
-                    DomainError::new(
+                    // Same rule as the full read: only a real NoSuchKey
-                        ErrorKind::NotFound,
+                    // is NotFound. Ranged reads feed CDC reassembly and
-                        "S3",
+                    // deep verification, so mislabelling an outage here
-                        format!("Failed to get blob range {}: {}", hash, e),
+                    // reads as "this chunk is gone" — a data-loss
-                    )
+                    // conclusion drawn from a network problem.
+                    if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e
+                        && svc.err().is_no_such_key()
+                    {
+                        return DomainError::new(
+                            ErrorKind::NotFound,
+                            "S3",
+                            format!("Failed to get blob range {hash}: no such key"),
+                        );
+                    }
+                    s3_domain_error("S3", format!("Failed to get blob range {hash}"), &e)
                 })?;
 
             let reader = output.body.into_async_read();
@@ -363,12 +409,7 @@ impl BlobStorageBackend for S3BlobBackend {
                 .key(&key)
                 .send()
                 .await
-                .map_err(|e| {
+                .map_err(|e| s3_domain_error("S3", format!("Failed to delete blob {hash}"), &e))?;
-                    DomainError::internal_error(
-                        "S3",
-                        format!("Failed to delete blob {}: {}", hash, e),
-                    )
-                })?;
 
             Ok(())
         })
@@ -392,15 +433,17 @@ impl BlobStorageBackend for S3BlobBackend {
             {
                 Ok(_) => Ok(true),
                 Err(e) => {
-                    // Check if it's a 404 (not found) vs an actual error
+                    // A 404 is the only answer that means "absent". Classify
-                    let service_err = e.into_service_error();
+                    // before consuming the SdkError so everything else keeps
-                    if service_err.is_not_found() {
+                    // its transient/permanent class: this is the migration's
+                    // source probe, and a refused connection reported as a
+                    // plain failure would be treated as permanent.
+                    let classified =
+                        s3_domain_error("S3", format!("Failed to check blob {hash}"), &e);
+                    if e.into_service_error().is_not_found() {
                         Ok(false)
                     } else {
-                        Err(DomainError::internal_error(
+                        Err(classified)
-                            "S3",
-                            format!("Failed to check blob {}: {}", hash, service_err),
-                        ))
                     }
                 }
             }
@@ -423,11 +466,19 @@ impl BlobStorageBackend for S3BlobBackend {
                 .send()
                 .await
                 .map_err(|e| {
-                    DomainError::new(
+                    // `head_object` reports a missing key as NotFound
-                        ErrorKind::NotFound,
+                    // rather than NoSuchKey, so match on the typed
-                        "S3",
+                    // variant the SDK actually returns here.
-                        format!("Failed to stat blob {}: {}", hash, e),
+                    if let aws_sdk_s3::error::SdkError::ServiceError(svc) = &e
-                    )
+                        && svc.err().is_not_found()
+                    {
+                        return DomainError::new(
+                            ErrorKind::NotFound,
+                            "S3",
+                            format!("Failed to stat blob {hash}: not found"),
+                        );
+                    }
+                    s3_domain_error("S3", format!("Failed to stat blob {hash}"), &e)
                 })?;
 
             Ok(output.content_length().unwrap_or(0) as u64)
@@ -565,11 +616,11 @@ impl BlobStorageBackend for S3BlobBackend {
                 }
 
                 let resp = req.send().await.map_err(|e| {
-                    DomainError::new(
+                    // Classified, because `backend_consistency` fails the
-                        ErrorKind::InternalError,
+                    // whole run on an enumeration error — a throttle
-                        "Blob",
+                    // midway through a million-object bucket should be
-                        format!("S3 ListObjectsV2 failed: {e}"),
+                    // retryable rather than throwing the sweep away.
-                    )
+                    s3_domain_error("Blob", "S3 ListObjectsV2 failed".to_string(), &e)
                 })?;
 
                 requests += 1;
@@ -638,6 +689,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 +761,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/infrastructure/services/timeout_blob_backend.rs

@@ -0,0 +1,632 @@
+//! `TimeoutBlobBackend` — bounds every backend call in wall-clock time.
+//!
+//! ## Why this exists
+//!
+//! A backend call that *fails* is handled: it is classified, retried if
+//! transient, and pauses the job at its cursor if it stays transient. A
+//! call that never returns is handled by nothing at all.
+//!
+//! That is not hypothetical. Pull the network on an established TCP
+//! connection and there is no RST and no ICMP — the peer simply stops
+//! answering, and a socket read blocks until the OS gives up on
+//! retransmission, on the order of fifteen minutes. For that whole
+//! window the job is neither running nor failed: no error, so no retry,
+//! no log line, no pause, nothing on the admin page. It looks exactly
+//! like a very slow migration.
+//!
+//! Refusing a connection is instant and *does* surface (that is what
+//! makes a `127.0.0.1` test look reassuring); losing a network mid-flight
+//! is the silent case, and it is also the realistic one.
+//!
+//! ## Why a decorator rather than per-SDK configuration
+//!
+//! The S3 SDK can express this natively, and does — see the
+//! `TimeoutConfig` in `s3_blob_backend.rs`, which is throughput-aware and
+//! therefore strictly better for streams. But it only covers S3, and
+//! Azure's 0.21 client has no equivalent knob short of supplying a custom
+//! transport. That asymmetry is the reason this lives in the chain
+//! instead of being configured twice.
+//!
+//! The local backend is a different story and deliberately not the
+//! justification for this decorator: a local path is reached through the
+//! kernel, and the kernel already owns that timeout. iSCSI gives up after
+//! `replacement_timeout` (120s by default) and returns an I/O error;
+//! NVMe-oF and soft-mounted NFS behave the same way. Those surface as an
+//! `io::Error` and are classified by `local_io_error`, which is where
+//! they belong. Local passes through this decorator only because a
+//! uniform chain is simpler than a conditional one, and a bound that
+//! never fires costs nothing.
+//!
+//! ## Operation classes
+//!
+//! A single timeout cannot fit both a `HEAD` and a multi-gigabyte upload,
+//! so calls are bounded by what they do:
+//!
+//! * **Metadata** — `blob_exists`, `blob_size`, `delete_blob`,
+//!   `initialize`, `health_check`, `list_blob_hashes`. Bounded tightly.
+//!   These are the calls a migration makes per blob, and `blob_exists` in
+//!   particular is its very first probe of the source.
+//! * **Open** — `get_blob_stream`, `get_blob_range_stream`. The future
+//!   resolves once the response *starts*; the body streams afterwards. So
+//!   this bounds time-to-first-byte, not transfer duration, and a slow
+//!   large read is never punished for being large.
+//! * **Write** — the `put_*` family and `sync_blobs`. The entire transfer
+//!   happens inside the future, so any wall-clock bound here is also a
+//!   maximum upload duration. Unbounded by default for that reason:
+//!   getting it wrong truncates legitimate uploads, which is a worse
+//!   failure than the hang it would prevent. S3 covers this properly
+//!   through stalled-stream protection, which measures throughput instead
+//!   of elapsed time.
+//!
+//! A timeout is reported as [`DomainError::transient_backend`], because
+//! that is what it is: no information was obtained about the blob. The
+//! job engine pauses at its cursor and the work resumes when the network
+//! does.
+
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::application::ports::blob_storage_ports::{
+    BlobListPage, BlobStorageBackend, BlobStream, StorageHealthStatus,
+};
+use crate::domain::errors::DomainError;
+use bytes::Bytes;
+
+// ── Timeout policy ─────────────────────────────────────────────────
+
+/// Per-operation-class wall-clock bounds.
+#[derive(Debug, Clone)]
+pub struct TimeoutPolicy {
+    /// Bound for metadata calls (exists, size, delete, init, health, list).
+    pub metadata: Option<Duration>,
+    /// Bound for time-to-first-byte on reads.
+    pub open: Option<Duration>,
+    /// Bound for the whole of a write. `None` (the default) leaves large
+    /// uploads unbounded — see the module docs.
+    pub write: Option<Duration>,
+}
+
+impl Default for TimeoutPolicy {
+    fn default() -> Self {
+        Self {
+            metadata: Some(Duration::from_secs(30)),
+            open: Some(Duration::from_secs(60)),
+            write: None,
+        }
+    }
+}
+
+impl TimeoutPolicy {
+    /// A policy that bounds nothing — the pre-decorator behaviour.
+    pub fn disabled() -> Self {
+        Self {
+            metadata: None,
+            open: None,
+            write: None,
+        }
+    }
+
+    /// True when at least one class is bounded, i.e. wrapping is worth it.
+    pub fn is_enabled(&self) -> bool {
+        self.metadata.is_some() || self.open.is_some() || self.write.is_some()
+    }
+}
+
+// ── TimeoutBlobBackend ─────────────────────────────────────────────
+
+/// Decorator that fails a call the backend never answers.
+pub struct TimeoutBlobBackend {
+    inner: Arc<dyn BlobStorageBackend>,
+    policy: TimeoutPolicy,
+}
+
+impl TimeoutBlobBackend {
+    pub fn new(inner: Arc<dyn BlobStorageBackend>, policy: TimeoutPolicy) -> Self {
+        Self { inner, policy }
+    }
+}
+
+/// Await `fut`, giving up after `limit`.
+///
+/// `name` is lazy for the same reason as the retry decorator's: the
+/// success path must not pay for a `format!` it will never print.
+async fn with_timeout<T, L>(
+    limit: Option<Duration>,
+    backend: &'static str,
+    name: L,
+    fut: impl std::future::Future<Output = Result<T, DomainError>>,
+) -> Result<T, DomainError>
+where
+    L: Fn() -> String,
+{
+    let Some(limit) = limit else {
+        return fut.await;
+    };
+    match tokio::time::timeout(limit, fut).await {
+        Ok(result) => result,
+        Err(_) => {
+            let op = name();
+            // The one log line that distinguishes "hung" from "slow".
+            // Without it a stalled backend is invisible until the job
+            // pauses, and the pause reason alone does not say which
+            // layer noticed.
+            tracing::warn!(
+                target: "oxicloud::storage",
+                wrapper = "timeout",
+                backend = backend,
+                operation = %op,
+                timeout_ms = limit.as_millis() as u64,
+                "⏱️ Backend call timed out — treating as transient"
+            );
+            Err(DomainError::transient_backend(
+                "Blob",
+                format!(
+                    "{op} on {backend} backend timed out after {:?} (no response)",
+                    limit
+                ),
+            ))
+        }
+    }
+}
+
+impl BlobStorageBackend for TimeoutBlobBackend {
+    fn initialize(
+        &self,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            with_timeout(
+                limit,
+                backend,
+                || "initialize".to_string(),
+                inner.initialize(),
+            )
+            .await
+        })
+    }
+
+    fn put_blob(
+        &self,
+        hash: &str,
+        source_path: &Path,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.write;
+        let hash = hash.to_string();
+        let path = source_path.to_path_buf();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("put_blob({label})"),
+                inner.put_blob(&hash, &path),
+            )
+            .await
+        })
+    }
+
+    fn put_blob_from_bytes(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.write;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("put_blob_from_bytes({label})"),
+                inner.put_blob_from_bytes(&hash, data),
+            )
+            .await
+        })
+    }
+
+    fn put_blob_from_bytes_unsynced(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.write;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("put_blob_from_bytes_unsynced({label})"),
+                inner.put_blob_from_bytes_unsynced(&hash, data),
+            )
+            .await
+        })
+    }
+
+    fn put_blob_from_bytes_replace(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.write;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("put_blob_from_bytes_replace({label})"),
+                inner.put_blob_from_bytes_replace(&hash, data),
+            )
+            .await
+        })
+    }
+
+    fn sync_blobs(
+        &self,
+        hashes: &[String],
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.write;
+        let hashes = hashes.to_vec();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let count = hashes.len();
+            with_timeout(
+                limit,
+                backend,
+                || format!("sync_blobs({count} hashes)"),
+                inner.sync_blobs(&hashes),
+            )
+            .await
+        })
+    }
+
+    fn get_blob_stream(
+        &self,
+        hash: &str,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
+    {
+        let inner = self.inner.clone();
+        let limit = self.policy.open;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("get_blob_stream({label})"),
+                inner.get_blob_stream(&hash),
+            )
+            .await
+        })
+    }
+
+    fn get_blob_range_stream(
+        &self,
+        hash: &str,
+        start: u64,
+        end: Option<u64>,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
+    {
+        let inner = self.inner.clone();
+        let limit = self.policy.open;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("get_blob_range_stream({label}, {start}..{end:?})"),
+                inner.get_blob_range_stream(&hash, start, end),
+            )
+            .await
+        })
+    }
+
+    fn delete_blob(
+        &self,
+        hash: &str,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("delete_blob({label})"),
+                inner.delete_blob(&hash),
+            )
+            .await
+        })
+    }
+
+    fn blob_exists(
+        &self,
+        hash: &str,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<bool, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("blob_exists({label})"),
+                inner.blob_exists(&hash),
+            )
+            .await
+        })
+    }
+
+    fn blob_size(
+        &self,
+        hash: &str,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        let hash = hash.to_string();
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            let label = hash.clone();
+            with_timeout(
+                limit,
+                backend,
+                || format!("blob_size({label})"),
+                inner.blob_size(&hash),
+            )
+            .await
+        })
+    }
+
+    fn health_check(
+        &self,
+    ) -> Pin<
+        Box<dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>> + Send + '_>,
+    > {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            with_timeout(
+                limit,
+                backend,
+                || "health_check".to_string(),
+                inner.health_check(),
+            )
+            .await
+        })
+    }
+
+    fn list_blob_hashes(
+        &self,
+        cursor: Option<String>,
+        limit_n: usize,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobListPage, DomainError>> + Send + '_>>
+    {
+        let inner = self.inner.clone();
+        let limit = self.policy.metadata;
+        Box::pin(async move {
+            let backend = inner.backend_type();
+            with_timeout(
+                limit,
+                backend,
+                || format!("list_blob_hashes(limit {limit_n})"),
+                inner.list_blob_hashes(cursor, limit_n),
+            )
+            .await
+        })
+    }
+
+    fn backend_type(&self) -> &'static str {
+        self.inner.backend_type()
+    }
+
+    fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
+        self.inner.local_blob_path(hash)
+    }
+
+    /// Forwarded, and then re-wrapped.
+    ///
+    /// `uncached()` exists so a verification pass can read past the
+    /// cache; an inner backend reached that way is no less able to hang
+    /// than the cached one, so it keeps the same bound.
+    fn uncached(&self) -> Option<Arc<dyn BlobStorageBackend>> {
+        self.inner.uncached().map(|inner| {
+            Arc::new(TimeoutBlobBackend::new(inner, self.policy.clone()))
+                as Arc<dyn BlobStorageBackend>
+        })
+    }
+
+    fn read_prefetch(&self) -> usize {
+        self.inner.read_prefetch()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::domain::errors::ErrorKind;
+
+    /// A backend whose every call parks forever — the network-pulled case.
+    struct HangingBackend;
+
+    impl BlobStorageBackend for HangingBackend {
+        fn initialize(
+            &self,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn put_blob(
+            &self,
+            _hash: &str,
+            _source_path: &Path,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn put_blob_from_bytes(
+            &self,
+            _hash: &str,
+            _data: Bytes,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn get_blob_stream(
+            &self,
+            _hash: &str,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn get_blob_range_stream(
+            &self,
+            _hash: &str,
+            _start: u64,
+            _end: Option<u64>,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn delete_blob(
+            &self,
+            _hash: &str,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn blob_exists(
+            &self,
+            _hash: &str,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<bool, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn blob_size(
+            &self,
+            _hash: &str,
+        ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>>
+        {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn health_check(
+            &self,
+        ) -> Pin<
+            Box<
+                dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>>
+                    + Send
+                    + '_,
+            >,
+        > {
+            Box::pin(async { std::future::pending().await })
+        }
+        fn backend_type(&self) -> &'static str {
+            "hanging"
+        }
+        fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
+            None
+        }
+    }
+
+    fn wrapped() -> TimeoutBlobBackend {
+        TimeoutBlobBackend::new(
+            Arc::new(HangingBackend),
+            TimeoutPolicy {
+                metadata: Some(Duration::from_millis(50)),
+                open: Some(Duration::from_millis(50)),
+                write: Some(Duration::from_millis(50)),
+            },
+        )
+    }
+
+    /// The whole point: a hang must become a *transient* error, not a
+    /// hang and not a permanent one. `NotFound` here would tell a
+    /// migration the blob is absent; a permanent error would fail the
+    /// run instead of pausing it.
+    #[tokio::test]
+    async fn a_hanging_backend_yields_a_transient_error_not_a_hang() {
+        let backend = wrapped();
+
+        let err = backend.blob_exists("abc").await.unwrap_err();
+        assert!(
+            err.is_transient(),
+            "a stalled probe must be transient so the job pauses and resumes: {err}"
+        );
+        assert_ne!(
+            err.kind,
+            ErrorKind::NotFound,
+            "a hang says nothing about whether the blob exists"
+        );
+
+        assert!(backend.blob_size("abc").await.unwrap_err().is_transient());
+        assert!(backend.delete_blob("abc").await.unwrap_err().is_transient());
+        assert!(backend.initialize().await.unwrap_err().is_transient());
+        // `BlobStream` is not `Debug`, so go through `.err()` rather than
+        // `unwrap_err()`.
+        assert!(
+            backend
+                .get_blob_stream("abc")
+                .await
+                .err()
+                .expect("a hanging read must not succeed")
+                .is_transient()
+        );
+    }
+
+    /// `write: None` is the default, and it must genuinely mean
+    /// "unbounded" — a large upload cannot be truncated by this
+    /// decorator.
+    #[tokio::test]
+    async fn an_unbounded_class_is_not_bounded() {
+        let backend = TimeoutBlobBackend::new(
+            Arc::new(HangingBackend),
+            TimeoutPolicy {
+                metadata: Some(Duration::from_millis(50)),
+                open: None,
+                write: None,
+            },
+        );
+
+        // Bounded class still fires...
+        assert!(backend.blob_exists("abc").await.unwrap_err().is_transient());
+
+        // ...while an unbounded one is still pending long after the
+        // bounded one would have given up. `write: None` is the default,
+        // and a truncated multi-gigabyte upload is a worse outcome than
+        // the hang the bound would have caught.
+        let open = backend.get_blob_stream("abc");
+        assert!(
+            tokio::time::timeout(Duration::from_millis(250), open)
+                .await
+                .is_err(),
+            "an unbounded class must never be cut short by the decorator"
+        );
+    }
+
+    #[test]
+    fn disabled_is_disabled() {
+        assert!(!TimeoutPolicy::disabled().is_enabled());
+        assert!(TimeoutPolicy::default().is_enabled());
+    }
+}

src/interfaces/api/handlers/admin_handler.rs

@@ -2570,41 +2570,94 @@ pub async fn list_jobs(State(state): State<Arc<AppState>>) -> impl IntoResponse
     // failures fall back to the pre-enrichment shape so the endpoint
     // stays useful when the jobs DB is temporarily unreachable.
     if let Some(pool) = state.db_pool.as_ref() {
-        let paused_rows: Vec<(String, uuid::Uuid, Option<i64>, Option<i64>)> = sqlx::query_as(
+        // The LATEST run per job, whatever its status — not just the
+        // paused ones.
+        //
+        // `last_outcome` is in-memory, written when a dispatch finishes
+        // through the engine. Anything that changes a run row WITHOUT
+        // running the handler leaves it stale: cancelling a Paused run
+        // is a direct SQL flip to `Cancelled`, so the panel kept
+        // rendering the outcome of the run that pause belonged to — a
+        // cancelled job still showing "blocked".
+        //
+        // `DISTINCT ON` is safe as "the current run": the
+        // `one_active_run_per_job` partial unique index allows only one
+        // non-terminal row per job, and a resume reuses it rather than
+        // starting a new one, so a non-terminal row is always the newest.
+        /// `(job_name, status, run_id, started_at, scanned, total)` — the
+        /// enrichment row shape, named so the query's type stays legible.
+        type LatestRunRow = (
+            String,
+            String,
+            uuid::Uuid,
+            chrono::DateTime<chrono::Utc>,
+            Option<i64>,
+            Option<i64>,
+        );
+        let latest_rows: Vec<LatestRunRow> = sqlx::query_as(
             r#"
-            SELECT
+            SELECT DISTINCT ON (job_name)
                 job_name,
+                status::TEXT,
                 id,
+                started_at,
                 (stats  ->> 'scanned_count')::BIGINT AS scanned,
                 (params ->> 'total_rows')::BIGINT   AS total
             FROM jobs.recoverable_runs
-            WHERE status = 'Paused'
+            ORDER BY job_name, started_at DESC
             "#,
         )
         .fetch_all(pool.as_ref())
         .await
         .unwrap_or_default();
 
-        let by_name: std::collections::HashMap<String, PausedRunBrief> = paused_rows
+        type LatestRun = (String, chrono::DateTime<chrono::Utc>, PausedRunBrief);
+        let by_name: std::collections::HashMap<String, LatestRun> = latest_rows
             .into_iter()
-            .map(|(name, id, scanned, total)| {
+            .map(|(name, status, id, started_at, scanned, total)| {
                 (
                     name,
-                    PausedRunBrief {
+                    (
-                        id,
+                        status,
-                        scanned: scanned.unwrap_or(0).max(0) as u64,
+                        started_at,
-                        total: total.filter(|t| *t > 0).map(|t| t as u64),
+                        PausedRunBrief {
-                    },
+                            id,
+                            scanned: scanned.unwrap_or(0).max(0) as u64,
+                            total: total.filter(|t| *t > 0).map(|t| t as u64),
+                        },
+                    ),
                 )
             })
             .collect();
 
         for job in summary.iter_mut() {
-            if job.recoverable
+            if !job.recoverable {
-                && !job.running
+                continue;
-                && let Some(paused) = by_name.get(&job.name)
+            }
-            {
+            let Some((status, started_at, brief)) = by_name.get(&job.name) else {
-                job.paused_run = Some(paused.clone());
+                continue;
+            };
+            // Always reported, so the panel can prefer the row's truth
+            // over the in-memory outcome rather than guessing which is
+            // fresher.
+            job.last_run_status = Some(status.clone());
+            // Fill the timestamp too when memory has none.
+            //
+            // `last_outcome` and `last_run_at` are both in-memory, so a
+            // restart empties them and the row read "never" for a job
+            // with real runs in the DB — the opposite failure to the
+            // stale-outcome one, and just as misleading. The row is
+            // authoritative for "did this ever run"; memory only adds
+            // the richer outcome detail when it happens to be warm.
+            //
+            // Only when absent: a warm `last_run_at` describes the last
+            // DISPATCH, which for a non-recoverable tick is finer-grained
+            // than any run row.
+            if job.last_run_at.is_none() {
+                job.last_run_at = Some(*started_at);
+            }
+            if !job.running && status == "Paused" {
+                job.paused_run = Some(brief.clone());
             }
         }
     }
@@ -2879,16 +2932,73 @@ pub async fn cancel_job(
         .request_terminal_cancel(&name)
         .await
     {
-        Ok(Some(run_id)) => (
+        Ok(Some(run_id)) => {
-            StatusCode::OK,
+            // Cancelling a PAUSED migration has to give writes back
-            Json(serde_json::json!({
+            // here, because nothing else will.
-                "cancelled": true,
+            //
-                "run_id": run_id.to_string(),
+            // A Running row re-enters the handler, which releases the
-                "note": "Running row → will land in Cancelled at next batch boundary; \
+            // gate itself at its next cancel poll. A Paused row does
-                         Paused row → flipped to Cancelled immediately.",
+            // not: `request_terminal_cancel` flips it straight to
-            })),
+            // Cancelled in SQL with no handler in the loop. That is the
-        )
+            // common case — a migration paused by an outage, holding
-            .into_response(),
+            // `migration_readonly`, which an operator cancels precisely
+            // TO get writes back. Without this the app stayed read-only
+            // forever: the flag is persisted, so even a restart reloaded
+            // it, and the only escape was editing admin_settings by
+            // hand.
+            //
+            // Safe because cancel ends the run with no swap — the source
+            // is still the active backend, so there is nothing left for
+            // the freeze to protect. Releasing on PAUSE would not be
+            // safe; see `release_readonly_on_terminal_cancel`.
+            //
+            // Idempotent and harmless for every other job: the flag is
+            // only ever set by backend_migration, so clearing it when it
+            // is already false is a no-op.
+            if name == crate::infrastructure::services::backend_migration_service::BACKEND_MIGRATION_JOB_NAME
+                && state
+                    .migration_readonly
+                    .load(std::sync::atomic::Ordering::Relaxed)
+            {
+                if let Some(pool) = state.db_pool.as_ref()
+                    && let Err(e) =
+                        crate::infrastructure::services::entry_backend::persist_migration_readonly(
+                            pool.as_ref(),
+                            false,
+                        )
+                        .await
+                {
+                    tracing::warn!(
+                        target: "oxicloud::migration",
+                        event = "storage.migration_readonly.release_persist_failed",
+                        run_id = %run_id,
+                        error = %e,
+                        "could not persist migration_readonly=false after cancelling a paused \
+                         migration; writes resume now but a restart will come up read-only"
+                    );
+                }
+                state
+                    .migration_readonly
+                    .store(false, std::sync::atomic::Ordering::Relaxed);
+                tracing::info!(
+                    target: "audit",
+                    event = "storage.migration_readonly.released",
+                    reason = "paused_migration_cancelled",
+                    run_id = %run_id,
+                    "🚧 migration_readonly released — writes resume, active backend unchanged"
+                );
+            }
+            (
+                StatusCode::OK,
+                Json(serde_json::json!({
+                    "cancelled": true,
+                    "run_id": run_id.to_string(),
+                    "note": "Running row → will land in Cancelled at next batch boundary; \
+                             Paused row → flipped to Cancelled immediately.",
+                })),
+            )
+                .into_response()
+        }
         Ok(None) => (
             StatusCode::OK,
             Json(serde_json::json!({

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 {

tests/api/backend_migration_blackhole.hurl

@@ -0,0 +1,166 @@
+# =============================================================
+# OxiCloud – backend_migration against an endpoint that never answers
+#
+# The regression test asked for by
+# `docs/plan/jobs-handling-recoverable-error.md` §Testing: assert that
+# an unreachable backend lands the run in **Paused**, that
+# `error_message` names the cause, and that it gets there in **bounded
+# time rather than hanging**.
+#
+# ## The failure this pins
+#
+# A backend that *fails* was always handled — classified, retried,
+# paused. A backend that never *answers* was handled by nothing. Pull a
+# network on an established connection and there is no RST and no ICMP;
+# the socket blocks until the OS abandons retransmission, on the order
+# of fifteen minutes. Throughout that window the job is neither running
+# nor failed: no error, so no retry, no log line, no pause. It looks
+# exactly like a slow migration.
+#
+# Worse, before the classification fixes the `blob_exists` source probe
+# reported such a failure as PERMANENT, which took the
+# record-a-finding-and-continue branch — the cursor advanced past the
+# blob, and with `failed` still 0 the run could reach `finish_completed`
+# and flip the pointer to a target missing everything the outage
+# covered. A migration reporting success having silently dropped
+# whatever was unreachable at the time.
+#
+# ## Why `s3_blackhole` and not `s3_stub`
+#
+# `s3_stub` points at `127.0.0.1:9999`, where nothing listens, so the
+# connection is REFUSED instantly. That path was never broken. A test
+# built on it would pass with no timeout configured anywhere and pin
+# nothing.
+#
+# `s3_blackhole` points at `192.0.2.1` — TEST-NET-1 (RFC 5737),
+# reserved for documentation and guaranteed unrouted. A SYN goes
+# unanswered, which is the hang. See `tests/common/server.env`.
+#
+# ## Why this is safe inside the shared suite
+#
+# The migration fails at `target.initialize()`, which runs BEFORE
+# `migration_readonly` is engaged (`backend_migration_service.rs`, the
+# comment on the target-init pause). So this file cannot leave the
+# server read-only for whatever runs after it — the reason this shape
+# was chosen over a mid-copy failure, which would hold the freeze.
+#
+# The run is cancelled at the end regardless, so the DB is left with no
+# non-terminal `backend_migration` row.
+#
+# Prerequisites: setup.hurl must have run (admin user exists).
+# =============================================================
+
+
+# ─────────────────────────────────────────────────────────────
+# Step 1 — Log in as admin.
+# ─────────────────────────────────────────────────────────────
+POST {{base_url}}/api/auth/login
+Content-Type: application/json
+{
+  "username": "{{username}}",
+  "password": "{{password}}"
+}
+
+HTTP 200
+[Captures]
+admin_token: jsonpath "$.access_token"
+
+
+# ─────────────────────────────────────────────────────────────
+# Step 2 — Dispatch the migration at the black hole.
+#
+# `backend_migration` is a DETACHED job: the handler is spawned and the
+# call returns 202 immediately, so this response carries no outcome and
+# no run_id. Step 3 polls for both.
+#
+# That is also why the bound is asserted as a polling budget rather
+# than with `duration` — the trigger returns in milliseconds no matter
+# how long the backend hangs, so timing THIS request would prove
+# nothing at all.
+# ─────────────────────────────────────────────────────────────
+POST {{base_url}}/api/admin/jobs/backend_migration/trigger?storage=s3_blackhole
+Authorization: Bearer {{admin_token}}
+
+HTTP 202
+[Asserts]
+jsonpath "$.ok" == true
+jsonpath "$.detached" == true
+
+
+# ─────────────────────────────────────────────────────────────
+# Step 3 — Poll until the run reaches Paused.
+#
+# THE LOAD-BEARING STEP. Runs are newest-first, so `$[0]` is ours.
+#
+# The retry budget IS the bounded-time assertion: 60 attempts × 2s =
+# 120s, after which hurl fails with the last assert error. Against the
+# old hanging behaviour the row would sit in `Running` for ~15 minutes
+# and this step would exhaust its budget — which is the entire point of
+# the file. Every other assertion here would eventually pass even
+# unbounded; this one would not.
+#
+# 120s is deliberately loose: comfortably above the observed ~31s (the
+# SDK's own attempts stacked on the 10s connect timeout) and three
+# orders of magnitude below the unbounded socket. A slow CI runner must
+# not make this flaky, and the failure it guards is nowhere near the
+# threshold.
+#
+# `Paused`, not `Failed`: that distinction is the whole plan. Failed is
+# terminal and needs a human; Paused resumes and finishes the migration
+# once the backend returns.
+# ─────────────────────────────────────────────────────────────
+GET {{base_url}}/api/admin/jobs/backend_migration/runs?limit=1
+Authorization: Bearer {{admin_token}}
+[Options]
+retry: 60
+retry-interval: 2000
+
+HTTP 200
+[Captures]
+blackhole_run_id: jsonpath "$[0].id"
+[Asserts]
+jsonpath "$[0].status" == "Paused"
+# The reason must name what went wrong, not merely that something did.
+# An operator reading only this string has to be able to tell an
+# unreachable backend from a wrong bucket — the first is worth waiting
+# out, the second never resolves on its own.
+jsonpath "$[0].error_message" contains "Transient Backend"
+jsonpath "$[0].error_message" contains "target backend init"
+# Absent, not null — the run is not over. A paused row carrying a
+# completion timestamp would make every "how long did this take" query
+# lie, and would read as finished in the admin panel.
+jsonpath "$[0].completed_at" not exists
+
+
+# ─────────────────────────────────────────────────────────────
+# Step 4 — Teardown: cancel the paused run.
+#
+# Mandatory, not tidiness. `open_or_start` picks up the latest
+# non-terminal row for a job name, so a `Paused` row left behind would
+# be RESUMED by the next `backend_migration` trigger in the suite —
+# silently retargeting that run at the black hole and failing a test
+# that has nothing to do with this file. Hurl files share one database.
+#
+# Cancel is also the path that releases `migration_readonly` for a
+# paused row (nothing to release here — this run never engaged it —
+# but the call is idempotent).
+# ─────────────────────────────────────────────────────────────
+POST {{base_url}}/api/admin/jobs/backend_migration/cancel
+Authorization: Bearer {{admin_token}}
+
+HTTP 200
+[Asserts]
+jsonpath "$.cancelled" == true
+jsonpath "$.run_id" == "{{blackhole_run_id}}"
+
+
+# ─────────────────────────────────────────────────────────────
+# Step 5 — Confirm the row is terminal, so the next trigger in the
+# suite starts fresh instead of resuming ours.
+# ─────────────────────────────────────────────────────────────
+GET {{base_url}}/api/admin/jobs/backend_migration/runs/{{blackhole_run_id}}
+Authorization: Bearer {{admin_token}}
+
+HTTP 200
+[Asserts]
+jsonpath "$.status" == "Cancelled"

tests/api/run.sh

@@ -227,6 +227,13 @@ hurl --variables-file "$API_DIR/test.env" --file-root "$REPO_ROOT/tests" --test
   "$API_DIR/nfc_normalization.hurl" \
   "$API_DIR/wopi_authz.hurl" \
   "$API_DIR/wopi_shared_drive.hurl" \
+  `# Second-to-last. The slowest file in the suite BY DESIGN: it waits` \
+  `# out an unreachable endpoint (~31s) to prove the wait is bounded, so` \
+  `# that cost belongs at the end rather than in the middle. It leaves no` \
+  `# read-only freeze behind — the migration fails at target init, before` \
+  `# the gate is engaged — and cancels its own run, so the shared DB is` \
+  `# clean for whatever follows.` \
+  "$API_DIR/backend_migration_blackhole.hurl" \
   `# LAST, deliberately — and kept last even though it no longer cuts` \
   `# the storage pointer over. It is the only scenario that depends on a` \
   `# second service (Azurite on 10000), so if that container is missing` \

tests/api/storage_multi_entry.hurl

@@ -43,8 +43,21 @@ Authorization: Bearer {{admin_token}}
 
 HTTP 200
 [Asserts]
-# Two entries declared, in _ENTRIES order.
+# Membership, asserted per entry rather than by counting them.
-jsonpath "$.entries" count == 3
+#
+# A `count ==` here breaks every time anyone declares a new entry in
+# server.env, and the failure says "expected 3, got 4" — which names
+# neither the entry that appeared nor whether it was supposed to. This
+# is a registry; it grows. `s3_blackhole` was the fourth and broke it
+# exactly this way.
+jsonpath "$.entries[*].name" contains "local_main"
+jsonpath "$.entries[*].name" contains "s3_stub"
+jsonpath "$.entries[*].name" contains "azurite"
+jsonpath "$.entries[*].name" contains "s3_blackhole"
+
+# Position, which is a SEPARATE property from membership and is
+# deliberately still pinned: entries come back in _ENTRIES order, and
+# the boot fallback below depends on `[0]` being the first declared.
 jsonpath "$.entries[0].name" == "local_main"
 jsonpath "$.entries[1].name" == "s3_stub"
 jsonpath "$.entries[2].name" == "azurite"

tests/common/server.env

@@ -42,7 +42,7 @@ OXICLOUD_NEXTCLOUD_ENABLED=true
 # `s3_stub` is declared but never activated — it lets storage_config.hurl
 # assert the entries table has more than one row without needing a
 # real S3 backend.
-OXICLOUD_STORAGE_ENTRIES=local_main,s3_stub,azurite
+OXICLOUD_STORAGE_ENTRIES=local_main,s3_stub,azurite,s3_blackhole
 OXICLOUD_STORAGE_local_main_BACKEND=local
 OXICLOUD_STORAGE_s3_stub_BACKEND=s3
 OXICLOUD_STORAGE_s3_stub_S3_BUCKET=oxicloud-test-stub
@@ -51,6 +51,36 @@ OXICLOUD_STORAGE_s3_stub_S3_ENDPOINT_URL=http://127.0.0.1:9999
 OXICLOUD_STORAGE_s3_stub_S3_ACCESS_KEY=stub
 OXICLOUD_STORAGE_s3_stub_S3_SECRET_KEY=stub
 
+# `s3_blackhole` — an endpoint that never answers, as opposed to
+# `s3_stub` above which refuses instantly.
+#
+# The distinction is the entire point. `127.0.0.1:9999` has nothing
+# listening, so a connection is REFUSED: the kernel returns ECONNREFUSED
+# immediately and the SDK reports an error straight away. That path was
+# always handled. A test built on it would pass even with no timeout
+# configured anywhere.
+#
+# `192.0.2.1` is TEST-NET-1 (RFC 5737), reserved for documentation and
+# guaranteed not to be routed. A SYN to it goes unanswered — no RST, no
+# ICMP — which is the failure that used to hang: with no bound, a socket
+# read blocks until the OS abandons retransmission, on the order of
+# fifteen minutes, during which the job is neither running nor failed.
+#
+# Declared but NEVER activated, like the other two. `backend_migration
+# ?storage=s3_blackhole` reaches it explicitly.
+#
+# If a CI network answers 192.0.2.1 with ICMP unreachable, the failure
+# degrades to the refused shape and `backend_migration_blackhole.hurl`
+# still passes — both classify transient and both pause. It would simply
+# stop pinning the timeout specifically. The assert on elapsed time in
+# that file is what would notice.
+OXICLOUD_STORAGE_s3_blackhole_BACKEND=s3
+OXICLOUD_STORAGE_s3_blackhole_S3_BUCKET=oxicloud-blackhole
+OXICLOUD_STORAGE_s3_blackhole_S3_REGION=us-east-1
+OXICLOUD_STORAGE_s3_blackhole_S3_ENDPOINT_URL=http://192.0.2.1:9999
+OXICLOUD_STORAGE_s3_blackhole_S3_ACCESS_KEY=blackhole
+OXICLOUD_STORAGE_s3_blackhole_S3_SECRET_KEY=blackhole
+
 # `azurite` — a REAL, reachable Azure backend, unlike `s3_stub` above.
 # It points at the Azurite emulator started by spawn-db.sh, which speaks
 # the actual Blob REST API, so this is the only way to exercise the Azure