diff --git a/justfile b/justfile index 1682b237..003f8734 100644 --- a/justfile +++ b/justfile @@ -71,14 +71,14 @@ coverage-integration filter='mount': # of this file would otherwise leak in) cannot point the tests at the # real dev DB. The test pool helpers also refuse non-`oxicloud_test` # URLs as defence in depth. -test-integration: +test-integration filter='': bash tests/common/spawn-db.sh PGHOST=localhost PGPORT=5433 PGUSER=oxicloud_test PGPASSWORD=oxicloud_test \ PGDATABASE=oxicloud_test \ bash tests/common/init-test-schema.sh DATABASE_URL='postgres://oxicloud_test:oxicloud_test@localhost:5433/oxicloud_test' \ RUSTFLAGS='--cfg integration_tests' \ - cargo test --workspace --tests + cargo test --workspace --tests {{filter}} bash tests/common/stop-db.sh test-one name: diff --git a/src/common/di.rs b/src/common/di.rs index 629029c8..85ebb0e8 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1274,6 +1274,21 @@ impl AppServiceFactory { .register(&core.job_registry) .await; + // First recoverable-run tenant (`docs/plan/job-registry.md` + // Part 2). Iterates `storage.drives` and reports each drive + // whose cached `used_bytes` differs from `SUM(files.size)`. + // On-demand only — read-only diagnostic, not periodic. + // Runs on the maintenance pool alongside the other sweeps. + let job_store_provider_dyn: Arc = + core.job_store_provider.clone(); + let _ = Arc::new( + crate::infrastructure::services::drives_consistency_service::DrivesConsistencyCheck::new( + maintenance_pool.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + // 2. Repository services (requires PgPool for all metadata) let repos = self.create_repository_services(&core, &pool); diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 7bebeff1..65d3d74b 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -462,12 +462,23 @@ impl PgJobStoreProvider { // second INSERT; on conflict we retry. let run_id = Uuid::new_v4(); let now = Utc::now(); + // ON CONFLICT here infers the partial unique index by + // matching `(job_name)` + the WHERE predicate that + // matches `one_active_run_per_job`. We cannot use + // `ON CONFLICT ON CONSTRAINT one_active_run_per_job` + // because `CREATE UNIQUE INDEX` produces an index, not + // a named constraint from PG's perspective; that + // syntax is reserved for `ALTER TABLE ... ADD CONSTRAINT + // UNIQUE`. Inference form is equivalent and works with + // partial indexes. let result = sqlx::query( r#" INSERT INTO jobs.recoverable_runs (id, job_name, status, started_at, last_progress_at) VALUES ($1, $2, 'Running', $3, $3) - ON CONFLICT ON CONSTRAINT one_active_run_per_job DO NOTHING + ON CONFLICT (job_name) + WHERE status IN ('Running', 'Paused', 'CancelRequested') + DO NOTHING "#, ) .bind(run_id) diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs index ed56dac6..396686c3 100644 --- a/src/infrastructure/scheduler/types.rs +++ b/src/infrastructure/scheduler/types.rs @@ -25,9 +25,19 @@ use serde::{Deserialize, Serialize}; /// - `dedup_gc` — skip the orphan grace window (grace = 0). /// - `grant_cleanup` — grace = 0. /// - Others (trash_cleanup, storage_reconcile, …) — ignored. +/// +/// Semantics of `deep`, per job: +/// - `consistency_batch` — propagate to sub-jobs; only `storage_consistency` +/// currently respects it. Wraps the "run all consistency checks +/// including the slow ones" case behind the same job_name lock as +/// the normal batch (Ed's Option B, 2026-07-29). +/// - `storage_consistency` (future) — enables per-blob re-BLAKE3 (bitrot +/// detection) + mime sniff alongside the fast orphan check. +/// - Others — ignored. #[derive(Debug, Clone, Default)] pub struct JobRunArgs { pub force: bool, + pub deep: bool, } /// Uniform outcome the supervisor logs and stores for every job dispatch. diff --git a/src/infrastructure/services/drives_consistency_service.rs b/src/infrastructure/services/drives_consistency_service.rs new file mode 100644 index 00000000..6aa0d232 --- /dev/null +++ b/src/infrastructure/services/drives_consistency_service.rs @@ -0,0 +1,727 @@ +//! First tenant of Part 2 (recoverable-run engine). +//! +//! Iterates `storage.drives` and reports each drive whose cached +//! `used_bytes` differs from `SUM(files.size) WHERE NOT is_trashed` +//! for that drive. **Read-only** — reports drift as findings but does +//! NOT fix it. The existing `storage_reconcile` job (Part 1) is what +//! corrects the counter; this check surfaces WHEN drift happens so +//! operators can trace it back to root cause (missed delta call, +//! delta failed silently, race, etc.). +//! +//! One check today — `used_bytes` drift — but structured so more +//! checks can slot in as per-row branches (quota-vs-usage inversion, +//! `kind` vs `default_for_user` invariants, ...). See memory note +//! `project_consistency_jobs_landscape`. +//! +//! Findings are LOGGED to `target: "oxicloud::consistency"` for now. +//! Persistence to `jobs.run_findings` lands with the findings-table +//! migration (deferred; see the plan doc). Once landed, this handler +//! swaps its `tracing::warn!` finding calls for +//! `store.record_finding(...)` — nothing else changes. + +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, +}; + +pub const DRIVES_CONSISTENCY_JOB_NAME: &str = "drives_consistency"; + +/// Rows per batch. Drives are few (dozens per install), so this only +/// matters for the cancel-poll cadence — smaller batch = more frequent +/// status polls but more DB round-trips. 100 is comfortably fast for +/// any realistic drive count. +const BATCH_SIZE: i64 = 100; + +pub struct DrivesConsistencyCheck { + pool: Arc, +} + +impl DrivesConsistencyCheck { + pub fn new(pool: Arc) -> Self { + Self { pool } + } + + /// Register self with the periodic-job scheduler as a recoverable + /// job, on-demand only (no periodic tick). Follows the same + /// chainable pattern as Part 1 tenants' `register_job` — DI stays + /// one line. + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } +} + +#[async_trait] +impl RecoverableJobHandler for DrivesConsistencyCheck { + fn name(&self) -> &str { + DRIVES_CONSISTENCY_JOB_NAME + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Decode cursor. Convention for this job: 16 raw UUID bytes, + // or empty/absent = start from the beginning. + 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 drift_count = 0u64; + + loop { + // Cancel poll BETWEEN batches — the cooperative cancel + // contract (`RecoverableJobHandler` trait doc). + match store.status().await { + Ok(RunStatus::CancelRequested) => { + tracing::info!( + target: "oxicloud::consistency", + event = "drives_consistency.cancelled", + run_id = %store.run_id(), + drift_count = drift_count, + "drives_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}"), + }; + } + } + + // Fetch next batch of drives + their actual SUM in one + // query. LEFT JOIN via correlated subquery gets us both + // sides in one round-trip; the storage_reconcile sweep + // uses the same shape. + let rows: Vec<(Uuid, i64, i64)> = match sqlx::query_as( + r#" + SELECT + d.id, + d.used_bytes, + COALESCE(( + SELECT SUM(size)::bigint + FROM storage.files + WHERE drive_id = d.id + AND NOT is_trashed + ), 0) AS actual_bytes + FROM storage.drives d + WHERE ($1::uuid IS NULL OR d.id > $1) + ORDER BY d.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 = "drives_consistency.completed", + run_id = %store.run_id(), + drift_count = drift_count, + "drives_consistency completed with {} drift finding(s)", + drift_count + ); + return RunOutcome::Completed; + } + + // Per-row check: cached vs actual. This is the ONE check + // in v1 — more per-row branches (quota inversion, kind vs + // default_for_user, …) slot in here. + 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 + ); + } + } + + // Advance cursor to the last row's id + checkpoint. + let last_id = rows.last().map(|(id, _, _)| *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}"), + }; + } + + // Short batch = drained the drives table. + if (rows.len() as i64) < BATCH_SIZE { + tracing::info!( + target: "oxicloud::consistency", + event = "drives_consistency.completed", + run_id = %store.run_id(), + drift_count = drift_count, + "drives_consistency completed with {} drift finding(s)", + drift_count + ); + return RunOutcome::Completed; + } + } + } +} + +// ─── Integration tests — real PG round-trip ───────────────────────────────── +// +// Gated on `--cfg integration_tests` (see `just test-integration`). +// Requires a running test PG on 5433 with `oxicloud_test` DB, schema +// applied via `tests/common/init-test-schema.sh`. Runs: +// just test-integration -- drives_consistency_service +// +// Tests exercise the full recoverable-run engine against real PG: +// - PgJobStoreProvider::open_or_start creates a run row. +// - Handler walks a seeded drive, checkpoints, marks Completed. +// - `stats.scanned_count` bumped, drive row untouched (read-only). +// - Drift is DETECTED — surfaced as a `consistency_finding` event +// on the `oxicloud::consistency` tracing target. Captured via a +// scoped subscriber. + +#[cfg(integration_tests)] +#[allow(dead_code, unused_imports)] // items are exercised by #[tokio::test] +// fns; cargo check --lib doesn't see the +// test entry-point call graph. +mod integration_tests { + use super::*; + 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(); + let pool = PgPoolOptions::new() + .max_connections(4) + .connect(&url) + .await + .expect("connect to test DB — run tests/common/spawn-db.sh first"); + Arc::new(pool) + } + + /// Seed a personal drive with `used_bytes = cached` and one file + /// of size `actual` (post-D7 schema — no `user_id` on files / + /// folders, drives created via the circular-FK dance). + /// + /// `default_for_user = NULL` on the drive so we don't collide + /// with the seeded user's real default (partial unique index). + /// Returns the drive id. + async fn seed_drift(pool: &sqlx::PgPool, cached: i64, actual: i64) -> Uuid { + let owner_id: Uuid = sqlx::query("SELECT id FROM auth.users LIMIT 1") + .fetch_one(pool) + .await + .expect("test DB must have at least one user") + .get(0); + + // Steps 1-3 MUST run inside one transaction because + // `trg_no_orphan_root_folder` is DEFERRABLE INITIALLY DEFERRED + // (fires at COMMIT). Autocommit-per-statement would trip the + // trigger on the folder INSERT before the drive UPDATE gets a + // chance to close the FK. Mirrors `DrivePgRepository:: + // create_personal_drive_atomic`. + let mut tx = pool.begin().await.expect("begin drive-create tx"); + + // 1. Drive with kind=personal, no default_for_user (avoids + // partial-unique conflict with owner's real default), + // used_bytes=0 for now — we set the fake value LAST. + let drive_id: Uuid = sqlx::query_scalar( + r#" + INSERT INTO storage.drives + (kind, default_for_user, quota_bytes, used_bytes) + VALUES ('personal', NULL, NULL, 0) + RETURNING id + "#, + ) + .fetch_one(&mut *tx) + .await + .expect("insert test drive"); + + // 2. Root folder for the drive (parent_id = NULL = drive root). + // Post-D7: only `name`, `parent_id`, `drive_id`, `created_by`, + // `updated_by` on the INSERT — `user_id`/`path`/`ltree` are + // dropped or derived. + let root_folder_id: Uuid = sqlx::query_scalar( + r#" + INSERT INTO storage.folders + (name, parent_id, drive_id, created_by, updated_by) + VALUES ('drift-test-root', NULL, $1, $2, $2) + RETURNING id + "#, + ) + .bind(drive_id) + .bind(owner_id) + .fetch_one(&mut *tx) + .await + .expect("insert root folder for test drive"); + + // 3. Close the circular FK: drive.root_folder_id points at + // the folder we just created. + sqlx::query("UPDATE storage.drives SET root_folder_id = $1 WHERE id = $2") + .bind(root_folder_id) + .bind(drive_id) + .execute(&mut *tx) + .await + .expect("wire drive.root_folder_id"); + + tx.commit() + .await + .expect("commit drive-create tx (deferred trigger fires here)"); + + // 4. Optionally insert a file summing to `actual`. Also insert + // a matching `storage.blobs` row so `trg_files_decrement_blob_ref` + // stays happy on cleanup. Fake hash is 64-char hex derived + // from a UUID — plausible shape, unique per fixture invocation. + if actual > 0 { + let fake_hash = format!( + "{:032x}{:032x}", + Uuid::new_v4().as_u128(), + Uuid::new_v4().as_u128() + ); + sqlx::query( + r#" + INSERT INTO storage.blobs (hash, size, ref_count, content_type) + VALUES ($1, $2, 1, 'application/octet-stream') + ON CONFLICT (hash) DO NOTHING + "#, + ) + .bind(&fake_hash) + .bind(actual) + .execute(pool) + .await + .expect("insert fixture blob"); + + sqlx::query( + r#" + INSERT INTO storage.files + (name, folder_id, drive_id, blob_hash, size, + mime_type, is_trashed, created_by, updated_by) + VALUES ($1, $2, $3, $4, $5, + 'application/octet-stream', false, $6, $6) + "#, + ) + .bind(format!("drift-fixture-{}.bin", Uuid::new_v4())) + .bind(root_folder_id) + .bind(drive_id) + .bind(&fake_hash) + .bind(actual) + .bind(owner_id) + .execute(pool) + .await + .expect("insert fixture file"); + } + + // 5. Set the artificially-wrong cached used_bytes. LAST, so + // no INSERT-side trigger overwrites our fake (there is no + // such trigger today, but ordering is cheap insurance). + sqlx::query("UPDATE storage.drives SET used_bytes = $1 WHERE id = $2") + .bind(cached) + .bind(drive_id) + .execute(pool) + .await + .expect("set fake used_bytes"); + + drive_id + } + + async fn cleanup_test_drive(pool: &sqlx::PgPool, drive_id: Uuid) { + // Cascading FK on files.drive_id, folders.drive_id kicks in. + sqlx::query("DELETE FROM storage.files WHERE drive_id = $1") + .bind(drive_id) + .execute(pool) + .await + .ok(); + sqlx::query("DELETE FROM storage.folders WHERE drive_id = $1") + .bind(drive_id) + .execute(pool) + .await + .ok(); + sqlx::query("DELETE FROM storage.drives WHERE id = $1") + .bind(drive_id) + .execute(pool) + .await + .ok(); + } + + async fn cleanup_run(pool: &sqlx::PgPool, run_id: Uuid) { + sqlx::query("DELETE FROM jobs.recoverable_runs WHERE id = $1") + .bind(run_id) + .execute(pool) + .await + .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 + // module all target `job_name = 'drives_consistency'` in + // `jobs.recoverable_runs`. Without serialization, + // `second_trigger_is_already_active` seeds a Running row that + // makes `detects_used_bytes_drift`'s `open_or_start` short-circuit + // with `AlreadyActive` — its handler never dispatches, no + // `consistency_finding` fires, and the drift assertion sees empty + // events. Holding `TEST_LOCK` for each test's full body prevents + // that interleaving; `wipe_our_runs` on entry defends against + // stale rows left by a crashed / cancelled prior run. + + static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + async fn wipe_our_runs(pool: &sqlx::PgPool) { + sqlx::query("DELETE FROM jobs.recoverable_runs WHERE job_name = 'drives_consistency'") + .execute(pool) + .await + .ok(); + } + + // ─── The tests ───────────────────────────────────────────────────────── + + #[tokio::test] + async fn drives_consistency_detects_used_bytes_drift() { + let _lock = TEST_LOCK.lock().await; + let pool = test_pool().await; + wipe_our_runs(pool.as_ref()).await; + // 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. + let provider: Arc = Arc::new( + crate::infrastructure::scheduler::PgJobStoreProvider::new(pool.clone()), + ); + let handler: Arc = + Arc::new(DrivesConsistencyCheck::new(pool.clone())); + let outcome = crate::infrastructure::scheduler::run_or_resume( + handler, + provider.clone(), + &JobRunArgs::default(), + ) + .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) + .fetch_one(pool.as_ref()) + .await + .expect("drive still exists") + .get(0); + assert_eq!(post_cached, 999, "drives_consistency must be read-only"); + + // Find the run row that was created and verify its state. + let latest_run: Option<(Uuid, String, i64)> = sqlx::query_as( + r#" + SELECT id, status, COALESCE((stats->>'scanned_count')::bigint, 0) + 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"); + let (run_id, status, scanned) = latest_run.expect("run row must exist after run_or_resume"); + assert_eq!(status, "Completed", "run must be Completed"); + assert!( + scanned >= 1, + "scanned_count must include at least our drive, got {scanned}" + ); + + // 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. + cleanup_run(pool.as_ref(), run_id).await; + cleanup_test_drive(pool.as_ref(), drive_id).await; + } + + #[tokio::test] + async fn drives_consistency_no_drift_emits_no_finding() { + let _lock = TEST_LOCK.lock().await; + let pool = test_pool().await; + wipe_our_runs(pool.as_ref()).await; + // 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()), + ); + let handler: Arc = + Arc::new(DrivesConsistencyCheck::new(pool.clone())); + let outcome = crate::infrastructure::scheduler::run_or_resume( + handler, + provider, + &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. + 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; + } + cleanup_test_drive(pool.as_ref(), drive_id).await; + } + + #[tokio::test] + async fn drives_consistency_second_trigger_is_already_active() { + let _lock = TEST_LOCK.lock().await; + let pool = test_pool().await; + wipe_our_runs(pool.as_ref()).await; + + // Directly INSERT a Running row for drives_consistency to + // simulate an in-flight prior dispatch, then observe that + // open_or_start refuses to spawn a parallel run. + let seeded_run_id = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO jobs.recoverable_runs (id, job_name, status, started_at, last_progress_at) + VALUES ($1, 'drives_consistency', 'Running', NOW(), NOW()) + ON CONFLICT (job_name) WHERE status IN ('Running', 'Paused', 'CancelRequested') DO NOTHING + "#, + ) + .bind(seeded_run_id) + .execute(pool.as_ref()) + .await + .expect("insert seed Running row"); + + let provider: Arc = Arc::new( + crate::infrastructure::scheduler::PgJobStoreProvider::new(pool.clone()), + ); + let opened = provider + .open_or_start("drives_consistency") + .await + .expect("open_or_start"); + match opened { + OpenedRun::AlreadyActive { status, .. } => { + assert_eq!(status, RunStatus::Running); + } + _ => panic!("expected AlreadyActive for a job with a Running row"), + } + + // Cleanup. + sqlx::query("DELETE FROM jobs.recoverable_runs WHERE job_name = 'drives_consistency'") + .execute(pool.as_ref()) + .await + .ok(); + } +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 7a28b6a0..ae2a2d40 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -5,6 +5,7 @@ pub mod chunked_upload_service; pub mod compression_service; pub mod db_pool_monitor; pub mod dedup_service; +pub mod drives_consistency_service; pub mod encrypted_blob_backend; pub mod exif_service; pub mod face_geometry; diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 0c69c39a..73156312 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -2095,10 +2095,16 @@ pub async fn list_jobs(State(state): State>) -> impl IntoResponse /// support it (dedup_gc → grace = 0, grant_cleanup → grace = 0). /// Silently ignored by handlers that don't (trash_cleanup, /// storage_reconcile). +/// +/// `deep=true` opts into slow variants — `consistency_batch` fans it +/// out to sub-jobs; `storage_consistency` (when implemented) will +/// re-BLAKE3 each blob for bitrot detection. See `JobRunArgs.deep`. #[derive(serde::Deserialize)] pub struct TriggerJobQuery { #[serde(default)] pub force: bool, + #[serde(default)] + pub deep: bool, } /// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule. @@ -2136,11 +2142,16 @@ pub async fn trigger_job( event = "job.trigger", job = %name, force = query.force, - "👮🏻‍♂️ Admin triggered job {} (force={})", + deep = query.deep, + "👮🏻‍♂️ Admin triggered job {} (force={}, deep={})", name, query.force, + query.deep, ); - let args = JobRunArgs { force: query.force }; + let args = JobRunArgs { + force: query.force, + deep: query.deep, + }; match state.core.job_registry.trigger(&name, &args).await { Some(outcome) => ( StatusCode::OK, diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index 64f6753a..b9c23394 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -89,7 +89,12 @@ jsonpath "$..interval_ms" contains 600000 jsonpath "$..interval_ms" count == 3 # Every entry carries a `running` bool — same aggregate primitive. -jsonpath "$..running" count == 4 +# Count matches the registered-tenant count: 4 Part 1 periodics +# (trash_cleanup, storage_reconcile, dedup_gc, grant_cleanup) + 1 +# Part 2 recoverable (drives_consistency, wrapped by RecoverableAdapter +# so it appears here alongside the periodics). Bump when a new +# tenant registers. +jsonpath "$..running" count == 5 # ───────────────────────────────────────────────────────────── diff --git a/tests/api/recoverable_jobs.hurl b/tests/api/recoverable_jobs.hurl new file mode 100644 index 00000000..b9f38736 --- /dev/null +++ b/tests/api/recoverable_jobs.hurl @@ -0,0 +1,188 @@ +# ============================================================= +# OxiCloud — Recoverable-run admin surface +# ============================================================= +# Pins the Part 2 (recoverable-run engine) admin endpoints: +# * POST /api/admin/jobs/{name}/trigger (RecoverableJobHandler +# path via RecoverableAdapter) +# * POST /api/admin/jobs/{name}/cancel +# * GET /api/admin/jobs/{name}/runs +# * GET /api/admin/jobs/{name}/runs/{id} +# +# Uses `drives_consistency` — the first recoverable tenant, on-demand +# only. Verifies: +# 1. Registered job appears in `GET /api/admin/jobs` with no +# interval (on-demand only). +# 2. Triggering creates a fresh row in `jobs.recoverable_runs`, +# handler completes, run terminates as Completed. +# 3. History endpoint returns the just-completed run. +# 4. Single-run detail endpoint returns the same row. +# 5. Cancel-on-idle is a no-op with `cancelled: false` (nothing +# running to cancel). +# 6. Unknown run id → 404 on the single-run endpoint. +# 7. Non-admin caller → 403 from the admin middleware on every +# recoverable endpoint (no bespoke role check in the handlers). +# +# Drift-finding assertions land alongside the `jobs.run_findings` +# migration — the current build LOGS findings to +# `oxicloud::consistency` without persisting them. Log-tail +# assertions from Hurl are fragile so we defer them. +# ============================================================= + + +# ───────────────────────────────────────────────────────────── +# Setup — admin login + rjobs_bob (non-admin) provisioning +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/auth/login +Content-Type: application/json +{ "username": "{{username}}", "password": "{{password}}" } + +HTTP 200 +[Captures] +admin_token: jsonpath "$.access_token" + + +# Anti-enum registration. +POST {{base_url}}/api/auth/register +Content-Type: application/json +{ + "username": "rjobs_bob", + "email": "rjobs_bob@example.com", + "password": "RjobsBobPassword1!" +} + +HTTP 200 + + +POST {{base_url}}/api/auth/login +Content-Type: application/json +{ "username": "rjobs_bob", "password": "RjobsBobPassword1!" } + +HTTP 200 +[Captures] +bob_token: jsonpath "$.access_token" + + +# ───────────────────────────────────────────────────────────── +# Step 1 — `drives_consistency` is registered on-demand only. +# Appears in the listing without an `interval_ms`. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$[*].name" contains "drives_consistency" +# On-demand → no interval_ms (`skip_serializing_if = Option::is_none`). +jsonpath "$[?(@.name=='drives_consistency')].interval_ms" not exists + + +# ───────────────────────────────────────────────────────────── +# Step 2 — Trigger the check. Handler dispatches through +# `RecoverableAdapter` → `run_or_resume`, which INSERTs +# a fresh `jobs.recoverable_runs` row, runs the scan, +# marks it Completed. Response envelope: +# { ok, outcome: { outcome: "ok", +# extra: { completed: true, run_id: "..." } } } +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/drives_consistency/trigger +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.extra.completed" == true +[Captures] +run_id: jsonpath "$.outcome.extra.run_id" + + +# ───────────────────────────────────────────────────────────── +# Step 3 — Run history returns at least the just-triggered +# run, newest first. Response is a JSON array of +# RunSummary; the top entry must be the run_id we +# captured above with status='Completed'. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs/drives_consistency/runs +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$" isCollection +jsonpath "$[0].id" == "{{run_id}}" +jsonpath "$[0].job_name" == "drives_consistency" +jsonpath "$[0].status" == "Completed" + + +# ───────────────────────────────────────────────────────────── +# Step 4 — Single-run detail. Returns the same row shape as +# the listing but for one id. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs/drives_consistency/runs/{{run_id}} +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.id" == "{{run_id}}" +jsonpath "$.job_name" == "drives_consistency" +jsonpath "$.status" == "Completed" +# scanned_count is bumped by the handler's checkpoint call — at +# least 0 (empty drives table) but ordinarily > 0 for any real +# fixture data. Present-ness of the field is what we pin. +jsonpath "$.stats.scanned_count" isNumber + + +# ───────────────────────────────────────────────────────────── +# Step 5 — Cancel-on-idle is a no-op. No Running row means no +# Running→CancelRequested flip. Response is 200 with +# `cancelled: false` (NOT a 404 — the job name is +# registered, cancel just found nothing to cancel). +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/drives_consistency/cancel +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.cancelled" == false +jsonpath "$.reason" == "no running run for this job" + + +# ───────────────────────────────────────────────────────────── +# Step 6 — Unknown run id → 404. UUID shape is valid; the id +# just isn't in `jobs.recoverable_runs`. +# ───────────────────────────────────────────────────────────── +GET {{base_url}}/api/admin/jobs/drives_consistency/runs/00000000-0000-0000-0000-000000000000 +Authorization: Bearer {{admin_token}} + +HTTP 404 +[Asserts] +jsonpath "$.error" == "run not found" + + +# ───────────────────────────────────────────────────────────── +# Step 7 — Non-admin caller is denied on every endpoint by +# the `/api/admin/*` middleware layer. Handlers have +# no bespoke role check — reaching them at all means +# the caller is admin. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/drives_consistency/trigger +Authorization: Bearer {{bob_token}} + +HTTP 403 + + +POST {{base_url}}/api/admin/jobs/drives_consistency/cancel +Authorization: Bearer {{bob_token}} + +HTTP 403 + + +GET {{base_url}}/api/admin/jobs/drives_consistency/runs +Authorization: Bearer {{bob_token}} + +HTTP 403 + + +GET {{base_url}}/api/admin/jobs/drives_consistency/runs/{{run_id}} +Authorization: Bearer {{bob_token}} + +HTTP 403 diff --git a/tests/api/run.sh b/tests/api/run.sh index b17419d2..1ec3157d 100755 --- a/tests/api/run.sh +++ b/tests/api/run.sh @@ -167,6 +167,7 @@ hurl --variables-file "$API_DIR/test.env" --file-root "$REPO_ROOT/tests" --test "$API_DIR/dedup_blob_cleanup.hurl" \ "$API_DIR/dedup_admin_gate.hurl" \ "$API_DIR/admin_jobs.hurl" \ + "$API_DIR/recoverable_jobs.hurl" \ "$API_DIR/default_caldav_carddav.hurl" \ "$API_DIR/dav_error_mapping.hurl" \ "$API_DIR/carddav_vcard_properties.hurl" \