diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 570b06b0..8aa65870 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -713,13 +713,13 @@ rationale + the merges/separations that fall out of the rule. |---|---|---|---|---| | `drives_consistency` | `storage.drives` | drive UUID | `used_bytes` drift (drive + user envelope) | Shipped Slice 3. | | `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. | +| `files_consistency` | `storage.files` | file UUID | `parent_folder_trashed` (live file under trashed folder), `missing_blob` (severity `data_loss` — file's `blob_hash` absent from `storage.blobs`), `blob_size_mismatch` (denormalised `files.size` diverges from `blobs.size`) | Shipped Slice 6. Missing-side of the old bidirectional blob check. `path` sub-check dropped — files carry no materialised path in the post-D7 schema. Room to grow: `drive_id_parent_mismatch`, mime-type reconciliation. | | `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 | | | `storage_migration` | `storage.blobs` (source) → target backend | blob hash | Copy bytes; failures → `stats.failed_blobs` (and eventually `jobs.run_findings`) | Retires `Arc>` in `migration_job.rs`. | | `reextract_audio` | `storage.files` where audio | file UUID | Re-run audio-tag parser, upsert `audio_metadata` | Retires synchronous admin-request execution. | | `reextract_image` | `storage.files` where image/video | file UUID | Re-run EXIF/container date parser, upsert capture date | Same shape as reextract_audio. | -| `consistency_batch` (wrapper) | Iterates registered `*_consistency` jobs | — (JobHandler, not RecoverableJobHandler) | Sequentially triggers each sub-job; `?deep=true` propagates | One-click "run all" without per-job clicks; exclusivity via `job_name` prevents concurrent batches from stepping on each other. | +| `consistency_batch` (wrapper) | Iterates registered `*_consistency` jobs | — (JobHandler, not RecoverableJobHandler) | Sequentially triggers each sub-job; `?deep=true` propagates | Shipped Slice 5. One-click "run all" without per-job clicks; exclusivity via `job_name` prevents concurrent batches from stepping on each other. Batch itself always returns `Ok` — child failures land in `outcome.extra.per_check[].outcome`. | **Not consistency**: `POST /api/admin/dedup/recalculate` is aggregate- stats-only (`unique_blobs`, `total_references`, `bytes_saved`) — one diff --git a/src/common/di.rs b/src/common/di.rs index 12535351..3d413f16 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1304,6 +1304,36 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Third recoverable-run tenant. Iterates `storage.files` + // and reports parent-folder-trashed cascade misses, + // `missing_blob` (data-loss indicator — file references + // absent blob row), and `blob_size_mismatch` (denormalised + // size drift). One SQL round-trip loads folder + blob via + // two LEFT JOINs; per-row branches key off the join + // results. + let _ = Arc::new( + crate::infrastructure::services::files_consistency_service::FilesConsistencyCheck::new( + maintenance_pool.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + + // "Run all consistency checks" coordinator. Plain JobHandler + // (not RecoverableJobHandler) — it dispatches, doesn't scan. + // MUST register AFTER every `*_consistency` tenant so the + // snapshot ordering in `GET /api/admin/jobs` shows children + // then wrapper; snapshot filtering happens at run time so + // late registration is fine. Weak internally + // breaks the Arc cycle. + let _ = Arc::new( + crate::infrastructure::services::consistency_batch_service::ConsistencyBatch::new( + &core.job_registry, + ), + ) + .register_job(&core.job_registry) + .await; + // 2. Repository services (requires PgPool for all metadata) let repos = self.create_repository_services(&core, &pool); diff --git a/src/infrastructure/services/consistency_batch_service.rs b/src/infrastructure/services/consistency_batch_service.rs new file mode 100644 index 00000000..1b66177e --- /dev/null +++ b/src/infrastructure/services/consistency_batch_service.rs @@ -0,0 +1,188 @@ +//! "Run all consistency checks" coordinator. +//! +//! A plain [`JobHandler`] (not [`RecoverableJobHandler`]) — it walks +//! nothing, holds no cursor. Its whole job is to snapshot the +//! registry, filter to names ending `_consistency`, and dispatch each +//! sequentially via `registry.trigger(name, args)`. Sub-jobs receive +//! the SAME `JobRunArgs` the batch was invoked with — so +//! `?deep=true` on the batch propagates to whichever tenants respect +//! it (currently future `storage_consistency`, but the plumbing is in +//! place). +//! +//! ### Why a wrapper instead of a bulk endpoint +//! +//! Operators want one click for "run all". Building a general-purpose +//! `POST /api/admin/jobs/*/trigger` group-endpoint would need its own +//! auth path, its own concurrency envelope, its own outcome shape. +//! A wrapper JobHandler reuses ALL of that infrastructure: +//! +//! - Same admin URL: `POST /api/admin/jobs/consistency_batch/trigger`. +//! - Same audit trail: one line per batch invocation. +//! - Same exclusivity primitive: the Part 1 per-job semaphore keeps +//! two `consistency_batch` runs from stomping each other. Two +//! batches (say, one `?deep=false` + one `?deep=true`) share the +//! same lock — an admin cannot accidentally start a deep pass +//! while a normal one is still walking. +//! - Same JSON outcome envelope — `per_check` lands under +//! `outcome.extra`, which the admin UI can drill into without +//! inventing a new response schema. +//! +//! ### Why the batch always returns `Ok` +//! +//! The batch's job is **dispatch**, not investigation. A child +//! failing means the child failed — not the batch. Failures surface +//! in `extra.per_check[].outcome = "err"`; the operator drills +//! in. Reporting the batch itself as `Err` would confuse the metric +//! "did the batch run" with "did all children succeed", which are +//! genuinely different questions. +//! +//! ### Registration ordering +//! +//! `consistency_batch` MUST register AFTER every tenant it dispatches +//! — but only for a debug-affordance reason: the ordering of the +//! `GET /api/admin/jobs` response mirrors registration order, and +//! having the wrapper sit at the end of the consistency block reads +//! more naturally. Snapshot filtering happens at RUN time, so a +//! reversed order would still work; DI's ordering is aesthetic. +//! +//! ### Arc cycle avoidance +//! +//! [`ConsistencyBatch`] holds a `Weak` — the registry +//! owns an `Arc` for the batch, and the batch needs +//! access back to `trigger`. A strong `Arc` inside the +//! handler would leak the registry forever. Upgrading the weak on +//! each `run()` is cheap (one refcount bump) and gracefully surfaces +//! "registry dropped mid-shutdown" as an error rather than a hang. + +use std::sync::{Arc, Weak}; + +use async_trait::async_trait; +use serde_json::json; + +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRegistry, JobRunArgs}; + +pub const CONSISTENCY_BATCH_JOB_NAME: &str = "consistency_batch"; + +pub struct ConsistencyBatch { + registry: Weak, +} + +impl ConsistencyBatch { + pub fn new(registry: &Arc) -> Self { + Self { + registry: Arc::downgrade(registry), + } + } + + /// Chainable self-registration — mirrors the per-tenant helpers. + /// On-demand only; there is no periodic tick (operators fire it + /// when they want to sweep, or the frontend "run all" button in + /// `/admin/jobs` triggers it once the UI ships). + pub async fn register_job(self: Arc, registry: &JobRegistry) -> Arc { + registry.register(self.clone(), None, None).await; + self + } +} + +#[async_trait] +impl JobHandler for ConsistencyBatch { + fn name(&self) -> &str { + CONSISTENCY_BATCH_JOB_NAME + } + + async fn run(&self, args: &JobRunArgs) -> JobOutcome { + // Upgrade the Weak. Only fails if the registry has been + // dropped — which can only happen during process shutdown, + // in which case the scheduler is winding down anyway. + let registry = match self.registry.upgrade() { + Some(r) => r, + None => { + return JobOutcome::err( + "consistency_batch: registry dropped (shutdown in progress?)", + ); + } + }; + + // Snapshot + filter. `snapshot_all` would give us Arc + // handles too, but we don't need them — `registry.trigger` + // does the lookup by name itself. `snapshot` returns the + // per-job public DTOs, which is exactly the shape we want. + let targets: Vec = registry + .snapshot() + .await + .into_iter() + .filter(|s| { + s.name.ends_with("_consistency") && s.name != CONSISTENCY_BATCH_JOB_NAME + }) + .map(|s| s.name) + .collect(); + + let mut per_check = serde_json::Map::new(); + let mut ok_count = 0u64; + let mut err_count = 0u64; + + // Sequential dispatch. Parallel would give us tail-latency + // wins but also multiplies DB pressure — the maintenance pool + // is shared with the periodic sweeps that keep running while + // the batch runs. Sequential keeps memory + IO envelope + // predictable; the batch is a "run once in a while, take as + // long as it takes" workflow, not a hot path. + for name in &targets { + let child = registry.trigger(name, args).await; + match &child { + Some(JobOutcome::Ok { count, extra }) => { + ok_count += 1; + let mut entry = json!({ + "outcome": "ok", + "count": count, + }); + if !extra.is_null() { + // Preserve per-check `extra` (e.g. + // drives_consistency emits drift counts here + // once `run_findings` lands). Nested under + // its own key so operators reading + // `per_check[name]` see a stable shape. + entry["extra"] = extra.clone(); + } + per_check.insert(name.clone(), entry); + } + Some(JobOutcome::Err { message }) => { + err_count += 1; + per_check.insert( + name.clone(), + json!({ + "outcome": "err", + "message": message, + }), + ); + } + None => { + // Race: the job disappeared between snapshot and + // trigger. In practice this only happens if some + // future code path deregisters a tenant at + // runtime. Report so operators see it in the + // batch outcome and can chase the cause. + err_count += 1; + per_check.insert( + name.clone(), + json!({ + "outcome": "err", + "message": "job no longer registered (race with deregistration)", + }), + ); + } + } + } + + JobOutcome::ok_with( + targets.len() as u64, + json!({ + "per_check": per_check, + "deep": args.deep, + "force": args.force, + "ok": ok_count, + "err": err_count, + }), + ) + } +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 09d91417..ee359a9d 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -3,6 +3,7 @@ pub mod azure_blob_backend; pub mod cached_blob_backend; pub mod chunked_upload_service; pub mod compression_service; +pub mod consistency_batch_service; pub mod db_pool_monitor; pub mod dedup_service; pub mod drives_consistency_service; @@ -13,6 +14,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 files_consistency_service; pub mod folders_consistency_service; pub mod grant_cleanup_service; pub mod image_transcode_service; diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index 13c23cb9..176a455d 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -90,13 +90,17 @@ 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) + 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 +# (trash_cleanup, storage_reconcile, dedup_gc, grant_cleanup) + 3 +# Part 2 recoverables (drives_consistency, folders_consistency, +# files_consistency — wrapped by RecoverableAdapter so they appear +# here alongside the periodics) + 1 coordinator (consistency_batch +# — a plain JobHandler that dispatches every registered +# `*_consistency`). Bump when a new tenant registers. +jsonpath "$..running" count == 8 jsonpath "$[*].name" contains "drives_consistency" jsonpath "$[*].name" contains "folders_consistency" +jsonpath "$[*].name" contains "files_consistency" +jsonpath "$[*].name" contains "consistency_batch" # ───────────────────────────────────────────────────────────── @@ -151,6 +155,49 @@ jsonpath "$.outcome.outcome" == "ok" jsonpath "$.outcome.count" exists +# ───────────────────────────────────────────────────────────── +# Step 4b2 — Trigger `files_consistency`. Third Part 2 recoverable +# tenant. Iterates `storage.files` and self-joins folder +# + blob. Same envelope shape as the earlier consistency +# tenants. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/files_consistency/trigger +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.count" exists + + +# ───────────────────────────────────────────────────────────── +# Step 4c — Trigger `consistency_batch`. Coordinator (plain +# JobHandler) — snapshots the registry, filters names +# ending `_consistency`, sequentially triggers each. +# `outcome.count` = number of children dispatched (3 as +# of Slice 6: drives + folders + files). `extra.per_check` +# carries a per-child outcome map. Batch itself always +# returns ok — child failures live inside per_check. +# `?deep=true` propagates as `extra.deep`. +# ───────────────────────────────────────────────────────────── +POST {{base_url}}/api/admin/jobs/consistency_batch/trigger?deep=true +Authorization: Bearer {{admin_token}} + +HTTP 200 +[Asserts] +jsonpath "$.ok" == true +jsonpath "$.outcome.outcome" == "ok" +jsonpath "$.outcome.count" == 3 +jsonpath "$.outcome.extra.deep" == true +jsonpath "$.outcome.extra.ok" == 3 +jsonpath "$.outcome.extra.err" == 0 +# per_check is keyed by child job name. +jsonpath "$.outcome.extra.per_check.drives_consistency.outcome" == "ok" +jsonpath "$.outcome.extra.per_check.folders_consistency.outcome" == "ok" +jsonpath "$.outcome.extra.per_check.files_consistency.outcome" == "ok" + + # ───────────────────────────────────────────────────────────── # Step 5 — Trigger a job that doesn't exist. 404 anti-enum on # `JobRegistry::trigger` returning `None`.