From 782a5c99bd8333608939783259ede9e0028f3bce Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Wed, 29 Jul 2026 01:33:47 +0200 Subject: [PATCH] feat(recoverable-job): add folder_consistency --- docs/plan/job-registry.md | 2 +- src/common/di.rs | 15 + .../services/folders_consistency_service.rs | 328 ++++++++++++++++++ src/infrastructure/services/mod.rs | 1 + tests/api/admin_jobs.hurl | 30 +- 5 files changed, 370 insertions(+), 6 deletions(-) create mode 100644 src/infrastructure/services/folders_consistency_service.rs diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 841a5165..570b06b0 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -712,7 +712,7 @@ rationale + the merges/separations that fall out of the rule. | Tenant | Iterates | Cursor | v1 checks | Notes | |---|---|---|---|---| | `drives_consistency` | `storage.drives` | drive UUID | `used_bytes` drift (drive + user envelope) | Shipped Slice 3. | -| `folders_consistency` | `storage.folders` | folder UUID | parent alive, `path` matches parent chain, `ltree` matches parent chain | | +| `folders_consistency` | `storage.folders` | folder UUID | `parent_trashed_mismatch` (live folder under trashed parent), `path_mismatch`, `lpath_mismatch` — both materialised columns compared to parent-chain reconstruction | Shipped Slice 4. Room to grow: `drive_id_parent_mismatch`, `orphan_root` (self-join already loads the fields). | | `files_consistency` | `storage.files` | file UUID | parent folder alive, `path` correct, `blob_hash` present in `storage.blobs` | Missing-side of the old bidirectional blob check. | | `storage_consistency` | Storage backend (fs / S3) | object key / path | Each blob has a `storage.blobs` row (orphan detection) | `?deep=true` adds re-BLAKE3 + mime sniff. Orphan-side of the old bidirectional blob check + former `blob_integrity` + former `thumbnail_consistency`. | | `grants_consistency` (future) | `storage.role_grants` | grant UUID | subject/resource/granted_by exist | | diff --git a/src/common/di.rs b/src/common/di.rs index 85ebb0e8..12535351 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1289,6 +1289,21 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Second recoverable-run tenant. Iterates `storage.folders` + // and reports each row whose materialised path/lpath or + // parent-trashed state has drifted from the parent-chain + // reconstruction — same subject-iteration pattern as drives. + // On-demand only; findings surface via the + // `oxicloud::consistency` tracing target until the + // `jobs.run_findings` table lands. + let _ = Arc::new( + crate::infrastructure::services::folders_consistency_service::FoldersConsistencyCheck::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/services/folders_consistency_service.rs b/src/infrastructure/services/folders_consistency_service.rs new file mode 100644 index 00000000..ec90d816 --- /dev/null +++ b/src/infrastructure/services/folders_consistency_service.rs @@ -0,0 +1,328 @@ +//! Second tenant of Part 2 (recoverable-run engine). +//! +//! Iterates `storage.folders` and reports each row whose maintained- +//! by-trigger materialised state has drifted from what walking its +//! `parent_id` chain would produce. **Read-only** — reports drift as +//! findings but does NOT rewrite anything; a curative sweep (or a +//! targeted repair endpoint) is a separate concern. +//! +//! Why this matters: the ltree cascade trigger `trg_folders_cascade_path` +//! is what keeps `path` and `lpath` in sync with `parent_id`. Any +//! bulk write path that bypasses the trigger (raw COPY, migration +//! backfill with FOR EACH STATEMENT triggers disabled, hand-rolled +//! UPDATE with `SET LOCAL session_replication_role = 'replica'`) can +//! leave the materialised columns wrong. Silent divergence breaks +//! every `WHERE lpath <@ ancestor` query — subtree list, breadcrumb +//! walk, recursive copy/move/delete. Detecting the drift is what +//! surfaces the underlying misuse. +//! +//! ### v1 checks (three per-row branches) +//! +//! * `parent_trashed_mismatch` — a non-trashed folder whose parent +//! IS trashed. FK enforcement + trash cascade should make this +//! impossible; when it does happen the cascade missed a row. +//! * `path_mismatch` — materialised `folders.path` differs from +//! the parent-chain reconstruction. +//! * `lpath_mismatch` — materialised `folders.lpath` differs from +//! the parent-chain reconstruction. +//! +//! Reconstruction convention (mirrors `storage.compute_folder_path` +//! from `20260307000000_initial_schema.sql`): +//! +//! ```text +//! my_label = replace(id::text, '-', '_') +//! root: path = name lpath = my_label +//! non-root: path = parent.path || '/' || name +//! lpath = parent.lpath || my_label +//! ``` +//! +//! Findings are LOGGED to `target: "oxicloud::consistency"` for now, +//! same as `drives_consistency`. Persistence to `jobs.run_findings` +//! lands with the findings-table migration (deferred); at that point +//! this handler swaps its `tracing::warn!` calls for +//! `store.record_finding(...)` — nothing else changes. +//! +//! ### Room to grow (already-cheap branches deferred) +//! +//! * `drive_id_parent_mismatch` — parent + child in different drives. +//! The self-join already loads `parent.drive_id`; one more per-row +//! `if` when the drive-membership rules stabilise post-D7. +//! * `orphan_root` — a non-trashed `parent_id IS NULL` folder no +//! `drives.root_folder_id` points at. The `check_no_orphan_root_folder` +//! trigger from `20260803000000_*` blocks new orphans; a check here +//! would surface pre-trigger legacy rows. +//! * Name-vs-path terminal drift (`path` ending in the folder's `name`). +//! Redundant with `path_mismatch` unless we ever start storing them +//! independently. + +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 FOLDERS_CONSISTENCY_JOB_NAME: &str = "folders_consistency"; + +/// Rows per batch. Folders can be numerous (millions on big +/// installs), each row is light. 500 keeps the cancel-poll cadence +/// sub-second on a warm cache while amortising round-trip overhead. +const BATCH_SIZE: i64 = 500; + +pub struct FoldersConsistencyCheck { + pool: Arc, +} + +impl FoldersConsistencyCheck { + pub fn new(pool: Arc) -> Self { + Self { pool } + } + + /// Chainable self-registration — mirrors `DrivesConsistencyCheck`. + /// On-demand only (no periodic tick); operators fire it from + /// `POST /api/admin/jobs/folders_consistency/trigger`. + 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 FolderRow { + id: Uuid, + parent_id: Option, + is_trashed: bool, + path: String, + lpath_text: String, + parent_is_trashed: Option, + parent_path: Option, + parent_lpath_text: Option, + expected_path: String, + expected_lpath_text: String, +} + +#[async_trait] +impl RecoverableJobHandler for FoldersConsistencyCheck { + fn name(&self) -> &str { + FOLDERS_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 `drives_consistency` so the resume path + // in `PgJobStoreProvider` treats every UUID-cursor tenant the + // same way. + 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 = "folders_consistency.cancelled", + run_id = %store.run_id(), + finding_count = finding_count, + "folders_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 a batch of folders + reconstruct expected + // path/lpath in SQL via LEFT JOIN on parent. The + // reconstruction formula mirrors `compute_folder_path` + // 1:1; if that trigger's convention ever changes both + // must move together. + let rows: Vec = match sqlx::query_as( + r#" + SELECT + f.id AS id, + f.parent_id AS parent_id, + f.is_trashed AS is_trashed, + f.path AS path, + f.lpath::text AS lpath_text, + parent.is_trashed AS parent_is_trashed, + parent.path AS parent_path, + parent.lpath::text AS parent_lpath_text, + CASE + WHEN f.parent_id IS NULL THEN f.name + ELSE parent.path || '/' || f.name + END AS expected_path, + CASE + WHEN f.parent_id IS NULL THEN replace(f.id::text, '-', '_') + ELSE parent.lpath::text || '.' || replace(f.id::text, '-', '_') + END AS expected_lpath_text + FROM storage.folders f + LEFT JOIN storage.folders parent ON parent.id = f.parent_id + 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 = "folders_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "folders_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + + // Per-row branches. Add new ones here — same pattern as + // `drives_consistency`. Emit at most one finding per + // (row, kind); multiple different kinds for the same row + // are fine and independent. + for row in &rows { + // (1) parent_trashed_mismatch: a live folder under a + // 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 + ); + } + + // (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 + ); + } + + // (3) lpath_mismatch: materialised lpath drifted from + // the parent-chain reconstruction. Independent of (2) + // — either can be wrong without the other, and both + // 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 + ); + } + } + + // Advance cursor + checkpoint. Batch length is what we + // report to `stats.scanned_count`; findings are separate + // (they arrive via the tracing subscriber / eventually + // `run_findings`). + 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}"), + }; + } + + // Short batch = drained the folders table. + if (rows.len() as i64) < BATCH_SIZE { + tracing::info!( + target: "oxicloud::consistency", + event = "folders_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "folders_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + } + } +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index ae2a2d40..09d91417 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -13,6 +13,7 @@ pub mod face_indexing_service; pub mod ffmpeg_video_frame_service; pub mod file_content_cache; pub mod file_system_i18n_service; +pub mod folders_consistency_service; pub mod grant_cleanup_service; pub mod image_transcode_service; pub mod jwt_service; diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index b9c23394..13c23cb9 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -90,11 +90,13 @@ jsonpath "$..interval_ms" count == 3 # Every entry carries a `running` bool — same aggregate primitive. # 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 +# (trash_cleanup, storage_reconcile, dedup_gc, grant_cleanup) + 2 +# Part 2 recoverables (drives_consistency, folders_consistency — +# wrapped by RecoverableAdapter so they appear here alongside the +# periodics). Bump when a new tenant registers. +jsonpath "$..running" count == 6 +jsonpath "$[*].name" contains "drives_consistency" +jsonpath "$[*].name" contains "folders_consistency" # ───────────────────────────────────────────────────────────── @@ -131,6 +133,24 @@ HTTP 200 jsonpath "$..last_outcome.outcome" contains "ok" +# ───────────────────────────────────────────────────────────── +# Step 4b — Trigger `folders_consistency`. Second Part 2 +# recoverable tenant. Goes through the same +# RecoverableAdapter → PgJobStoreProvider path as +# drives_consistency (opens a run row, walks the +# `storage.folders` cursor, marks Completed). Success +# envelope shape identical. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/folders_consistency/trigger +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.count" exists + + # ───────────────────────────────────────────────────────────── # Step 5 — Trigger a job that doesn't exist. 404 anti-enum on # `JobRegistry::trigger` returning `None`.