feat(recoverable-job): add consistency_batch (runs all consistency check)
This commit is contained in:
Edouard Vanbelle
@@ -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<RwLock<MigrationState>>` 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[<name>].outcome`. |
|
||||
|
||||
**Not consistency**: `POST /api/admin/dedup/recalculate` is aggregate-
|
||||
stats-only (`unique_blobs`, `total_references`, `bytes_saved`) — one
|
||||
|
||||
@@ -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<JobRegistry> 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);
|
||||
|
||||
|
||||
@@ -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[<name>].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<JobRegistry>` — the registry
|
||||
//! owns an `Arc<dyn JobHandler>` for the batch, and the batch needs
|
||||
//! access back to `trigger`. A strong `Arc<JobRegistry>` 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<JobRegistry>,
|
||||
}
|
||||
|
||||
impl ConsistencyBatch {
|
||||
pub fn new(registry: &Arc<JobRegistry>) -> 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<Self>, registry: &JobRegistry) -> Arc<Self> {
|
||||
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<JobEntry>
|
||||
// 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<String> = 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,
|
||||
}),
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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`.
|
||||
|
||||
Reference in New Issue
Block a user