diff --git a/src/common/di.rs b/src/common/di.rs index bae22651..92b1b572 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1494,13 +1494,13 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; - // Finds derived mappings whose Blob is gone on either side. Nothing - // else can: a row whose SOURCE was reaped still holds a valid + // Both satellite tables, checked for mappings whose Blob is gone. + // Nothing else can: a row whose SOURCE was reaped still holds a valid // reference to a real artifact with a correct refcount, so every // other check agrees the system is healthy while the artifact is // pinned forever. Read-only. let _ = Arc::new( - crate::infrastructure::services::derived_consistency_service::DerivedConsistencyCheck::new( + crate::infrastructure::services::satellites_consistency_service::SatellitesConsistencyCheck::new( maintenance_pool.clone(), ), ) diff --git a/src/infrastructure/services/derived_consistency_service.rs b/src/infrastructure/services/derived_consistency_service.rs deleted file mode 100644 index f22380f4..00000000 --- a/src/infrastructure/services/derived_consistency_service.rs +++ /dev/null @@ -1,328 +0,0 @@ -//! `derived_consistency` — the last unbuilt row of the coverage matrix. -//! -//! Walks `storage.content_derived_blobs` and reports mappings that point at -//! Blobs which no longer exist, in either direction. -//! -//! ### Why nothing else finds these -//! -//! Every other job reasons from a Blob outwards: `blobs_consistency` and -//! `manifests_consistency` recompute refcounts for rows that exist, -//! `backend_consistency` merge-joins the registry against the backend. A -//! derived row whose SOURCE is gone breaks none of those invariants — the -//! row holds a perfectly valid reference to a real artifact, the refcount is -//! exactly right, and the bytes are present on the backend. Every check -//! agrees the system is healthy. -//! -//! It is only wrong one level up: nothing will ever reap that source again, -//! so `purge_derived_blobs` can never fire, so the mapping is unreachable and -//! its artifact is pinned forever. A leak that looks like correctness. -//! -//! That is not hypothetical — it shipped. Background thumbnail generation is -//! spawned and unawaited, so an upload deleted promptly had its render -//! complete after GC reaped the blob and then record three mappings to a -//! corpse (fixed at the write side in `store_derived_blob`, which now -//! refuses a mapping whose source is gone). This job finds the ones already -//! on disk, which that fix cannot reach. -//! -//! ### Per-row checks -//! -//! * `derived_orphan_mapping` (severity `inconsistent`) — `source_hash` has -//! neither a manifest nor a blob row. Storage overhead that grows and never -//! reclaims. Recovery = delete the row, which releases the artifact. -//! * `derived_dangling_blob` (severity `data_loss`) — `blob_hash` has no Blob -//! behind it. The opposite and the more serious one: the mapping promises -//! an artifact that is gone, so a read finds a row and then fails. -//! -//! Read-only, per the house default. Both findings name a row rather than a -//! range, so recovery can act on them individually. - -use std::sync::Arc; - -use async_trait::async_trait; -use sqlx::PgPool; - -use crate::infrastructure::scheduler::{ - JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, - RunStatus, record_or_log, -}; - -pub const DERIVED_CONSISTENCY_JOB_NAME: &str = "derived_consistency"; - -/// Rows per page. Each is two indexed existence probes folded into the page -/// query, so this can be larger than a job doing per-row I/O. -const BATCH_SIZE: i64 = 500; - -pub struct DerivedConsistencyCheck { - pool: Arc, -} - -/// One row plus the two existence answers, resolved server-side so a page -/// costs one round-trip rather than `2 × rows`. -#[derive(Debug, sqlx::FromRow)] -struct DerivedRow { - source_hash: String, - kind: String, - variant: String, - blob_hash: String, - source_exists: bool, - artifact_exists: bool, -} - -impl DerivedConsistencyCheck { - pub fn new(pool: Arc) -> Self { - Self { pool } - } - - pub async fn register_recoverable_job( - self: Arc, - registry: &JobRegistry, - provider: &Arc, - ) -> Arc { - registry - .register_recoverable_job(self.clone(), provider.clone(), None) - .await; - self - } - - /// Page query, keyed on the full primary key. - /// - /// Row-value comparison (`(a,b,c) > ($1,$2,$3)`) rather than - /// `source_hash > $1`: a source has several variants, so a page boundary - /// can fall inside one, and advancing by source alone would skip its - /// remaining rows. The tuple form is also index-friendly — it matches the - /// primary key's own ordering. - /// - /// "Exists" means EITHER table, because `source_hash` and `blob_hash` both - /// name a Blob: a manifest for CDC content, a bare `storage.blobs` row for - /// legacy whole-file content. Checking only one would report every legacy - /// blob as missing. - const PAGE_SQL: &'static str = r#" - SELECT d.source_hash, - d.kind, - d.variant, - d.blob_hash, - (EXISTS (SELECT 1 FROM storage.chunk_manifests m WHERE m.file_hash = d.source_hash) - OR EXISTS (SELECT 1 FROM storage.blobs b WHERE b.hash = d.source_hash)) - AS source_exists, - (EXISTS (SELECT 1 FROM storage.chunk_manifests m WHERE m.file_hash = d.blob_hash) - OR EXISTS (SELECT 1 FROM storage.blobs b WHERE b.hash = d.blob_hash)) - AS artifact_exists - FROM storage.content_derived_blobs d - WHERE ($1::text IS NULL - OR (d.source_hash, d.kind, d.variant) > ($1::text, $2::text, $3::text)) - ORDER BY d.source_hash, d.kind, d.variant - LIMIT $4"#; -} - -/// Cursor is the primary-key triple, newline-joined. -/// -/// Safe as a delimiter: `source_hash` is hex, `kind` comes from a CHECK -/// constraint, and `variant` is a size/format token — none can contain a -/// newline. -fn encode_cursor(r: &DerivedRow) -> Vec { - format!("{}\n{}\n{}", r.source_hash, r.kind, r.variant).into_bytes() -} - -fn decode_cursor(bytes: Vec) -> Result, String> { - if bytes.is_empty() { - return Ok(None); - } - let s = String::from_utf8(bytes).map_err(|e| format!("not valid UTF-8: {e}"))?; - let mut parts = s.splitn(3, '\n'); - match (parts.next(), parts.next(), parts.next()) { - (Some(a), Some(b), Some(c)) => Ok(Some((a.into(), b.into(), c.into()))), - _ => Err(format!( - "expected three newline-separated fields, got {s:?}" - )), - } -} - -#[async_trait] -impl RecoverableJobHandler for DerivedConsistencyCheck { - fn name(&self) -> &str { - DERIVED_CONSISTENCY_JOB_NAME - } - - async fn count_total(&self) -> Option { - sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM storage.content_derived_blobs") - .fetch_one(self.pool.as_ref()) - .await - .ok() - .map(|(n,)| n.max(0) as u64) - } - - async fn run_resumable( - &self, - store: &dyn JobStore, - _args: &JobRunArgs, - resume_cursor: Option>, - ) -> RunOutcome { - let mut cursor = match resume_cursor.map(decode_cursor).transpose() { - Ok(c) => c.flatten(), - Err(message) => return RunOutcome::Failed { message }, - }; - - let mut finding_count = 0u64; - - loop { - match store.status().await { - Ok(RunStatus::CancelRequested) => { - return RunOutcome::Paused { - cursor: cursor - .as_ref() - .map(|(a, b, c)| format!("{a}\n{b}\n{c}").into_bytes()) - .unwrap_or_default(), - }; - } - Ok(_) => {} - Err(e) => { - return RunOutcome::Failed { - message: format!("status poll: {e}"), - }; - } - } - - let (ch, ck, cv) = match &cursor { - Some((a, b, c)) => (Some(a.as_str()), Some(b.as_str()), Some(c.as_str())), - None => (None, None, None), - }; - - let rows: Vec = match sqlx::query_as(Self::PAGE_SQL) - .bind(ch) - .bind(ck) - .bind(cv) - .bind(BATCH_SIZE) - .fetch_all(self.pool.as_ref()) - .await - { - Ok(r) => r, - Err(e) => { - return RunOutcome::Failed { - message: format!("derived page: {e}"), - }; - } - }; - - if rows.is_empty() { - break; - } - - for row in &rows { - if !row.source_exists { - finding_count += 1; - record_or_log( - store, - DERIVED_CONSISTENCY_JOB_NAME, - "derived_orphan_mapping", - "inconsistent", - None, - serde_json::json!({ - "source_hash": row.source_hash, - "kind": row.kind, - "variant": row.variant, - "blob_hash": row.blob_hash, - "note": "source Blob is gone, so purge_derived_blobs can never fire; \ - this row pins its artifact forever", - }), - ) - .await; - } - - if !row.artifact_exists { - finding_count += 1; - record_or_log( - store, - DERIVED_CONSISTENCY_JOB_NAME, - "derived_dangling_blob", - "data_loss", - None, - serde_json::json!({ - "source_hash": row.source_hash, - "kind": row.kind, - "variant": row.variant, - "blob_hash": row.blob_hash, - "note": "mapping promises an artifact with no Blob behind it; \ - a read finds the row and then fails", - }), - ) - .await; - } - } - - let scanned = rows.len() as u64; - cursor = rows - .last() - .map(|r| (r.source_hash.clone(), r.kind.clone(), r.variant.clone())); - - let checkpoint = rows.last().map(encode_cursor).unwrap_or_default(); - if let Err(e) = store.checkpoint(checkpoint, scanned).await { - return RunOutcome::Failed { - message: format!("checkpoint: {e}"), - }; - } - - if scanned < BATCH_SIZE as u64 { - break; - } - } - - tracing::info!( - target: "oxicloud::consistency", - event = "derived_consistency.completed", - run_id = %store.run_id(), - finding_count = finding_count, - "derived_consistency completed with {} finding(s)", - finding_count - ); - - RunOutcome::completed() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn row(source: &str, kind: &str, variant: &str) -> DerivedRow { - DerivedRow { - source_hash: source.into(), - kind: kind.into(), - variant: variant.into(), - blob_hash: "b".into(), - source_exists: true, - artifact_exists: true, - } - } - - /// The cursor must survive the round trip, or a resumed run silently - /// restarts or skips — the failure mode a paged audit job can least - /// afford, since it would under-report rather than error. - #[test] - fn cursor_round_trips() { - let r = row("0a1b", "thumbnail", "preview.webp"); - let decoded = decode_cursor(encode_cursor(&r)).unwrap(); - assert_eq!( - decoded, - Some(( - "0a1b".to_string(), - "thumbnail".to_string(), - "preview.webp".to_string() - )) - ); - } - - /// An empty cursor means "from the beginning", not a parse error — the - /// scheduler hands one back for a fresh run. - #[test] - fn empty_cursor_starts_from_the_beginning() { - assert_eq!(decode_cursor(Vec::new()).unwrap(), None); - } - - /// A malformed cursor must fail loudly. Silently treating it as "start - /// over" would turn a corrupt checkpoint into a job that never finishes - /// and never says why. - #[test] - fn malformed_cursor_is_an_error() { - assert!(decode_cursor(b"only-one-field".to_vec()).is_err()); - } -} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 1e6d9f83..5ad26293 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -10,7 +10,6 @@ pub mod compression_service; pub mod consistency_batch_service; pub mod db_pool_monitor; pub mod dedup_service; -pub mod derived_consistency_service; pub mod dpop_nonce_service; pub mod dpop_replay_cache; pub mod dpop_verifier; @@ -52,6 +51,7 @@ pub mod plugins; pub mod recent_recording_hook; pub mod retry_blob_backend; pub mod s3_blob_backend; +pub mod satellites_consistency_service; pub mod search_index; pub mod session_cleanup_service; pub mod session_liveness_gauges; diff --git a/src/infrastructure/services/satellites_consistency_service.rs b/src/infrastructure/services/satellites_consistency_service.rs new file mode 100644 index 00000000..dbdea998 --- /dev/null +++ b/src/infrastructure/services/satellites_consistency_service.rs @@ -0,0 +1,506 @@ +//! `satellites_consistency` — the last unbuilt row of the coverage matrix. +//! +//! Walks both satellite tables and reports mappings pointing at Blobs that no +//! longer exist. One job rather than two, because the tables are one concept +//! — the content-keyed and file-keyed halves of "things attached to a Blob" — +//! and the vocabulary already exists in `storage.copy_file_satellites`. +//! +//! ### Why nothing else finds these +//! +//! Every other job reasons from a Blob outwards: `blobs_consistency` and +//! `manifests_consistency` recompute refcounts for rows that exist, +//! `backend_consistency` merge-joins the registry against the backend. A +//! satellite row whose SOURCE is gone breaks none of those invariants — the +//! row holds a valid reference to a real artifact, the refcount is exactly +//! right, and the bytes are present on the backend. Every check agrees the +//! system is healthy. +//! +//! It is only wrong one level up: nothing will ever reap that source again, +//! so `purge_derived_blobs` can never fire, so the mapping is unreachable and +//! its artifact is pinned forever. A leak that looks like correctness, which +//! is why it survived four full suite runs before being named. +//! +//! That is not hypothetical — it shipped. Background thumbnail generation is +//! spawned and unawaited, so an upload deleted promptly had its render +//! complete after GC reaped the blob and then record three mappings to a +//! corpse. Fixed at the write side in `store_derived_blob`, which now refuses +//! a mapping whose source is gone; this job finds the ones already on disk, +//! which that fix cannot reach. +//! +//! ### Per-row checks +//! +//! * `derived_orphan_mapping` (`inconsistent`) — a `content_derived_blobs` +//! row whose `source_hash` has neither a manifest nor a blob row. Storage +//! that grows and never reclaims. +//! * `derived_dangling_blob` (`data_loss`) — its `blob_hash` has no Blob. +//! The mapping promises an artifact that is gone, so a read finds the row +//! and then fails. Recoverable in practice: a derived artifact is a pure +//! function of its source, so re-rendering restores it. +//! * `attached_dangling_blob` (`data_loss`) — the same for +//! `file_attached_blobs`, and **the one that cannot be recovered**. These +//! bytes are user-supplied — a client-generated PDF preview has no +//! server-side render path — so there is nothing to regenerate from. Same +//! finding shape as the derived case, materially higher stakes. +//! +//! There is deliberately no orphan-mapping check for the attached table: +//! `file_id` is `REFERENCES storage.files(id) ON DELETE CASCADE`, so a row +//! cannot outlive its file. The database enforces what the derived table +//! cannot, since a content hash has no row to point a foreign key at — which +//! is precisely why only that half could rot. +//! +//! Read-only, per the house default. Findings name a row rather than a range, +//! so recovery can act on them individually. + +use std::sync::Arc; + +use async_trait::async_trait; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; + +pub const SATELLITES_CONSISTENCY_JOB_NAME: &str = "satellites_consistency"; + +/// Rows per page. Existence probes fold into the page query, so a page costs +/// one round-trip rather than `2 × rows`. +const BATCH_SIZE: i64 = 500; + +/// "Does this hash name a Blob?" — either table, because a Blob is a manifest +/// for CDC content and a bare `storage.blobs` row for legacy whole-file +/// content. Checking one would report every legacy blob as missing. +macro_rules! blob_exists { + ($col:literal) => { + concat!( + "(EXISTS (SELECT 1 FROM storage.chunk_manifests m WHERE m.file_hash = ", + $col, + ") OR EXISTS (SELECT 1 FROM storage.blobs b WHERE b.hash = ", + $col, + "))" + ) + }; +} + +pub struct SatellitesConsistencyCheck { + pool: Arc, +} + +#[derive(Debug, sqlx::FromRow)] +struct DerivedRow { + source_hash: String, + kind: String, + variant: String, + blob_hash: String, + source_exists: bool, + artifact_exists: bool, +} + +#[derive(Debug, sqlx::FromRow)] +struct AttachedRow { + file_id: Uuid, + kind: String, + variant: String, + blob_hash: String, + uploaded_by: Uuid, + artifact_exists: bool, +} + +impl SatellitesConsistencyCheck { + pub fn new(pool: Arc) -> Self { + Self { pool } + } + + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } + + /// Both page queries key on the full primary key with a row-value + /// comparison, not on the first column: a source (or file) has several + /// variants, so a page boundary can fall inside one and advancing by the + /// first column alone would skip the rest. The tuple form also matches + /// the primary key's own ordering, so it stays index-friendly. + const DERIVED_PAGE_SQL: &'static str = concat!( + "SELECT d.source_hash, d.kind, d.variant, d.blob_hash, ", + blob_exists!("d.source_hash"), + " AS source_exists, ", + blob_exists!("d.blob_hash"), + " AS artifact_exists + FROM storage.content_derived_blobs d + WHERE ($1::text IS NULL + OR (d.source_hash, d.kind, d.variant) > ($1::text, $2::text, $3::text)) + ORDER BY d.source_hash, d.kind, d.variant + LIMIT $4" + ); + + const ATTACHED_PAGE_SQL: &'static str = concat!( + "SELECT a.file_id, a.kind, a.variant, a.blob_hash, a.uploaded_by, ", + blob_exists!("a.blob_hash"), + " AS artifact_exists + FROM storage.file_attached_blobs a + WHERE ($1::uuid IS NULL + OR (a.file_id, a.kind, a.variant) > ($1::uuid, $2::text, $3::text)) + ORDER BY a.file_id, a.kind, a.variant + LIMIT $4" + ); +} + +/// Cursor is `{phase}\n{a}\n{b}\n{c}`. +/// +/// The phase is what lets one job walk two tables and still resume exactly: +/// without it, a cursor from the attached pass would be replayed against the +/// derived table and silently re-scan or skip. Newline is a safe delimiter — +/// hashes are hex, uuids are uuids, `kind` comes from a CHECK constraint, and +/// `variant` is a size/format token. +#[derive(Debug, PartialEq, Clone, Copy)] +enum Phase { + Derived, + Attached, +} + +impl Phase { + fn as_str(self) -> &'static str { + match self { + Phase::Derived => "derived", + Phase::Attached => "attached", + } + } +} + +fn encode_cursor(phase: Phase, a: &str, b: &str, c: &str) -> Vec { + format!("{}\n{a}\n{b}\n{c}", phase.as_str()).into_bytes() +} + +type Cursor = Option<(Phase, String, String, String)>; + +fn decode_cursor(bytes: Vec) -> Result { + if bytes.is_empty() { + return Ok(None); + } + let s = String::from_utf8(bytes).map_err(|e| format!("not valid UTF-8: {e}"))?; + let mut parts = s.splitn(4, '\n'); + match (parts.next(), parts.next(), parts.next(), parts.next()) { + (Some("derived"), Some(a), Some(b), Some(c)) => { + Ok(Some((Phase::Derived, a.into(), b.into(), c.into()))) + } + (Some("attached"), Some(a), Some(b), Some(c)) => { + Ok(Some((Phase::Attached, a.into(), b.into(), c.into()))) + } + _ => Err(format!("malformed cursor: {s:?}")), + } +} + +#[async_trait] +impl RecoverableJobHandler for SatellitesConsistencyCheck { + fn name(&self) -> &str { + SATELLITES_CONSISTENCY_JOB_NAME + } + + async fn count_total(&self) -> Option { + sqlx::query_as::<_, (i64,)>( + "SELECT (SELECT COUNT(*) FROM storage.content_derived_blobs) + + (SELECT COUNT(*) FROM storage.file_attached_blobs)", + ) + .fetch_one(self.pool.as_ref()) + .await + .ok() + .map(|(n,)| n.max(0) as u64) + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + let start = match resume_cursor.map(decode_cursor).transpose() { + Ok(c) => c.flatten(), + Err(message) => return RunOutcome::Failed { message }, + }; + + let mut finding_count = 0u64; + + // ── Phase 1: content-keyed ─────────────────────────────────────── + // Skipped entirely when resuming mid-attached, since that phase runs + // strictly after this one. + let mut derived_cursor = match &start { + Some((Phase::Attached, ..)) => None, + Some((Phase::Derived, a, b, c)) => Some((a.clone(), b.clone(), c.clone())), + None => None, + }; + let skip_derived = matches!(&start, Some((Phase::Attached, ..))); + + if !skip_derived { + loop { + if let Some(outcome) = poll_cancel( + store, + derived_cursor + .as_ref() + .map(|(a, b, c)| encode_cursor(Phase::Derived, a, b, c)), + ) + .await + { + return outcome; + } + + let (ch, ck, cv) = match &derived_cursor { + Some((a, b, c)) => (Some(a.as_str()), Some(b.as_str()), Some(c.as_str())), + None => (None, None, None), + }; + + let rows: Vec = match sqlx::query_as(Self::DERIVED_PAGE_SQL) + .bind(ch) + .bind(ck) + .bind(cv) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + return RunOutcome::Failed { + message: format!("derived page: {e}"), + }; + } + }; + if rows.is_empty() { + break; + } + + for row in &rows { + if !row.source_exists { + finding_count += 1; + record_or_log( + store, + SATELLITES_CONSISTENCY_JOB_NAME, + "derived_orphan_mapping", + "inconsistent", + None, + serde_json::json!({ + "source_hash": row.source_hash, + "kind": row.kind, + "variant": row.variant, + "blob_hash": row.blob_hash, + "note": "source Blob is gone, so purge_derived_blobs can never \ + fire; this row pins its artifact forever", + }), + ) + .await; + } + if !row.artifact_exists { + finding_count += 1; + record_or_log( + store, + SATELLITES_CONSISTENCY_JOB_NAME, + "derived_dangling_blob", + "data_loss", + None, + serde_json::json!({ + "source_hash": row.source_hash, + "kind": row.kind, + "variant": row.variant, + "blob_hash": row.blob_hash, + "recoverable": true, + "note": "artifact missing; derived content is a pure function of \ + its source, so re-rendering restores it", + }), + ) + .await; + } + } + + let scanned = rows.len() as u64; + let last = rows.last().unwrap(); + derived_cursor = Some(( + last.source_hash.clone(), + last.kind.clone(), + last.variant.clone(), + )); + if let Err(e) = store + .checkpoint( + encode_cursor(Phase::Derived, &last.source_hash, &last.kind, &last.variant), + scanned, + ) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + if scanned < BATCH_SIZE as u64 { + break; + } + } + } + + // ── Phase 2: file-keyed ────────────────────────────────────────── + // No orphan-mapping check here: `file_id` is ON DELETE CASCADE, so a + // row cannot outlive its file. Only the artifact side can rot. + let mut attached_cursor: Option<(Uuid, String, String)> = match &start { + Some((Phase::Attached, a, b, c)) => match Uuid::parse_str(a) { + Ok(id) => Some((id, b.clone(), c.clone())), + Err(e) => { + return RunOutcome::Failed { + message: format!("attached cursor is not a uuid: {e}"), + }; + } + }, + _ => None, + }; + + loop { + if let Some(outcome) = poll_cancel( + store, + attached_cursor + .as_ref() + .map(|(a, b, c)| encode_cursor(Phase::Attached, &a.to_string(), b, c)), + ) + .await + { + return outcome; + } + + let (ch, ck, cv) = match &attached_cursor { + Some((a, b, c)) => (Some(*a), Some(b.as_str()), Some(c.as_str())), + None => (None, None, None), + }; + + let rows: Vec = match sqlx::query_as(Self::ATTACHED_PAGE_SQL) + .bind(ch) + .bind(ck) + .bind(cv) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + return RunOutcome::Failed { + message: format!("attached page: {e}"), + }; + } + }; + if rows.is_empty() { + break; + } + + for row in &rows { + if !row.artifact_exists { + finding_count += 1; + record_or_log( + store, + SATELLITES_CONSISTENCY_JOB_NAME, + "attached_dangling_blob", + "data_loss", + None, + serde_json::json!({ + "file_id": row.file_id, + "kind": row.kind, + "variant": row.variant, + "blob_hash": row.blob_hash, + "uploaded_by": row.uploaded_by, + "recoverable": false, + "note": "UNRECOVERABLE: these bytes were user-supplied and have no \ + server-side render path, so nothing can regenerate them", + }), + ) + .await; + } + } + + let scanned = rows.len() as u64; + let last = rows.last().unwrap(); + attached_cursor = Some((last.file_id, last.kind.clone(), last.variant.clone())); + if let Err(e) = store + .checkpoint( + encode_cursor( + Phase::Attached, + &last.file_id.to_string(), + &last.kind, + &last.variant, + ), + scanned, + ) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + if scanned < BATCH_SIZE as u64 { + break; + } + } + + tracing::info!( + target: "oxicloud::consistency", + event = "satellites_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "satellites_consistency completed with {} finding(s)", + finding_count + ); + + RunOutcome::completed() + } +} + +/// Cooperative cancel, shared by both phases so neither can forget it. +async fn poll_cancel(store: &dyn JobStore, cursor: Option>) -> Option { + match store.status().await { + Ok(RunStatus::CancelRequested) => Some(RunOutcome::Paused { + cursor: cursor.unwrap_or_default(), + }), + Ok(_) => None, + Err(e) => Some(RunOutcome::Failed { + message: format!("status poll: {e}"), + }), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The phase is what lets one job walk two tables and resume exactly. + /// Without it an attached cursor would be replayed against the derived + /// table, silently re-scanning or skipping — an audit job under-reporting + /// is the worst failure available to it. + #[test] + fn cursor_round_trips_and_keeps_its_phase() { + for phase in [Phase::Derived, Phase::Attached] { + let encoded = encode_cursor(phase, "0a1b", "thumbnail", "preview.webp"); + assert_eq!( + decode_cursor(encoded).unwrap(), + Some(( + phase, + "0a1b".to_string(), + "thumbnail".to_string(), + "preview.webp".to_string() + )) + ); + } + } + + #[test] + fn empty_cursor_starts_from_the_beginning() { + assert_eq!(decode_cursor(Vec::new()).unwrap(), None); + } + + /// Loudly, rather than silently restarting: a corrupt checkpoint that + /// reads as "start over" gives a job that never finishes and never says + /// why. + #[test] + fn malformed_cursor_is_an_error() { + assert!(decode_cursor(b"only-one-field".to_vec()).is_err()); + assert!(decode_cursor(b"bogus\na\nb\nc".to_vec()).is_err()); + } +} diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index 92464c9a..82e4121a 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -225,13 +225,22 @@ jsonpath "$.outcome.count" exists # Step 4c — Trigger `consistency_batch`. Coordinator (plain # JobHandler) — snapshots the registry, filters names # ending `_consistency`, sequentially triggers each. -# `outcome.count` = number of children dispatched (6 as -# of the refcount_cascade fix: drives + folders + -# files + blobs + manifests + backend). `extra.per_check` -# carries a per-child outcome -# map. Batch itself always returns ok — child failures -# live inside per_check. `?deep=true` propagates as -# `extra.deep`. +# `extra.per_check` carries a per-child outcome map. Batch +# itself always returns ok — child failures live inside +# per_check. `?deep=true` propagates as `extra.deep`. +# +# NO assertion on `outcome.count`. The batch auto-discovers +# tenants via `.ends_with("_consistency")`, so a hardcoded +# total breaks every time one is added — it broke on +# `manifests_consistency` and again on +# `satellites_consistency`, each time asserting arithmetic +# rather than behaviour. Per the house rule: `contains` per +# item, never a total. +# +# What matters is that every child SUCCEEDED, which +# `err == 0` states directly and without a magic number, plus +# a named check per tenant below so a job silently dropping +# out of the batch is still caught. # ───────────────────────────────────────────────────────────── POST {{base_url}}/api/admin/jobs/consistency_batch/trigger?deep=true Authorization: Bearer {{admin_token}} @@ -240,9 +249,9 @@ HTTP 200 [Asserts] jsonpath "$.ok" == true jsonpath "$.outcome.outcome" == "ok" -jsonpath "$.outcome.count" == 6 jsonpath "$.outcome.extra.deep" == true -jsonpath "$.outcome.extra.ok" == 6 +# Zero failures, whatever the tenant count happens to be. `ok` is not +# asserted against a number for the same reason `count` is not. jsonpath "$.outcome.extra.err" == 0 # per_check is keyed by child job name. `manifests_consistency` was added # by the refcount_cascade fix — see docs/plan/derived-blobs.md and @@ -255,6 +264,11 @@ jsonpath "$.outcome.extra.per_check.files_consistency.outcome" == "ok" jsonpath "$.outcome.extra.per_check.blobs_consistency.outcome" == "ok" jsonpath "$.outcome.extra.per_check.manifests_consistency.outcome" == "ok" jsonpath "$.outcome.extra.per_check.backend_consistency.outcome" == "ok" +# Finds satellite mappings whose Blob is gone — the one class every +# refcount-based check above reports as healthy, because the row holds a +# valid reference with an exactly correct count while pinning an artifact +# that can never be reclaimed. +jsonpath "$.outcome.extra.per_check.satellites_consistency.outcome" == "ok" # ───────────────────────────────────────────────────────────── diff --git a/tests/api/storage_cleanup_check.sh b/tests/api/storage_cleanup_check.sh index f9b30a8e..74f2fef4 100755 --- a/tests/api/storage_cleanup_check.sh +++ b/tests/api/storage_cleanup_check.sh @@ -527,24 +527,37 @@ fi # Zero findings is the assertion. These jobs are read-only, so a finding # here is a real invariant violation, not a repair opportunity. +# EVERY registered consistency tenant. Keep this list exhaustive: two of +# these (drives, folders) were missing until 2026-08-28 and had never run +# under test at all. CONSISTENCY_JOBS=( files_consistency + folders_consistency + drives_consistency blobs_consistency manifests_consistency backend_consistency - # Catches what the others structurally cannot: a derived mapping whose + # Catches what the others structurally cannot: a satellite mapping whose # source Blob is gone looks healthy to every refcount-based check — valid # reference, correct count, bytes present — while pinning its artifact # forever. That leak reached this suite as three unreclaimable blobs and # took four runs to identify. - derived_consistency + satellites_consistency ) CONSISTENCY_FAILED=0 for job in "${CONSISTENCY_JOBS[@]}"; do + # FAIL on an unknown job rather than warn-and-skip. + # + # The warning was there so a feature-gated build would not break, but the + # cost is worse than the case it protects: renaming a job (or a typo) + # silently removes it from the sweep, and the suite goes on reporting + # green over a check that no longer runs. This list said + # `derived_consistency` for exactly one commit after the rename and would + # have skipped it without comment. TRIGGER=$(curl -sf -X POST -H "$AUTH" "$base_url/api/admin/jobs/$job/trigger") \ - || { log "WARNING: $job not registered in this build — skipped"; continue; } - [[ -z "$TRIGGER" ]] && { log "WARNING: $job returned an empty body — skipped"; continue; } + || fail "$job could not be triggered — renamed, unregistered, or a typo in CONSISTENCY_JOBS" + [[ -z "$TRIGGER" ]] && fail "$job returned an empty body" # The trigger is synchronous for these tenants, but the run row is what # carries the findings, so read it back rather than trusting the