From 0f12399a484af1c847b675c93e82c3d5cb963824 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Wed, 29 Jul 2026 22:17:15 +0200 Subject: [PATCH] feat(recoverable-job): fix files_consistency to check blob chunk consistency --- docs/plan/job-registry.md | 2 +- .../src/lib/components/AdminJobsPanel.svelte | 128 +++++++++++++- frontend/static/locales/en.json | 5 + src/infrastructure/scheduler/recoverable.rs | 43 ++++- .../services/drives_consistency_service.rs | 13 +- .../services/files_consistency_service.rs | 157 ++++++++++++++++-- .../services/folders_consistency_service.rs | 9 + 7 files changed, 326 insertions(+), 31 deletions(-) diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 291349a4..0a82658e 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -713,7 +713,7 @@ 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_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. | +| `files_consistency` | `storage.files` | file UUID | `parent_folder_trashed` (live file under trashed folder), `missing_blob` (severity `data_loss` — `blob_hash` present in neither `storage.blobs` nor `storage.chunk_manifests`), `chunk_missing` (severity `data_loss` — manifest exists but points at chunks absent from `storage.blobs`; typical dedup GC race), `blob_size_mismatch` (denormalised `files.size` diverges from the authoritative size — manifest first, blob fallback) | Shipped Slice 6, CDC-aware Slice 10. Handles both storage paths: `storage.chunk_manifests` (post-Apr-2026 FastCDC ingest, dominant path) and `storage.blobs` (pre-CDC whole-file blob, legacy fallback). Physical backend-existence checks (chunk bytes actually on disk) belong in `storage_consistency`. 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`. | diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index 9086d07c..b56b8551 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -230,19 +230,39 @@ return t('admin.jobs.every_sec', { n: secs }, 'every {{n}} s'); } + /** + * Number of findings the last completed run surfaced, from + * `last_outcome.extra.finding_count` (populated by `run_or_resume` + * on Completed/Paused). Returns 0 for jobs without a recoverable + * shape, jobs that haven't run yet, or runs pre-dating the field. + */ + function lastFindingCount(job: JobSummary): number { + if (!job.last_outcome || job.last_outcome.outcome !== 'ok') return 0; + const extra = job.last_outcome.extra as { finding_count?: number } | undefined; + return extra?.finding_count ?? 0; + } + function outcomeLabel(job: JobSummary): string { if (!job.last_outcome) return t('admin.jobs.never', 'never'); if (job.last_outcome.outcome === 'ok') { - return t('admin.jobs.outcome_ok', 'ok'); + // `ok` on the wire = dispatch completed. But if findings + // were surfaced, "ok" reads as "all good" to the operator, + // which is misleading — flip the label + colour to warn. + return lastFindingCount(job) > 0 + ? t('admin.jobs.outcome_issues', 'issues') + : t('admin.jobs.outcome_ok', 'ok'); } return t('admin.jobs.outcome_err', 'err'); } function outcomeClass(job: JobSummary): string { if (!job.last_outcome) return 'jobs-panel__pill jobs-panel__pill--neutral'; - return job.last_outcome.outcome === 'ok' - ? 'jobs-panel__pill jobs-panel__pill--ok' - : 'jobs-panel__pill jobs-panel__pill--err'; + if (job.last_outcome.outcome !== 'ok') { + return 'jobs-panel__pill jobs-panel__pill--err'; + } + return lastFindingCount(job) > 0 + ? 'jobs-panel__pill jobs-panel__pill--paused' + : 'jobs-panel__pill jobs-panel__pill--ok'; } function statusClass(status: RunStatus): string { @@ -412,7 +432,23 @@ {cadenceLabel(job)} {timeAgo(job.last_run_at)} - {outcomeLabel(job)} + +
+ {outcomeLabel(job)} + {#if lastFindingCount(job) > 0} + {@const findings = lastFindingCount(job)} + + {t('admin.jobs.n_findings', { n: findings }, '{{n}} findings')} + + {/if} +
+ {#if isRunning(job)} @@ -562,12 +598,38 @@ {run.progress.scanned}/{run.progress.total} + {:else if scanned != null} + + {t( + 'admin.jobs.progress_scanned_only', + { n: scanned }, + '{{n}} scanned' + )} + {:else} - {scanned ?? '—'} + — {/if} - {findingCount ?? 0} + {#if findingCount && findingCount > 0} + + {findingCount} + + {:else} + 0 + {/if} {#if run.error_message} @@ -653,6 +715,14 @@ {#each findings as f (f.id)} + {@const detail = (f.detail ?? {}) as Record< + string, + unknown + >} + {@const label = + (detail.path as string | undefined) ?? + (detail.name as string | undefined) ?? + null} {f.kind} @@ -665,8 +735,23 @@ {f.severity} - - {f.resource_id ?? '—'} + + {#if label} +
+ {label} + {#if f.resource_id} + {f.resource_id} + {/if} +
+ {:else} + {f.resource_id ?? '—'} + {/if} { log_terminal_write_err("mark_completed", run_id, store.mark_completed().await); JobOutcome::ok_with( - 0, + finding_count, serde_json::json!({ "completed": true, "run_id": run_id.to_string(), + "finding_count": finding_count, + "scanned_count": scanned_count, }), ) } @@ -579,11 +592,13 @@ pub async fn run_or_resume( let cursor_hex = hex::encode(&cursor); log_terminal_write_err("mark_paused", run_id, store.mark_paused(Some(cursor)).await); JobOutcome::ok_with( - 0, + finding_count, serde_json::json!({ "paused": true, "run_id": run_id.to_string(), "cursor_hex": cursor_hex, + "finding_count": finding_count, + "scanned_count": scanned_count, }), ) } @@ -594,6 +609,28 @@ pub async fn run_or_resume( } } +/// Read `finding_count` + `scanned_count` from the just-completed +/// run's `stats`. Missing/failed → `(0, 0)` — the outer outcome +/// simply won't badge findings, which is the right fallback. +async fn fetch_outcome_stats(provider: &dyn JobStoreProvider, run_id: Uuid) -> (u64, u64) { + match provider.get_run_by_id(run_id).await { + Ok(Some(summary)) => { + let finding_count = summary + .stats + .get("finding_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + let scanned_count = summary + .stats + .get("scanned_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + (finding_count, scanned_count) + } + _ => (0, 0), + } +} + fn log_terminal_write_err(op: &str, run_id: Uuid, res: Result<(), DomainError>) { if let Err(e) = res { tracing::warn!( diff --git a/src/infrastructure/services/drives_consistency_service.rs b/src/infrastructure/services/drives_consistency_service.rs index 4781f766..1216a8bb 100644 --- a/src/infrastructure/services/drives_consistency_service.rs +++ b/src/infrastructure/services/drives_consistency_service.rs @@ -144,10 +144,11 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { // 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( + let rows: Vec<(Uuid, String, i64, i64)> = match sqlx::query_as( r#" SELECT d.id, + d.name, d.used_bytes, COALESCE(( SELECT SUM(size)::bigint @@ -189,7 +190,7 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { // 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 { + for (drive_id, drive_name, cached, actual) in &rows { if *cached != *actual { drift_count += 1; // Persisted finding via the shared helper. @@ -203,9 +204,10 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { "inconsistent", Some(*drive_id), serde_json::json!({ + "name": drive_name, "cached": cached, "actual": actual, - "delta": cached - actual, + "delta": cached - actual, }), ) .await; @@ -213,7 +215,10 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { } // Advance cursor to the last row's id + checkpoint. - let last_id = rows.last().map(|(id, _, _)| *id).expect("non-empty rows"); + 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 diff --git a/src/infrastructure/services/files_consistency_service.rs b/src/infrastructure/services/files_consistency_service.rs index 541a8a39..94a9bd83 100644 --- a/src/infrastructure/services/files_consistency_service.rs +++ b/src/infrastructure/services/files_consistency_service.rs @@ -95,6 +95,9 @@ impl FilesConsistencyCheck { #[derive(Debug, sqlx::FromRow)] struct FileRow { id: Uuid, + /// File name (basename). Captured into finding `detail` so + /// operators see a human identifier next to the UUID. + name: String, folder_id: Option, is_trashed: bool, size: i64, @@ -102,9 +105,41 @@ struct FileRow { /// `None` when `folder_id IS NULL` (file at drive root) — the /// LEFT JOIN yields no parent row. parent_is_trashed: Option, - /// `None` when the blob row is missing — the LEFT JOIN yields - /// no `blobs` side. This IS the `missing_blob` signal. + /// Parent folder's materialised `path` (post-D7 files carry no + /// path themselves). `None` for root files. + parent_path: Option, + /// Legacy whole-file blob row size (pre-CDC). `None` when the + /// file was ingested via CDC (`chunk_manifests` path) OR when + /// the blob is truly missing — disambiguated by `manifest_size`. blob_size: Option, + /// CDC manifest total size. `Some` when the file was ingested + /// via FastCDC (its bytes live as chunks referenced by + /// `storage.chunk_manifests.chunk_hashes`, not as one + /// `storage.blobs` row). `None` when there is no manifest for + /// this hash. + manifest_size: Option, + /// Total chunks the manifest claims. `None` when the file is + /// pre-CDC (whole-file blob path) or has no manifest. + manifest_chunk_count: Option, + /// Count of chunks referenced by the manifest that have NO + /// matching row in `storage.blobs`. `None` when there's no + /// manifest to check. `Some(n)` with `n > 0` means the manifest + /// points at reaped chunks — a real data-loss condition, more + /// precise than plain `missing_blob` (which only fires when the + /// whole-file registry entry is absent). This is a DB-registry + /// check; physical backend-existence checks belong in the + /// future `storage_consistency` tenant. + chunks_missing: Option, +} + +/// Build the file's display path from its folder's `path` and its +/// own `name`. Root files just show the name. Trashed folder paths +/// still work (ltree keeps them intact under `is_trashed`). +fn display_path(folder_path: Option<&str>, name: &str) -> String { + match folder_path { + Some(p) if !p.is_empty() => format!("{p}/{name}"), + _ => name.to_string(), + } } #[async_trait] @@ -192,19 +227,56 @@ impl RecoverableJobHandler for FilesConsistencyCheck { // row). Left-joining the blob is what lets us detect // `missing_blob` — a matched row has `blob.size` // populated; a miss surfaces as NULL. + // Three LEFT JOINs — the blob-existence check has to + // handle BOTH storage paths OxiCloud uses: + // + // * `storage.chunk_manifests` (CDC / FastCDC) — the + // dominant path for anything ingested after Apr 2026. + // Whole-file hash lives here; actual bytes are chunks + // referenced by `chunk_hashes[]`. + // * `storage.blobs` (legacy pre-CDC whole-file blob) — + // still supported via the read path's fallback for + // pre-CDC uploads. + // + // A file is "missing_blob" ONLY when NEITHER row exists. + // Deep chunk validation (every chunk in `chunk_hashes[]` + // present in `storage.blobs`) is out of scope here — it + // belongs in the future `storage_consistency` tenant that + // walks the backend against the blob registry. + // Correlated subquery `chunks_missing` runs per-row over + // the manifest's chunk_hashes array. `hash` is indexed + // (PRIMARY KEY on storage.blobs), so each `NOT EXISTS` + // probe is O(log n). NULL (not zero) when the file is + // pre-CDC or has no manifest — the LEFT JOIN result on + // `m` is NULL and `unnest(NULL::text[])` yields zero rows. let rows: Vec = match sqlx::query_as( r#" SELECT f.id AS id, + f.name AS name, 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 + parent.path AS parent_path, + b.size AS blob_size, + m.total_size AS manifest_size, + m.chunk_count AS manifest_chunk_count, + CASE WHEN m.chunk_hashes IS NULL THEN NULL + ELSE ( + SELECT COUNT(*)::bigint + FROM unnest(m.chunk_hashes) AS ch(hash) + WHERE NOT EXISTS ( + SELECT 1 FROM storage.blobs bb + WHERE bb.hash = ch.hash + ) + ) + END AS chunks_missing 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 + LEFT JOIN storage.folders parent ON parent.id = f.folder_id + LEFT JOIN storage.blobs b ON b.hash = f.blob_hash + LEFT JOIN storage.chunk_manifests m ON m.file_hash = f.blob_hash WHERE ($1::uuid IS NULL OR f.id > $1) ORDER BY f.id LIMIT $2 @@ -236,6 +308,12 @@ impl RecoverableJobHandler for FilesConsistencyCheck { } for row in &rows { + // Human-readable path captured once per row and folded + // into every finding on this row. `name` is the raw + // basename (useful even when the parent is orphaned + // and `parent_path` is None). + let path = display_path(row.parent_path.as_deref(), &row.name); + // (1) parent_folder_trashed: live file under a // soft-deleted folder. Root files (`folder_id IS // NULL`) are exempt — `parent_is_trashed` is None @@ -249,16 +327,28 @@ impl RecoverableJobHandler for FilesConsistencyCheck { "inconsistent", Some(row.id), serde_json::json!({ + "name": row.name, + "path": path, "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() { + // Content-bearing size for this file, in priority + // order: CDC manifest (dominant path — every file + // uploaded after Apr 2026), then legacy pre-CDC + // whole-file blob. `None` = no registry entry on + // either path → real `missing_blob`. + let content_size = row.manifest_size.or(row.blob_size); + + // (2) missing_blob: NEITHER the CDC manifest nor the + // legacy blob row exists for this hash. Real data-loss + // indicator — the read path checks manifest first and + // falls back to blob; if both are missing, reading + // the file will fail. NOT a false positive for CDC + // files, because the manifest check catches them. + if content_size.is_none() { finding_count += 1; record_or_log( store, @@ -267,19 +357,55 @@ impl RecoverableJobHandler for FilesConsistencyCheck { "data_loss", Some(row.id), serde_json::json!({ + "name": row.name, + "path": path, "blob_hash": row.blob_hash, }), ) .await; - // No point checking size when the blob row is - // gone — skip (3) for this row. + // No point checking size when neither registry + // entry exists — skip (3) for this row. continue; } + // (2b) chunk_missing: the file's CDC manifest exists + // and points at N chunks, but K of them have no row + // in `storage.blobs`. Real data-loss condition — the + // read path will fail reassembly when it tries to + // fetch a reaped chunk. Typically caused by a dedup + // GC race (chunk reaped while a manifest still held + // a reference) or partial pg_dump/restore that + // dropped `storage.blobs` rows. + if let Some(missing) = row.chunks_missing + && missing > 0 + { + finding_count += 1; + record_or_log( + store, + FILES_CONSISTENCY_JOB_NAME, + "chunk_missing", + "data_loss", + Some(row.id), + serde_json::json!({ + "name": row.name, + "path": path, + "blob_hash": row.blob_hash, + "chunks_missing": missing, + "chunks_total": row.manifest_chunk_count, + }), + ) + .await; + // Deliberately DON'T `continue` — a + // chunk_missing finding does not preclude a + // size mismatch, and the two are independent + // signals worth surfacing separately. + } + // (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 + // from the content-registry's authoritative size. + // Prefers manifest.total_size when present (post-CDC + // ingest path); falls back to blob.size (legacy). + if let Some(bs) = content_size && bs != row.size { finding_count += 1; @@ -290,10 +416,13 @@ impl RecoverableJobHandler for FilesConsistencyCheck { "inconsistent", Some(row.id), serde_json::json!({ + "name": row.name, + "path": path, "blob_hash": row.blob_hash, "stored": row.size, "actual": bs, "delta": row.size - bs, + "source": if row.manifest_size.is_some() { "manifest" } else { "blob" }, }), ) .await; diff --git a/src/infrastructure/services/folders_consistency_service.rs b/src/infrastructure/services/folders_consistency_service.rs index c3519197..7bc993b4 100644 --- a/src/infrastructure/services/folders_consistency_service.rs +++ b/src/infrastructure/services/folders_consistency_service.rs @@ -100,6 +100,9 @@ impl FoldersConsistencyCheck { #[derive(Debug, sqlx::FromRow)] struct FolderRow { id: Uuid, + /// Folder basename — surfaced in finding `detail` so operators + /// see a human identifier next to the UUID. + name: String, parent_id: Option, is_trashed: bool, path: String, @@ -200,6 +203,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { r#" SELECT f.id AS id, + f.name AS name, f.parent_id AS parent_id, f.is_trashed AS is_trashed, f.path AS path, @@ -263,6 +267,8 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { "inconsistent", Some(row.id), serde_json::json!({ + "name": row.name, + "path": row.path, "parent_id": row.parent_id, }), ) @@ -280,6 +286,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { "inconsistent", Some(row.id), serde_json::json!({ + "name": row.name, "stored": row.path, "expected": row.expected_path, "parent_path": row.parent_path, @@ -301,6 +308,8 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { "inconsistent", Some(row.id), serde_json::json!({ + "name": row.name, + "path": row.path, "stored": row.lpath_text, "expected": row.expected_lpath_text, "parent_lpath": row.parent_lpath_text,