diff --git a/frontend/src/lib/api/endpoints/adminJobs.ts b/frontend/src/lib/api/endpoints/adminJobs.ts index 71bf7d65..5722462e 100644 --- a/frontend/src/lib/api/endpoints/adminJobs.ts +++ b/frontend/src/lib/api/endpoints/adminJobs.ts @@ -60,11 +60,21 @@ export function listJobs(): Promise { } /** - * `POST /api/admin/jobs/{name}/trigger?force=X&deep=X` — dispatch a job - * on-demand. `force` bypasses per-tenant idempotency checks (e.g. - * `trash_cleanup` skipping when nothing is due). `deep` opts into slow - * variants (currently only `storage_consistency`, propagated by - * `consistency_batch` to every child). + * `POST /api/admin/jobs/{name}/trigger?force=X&deep=X&repair=X` — + * dispatch a job on-demand. + * + * - `force` bypasses per-tenant idempotency checks (e.g. `trash_cleanup` + * skipping when nothing is due). + * - `deep` opts into slow variants (currently only `storage_consistency`, + * propagated by `consistency_batch` to every child). + * - `repair` opts into corrective action on the refcount consistency + * tenants (`blobs_consistency`, `manifests_consistency`, and + * `consistency_batch` which fans out to both). Content-safe: only the + * stored counter changes to match the auditor's computed value. Race- + * safe: the corrective UPDATE recomputes the auditor formula in the + * same statement, so a concurrent write can't leave a stale value. + * Default `false` preserves discovery-only behaviour — surface a + * confirm-first flow when calling with `repair: true`. * * Throws on 4xx / 5xx with the backend's error message when present. * A 404 means the job name isn't registered — surface that specifically @@ -72,11 +82,12 @@ export function listJobs(): Promise { */ export async function triggerJob( name: string, - opts: { force?: boolean; deep?: boolean; storage?: string } = {} + opts: { force?: boolean; deep?: boolean; storage?: string; repair?: boolean } = {} ): Promise { const params = new URLSearchParams(); if (opts.force) params.set('force', 'true'); if (opts.deep) params.set('deep', 'true'); + if (opts.repair) params.set('repair', 'true'); // `storage` scopes tenants that respect JobRunArgs.storage — // currently blobs_consistency / backend_consistency (probes the // named entry instead of the live backend). See diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index 7a70be95..d247ba58 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -21,6 +21,7 @@ import { SvelteSet } from 'svelte/reactivity'; import Icon from '$lib/icons/Icon.svelte'; import Modal from '$lib/components/Modal.svelte'; + import { confirmDialog } from '$lib/stores/dialogs.svelte'; import { t } from '$lib/i18n/index.svelte'; import { errorMessage } from '$lib/utils/errors'; import { ui } from '$lib/stores/ui.svelte'; @@ -66,6 +67,43 @@ else busyKeys.delete(key); } + // Per-job "Run" split-button menu state. Keyed by job name so + // two rows can open their menus independently (though the + // outside-click handler below closes all on any click outside + // any menu — matching the /files upload dropdown pattern). Only + // rows with `supportsDeep` OR `supportsRepair` render a chevron; + // the plain-Run rows (drives/folders/files/backend/… consistency, + // trash_cleanup, dedup_gc, …) show a bare "Run" button with no + // menu, keeping the common case one-click. + let runMenuOpen = $state>({}); + function toggleRunMenu(name: string) { + runMenuOpen = { ...runMenuOpen, [name]: !runMenuOpen[name] }; + } + function closeAllRunMenus() { + runMenuOpen = {}; + } + // Global outside-click + Escape dismiss. Only registered while at + // least one menu is open — a background admin tab doesn't hold + // listeners. + $effect(() => { + const anyOpen = Object.values(runMenuOpen).some((v) => v); + if (!anyOpen) return; + const onDown = (e: MouseEvent) => { + if (!(e.target as HTMLElement).closest('.jobs-panel__split')) { + closeAllRunMenus(); + } + }; + const onKey = (e: KeyboardEvent) => { + if (e.key === 'Escape') closeAllRunMenus(); + }; + window.addEventListener('pointerdown', onDown); + window.addEventListener('keydown', onKey); + return () => { + window.removeEventListener('pointerdown', onDown); + window.removeEventListener('keydown', onKey); + }; + }); + // Purge-modal state. Null = closed; otherwise carries the // draft retention days the operator's picking. Kept separate // from the top-bar action state so mouse-away doesn't lose @@ -234,8 +272,10 @@ // ─── Actions ─────────────────────────────────────────────────────── - async function onTrigger(name: string, opts: { deep?: boolean } = {}) { - const key = `trigger:${name}${opts.deep ? ':deep' : ''}`; + async function onTrigger(name: string, opts: { deep?: boolean; repair?: boolean } = {}) { + // Key suffix has to keep every dispatched variant distinct so the + // button-disabled state of one doesn't lock out another mid-flight. + const key = `trigger:${name}${opts.deep ? ':deep' : ''}${opts.repair ? ':repair' : ''}`; markBusy(key, true); try { // Fire the trigger + a follow-up loadJobs after a short delay @@ -265,10 +305,49 @@ if (!res.outcome) { // dispatched (detached) — no outcome to render } else if (res.outcome.outcome === 'ok') { - ui.notify( - t('admin.jobs.triggered_ok', { name }, '{{name}} triggered successfully'), - 'success' - ); + // Repair runs surface a rollup so the operator sees + // whether corrective UPDATEs actually fired. `extra` + // carries `repaired_count` on the two refcount tenants + // directly, and nested under `per_check[*].extra` when + // dispatched via `consistency_batch`. Sum across the + // per_check dict if present, else read the top-level. + let repairedTotal = 0; + let sawRepair = false; + const extra = (res.outcome.extra ?? {}) as { + repair_requested?: boolean; + repaired_count?: number; + per_check?: Record< + string, + { extra?: { repair_requested?: boolean; repaired_count?: number } } + >; + }; + if (extra.repair_requested) { + sawRepair = true; + repairedTotal += extra.repaired_count ?? 0; + } + if (extra.per_check) { + for (const child of Object.values(extra.per_check)) { + if (child?.extra?.repair_requested) { + sawRepair = true; + repairedTotal += child.extra.repaired_count ?? 0; + } + } + } + if (sawRepair) { + ui.notify( + t( + 'admin.jobs.triggered_ok_repair', + { name, n: repairedTotal }, + '{{name}}: {{n}} counter(s) repaired' + ), + 'success' + ); + } else { + ui.notify( + t('admin.jobs.triggered_ok', { name }, '{{name}} triggered successfully'), + 'success' + ); + } } else { ui.notify( t( @@ -557,6 +636,31 @@ return name === 'consistency_batch' || name === 'blobs_consistency'; } + // Jobs whose handler consults `args.repair` and applies a + // corrective UPDATE against the finding it just emitted. Only the + // two ref_count tenants today; `consistency_batch` also accepts + // the flag (fans out to both) and is surfaced separately as the + // top-bar "Repair ref_counts" button. Keep this list narrow — + // adding a job here without a matching backend handler produces a + // silently no-op button that confuses operators. + function supportsRepair(name: string): boolean { + return name === 'blobs_consistency' || name === 'manifests_consistency'; + } + + async function onTriggerWithRepairConfirm(name: string) { + const ok = await confirmDialog({ + title: t('admin.jobs.run_repair_confirm_title', 'Repair drifted ref_counts?'), + message: t( + 'admin.jobs.run_repair_confirm_body_scoped', + { name }, + 'Runs {{name}} and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only counters change; blob content and file rows are untouched.' + ), + confirmText: t('admin.jobs.run_repair_confirm', 'Repair'), + danger: true + }); + if (ok) await onTrigger(name, { repair: true }); + } + function isRunning(job: JobSummary): boolean { return job.running; } @@ -614,6 +718,39 @@ {t('admin.jobs.run_deep', 'Run deep')} + + {/if} + {@const hasRunVariants = supportsDeep(job.name) || supportsRepair(job.name)} + - {/if} + {#if hasRunVariants} + + {#if runMenuOpen[job.name]} + + {/if} + {/if} + {/if} {#if isRunning(job) && canExpand} {#if isRecoverable(job)} @@ -1304,6 +1499,90 @@ color: var(--color-danger-text-alt); } + /* Warn variant — used for actions that mutate data but are content- + safe / reversible-in-outcome (e.g. Repair ref_counts). Signals + "read the tooltip and the confirm before clicking" without the + danger red reserved for destructive delete-style buttons. */ + .jobs-panel__btn--warn { + border-color: var(--color-warning-border); + color: var(--color-warning-text); + } + + /* Split-button — inline flex holding a primary "Run" (fires default + action) and a chevron (opens the variants menu). `position: + relative` anchors the menu below the toggle. Only rendered on + rows whose job supports at least one variant; plain-Run rows + sidestep this whole structure. */ + .jobs-panel__split { + display: inline-flex; + position: relative; + } + + /* Attached-button trick: main loses its right border-radius, toggle + loses its left. Toggle also loses its left border so the two + don't render a double-thick divider. */ + .jobs-panel__split-main { + border-top-right-radius: 0; + border-bottom-right-radius: 0; + } + + .jobs-panel__split-toggle { + border-top-left-radius: 0; + border-bottom-left-radius: 0; + border-left: none; + padding-left: 0.35rem; + padding-right: 0.35rem; + } + + /* The variants menu — dropdown below the toggle, right-aligned so + it doesn't overflow the Actions column edge into the next row's + badge cell. Shadow + surface bg mirror the /files upload + dropdown (`upload-dropdown-menu`); using local CSS here rather + than the ported class so the jobs-panel keeps its scoped styling. */ + .jobs-panel__run-menu { + position: absolute; + top: calc(100% + 2px); + right: 0; + z-index: 30; + min-width: 10rem; + background: var(--color-bg-surface); + border: 1px solid var(--color-border); + border-radius: var(--radius-md, 6px); + box-shadow: var(--shadow-md); + padding: 0.25rem 0; + } + + .jobs-panel__run-menu-item { + display: flex; + align-items: center; + gap: 0.5rem; + width: 100%; + padding: 0.4rem 0.75rem; + background: transparent; + border: none; + text-align: left; + font: inherit; + color: var(--color-text); + cursor: pointer; + white-space: nowrap; + } + + .jobs-panel__run-menu-item:hover:not(:disabled) { + background: var(--color-bg-hover); + } + + .jobs-panel__run-menu-item:disabled { + opacity: 0.5; + cursor: not-allowed; + } + + /* Warn colour on the menu item mirrors the button variant so the + Repair option carries the same "attention-worthy but not + destructive" visual weight as its top-bar counterpart. */ + .jobs-panel__run-menu-item--warn { + color: var(--color-warning-text); + } + .jobs-panel__pill { display: inline-block; padding: 0.1rem 0.5rem; diff --git a/frontend/static/locales/en.json b/frontend/static/locales/en.json index fc20b4d3..09a10822 100644 --- a/frontend/static/locales/en.json +++ b/frontend/static/locales/en.json @@ -1275,6 +1275,14 @@ "run_all_consistency": "Run all consistency checks", "run_deep": "Run deep", "run_deep_hint": "Also runs slow variants (blob re-hash, bitrot detection).", + "run_repair": "Repair ref_counts", + "run_repair_hint": "Corrects any drifted ref_counts (blobs + manifests) found by the audit. Content-safe — only counters change, not data.", + "run_repair_confirm_title": "Repair drifted ref_counts?", + "run_repair_confirm_body": "Runs the audit against every blob + manifest and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only the counter changes; blob content and file rows are untouched. Safe to run at any time; a discovery-only run happens first so you can see the drift before this repair overwrites it.", + "run_repair_confirm_body_scoped": "Runs {{name}} and applies a corrective UPDATE to any counter that disagrees with its live reference count. Content-safe: only counters change; blob content and file rows are untouched.", + "run_variants_menu": "Run variants menu", + "run_repair_confirm": "Repair", + "triggered_ok_repair": "{{name}}: {{n}} counter(s) repaired", "col_name": "Name", "col_cadence": "Cadence", "col_last_run": "Last run", diff --git a/frontend/static/locales/fr.json b/frontend/static/locales/fr.json index 7b51a642..15b38fbe 100644 --- a/frontend/static/locales/fr.json +++ b/frontend/static/locales/fr.json @@ -1197,6 +1197,14 @@ "run_all_consistency": "Exécuter tous les contrôles de cohérence", "run_deep": "Analyse approfondie", "run_deep_hint": "Exécute également des variantes lentes (re-hachage de blob, détection bitrot).", + "run_repair": "Réparer les compteurs", + "run_repair_hint": "Corrige les compteurs de références (blobs + manifestes) désynchronisés détectés par l'audit. Sûr pour les données — seuls les compteurs changent, pas le contenu.", + "run_repair_confirm_title": "Réparer les compteurs de références ?", + "run_repair_confirm_body": "Lance l'audit sur chaque blob et manifeste, puis applique un UPDATE correctif à chaque compteur qui ne correspond pas au nombre réel de références. Sans risque pour les données : seuls les compteurs changent ; le contenu des blobs et les enregistrements de fichiers ne sont pas touchés. Vous pouvez exécuter cela à tout moment ; un passage en lecture seule s'exécute d'abord pour visualiser l'écart avant que la réparation ne l'écrase.", + "run_repair_confirm_body_scoped": "Lance {{name}} et applique un UPDATE correctif à chaque compteur qui ne correspond pas au nombre réel de références. Sans risque pour les données : seuls les compteurs changent ; le contenu et les enregistrements de fichiers ne sont pas touchés.", + "run_variants_menu": "Menu des variantes d'exécution", + "run_repair_confirm": "Réparer", + "triggered_ok_repair": "{{name}} : {{n}} compteur(s) réparé(s)", "col_name": "Nom", "col_cadence": "Cadence", "col_last_run": "Dernière exécution", diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs index 0cb22375..57f48f8c 100644 --- a/src/infrastructure/scheduler/types.rs +++ b/src/infrastructure/scheduler/types.rs @@ -45,11 +45,25 @@ use serde::{Deserialize, Serialize}; /// of the entry to probe instead of the currently-active backend. /// `None` falls through to the live backend (today's behaviour). /// - Others — ignored. +/// +/// Semantics of `repair` (added 2026-10-17 for the refcount fix): +/// - `blobs_consistency` / `manifests_consistency` — when `true`, +/// after each `refcount_mismatch` / `manifest_refcount_mismatch` +/// finding is recorded, apply the corrective UPDATE that sets the +/// stored counter to the auditor's computed `actual_ref_count`. +/// Content-safe: the row itself is fine, only the counter is +/// wrong. Race-safe: each UPDATE recomputes the auditor formula +/// in the same statement, so a concurrent write can't leave a +/// stale value. Default `false` preserves discovery-only +/// behaviour. Also propagates through `consistency_batch` to +/// both tenants — one `?repair=true` call fixes both counters. +/// - Others — ignored. #[derive(Debug, Clone, Default)] pub struct JobRunArgs { pub force: bool, pub deep: bool, pub storage: Option, + pub repair: bool, } /// Uniform outcome the supervisor logs and stores for every job dispatch. diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index fed22da9..a2eae950 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -347,6 +347,10 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { // stats.finding_count — actual persistence happens in // `record_finding` on each emission). let mut finding_count = 0u64; + // Only touched when `args.repair == true`. Symmetric with + // `manifests_consistency`; reported in completion log + + // `extra_stats` so operators see "found N, fixed M" in one line. + let mut repaired_count = 0u64; // Deep mode is a per-run flag with two consumers: // 1. This handler — decides whether to re-hash bytes. @@ -399,6 +403,41 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { ); } + // Repair mode: same shape as `deep` above so the admin run- + // detail view can display `params.repair = "true"` alongside + // `params.deep`. Fresh persists what the trigger asked for; + // Resume reads back so a paused repair scan stays a repair + // scan (a mid-scan crash mustn't silently downgrade to + // discovery-only for the remaining rows). + let repair = if is_fresh { + let v = if args.repair { "true" } else { "false" }; + if let Err(e) = store.set_string_param("repair", v).await { + return RunOutcome::Failed { + message: format!("failed to persist repair flag to params: {e}"), + }; + } + args.repair + } else { + match store.get_string_param("repair").await { + Ok(Some(v)) => v == "true", + Ok(None) => false, + Err(e) => { + return RunOutcome::Failed { + message: format!("read `repair` from params: {e}"), + }; + } + } + }; + + if repair { + tracing::info!( + target: "oxicloud::consistency", + event = "blobs_consistency.repair_mode_active", + run_id = %store.run_id(), + "repair mode: refcount_mismatch findings will trigger corrective UPDATE" + ); + } + loop { // Cooperative cancel poll between batches. match store.status().await { @@ -448,11 +487,17 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { event = "blobs_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + repaired_count = repaired_count, + repair_requested = repair, deep = deep, - "blobs_consistency completed with {} finding(s)", - finding_count + "blobs_consistency completed with {} finding(s), {} repaired", + finding_count, + repaired_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "repair_requested": repair, + "repaired_count": repaired_count, + })); } let grace_cutoff = Utc::now() - CREATE_GRACE; @@ -480,6 +525,67 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { }), ) .await; + + // Repair pass — content-safe corrective UPDATE. Sets + // `stored` to the value the auditor's two-term formula + // would compute at UPDATE time (subquery mirrors + // `chunk_page_sql`'s `actual_ref_count`), so a + // concurrent write between our page fetch and this + // UPDATE can't leave a stale value — the subquery + // re-reads inside the same statement. The + // `<> (subquery)` guard makes the UPDATE a no-op if + // the drift has healed, making this idempotent under + // retry. + if repair { + let expected = "( \ + (SELECT COUNT(*) FROM storage.files f \ + WHERE f.blob_hash = b.hash \ + AND NOT EXISTS ( \ + SELECT 1 FROM storage.chunk_manifests m \ + WHERE m.file_hash = f.blob_hash \ + )) \ + + (SELECT COUNT(*) FROM storage.chunk_manifests m \ + WHERE b.hash = ANY(m.chunk_hashes)) \ + )"; + let update_sql = format!( + "UPDATE storage.blobs b \ + SET ref_count = {expected} \ + WHERE b.hash = $1 \ + AND b.ref_count <> {expected}", + ); + match sqlx::query(&update_sql) + .bind(&row.hash) + .execute(self.pool.as_ref()) + .await + { + Ok(res) if res.rows_affected() > 0 => { + repaired_count += 1; + tracing::info!( + target: "audit", + event = "blobs_consistency.repaired", + run_id = %store.run_id(), + hash = %row.hash, + stored_was = row.ref_count, + actual = row.actual_ref_count, + "🩹 blob ref_count repaired" + ); + } + Ok(_) => { + // No row touched — concurrent repair or + // self-healing drift. Silent no-op. + } + Err(e) => { + tracing::warn!( + target: "oxicloud::consistency", + event = "blobs_consistency.repair_failed", + run_id = %store.run_id(), + hash = %row.hash, + error = %e, + "blob ref_count repair UPDATE failed — finding stays" + ); + } + } + } } // Skip physical probes for rows within the write @@ -628,11 +734,17 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { event = "blobs_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + repaired_count = repaired_count, + repair_requested = repair, deep = deep, - "blobs_consistency completed with {} finding(s)", - finding_count + "blobs_consistency completed with {} finding(s), {} repaired", + finding_count, + repaired_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "repair_requested": repair, + "repaired_count": repaired_count, + })); } } } diff --git a/src/infrastructure/services/consistency_batch_service.rs b/src/infrastructure/services/consistency_batch_service.rs index ad174da1..f167615a 100644 --- a/src/infrastructure/services/consistency_batch_service.rs +++ b/src/infrastructure/services/consistency_batch_service.rs @@ -178,6 +178,7 @@ impl JobHandler for ConsistencyBatch { "per_check": per_check, "deep": args.deep, "force": args.force, + "repair": args.repair, "ok": ok_count, "err": err_count, }), diff --git a/src/infrastructure/services/manifests_consistency_service.rs b/src/infrastructure/services/manifests_consistency_service.rs index 7e4ea4a7..3ee5b016 100644 --- a/src/infrastructure/services/manifests_consistency_service.rs +++ b/src/infrastructure/services/manifests_consistency_service.rs @@ -161,9 +161,11 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { async fn run_resumable( &self, store: &dyn JobStore, - _args: &JobRunArgs, + args: &JobRunArgs, resume_cursor: Option>, ) -> RunOutcome { + let is_fresh = resume_cursor.is_none(); + // Cursor: the last `file_hash` as UTF-8. Same convention as // `blobs_consistency`, which also pages a hash-keyed table. let mut cursor: Option = match resume_cursor { @@ -179,7 +181,48 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { }, }; + // Persist the repair flag into `params.repair` so the admin + // run-detail view can display whether the run was a discovery + // scan or an active repair. Fresh takes it from args; Resume + // reads back so a paused repair scan stays a repair scan (a + // mid-scan crash mustn't silently downgrade the remaining + // rows to discovery-only). Same shape as + // `blobs_consistency_service.rs`'s `deep` handling — see the + // reasoning documented there. + let repair = if is_fresh { + let v = if args.repair { "true" } else { "false" }; + if let Err(e) = store.set_string_param("repair", v).await { + return RunOutcome::Failed { + message: format!("failed to persist repair flag to params: {e}"), + }; + } + args.repair + } else { + match store.get_string_param("repair").await { + Ok(Some(v)) => v == "true", + Ok(None) => false, + Err(e) => { + return RunOutcome::Failed { + message: format!("read `repair` from params: {e}"), + }; + } + } + }; + + if repair { + tracing::info!( + target: "oxicloud::consistency", + event = "manifests_consistency.repair_mode_active", + run_id = %store.run_id(), + "repair mode: manifest_refcount_mismatch findings will trigger corrective UPDATE" + ); + } + let mut finding_count = 0u64; + // Only relevant when `repair == true`. Reported inline in + // the completion log + the `extra_stats` payload so operators + // can see "we found N and fixed M" in one line. + let mut repaired_count = 0u64; loop { // Cooperative cancel poll between batches. @@ -227,10 +270,16 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { event = "manifests_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, - "manifests_consistency completed with {} finding(s)", - finding_count + repaired_count = repaired_count, + repair_requested = repair, + "manifests_consistency completed with {} finding(s), {} repaired", + finding_count, + repaired_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "repair_requested": repair, + "repaired_count": repaired_count, + })); } for row in &rows { @@ -258,6 +307,64 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { }), ) .await; + + // Repair pass — content-safe corrective UPDATE. The + // stored counter is set to what the auditor formula + // would compute at UPDATE time (subquery matches + // `manifest_page_sql`'s `actual_ref_count` predicate), + // so a concurrent file insert/delete between our page + // fetch and this UPDATE can't leave a stale value — + // the subquery re-reads inside the same statement. + // The `<> (subquery)` guard makes the UPDATE a no-op + // if the value is already correct, so this is + // idempotent under retry. + if repair { + match sqlx::query( + "UPDATE storage.chunk_manifests m \ + SET ref_count = ( \ + SELECT COUNT(*) FROM storage.files \ + WHERE blob_hash = m.file_hash \ + ) \ + WHERE m.file_hash = $1 \ + AND m.ref_count <> ( \ + SELECT COUNT(*) FROM storage.files \ + WHERE blob_hash = m.file_hash \ + )", + ) + .bind(&row.file_hash) + .execute(self.pool.as_ref()) + .await + { + Ok(res) if res.rows_affected() > 0 => { + repaired_count += 1; + tracing::info!( + target: "audit", + event = "manifests_consistency.repaired", + run_id = %store.run_id(), + file_hash = %row.file_hash, + stored_was = row.ref_count, + actual = row.actual_ref_count, + "🩹 manifest ref_count repaired" + ); + } + Ok(_) => { + // Row not touched — either another concurrent + // repair fixed it first, or the drift healed + // itself between page fetch and UPDATE. + // Silent no-op. + } + Err(e) => { + tracing::warn!( + target: "oxicloud::consistency", + event = "manifests_consistency.repair_failed", + run_id = %store.run_id(), + file_hash = %row.file_hash, + error = %e, + "manifest ref_count repair UPDATE failed — finding stays" + ); + } + } + } } // Advance cursor + checkpoint. @@ -279,10 +386,16 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { event = "manifests_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, - "manifests_consistency completed with {} finding(s)", - finding_count + repaired_count = repaired_count, + repair_requested = repair, + "manifests_consistency completed with {} finding(s), {} repaired", + finding_count, + repaired_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "repair_requested": repair, + "repaired_count": repaired_count, + })); } } } diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index e1985f77..43f75068 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -2575,6 +2575,12 @@ pub async fn list_jobs(State(state): State>) -> impl IntoResponse /// `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`. +/// +/// `repair=true` opts into corrective action on the refcount +/// consistency tenants (`blobs_consistency`, `manifests_consistency`, +/// and `consistency_batch` which fans out to both). Default `false` +/// preserves discovery-only. See `JobRunArgs.repair` for the +/// content-safety and race-safety guarantees. #[derive(serde::Deserialize)] pub struct TriggerJobQuery { #[serde(default)] @@ -2591,6 +2597,8 @@ pub struct TriggerJobQuery { /// `AppConfig.storage_entries`. #[serde(default)] pub storage: Option, + #[serde(default)] + pub repair: bool, } /// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule. @@ -2629,15 +2637,18 @@ pub async fn trigger_job( job = %name, force = query.force, deep = query.deep, - "👮🏻‍♂️ Admin triggered job {} (force={}, deep={})", + repair = query.repair, + "👮🏻‍♂️ Admin triggered job {} (force={}, deep={}, repair={})", name, query.force, query.deep, + query.repair, ); let args = JobRunArgs { force: query.force, deep: query.deep, storage: query.storage.clone(), + repair: query.repair, }; // Jobs that can run for hours (backend_migration, future