diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 8aa65870..0c78127b 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -853,17 +853,20 @@ what already exists. - Status colour: `ok` = green, `err` = red, `Running` = blue-pulse, `Paused` = amber, `CancelRequested` = amber-flash, `Completed` = neutral grey, `Failed` = red. -- Findings surfacing waits for `jobs.run_findings` — until then the - drawer's "Findings" tab is disabled with a tooltip explaining - drift shows up in the `oxicloud::consistency` log stream today. +- Findings surfacing is live as of Slice 7 (`jobs.run_findings` + + `store.record_finding` + `GET /api/admin/jobs/{name}/runs/{id}/findings`). + Drawer's "Findings" tab renders `kind`, `severity`, `resource_id`, + and per-tenant `detail` JSON. **Slice ordering:** frontend page is a follow-up PR, not blocking any backend slice. Order of appearance: 1. Backend Part 2 slices (engine, admin surface, first tenant) — done. -2. `jobs.run_findings` table + `store.record_finding` API. -3. `consistency_batch` + more tenants. +2. `jobs.run_findings` table + `store.record_finding` API — done (Slice 7). +3. `consistency_batch` + more tenants — done (Slices 5–6: drives + folders + files, plus batch). 4. Frontend `/admin/jobs` page — takes the completed backend surface as-is; no backend changes required by the UI landing. +5. **Deferred, post-UI:** progress estimation (`fraction`, `kind` on + `RunSummary`). See memory `project_job_progress_estimation`. ### Notifications & alerting diff --git a/migrations/20260930000001_jobs_run_findings.sql b/migrations/20260930000001_jobs_run_findings.sql new file mode 100644 index 00000000..051df8b4 --- /dev/null +++ b/migrations/20260930000001_jobs_run_findings.sql @@ -0,0 +1,62 @@ +-- ============================================================================ +-- Slice 7 of Part 2 — persistent finding storage. +-- +-- Findings from consistency jobs (and eventually storage_migration failure +-- rows, reextract failures) currently flow into `tracing::warn!` targeted +-- at `oxicloud::consistency`. That works for live tailing but rotates +-- away — an operator opening the admin UI a day after a run has nothing +-- to drill into. This table makes findings first-class + queryable. +-- +-- `run_findings` is INTENTIONALLY generic. The consistency-check plan and +-- the plan doc's tenant table both list `kind` + `severity` + `resource_id` +-- + `detail` as the union of what every current tenant needs. Adding a +-- tenant with a novel per-finding field means widening `detail` +-- (JSONB, per-tenant shape), not adding a column. +-- +-- Cascade rule: findings live and die with their parent run. Deleting a +-- terminal run row (retention pruning, planned) also drops its findings. +-- ============================================================================ + +CREATE TABLE jobs.run_findings ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + run_id UUID NOT NULL + REFERENCES jobs.recoverable_runs(id) + ON DELETE CASCADE, + -- Machine-readable kind (e.g. 'stale_used_bytes', 'missing_blob'). + -- Stable across releases per the audit-log convention — see + -- feedback memory `enum_over_string_literals_in_logs`. New failure + -- mode = new value, never repurpose an existing one. + kind TEXT NOT NULL, + -- Severity spectrum: + -- 'data_loss' — bytes / rows unreachable or gone. + -- 'inconsistent' — counters / materialised values wrong, + -- content intact. + -- 'anomaly' — surprising state worth surfacing, no known impact. + -- TEXT (not ENUM) so tenants can grow the vocabulary without a schema + -- migration; the app-layer types.rs is where the canonical set lives. + severity TEXT NOT NULL, + -- Nullable — some findings pertain to the run as a whole + -- (e.g. "backend enumeration truncated after 1M keys") rather than + -- one specific resource. + resource_id UUID, + -- Per-tenant per-finding structured detail (cached/actual/delta, + -- blob_hash, expected/stored path, ...). Consumers key off `kind` + -- to know the shape. + detail JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Time-ordered listing per run: `GET /api/admin/jobs/{name}/runs/{id}/findings` +-- pages by (run_id, created_at) — index-only scan. +CREATE INDEX ON jobs.run_findings (run_id, created_at); + +-- Aggregation queries — "how many `missing_blob` findings across all +-- runs of `files_consistency`" or "how many `data_loss`-severity +-- findings today". Both need `(kind)` and `(severity)` predicates; a +-- composite works for either since the leading column is selective. +CREATE INDEX ON jobs.run_findings (kind, severity, created_at); + +COMMENT ON TABLE jobs.run_findings IS + 'Structured per-finding records emitted by recoverable jobs. Replaces ' + 'the transitional `tracing::warn!(event=consistency_finding)` calls ' + 'in the consistency tenants — see docs/plan/job-registry.md Part 2.'; diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index 9afc6426..f48f40de 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/src/infrastructure/scheduler/mod.rs @@ -33,8 +33,8 @@ pub use engine::SchedulerEngine; pub use handler::JobHandler; pub use pg_job_store::{PgJobStore, PgJobStoreProvider}; pub use recoverable::{ - JobStore, JobStoreProvider, OpenedRun, RecoverableAdapter, RecoverableJobHandler, RunOutcome, - RunStatus, RunSummary, run_or_resume, + Finding, JobStore, JobStoreProvider, OpenedRun, RecoverableAdapter, RecoverableJobHandler, + RunOutcome, RunStatus, RunSummary, record_or_log, run_or_resume, }; pub use registry::{JobEntry, JobRegistry, JobSummary, RegisterError}; pub use types::{ErrCause, JobOutcome, JobRunArgs}; diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 65d3d74b..6e31ae26 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -16,7 +16,7 @@ use uuid::Uuid; use crate::common::errors::DomainError; -use super::recoverable::{JobStore, JobStoreProvider, OpenedRun, RunStatus, RunSummary}; +use super::recoverable::{Finding, JobStore, JobStoreProvider, OpenedRun, RunStatus, RunSummary}; // ─── PgJobStore — bound to one run ────────────────────────────────────────── @@ -99,6 +99,68 @@ impl JobStore for PgJobStore { Ok(()) } + async fn record_finding( + &self, + kind: &str, + severity: &str, + resource_id: Option, + detail: serde_json::Value, + ) -> Result<(), DomainError> { + // Two writes in one round-trip via CTE: INSERT the finding + // row + UPDATE stats.finding_count on the parent run. The + // counter stays a coarse UI hint — the source of truth is + // the `jobs.run_findings` table itself. Even if the counter + // drifts (crash mid-statement, hand-edited row), aggregations + // stay accurate. Bumping in the same statement avoids two + // round trips per finding on a hot scan; on a normal + // consistency run the ratio of findings-to-batches is low + // enough that a round trip either way is fine, but this shape + // scales to a bulk-finding tenant without change. + // + // `resource_id` is nullable in the schema; when None here we + // bind `Option::::None` and sqlx encodes it as SQL NULL. + let bumped = sqlx::query( + r#" + WITH inserted AS ( + INSERT INTO jobs.run_findings + (run_id, kind, severity, resource_id, detail) + VALUES ($1, $2, $3, $4, $5) + RETURNING run_id + ) + UPDATE jobs.recoverable_runs + SET stats = jsonb_set( + stats, + '{finding_count}', + ((COALESCE(stats->>'finding_count', '0')::bigint + 1)::text)::jsonb + ) + WHERE id = (SELECT run_id FROM inserted) + "#, + ) + .bind(self.run_id) + .bind(kind) + .bind(severity) + .bind(resource_id) + .bind(&detail) + .execute(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("record_finding", e))?; + + // A rows_affected == 0 on the UPDATE would mean the parent run + // row vanished between INSERT and UPDATE — theoretically + // impossible under our CASCADE FK (deleting the run drops the + // finding), so we don't error, but a debug log covers the + // defensive path. + if bumped.rows_affected() == 0 { + tracing::debug!( + target: "oxicloud::scheduler", + event = "record_finding.counter_bump_noop", + run_id = %self.run_id, + "record_finding: parent run row missing during counter bump" + ); + } + Ok(()) + } + async fn mark_completed(&self) -> Result<(), DomainError> { sqlx::query( r#" @@ -262,6 +324,54 @@ impl JobStoreProvider for PgJobStoreProvider { row.map(row_to_summary).transpose() } + async fn list_findings( + &self, + run_id: Uuid, + limit: u32, + offset: u32, + ) -> Result, DomainError> { + let capped = limit.min(500) as i64; + let off = offset as i64; + let rows: Vec<( + Uuid, + Uuid, + String, + String, + Option, + serde_json::Value, + DateTime, + )> = sqlx::query_as( + r#" + SELECT id, run_id, kind, severity, resource_id, detail, created_at + FROM jobs.run_findings + WHERE run_id = $1 + ORDER BY created_at, id + LIMIT $2 OFFSET $3 + "#, + ) + .bind(run_id) + .bind(capped) + .bind(off) + .fetch_all(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("list_findings", e))?; + + Ok(rows + .into_iter() + .map( + |(id, run_id, kind, severity, resource_id, detail, created_at)| Finding { + id, + run_id, + kind, + severity, + resource_id, + detail, + created_at, + }, + ) + .collect()) + } + async fn request_cancel(&self, job_name: &str) -> Result, DomainError> { // Only Running → CancelRequested flips. `Paused` can be // cancelled by not resuming — no need for a state change. diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index e042f97f..ffb39bd0 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -209,6 +209,37 @@ pub trait JobStore: Send + Sync { /// batches — the run's heartbeat. async fn checkpoint(&self, cursor: Vec, delta_count: u64) -> Result<(), DomainError>; + /// Persist one finding to `jobs.run_findings` and bump + /// `stats.finding_count` on the parent run. Consistency handlers + /// call this in place of the transitional + /// `tracing::warn!(event = "consistency_finding", …)` — see + /// `docs/plan/job-registry.md` Part 2 §Findings. + /// + /// `kind` — stable machine-readable enum-style key (e.g. + /// `"stale_used_bytes"`, `"missing_blob"`). Never rename across + /// releases; new failure modes get new values. + /// + /// `severity` — one of `"data_loss"`, `"inconsistent"`, `"anomaly"`. + /// + /// `resource_id` — the file / folder / drive / blob the finding + /// pertains to. `None` for run-wide findings (e.g. "backend + /// enumeration truncated at 1M keys"). + /// + /// `detail` — per-tenant per-kind JSON blob. Consumers key off + /// `kind` to know the shape (cached/actual/delta for + /// `stale_used_bytes`, blob_hash for `missing_blob`, etc.). + /// + /// Failure surfaces to the caller as `Err`. Handlers should + /// log-and-continue rather than fail the whole run — a lost + /// finding is bad but not worse than aborting the walk. + async fn record_finding( + &self, + kind: &str, + severity: &str, + resource_id: Option, + detail: serde_json::Value, + ) -> Result<(), DomainError>; + // ─── Terminal writes — engine-only. Do not call from handler code. /// Engine-only. Called by [`run_or_resume`] on @@ -278,6 +309,33 @@ pub trait JobStoreProvider: Send + Sync { /// the handler doesn't poll, cancel is a no-op until the run /// completes naturally. async fn request_cancel(&self, job_name: &str) -> Result, DomainError>; + + /// Findings for a specific run, newest-last, paginated. + /// Powers `GET /api/admin/jobs/{name}/runs/{id}/findings`. + /// `limit` caps rows; the API layer clamps it too. `offset` is + /// simple integer paging — findings-per-run is typically small + /// enough that cursor pagination is overkill. + async fn list_findings( + &self, + run_id: Uuid, + limit: u32, + offset: u32, + ) -> Result, DomainError>; +} + +/// Serialisable snapshot of one `jobs.run_findings` row, returned by +/// `GET /api/admin/jobs/{name}/runs/{id}/findings`. Consumers key off +/// `kind` to know the shape of `detail`. +#[derive(Debug, Clone, Serialize)] +pub struct Finding { + pub id: Uuid, + pub run_id: Uuid, + pub kind: String, + pub severity: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub resource_id: Option, + pub detail: serde_json::Value, + pub created_at: DateTime, } /// Serialisable snapshot of one `jobs.recoverable_runs` row, returned @@ -402,6 +460,40 @@ fn log_terminal_write_err(op: &str, run_id: Uuid, res: Result<(), DomainError>) } } +// ─── Recording helper — used by every consistency tenant ─────────────────── + +/// Persist a finding via `store.record_finding` and, if the write +/// fails, drop a `record_finding.failed` line to the tenant's +/// tracing target so operators don't lose the event silently. +/// +/// Exists because every consistency tenant needs the same +/// log-and-continue shape — extracting it here keeps each tenant's +/// per-row branch a single call. +pub async fn record_or_log( + store: &dyn JobStore, + job: &str, + kind: &str, + severity: &str, + resource_id: Option, + detail: serde_json::Value, +) { + if let Err(e) = store + .record_finding(kind, severity, resource_id, detail) + .await + { + tracing::warn!( + target: "oxicloud::consistency", + event = "record_finding.failed", + run_id = %store.run_id(), + job = job, + kind = kind, + resource_id = ?resource_id, + error = %e, + "failed to persist finding; dropped (walk continues)" + ); + } +} + // ─── Adapter — bridge to Part 1's JobHandler ──────────────────────────────── /// Wraps a `RecoverableJobHandler` behind a `JobHandler` face so it @@ -488,6 +580,7 @@ mod tests { cursor: Option>, scanned_count: u64, error_message: Option, + findings: Vec, } #[async_trait] @@ -507,6 +600,25 @@ mod tests { s.scanned_count += delta_count; Ok(()) } + async fn record_finding( + &self, + kind: &str, + severity: &str, + resource_id: Option, + detail: serde_json::Value, + ) -> Result<(), DomainError> { + let mut s = self.state.lock().unwrap(); + s.findings.push(Finding { + id: Uuid::new_v4(), + run_id: self.run_id, + kind: kind.to_string(), + severity: severity.to_string(), + resource_id, + detail, + created_at: Utc::now(), + }); + Ok(()) + } async fn mark_completed(&self) -> Result<(), DomainError> { self.state.lock().unwrap().status = RunStatus::Completed; Ok(()) @@ -555,6 +667,7 @@ mod tests { cursor: None, scanned_count: 0, error_message: None, + findings: Vec::new(), }), }); let id = store.run_id; @@ -610,6 +723,7 @@ mod tests { cursor: None, scanned_count: 0, error_message: None, + findings: Vec::new(), }), }); stores.push(store.clone()); @@ -683,6 +797,26 @@ mod tests { })) } + async fn list_findings( + &self, + run_id: Uuid, + limit: u32, + offset: u32, + ) -> Result, DomainError> { + let stores = self.stores.lock().unwrap(); + let Some(store) = stores.iter().find(|s| s.run_id == run_id) else { + return Ok(Vec::new()); + }; + let state = store.state.lock().unwrap(); + Ok(state + .findings + .iter() + .skip(offset as usize) + .take(limit as usize) + .cloned() + .collect()) + } + async fn request_cancel(&self, _job_name: &str) -> Result, DomainError> { let stores = self.stores.lock().unwrap(); if let Some(s) = stores.last() { diff --git a/src/infrastructure/services/drives_consistency_service.rs b/src/infrastructure/services/drives_consistency_service.rs index 6aa0d232..29516841 100644 --- a/src/infrastructure/services/drives_consistency_service.rs +++ b/src/infrastructure/services/drives_consistency_service.rs @@ -27,7 +27,7 @@ use uuid::Uuid; use crate::infrastructure::scheduler::{ JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, - RunStatus, + RunStatus, record_or_log, }; pub const DRIVES_CONSISTENCY_JOB_NAME: &str = "drives_consistency"; @@ -170,31 +170,23 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { for (drive_id, cached, actual) in &rows { if *cached != *actual { drift_count += 1; - // Finding — logged for now, will migrate to - // `store.record_finding(...)` when `jobs.run_findings` - // lands. Kind + severity chosen to match the - // consistency-check plan's convention: - // kind = 'stale_used_bytes' - // severity = 'inconsistent' (counters wrong, - // content intact — the reconciliation sweep - // will fix on its next tick). - tracing::warn!( - target: "oxicloud::consistency", - event = "consistency_finding", - run_id = %store.run_id(), - job = DRIVES_CONSISTENCY_JOB_NAME, - kind = "stale_used_bytes", - severity = "inconsistent", - resource_id = %drive_id, - cached = *cached, - actual = *actual, - delta = *cached - *actual, - "drive {} used_bytes drift: cached={} actual={} (delta={})", - drive_id, - cached, - actual, - cached - actual - ); + // Persisted finding via the shared helper. + // `stale_used_bytes` + severity `inconsistent` + // (counters wrong, content intact — the + // reconciliation sweep will fix on its next tick). + record_or_log( + store, + DRIVES_CONSISTENCY_JOB_NAME, + "stale_used_bytes", + "inconsistent", + Some(*drive_id), + serde_json::json!({ + "cached": cached, + "actual": actual, + "delta": cached - actual, + }), + ) + .await; } } @@ -251,8 +243,6 @@ mod integration_tests { use crate::infrastructure::scheduler::{JobStoreProvider, OpenedRun, RunStatus}; use sqlx::Row; use sqlx::postgres::PgPoolOptions; - use std::collections::HashMap; - use std::sync::Mutex; async fn test_pool() -> Arc { let url = crate::integration_test_support::test_db_url(); @@ -415,77 +405,6 @@ mod integration_tests { .ok(); } - // ─── Scoped tracing capture — Layer over Registry ───────────────────── - // - // Hand-rolling `Subscriber` from scratch is fragile (callsite - // registration, level filtering, missing default impls). Layer over - // `tracing_subscriber::Registry` is the blessed pattern — Registry - // handles span storage + callsite management, our Layer just captures - // events on the target we care about. Installed per-test via - // `tracing::subscriber::set_default` (returns a drop-guard). - - use tracing_subscriber::{Layer, Registry, layer::SubscriberExt}; - - #[derive(Default, Debug)] - struct CapturedFields { - strings: HashMap, - signed: HashMap, - } - - impl tracing::field::Visit for CapturedFields { - fn record_str(&mut self, field: &tracing::field::Field, value: &str) { - self.strings - .insert(field.name().to_string(), value.to_string()); - } - fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { - self.signed.insert(field.name().to_string(), value); - } - fn record_u64(&mut self, field: &tracing::field::Field, value: u64) { - self.signed.insert(field.name().to_string(), value as i64); - } - fn record_bool(&mut self, _: &tracing::field::Field, _: bool) {} - fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { - self.strings - .insert(field.name().to_string(), format!("{value:?}")); - } - } - - struct CaptureLayer { - target: &'static str, - events: Arc>>, - } - - impl Layer for CaptureLayer { - fn on_event( - &self, - event: &tracing::Event<'_>, - _ctx: tracing_subscriber::layer::Context<'_, S>, - ) { - if event.metadata().target() != self.target { - return; - } - let mut fields = CapturedFields::default(); - event.record(&mut fields); - self.events.lock().unwrap().push(fields); - } - } - - fn install_capture( - target: &'static str, - ) -> ( - Arc>>, - tracing::subscriber::DefaultGuard, - ) { - let events = Arc::new(Mutex::new(Vec::new())); - let layer = CaptureLayer { - target, - events: events.clone(), - }; - let subscriber = Registry::default().with(layer); - let guard = tracing::subscriber::set_default(subscriber); - (events, guard) - } - // ─── Parallel-test serialization ─────────────────────────────────────── // // Cargo runs `#[test]` fns in parallel; the three tests in this @@ -518,12 +437,12 @@ mod integration_tests { // Cached = 999, actual = 200 → delta = 799 (positive = cached over-reports). let drive_id = seed_drift(pool.as_ref(), 999, 200).await; - // Install scoped capture BEFORE dispatch. - let (events, guard) = install_capture("oxicloud::consistency"); - // Run end-to-end through the recoverable engine: PgJobStoreProvider // creates a run row, run_or_resume dispatches DrivesConsistencyCheck, - // handler walks the drive, marks Completed. + // handler walks the drive, marks Completed. Findings land in + // `jobs.run_findings` via `store.record_finding()` — asserted + // via `provider.list_findings(run_id, ...)` below (post-Slice 7, + // no more tracing-event capture). let provider: Arc = Arc::new( crate::infrastructure::scheduler::PgJobStoreProvider::new(pool.clone()), ); @@ -536,56 +455,9 @@ mod integration_tests { ) .await; - drop(guard); - // Framework assertions. assert!(outcome.is_ok(), "run must complete: {outcome:?}"); - // Drift-detection assertion — find the finding event for our drive. - let events = events.lock().unwrap(); - let finding = events - .iter() - .find(|e| { - e.strings - .get("event") - .map(|v| v == "consistency_finding") - .unwrap_or(false) - && e.strings - .get("resource_id") - .map(|v| v == &drive_id.to_string()) - .unwrap_or(false) - }) - .unwrap_or_else(|| { - panic!( - "expected a consistency_finding for drive {drive_id}, got events: {events:?}" - ); - }); - assert_eq!( - finding.strings.get("kind").map(String::as_str), - Some("stale_used_bytes"), - "wrong kind on finding: {finding:?}" - ); - assert_eq!( - finding.strings.get("severity").map(String::as_str), - Some("inconsistent"), - "wrong severity on finding: {finding:?}" - ); - assert_eq!( - finding.signed.get("cached").copied(), - Some(999), - "cached mismatch: {finding:?}" - ); - assert_eq!( - finding.signed.get("actual").copied(), - Some(200), - "actual mismatch: {finding:?}" - ); - assert_eq!( - finding.signed.get("delta").copied(), - Some(799), - "delta mismatch: {finding:?}" - ); - // Read-only invariant — drive's used_bytes is UNCHANGED by the check. let post_cached: i64 = sqlx::query("SELECT used_bytes FROM storage.drives WHERE id = $1") .bind(drive_id) @@ -615,6 +487,51 @@ mod integration_tests { "scanned_count must include at least our drive, got {scanned}" ); + // Drift-detection assertion — the finding for our seeded drive + // is now a persisted row. Query via the same interface the + // admin endpoint uses so the test also pins the read path. + let findings = provider + .list_findings(run_id, 500, 0) + .await + .expect("list_findings"); + let ours = findings + .iter() + .find(|f| f.resource_id == Some(drive_id)) + .unwrap_or_else(|| { + panic!("expected a persisted finding for drive {drive_id}, got: {findings:?}") + }); + assert_eq!(ours.kind, "stale_used_bytes", "wrong kind: {ours:?}"); + assert_eq!(ours.severity, "inconsistent", "wrong severity: {ours:?}"); + assert_eq!( + ours.detail.get("cached").and_then(|v| v.as_i64()), + Some(999), + "cached mismatch in detail: {ours:?}" + ); + assert_eq!( + ours.detail.get("actual").and_then(|v| v.as_i64()), + Some(200), + "actual mismatch in detail: {ours:?}" + ); + assert_eq!( + ours.detail.get("delta").and_then(|v| v.as_i64()), + Some(799), + "delta mismatch in detail: {ours:?}" + ); + + // Counter mirror — record_finding also bumps stats.finding_count. + let stored_finding_count: i64 = sqlx::query( + "SELECT COALESCE((stats->>'finding_count')::bigint, 0) FROM jobs.recoverable_runs WHERE id = $1", + ) + .bind(run_id) + .fetch_one(pool.as_ref()) + .await + .expect("query stats.finding_count") + .get(0); + assert!( + stored_finding_count >= 1, + "finding_count must be bumped, got {stored_finding_count}" + ); + // Cleanup — even on assertion failure the test panics before this, // leaving the test DB slightly dirty. That's fine per session; the // next spawn-db.sh reset clears everything. @@ -630,8 +547,6 @@ mod integration_tests { // cached == actual → no drift. let drive_id = seed_drift(pool.as_ref(), 500, 500).await; - let (events, guard) = install_capture("oxicloud::consistency"); - let provider: Arc = Arc::new( crate::infrastructure::scheduler::PgJobStoreProvider::new(pool.clone()), ); @@ -639,46 +554,39 @@ mod integration_tests { Arc::new(DrivesConsistencyCheck::new(pool.clone())); let outcome = crate::infrastructure::scheduler::run_or_resume( handler, - provider, + provider.clone(), &JobRunArgs::default(), ) .await; - drop(guard); assert!(outcome.is_ok()); - // For THIS drive, no finding event. Other drives in the test DB - // may still surface findings (unrelated fixture data); we only - // assert the invariant scoped to our drive_id. - let events = events.lock().unwrap(); - let our_findings = events - .iter() - .filter(|e| { - e.strings - .get("event") - .map(|v| v == "consistency_finding") - .unwrap_or(false) - && e.strings - .get("resource_id") - .map(|v| v == &drive_id.to_string()) - .unwrap_or(false) - }) - .count(); - assert_eq!( - our_findings, 0, - "no drift on this drive, expected 0 findings, got {our_findings}" - ); - - // Cleanup. + // Locate the run row we just wrote. let latest_run: Option<(Uuid,)> = sqlx::query_as( "SELECT id FROM jobs.recoverable_runs WHERE job_name='drives_consistency' ORDER BY started_at DESC LIMIT 1", ) .fetch_optional(pool.as_ref()) .await .expect("query recoverable_runs"); - if let Some((run_id,)) = latest_run { - cleanup_run(pool.as_ref(), run_id).await; - } + let (run_id,) = latest_run.expect("run row must exist"); + + // For THIS drive, no persisted finding. Other drives in the test + // DB may still surface findings (unrelated fixture data); we only + // assert the invariant scoped to our drive_id. + let findings = provider + .list_findings(run_id, 500, 0) + .await + .expect("list_findings"); + let ours = findings + .iter() + .filter(|f| f.resource_id == Some(drive_id)) + .count(); + assert_eq!( + ours, 0, + "no drift on this drive, expected 0 findings, got {ours}: {findings:?}" + ); + + cleanup_run(pool.as_ref(), run_id).await; cleanup_test_drive(pool.as_ref(), drive_id).await; } diff --git a/src/infrastructure/services/files_consistency_service.rs b/src/infrastructure/services/files_consistency_service.rs new file mode 100644 index 00000000..cc587b56 --- /dev/null +++ b/src/infrastructure/services/files_consistency_service.rs @@ -0,0 +1,306 @@ +//! Third tenant of Part 2 (recoverable-run engine). +//! +//! Iterates `storage.files` and reports each row whose parent-folder +//! state, blob reference, or denormalised size has drifted from +//! what the join with `storage.folders` + `storage.blobs` says is +//! true. **Read-only** — the fix path is other jobs (trash cascade +//! repair, dedup GC, blob resurrection). +//! +//! Post-D7 files schema notable columns: +//! +//! * `folder_id` — nullable; `NULL` = file at drive root. Cascade FK +//! to `storage.folders`. +//! * `blob_hash` — `NOT NULL`; MUST reference a row in +//! `storage.blobs.hash`. +//! * `size` — denormalised copy of the blob's byte length; the +//! original source of truth is `storage.blobs.size` (upload path +//! sets both; a mismatch is drift). +//! * NO `path` column and NO `user_id` column (dropped in D7). The +//! memory note's "path matches parent chain" check from the +//! earlier taxonomy does NOT apply here — files carry no +//! materialised path. +//! +//! ### v1 checks (three per-row branches) +//! +//! * `parent_folder_trashed` — a live file under a soft-deleted +//! parent folder. FK cascade + trash cascade should make this +//! impossible; occurrence means the cascade missed the row. +//! Files at drive root (`folder_id IS NULL`) are exempt — there is +//! no parent to check. +//! * `missing_blob` — file's `blob_hash` has no row in +//! `storage.blobs`. **Severity `data_loss`**: the file record +//! points at bytes the blob table doesn't know about, so any read +//! attempt fails. Historically this happens when the dedup GC +//! reaped a blob whose ref-count decrement raced with a fresh +//! file INSERT — the two-phase mark/sweep is meant to prevent +//! this, but the check surfaces regressions immediately. +//! * `blob_size_mismatch` — `files.size != blobs.size`. Cheap +//! because the same LEFT JOIN already loads `blobs.size`. Would +//! indicate the denormalised copy was ever set by a code path that +//! didn't read the blob's real length — a bug we want to see fast. +//! +//! ### Room to grow (same self-join, one more `if`) +//! +//! * `drive_id_parent_mismatch` — `files.drive_id` differs from +//! `parent.drive_id`. The join already loads it; adding this once +//! drive-membership rules stabilise post-D7 costs one branch. +//! * `mime_type_reconciliation` — compare `files.mime_type` against +//! the blob's `content_type`. Requires deciding which is +//! authoritative first. + +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 FILES_CONSISTENCY_JOB_NAME: &str = "files_consistency"; + +/// Rows per batch. Files can be very numerous — hundreds of +/// thousands on medium installs, millions on large — but the per-row +/// work is a couple of comparisons. 500 keeps the cancel-poll cadence +/// sub-second while amortising round-trip overhead. +const BATCH_SIZE: i64 = 500; + +pub struct FilesConsistencyCheck { + pool: Arc, +} + +impl FilesConsistencyCheck { + pub fn new(pool: Arc) -> Self { + Self { pool } + } + + /// Chainable self-registration — mirrors the other consistency + /// tenants. On-demand only (operators trigger from + /// `POST /api/admin/jobs/files_consistency/trigger` or via + /// `consistency_batch`). + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } +} + +#[derive(Debug, sqlx::FromRow)] +struct FileRow { + id: Uuid, + folder_id: Option, + is_trashed: bool, + size: i64, + blob_hash: String, + /// `None` when `folder_id IS NULL` (file at drive root) — the + /// LEFT JOIN yields no parent row. + parent_is_trashed: Option, + /// `None` when the blob row is missing — the LEFT JOIN yields + /// no `blobs` side. This IS the `missing_blob` signal. + blob_size: Option, +} + +#[async_trait] +impl RecoverableJobHandler for FilesConsistencyCheck { + fn name(&self) -> &str { + FILES_CONSISTENCY_JOB_NAME + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor: 16 raw UUID bytes, empty/absent = start from + // beginning. Same convention as the other UUID-cursor + // tenants so the resume path in `PgJobStoreProvider` is + // uniform. + let mut cursor: Option = match resume_cursor { + None => None, + Some(bytes) if bytes.is_empty() => None, + Some(bytes) if bytes.len() == 16 => { + let mut arr = [0u8; 16]; + arr.copy_from_slice(&bytes); + Some(Uuid::from_bytes(arr)) + } + Some(bytes) => { + return RunOutcome::Failed { + message: format!("invalid cursor: expected 16 bytes, got {}", bytes.len()), + }; + } + }; + + let mut finding_count = 0u64; + + loop { + // Cancel poll BETWEEN batches — the cooperative cancel + // contract (see `RecoverableJobHandler` trait doc). + match store.status().await { + Ok(RunStatus::CancelRequested) => { + tracing::info!( + target: "oxicloud::consistency", + event = "files_consistency.cancelled", + run_id = %store.run_id(), + finding_count = finding_count, + "files_consistency cancelled cooperatively, pausing" + ); + return RunOutcome::Paused { + cursor: cursor.map(|u| u.as_bytes().to_vec()).unwrap_or_default(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + // One query, two LEFT JOINs: (parent folder) + (blob + // row). Left-joining the blob is what lets us detect + // `missing_blob` — a matched row has `blob.size` + // populated; a miss surfaces as NULL. + let rows: Vec = match sqlx::query_as( + r#" + SELECT + f.id AS id, + f.folder_id AS folder_id, + f.is_trashed AS is_trashed, + f.size AS size, + f.blob_hash AS blob_hash, + parent.is_trashed AS parent_is_trashed, + b.size AS blob_size + FROM storage.files f + LEFT JOIN storage.folders parent ON parent.id = f.folder_id + LEFT JOIN storage.blobs b ON b.hash = f.blob_hash + WHERE ($1::uuid IS NULL OR f.id > $1) + ORDER BY f.id + LIMIT $2 + "#, + ) + .bind(cursor) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + return RunOutcome::Failed { + message: format!("batch fetch: {e}"), + }; + } + }; + + if rows.is_empty() { + tracing::info!( + target: "oxicloud::consistency", + event = "files_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "files_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + + for row in &rows { + // (1) parent_folder_trashed: live file under a + // soft-deleted folder. Root files (`folder_id IS + // NULL`) are exempt — `parent_is_trashed` is None + // there. + if !row.is_trashed && row.parent_is_trashed == Some(true) { + finding_count += 1; + record_or_log( + store, + FILES_CONSISTENCY_JOB_NAME, + "parent_folder_trashed", + "inconsistent", + Some(row.id), + serde_json::json!({ + "folder_id": row.folder_id, + }), + ) + .await; + } + + // (2) missing_blob: `blob_hash` has no `storage.blobs` + // row. Real data-loss indicator — reading the file + // will fail. + if row.blob_size.is_none() { + finding_count += 1; + record_or_log( + store, + FILES_CONSISTENCY_JOB_NAME, + "missing_blob", + "data_loss", + Some(row.id), + serde_json::json!({ + "blob_hash": row.blob_hash, + }), + ) + .await; + // No point checking size when the blob row is + // gone — skip (3) for this row. + continue; + } + + // (3) blob_size_mismatch: denormalised size drifted + // from the blob's real length. Cheap because we've + // already loaded both. + if let Some(bs) = row.blob_size + && bs != row.size + { + finding_count += 1; + record_or_log( + store, + FILES_CONSISTENCY_JOB_NAME, + "blob_size_mismatch", + "inconsistent", + Some(row.id), + serde_json::json!({ + "blob_hash": row.blob_hash, + "stored": row.size, + "actual": bs, + "delta": row.size - bs, + }), + ) + .await; + } + } + + // Advance cursor + checkpoint. `batch_len` feeds + // `stats.scanned_count`. + let last_id = rows.last().map(|r| r.id).expect("non-empty rows"); + cursor = Some(last_id); + let batch_len = rows.len() as u64; + if let Err(e) = store + .checkpoint(last_id.as_bytes().to_vec(), batch_len) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + + if (rows.len() as i64) < BATCH_SIZE { + tracing::info!( + target: "oxicloud::consistency", + event = "files_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "files_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + } + } +} diff --git a/src/infrastructure/services/folders_consistency_service.rs b/src/infrastructure/services/folders_consistency_service.rs index ec90d816..8494d73d 100644 --- a/src/infrastructure/services/folders_consistency_service.rs +++ b/src/infrastructure/services/folders_consistency_service.rs @@ -63,7 +63,7 @@ use uuid::Uuid; use crate::infrastructure::scheduler::{ JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, - RunStatus, + RunStatus, record_or_log, }; pub const FOLDERS_CONSISTENCY_JOB_NAME: &str = "folders_consistency"; @@ -233,41 +233,36 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { // soft-deleted parent. Cascade missed. if !row.is_trashed && row.parent_is_trashed == Some(true) { finding_count += 1; - tracing::warn!( - target: "oxicloud::consistency", - event = "consistency_finding", - run_id = %store.run_id(), - job = FOLDERS_CONSISTENCY_JOB_NAME, - kind = "parent_trashed_mismatch", - severity = "inconsistent", - resource_id = %row.id, - parent_id = ?row.parent_id, - "folder {} is live but its parent {:?} is trashed", - row.id, - row.parent_id - ); + record_or_log( + store, + FOLDERS_CONSISTENCY_JOB_NAME, + "parent_trashed_mismatch", + "inconsistent", + Some(row.id), + serde_json::json!({ + "parent_id": row.parent_id, + }), + ) + .await; } // (2) path_mismatch: materialised path drifted from // the parent-chain reconstruction. if row.path != row.expected_path { finding_count += 1; - tracing::warn!( - target: "oxicloud::consistency", - event = "consistency_finding", - run_id = %store.run_id(), - job = FOLDERS_CONSISTENCY_JOB_NAME, - kind = "path_mismatch", - severity = "inconsistent", - resource_id = %row.id, - stored = %row.path, - expected = %row.expected_path, - parent_path = ?row.parent_path, - "folder {} path drift: stored={:?} expected={:?}", - row.id, - row.path, - row.expected_path - ); + record_or_log( + store, + FOLDERS_CONSISTENCY_JOB_NAME, + "path_mismatch", + "inconsistent", + Some(row.id), + serde_json::json!({ + "stored": row.path, + "expected": row.expected_path, + "parent_path": row.parent_path, + }), + ) + .await; } // (3) lpath_mismatch: materialised lpath drifted from @@ -276,22 +271,19 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { // silently break different query shapes. if row.lpath_text != row.expected_lpath_text { finding_count += 1; - tracing::warn!( - target: "oxicloud::consistency", - event = "consistency_finding", - run_id = %store.run_id(), - job = FOLDERS_CONSISTENCY_JOB_NAME, - kind = "lpath_mismatch", - severity = "inconsistent", - resource_id = %row.id, - stored = %row.lpath_text, - expected = %row.expected_lpath_text, - parent_lpath = ?row.parent_lpath_text, - "folder {} lpath drift: stored={:?} expected={:?}", - row.id, - row.lpath_text, - row.expected_lpath_text - ); + record_or_log( + store, + FOLDERS_CONSISTENCY_JOB_NAME, + "lpath_mismatch", + "inconsistent", + Some(row.id), + serde_json::json!({ + "stored": row.lpath_text, + "expected": row.expected_lpath_text, + "parent_lpath": row.parent_lpath_text, + }), + ) + .await; } } diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 73156312..bcd3ff11 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -160,6 +160,7 @@ pub fn admin_routes() -> Router> { .route("/jobs/{name}/cancel", post(cancel_job)) .route("/jobs/{name}/runs", get(list_job_runs)) .route("/jobs/{name}/runs/{id}", get(get_job_run)) + .route("/jobs/{name}/runs/{id}/findings", get(list_job_run_findings)) // Drives — admin-wide view (distinct from `/api/drives` which // is filtered to the caller's role grants). .route("/drives", get(list_all_drives)) @@ -2313,3 +2314,78 @@ pub async fn get_job_run( Err(e) => AppError::internal_error(format!("get_run failed: {e}")).into_response(), } } + +/// Query parameters for `GET /api/admin/jobs/{name}/runs/{id}/findings`. +#[derive(serde::Deserialize)] +pub struct ListFindingsQuery { + /// Page size — server clamps to 500 defensively. + #[serde(default = "default_findings_limit")] + pub limit: u32, + #[serde(default)] + pub offset: u32, +} + +fn default_findings_limit() -> u32 { + 100 +} + +/// `GET /api/admin/jobs/{name}/runs/{id}/findings?limit=N&offset=M` — +/// paginated findings emitted by a specific run of a recoverable job. +/// +/// 404 when the run id doesn't exist (anti-enum: caller knew the id +/// somehow; we don't leak whether it was pruned vs never-existed). +/// Read-only, no audit — standard admin-middleware auth is enough. +/// +/// `{name}` is not validated against the run's `job_name` (the id is +/// globally unique) but keeps the URL path consistent with the other +/// per-run endpoints for stable per-job history links. +#[utoipa::path( + get, + path = "/api/admin/jobs/{name}/runs/{id}/findings", + params( + ("name" = String, Path, description = "Registered job name"), + ("id" = String, Path, description = "Run UUID"), + ("limit" = Option, Query, description = "Max rows (default 100, capped at 500)"), + ("offset" = Option, Query, description = "Rows to skip (default 0)"), + ), + responses( + (status = 200, description = "Findings listed (may be empty)"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + (status = 404, description = "Run not found"), + (status = 500, description = "DB error"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn list_job_run_findings( + State(state): State>, + axum::extract::Path((_name, id)): axum::extract::Path<(String, uuid::Uuid)>, + axum::extract::Query(query): axum::extract::Query, +) -> impl IntoResponse { + use crate::infrastructure::scheduler::JobStoreProvider as _; + // Existence check first — otherwise a paged listing of a + // nonexistent run returns 200 [] which is indistinguishable from + // "run exists, no findings" and breaks operator drill-down. + match state.core.job_store_provider.get_run_by_id(id).await { + Ok(Some(_)) => {} + Ok(None) => { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "run not found", "id": id.to_string() })), + ) + .into_response(); + } + Err(e) => return AppError::internal_error(format!("get_run failed: {e}")).into_response(), + } + let limit = query.limit.clamp(1, 500); + match state + .core + .job_store_provider + .list_findings(id, limit, query.offset) + .await + { + Ok(findings) => (StatusCode::OK, Json(findings)).into_response(), + Err(e) => AppError::internal_error(format!("list_findings failed: {e}")).into_response(), + } +} diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index 176a455d..082423cf 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -144,15 +144,54 @@ jsonpath "$..last_outcome.outcome" contains "ok" # drives_consistency (opens a run row, walks the # `storage.folders` cursor, marks Completed). Success # envelope shape identical. +# +# Captures the run_id so Step 4b-findings can pin the +# `GET /runs/{id}/findings` endpoint. # ───────────────────────────────────────────────────────────── POST {{base_url}}/api/admin/jobs/folders_consistency/trigger Authorization: Bearer {{admin_token}} HTTP 200 +[Captures] +folders_run_id: jsonpath "$.outcome.extra.run_id" [Asserts] jsonpath "$.ok" == true jsonpath "$.outcome.outcome" == "ok" jsonpath "$.outcome.count" exists +jsonpath "$.outcome.extra.completed" == true +jsonpath "$.outcome.extra.run_id" exists + + +# ───────────────────────────────────────────────────────────── +# Step 4b-findings — List findings for the run we just kicked. +# The response body is a JSON array (possibly empty on a clean +# test DB — the fresh Hurl DB has no folder-tree drift). Assert: +# * 200 on a real run_id. +# * response is an array (isCollection covers both empty + non- +# empty). Structural shape of individual finding rows is pinned +# by the drives_consistency_service integration test which seeds +# drift and asserts kind/severity/detail — not repeated here. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs/folders_consistency/runs/{{folders_run_id}}/findings +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$" isCollection + + +# ───────────────────────────────────────────────────────────── +# Step 4b-findings-404 — Findings for a run_id that doesn't +# exist. Endpoint returns 404, not 200 [] — otherwise a broken +# link from the admin UI would look like "no findings" instead +# of "run missing". +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs/folders_consistency/runs/00000000-0000-0000-0000-000000000000/findings +Authorization: Bearer {{admin_token}} + +HTTP 404 +[Asserts] +jsonpath "$.error" == "run not found" # ─────────────────────────────────────────────────────────────