diff --git a/docs/plan/derived-blobs.md b/docs/plan/derived-blobs.md index f96b9bdf..e35f67ce 100644 --- a/docs/plan/derived-blobs.md +++ b/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 | -| 10 | `content_derived_blobs.source_hash` → Blob | orphan mapping | — | ✗ 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 | `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 | diff --git a/docs/plan/jobs-handling-recoverable-error.md b/docs/plan/jobs-handling-recoverable-error.md index 3337c0d4..16b0902b 100644 --- a/docs/plan/jobs-handling-recoverable-error.md +++ b/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 -diagnosis (see [Motivating incident](#motivating-incident)). +**Status: implemented 2026-09-08**, except the online-migration +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`, diff --git a/frontend/src/lib/api/types.ts b/frontend/src/lib/api/types.ts index 200112b7..4758bf7c 100644 --- a/frontend/src/lib/api/types.ts +++ b/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 diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index 7b0a0770..d30f1367 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/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} + {#if isRunning(job)} {t('admin.jobs.state_running', 'running')} + {:else if job.last_run_status} + + + {runStatusLabel(job.last_run_status)} + {:else} — {/if} diff --git a/src/common/config.rs b/src/common/config.rs index c240fba1..f006fb99 100644 --- a/src/common/config.rs +++ b/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::() + { + *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::().unwrap_or(false); diff --git a/src/common/di.rs b/src/common/di.rs index fe88a3fa..d65377de 100644 --- a/src/common/di.rs +++ b/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 = 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::{ diff --git a/src/domain/errors.rs b/src/domain/errors.rs index 0d6f2f44..540f1a71 100644 --- a/src/domain/errors.rs +++ b/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>(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>(entity_type: &'static str, message: S) -> Self { Self { @@ -317,3 +375,41 @@ impl From 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"); + } +} diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index 58dc0b5c..88e18cf4 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -275,6 +275,30 @@ fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option, 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", diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index a78b9310..f1077050 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/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, - RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, RunSummary, derive_progress, - record_or_log, run_or_resume, + CANCEL_INTENT_PARAM, CANCEL_INTENT_TERMINATE, Finding, JobStore, JobStoreProvider, OpenedRun, + ProgressKind, RecoverableAdapter, RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, + RunSummary, derive_progress, record_or_log, run_or_resume, }; pub use registry::{ JobEntry, JobRegistry, JobSummary, PausedRunBrief, RegisterError, StartupTrigger, diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 2f6ee98c..8e06e306 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/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 { - // `(stats->>'scanned_count')::BIGINT` — text cast rather than - // `->` numeric extraction because the stored value has been - // written via `((...)::text)::jsonb` in `checkpoint`, which - // may present as either a JSON number or a JSON string - // depending on prior versions. `::BIGINT` handles both. + // `(stats ->> $2)::BIGINT` — text extraction then cast, rather + // than `->` numeric extraction, because the stored value has been + // written via `((...)::text)::jsonb` in `checkpoint` and may + // present as either a JSON number or a JSON string 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 { let row: Option<(Option,)> = 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>, + 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, Option>)> = 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 diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index 98db6dea..ff3728f8 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -157,6 +157,32 @@ pub enum RunOutcome { Paused { cursor: Vec, }, + /// 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, + 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; + async fn scanned_count(&self) -> Result { + 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; + + /// 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, + ) -> 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>) -> 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>, + 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, DomainError> { Ok(self.state.lock().unwrap().string_params.get(key).cloned()) } - async fn scanned_count(&self) -> Result { - Ok(self.state.lock().unwrap().scanned_count) + /// Mirrors the PG row: `scanned_count` is its own column-like + /// field, every other counter lives in the merged stats map. + async fn stat_u64(&self, key: &str) -> Result { + 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>, + 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 { + 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>, + ) -> 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>, + ) -> 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 = 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 = 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 = 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()); diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 951c23c8..7692d982 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/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, #[serde(skip_serializing_if = "Option::is_none")] pub interval_ms: Option, #[serde(skip_serializing_if = "Option::is_none")] diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index e8a18b43..f5e29587 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/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| { - DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}")) - })?; + client + .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| { - DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}")) - })?; + client + .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| { - DomainError::internal_error("Azure", format!("Failed to upload blob {hash}: {e}")) - })?; + client + .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( - ErrorKind::NotFound, - "Azure", - format!("Failed to get blob {hash}: {e}"), - )); + return Err(azure_read_error(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( - ErrorKind::NotFound, - "Azure", - format!("Failed to get blob range {hash}: {e}"), + return Err(azure_read_error( + 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( - "Azure", - format!("Failed to delete blob {hash}: {e}"), + Err(azure_domain_error( + format!("Failed to delete blob {hash}"), + &e, )) } } @@ -395,9 +393,13 @@ impl BlobStorageBackend for AzureBlobBackend { if status == Some(azure_core::StatusCode::NotFound) { Ok(false) } else { - Err(DomainError::internal_error( - "Azure", - format!("Failed to check blob {hash}: {e}"), + // Only the 404 means "absent"; everything else keeps + // its transient/permanent class so the migration's + // 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| { - DomainError::new( - ErrorKind::NotFound, - "Azure", - format!("Failed to stat blob {hash}: {e}"), - ) - })?; + let props = client + .get_properties() + .await + .map_err(|e| azure_read_error(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( - "Blob", + // Classified: `backend_consistency` fails the whole + // 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::*; diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 6107e193..0ee17d39 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/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 { - message: format!("probed backend init: {e}"), - }; + // Nothing has been scanned yet, so there is no cursor to keep + // — 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 { - message: format!( - "backend enumeration failed on {}: {e}", - backend.backend_type() - ), - }; + // Transient (throttle, 5xx, connection reset) pauses at + // the cursor so a resume continues the sweep; + // everything else fails terminally. Losing a + // half-finished audit of a million-object bucket to a + // brief 503 is the case this distinction exists for — + // the retry decorator has already given up by the time + // the error arrives here. + return RunOutcome::from_domain_error( + cursor.as_ref().map(|s| s.as_bytes()), + &format!("backend enumeration failed on {}", backend.backend_type()), + &e, + ); } }; diff --git a/src/infrastructure/services/backend_migration_service.rs b/src/infrastructure/services/backend_migration_service.rs index aa6b4c15..7ca33df7 100644 --- a/src/infrastructure/services/backend_migration_service.rs +++ b/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 { - message: format!("target backend init: {e}"), - }; + // Runs BEFORE `migration_readonly` is engaged, so pausing + // 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 failed_count = 0u64; - let mut source_missing_count = 0u64; + let mut skipped_count: u64 = restore("skipped").await; + let mut failed_count = restore("failed").await; + 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 - // finding — treat like a network blip. - // Skipping this row on this run; a re-run - // will re-probe. If the failure is - // persistent, `blobs_consistency` catches - // it. + // Permanent probe failure. Still skipped rather + // than fatal — one unprobeable blob must not + // abort the migration — but it now records a + // finding, so the run cannot report clean while + // having skipped rows, and `blobs_consistency` + // 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 = 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). /// diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 4f0fac15..d089f1ad 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/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| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to open blob {}: {}", hash, e), - ) - })?; + // Was unconditional NotFound: a stale NFS handle or an + // unmounted iSCSI target reported the blob as missing. + let file = File::open(&blob_path) + .await + .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| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to open blob {}: {}", hash, e), - ) - })?; + let mut file = File::open(&blob_path) + .await + .map_err(|e| local_io_error("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| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to stat blob {}: {}", hash, e), - ) - })?; + let meta = fs::metadata(&blob_path) + .await + .map_err(|e| local_io_error("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" + ); + } + } } diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index b872a0bb..1522429e 100644 --- a/src/infrastructure/services/mod.rs +++ b/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; diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index 0141a36f..1622d22a 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/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") diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index 6f877f2c..4bee0f2d 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/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( - "S3", - format!("Cannot access bucket '{}': {}", self.bucket, e), - ) + // Classified like every other SDK call. A refused + // connection or a 5xx here is the endpoint being + // 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| { - DomainError::internal_error( - "S3", - format!("Failed to upload blob {}: {}", hash, e), - ) - })?; + .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?; // Clean up local source after successful upload let _ = fs::remove_file(&source_path).await; @@ -215,12 +242,7 @@ impl BlobStorageBackend for S3BlobBackend { .body(body) .send() .await - .map_err(|e| { - DomainError::internal_error( - "S3", - format!("Failed to upload blob {}: {}", hash, e), - ) - })?; + .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?; Ok(size) }) @@ -253,12 +275,7 @@ impl BlobStorageBackend for S3BlobBackend { .body(ByteStream::from(data)) .send() .await - .map_err(|e| { - DomainError::internal_error( - "S3", - format!("Failed to upload blob {}: {}", hash, e), - ) - })?; + .map_err(|e| s3_domain_error("S3", format!("Failed to upload blob {hash}"), &e))?; Ok(size) }) } @@ -297,11 +314,30 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to get blob {}: {}", hash, e), - ) + // Only a real NoSuchKey is NotFound. This used to + // label EVERY read failure that way — a refused + // connection, a 503, an expired credential all + // 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> @@ -336,11 +372,21 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to get blob range {}: {}", hash, e), - ) + // Same rule as the full read: only a real NoSuchKey + // is NotFound. Ranged reads feed CDC reassembly and + // deep verification, so mislabelling an outage here + // 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| { - DomainError::internal_error( - "S3", - format!("Failed to delete blob {}: {}", hash, e), - ) - })?; + .map_err(|e| s3_domain_error("S3", format!("Failed to delete blob {hash}"), &e))?; Ok(()) }) @@ -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 - let service_err = e.into_service_error(); - if service_err.is_not_found() { + // A 404 is the only answer that means "absent". Classify + // before consuming the SdkError so everything else keeps + // 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( - "S3", - format!("Failed to check blob {}: {}", hash, service_err), - )) + Err(classified) } } } @@ -423,11 +466,19 @@ impl BlobStorageBackend for S3BlobBackend { .send() .await .map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "S3", - format!("Failed to stat blob {}: {}", hash, e), - ) + // `head_object` reports a missing key as NotFound + // rather than NoSuchKey, so match on the typed + // variant the SDK actually returns here. + 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( - ErrorKind::InternalError, - "Blob", - format!("S3 ListObjectsV2 failed: {e}"), - ) + // Classified, because `backend_consistency` fails the + // whole run on an enumeration error — a throttle + // midway through a million-object bucket should be + // retryable rather than throwing the sweep away. + s3_domain_error("Blob", "S3 ListObjectsV2 failed".to_string(), &e) })?; requests += 1; @@ -638,6 +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( + entity: &'static str, + context: String, + err: &aws_sdk_s3::error::SdkError, +) -> 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: ` — 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(err: &aws_sdk_s3::error::SdkError) -> String where E: aws_sdk_s3::error::ProvideErrorMetadata + std::fmt::Debug, diff --git a/src/infrastructure/services/timeout_blob_backend.rs b/src/infrastructure/services/timeout_blob_backend.rs new file mode 100644 index 00000000..3bdd9402 --- /dev/null +++ b/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, + /// Bound for time-to-first-byte on reads. + pub open: Option, + /// Bound for the whole of a write. `None` (the default) leaves large + /// uploads unbounded — see the module docs. + pub write: Option, +} + +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, + policy: TimeoutPolicy, +} + +impl TimeoutBlobBackend { + pub fn new(inner: Arc, 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( + limit: Option, + backend: &'static str, + name: L, + fut: impl std::future::Future>, +) -> Result +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> + 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> + 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> + 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> + 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> + 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> + 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> + 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, + ) -> Pin> + 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> + 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> + 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> + 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> + 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, + limit_n: usize, + ) -> Pin> + 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 { + 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> { + self.inner.uncached().map(|inner| { + Arc::new(TimeoutBlobBackend::new(inner, self.policy.clone())) + as Arc + }) + } + + 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> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn put_blob( + &self, + _hash: &str, + _source_path: &Path, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn put_blob_from_bytes( + &self, + _hash: &str, + _data: Bytes, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn get_blob_stream( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn get_blob_range_stream( + &self, + _hash: &str, + _start: u64, + _end: Option, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn delete_blob( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn blob_exists( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn blob_size( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn health_check( + &self, + ) -> Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + Box::pin(async { std::future::pending().await }) + } + fn backend_type(&self) -> &'static str { + "hanging" + } + fn local_blob_path(&self, _hash: &str) -> Option { + 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()); + } +} diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index e6c699ab..20b03f45 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -2570,41 +2570,94 @@ pub async fn list_jobs(State(state): State>) -> 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, Option)> = 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, + Option, + Option, + ); + let latest_rows: Vec = 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 = paused_rows + type LatestRun = (String, chrono::DateTime, PausedRunBrief); + let by_name: std::collections::HashMap = latest_rows .into_iter() - .map(|(name, id, scanned, total)| { + .map(|(name, status, id, started_at, scanned, total)| { ( name, - PausedRunBrief { - id, - scanned: scanned.unwrap_or(0).max(0) as u64, - total: total.filter(|t| *t > 0).map(|t| t as u64), - }, + ( + status, + started_at, + 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 - && !job.running - && let Some(paused) = by_name.get(&job.name) - { - job.paused_run = Some(paused.clone()); + if !job.recoverable { + continue; + } + let Some((status, started_at, brief)) = by_name.get(&job.name) else { + 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)) => ( - 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(Some(run_id)) => { + // Cancelling a PAUSED migration has to give writes back + // here, because nothing else will. + // + // A Running row re-enters the handler, which releases the + // gate itself at its next cancel poll. A Paused row does + // 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 + // `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!({ diff --git a/src/interfaces/errors.rs b/src/interfaces/errors.rs index a859e261..e5711950 100644 --- a/src/interfaces/errors.rs +++ b/src/interfaces/errors.rs @@ -132,6 +132,11 @@ impl From 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 { diff --git a/tests/api/backend_migration_blackhole.hurl b/tests/api/backend_migration_blackhole.hurl new file mode 100644 index 00000000..acae505c --- /dev/null +++ b/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" diff --git a/tests/api/run.sh b/tests/api/run.sh index 10643e8d..7aa66375 100755 --- a/tests/api/run.sh +++ b/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` \ diff --git a/tests/api/storage_multi_entry.hurl b/tests/api/storage_multi_entry.hurl index 847e93fb..a5d11b28 100644 --- a/tests/api/storage_multi_entry.hurl +++ b/tests/api/storage_multi_entry.hurl @@ -43,8 +43,21 @@ Authorization: Bearer {{admin_token}} HTTP 200 [Asserts] -# Two entries declared, in _ENTRIES order. -jsonpath "$.entries" count == 3 +# Membership, asserted per entry rather than by counting them. +# +# 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" diff --git a/tests/common/server.env b/tests/common/server.env index a661242d..77c6760d 100644 --- a/tests/common/server.env +++ b/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