cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(recoverable-job): add findings

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-07-29 01:50:26 +02:00
parent 41d83b3053
commit e1556e3d36
10 changed files with 864 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plan/job-registry.md
+8 -5
View File
@@ -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
migrations/20260930000001_jobs_run_findings.sql
+62
View File
@@ -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.';
src/infrastructure/scheduler/mod.rs
+2 -2
View File
@@ -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};
src/infrastructure/scheduler/pg_job_store.rs
+111 -1
View File
@@ -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<Uuid>,
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::<Uuid>::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<Vec<Finding>, DomainError> {
let capped = limit.min(500) as i64;
let off = offset as i64;
let rows: Vec<(
Uuid,
Uuid,
String,
String,
Option<Uuid>,
serde_json::Value,
DateTime<Utc>,
)> = 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<Option<Uuid>, DomainError> {
// Only Running → CancelRequested flips. `Paused` can be
// cancelled by not resuming — no need for a state change.
src/infrastructure/scheduler/recoverable.rs
+134
View File
@@ -209,6 +209,37 @@ pub trait JobStore: Send + Sync {
/// batches — the run's heartbeat.
async fn checkpoint(&self, cursor: Vec<u8>, 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<Uuid>,
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<Option<Uuid>, 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<Vec<Finding>, 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<Uuid>,
pub detail: serde_json::Value,
pub created_at: DateTime<Utc>,
}
/// 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<Uuid>,
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<Vec<u8>>,
scanned_count: u64,
error_message: Option<String>,
findings: Vec<Finding>,
}
#[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<Uuid>,
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<Vec<Finding>, 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<Option<Uuid>, DomainError> {
let stores = self.stores.lock().unwrap();
if let Some(s) = stores.last() {
src/infrastructure/services/drives_consistency_service.rs
+88 -180
View File
@@ -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<sqlx::PgPool> {
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<String, String>,
signed: HashMap<String, i64>,
}
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<Mutex<Vec<CapturedFields>>>,
}
impl<S: tracing::Subscriber> Layer<S> 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<Mutex<Vec<CapturedFields>>>,
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<dyn JobStoreProvider> = 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<dyn JobStoreProvider> = 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;
}
src/infrastructure/services/files_consistency_service.rs
+306
View File
@@ -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<PgPool>,
}
impl FilesConsistencyCheck {
pub fn new(pool: Arc<PgPool>) -> 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<Self>,
registry: &JobRegistry,
provider: &Arc<dyn JobStoreProvider>,
) -> Arc<Self> {
registry
.register_recoverable_job(self.clone(), provider.clone(), None)
.await;
self
}
}
#[derive(Debug, sqlx::FromRow)]
struct FileRow {
id: Uuid,
folder_id: Option<Uuid>,
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<bool>,
/// `None` when the blob row is missing — the LEFT JOIN yields
/// no `blobs` side. This IS the `missing_blob` signal.
blob_size: Option<i64>,
}
#[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<Vec<u8>>,
) -> 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<Uuid> = 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<FileRow> = 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;
}
}
}
}
src/infrastructure/services/folders_consistency_service.rs
+38 -46
View File
@@ -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;
}
}
src/interfaces/api/handlers/admin_handler.rs
+76
View File
@@ -160,6 +160,7 @@ pub fn admin_routes() -> Router<Arc<AppState>> {
.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<u32>, Query, description = "Max rows (default 100, capped at 500)"),
("offset" = Option<u32>, 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<Arc<AppState>>,
axum::extract::Path((_name, id)): axum::extract::Path<(String, uuid::Uuid)>,
axum::extract::Query(query): axum::extract::Query<ListFindingsQuery>,
) -> 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(),
}
}
tests/api/admin_jobs.hurl
+39
View File
@@ -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"
# ─────────────────────────────────────────────────────────────