diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 2f6ee98c..09d49b4e 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -327,6 +327,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#" diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index 98db6dea..c13532ed 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, }, @@ -554,6 +580,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 +1065,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})")) @@ -1340,6 +1428,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; diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index e8a18b43..2188a604 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) }) } @@ -371,9 +374,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, )) } } @@ -558,12 +561,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 +661,50 @@ 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. +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::*;