From 5527d09618f6b57bfc6ac0e79d61711e051175bb Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Wed, 29 Jul 2026 22:47:07 +0200 Subject: [PATCH] feat(recoverable-job): check if old blob (no cdc) still remains: notice only --- .../src/lib/components/AdminJobsPanel.svelte | 111 +++++++++++++---- frontend/static/locales/en.json | 3 + src/infrastructure/scheduler/pg_job_store.rs | 22 ++++ src/infrastructure/scheduler/recoverable.rs | 113 ++++++++++++++---- .../services/blobs_consistency_service.rs | 5 +- .../services/drives_consistency_service.rs | 16 ++- .../services/files_consistency_service.rs | 30 +++++ 7 files changed, 245 insertions(+), 55 deletions(-) diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index b56b8551..8ae1587c 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -231,26 +231,71 @@ } /** - * 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. + * Per-severity finding counts from `last_outcome.extra.severity_counts` + * (a JSON object populated by `run_or_resume`). Missing / older + * runs return an empty record — callers should tolerate absent keys. + * The three severity values are the ones consistency tenants emit + * today: `data_loss`, `inconsistent`, `anomaly`. */ - 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 lastSeverityCounts(job: JobSummary): Record { + if (!job.last_outcome || job.last_outcome.outcome !== 'ok') return {}; + const extra = job.last_outcome.extra as + | { severity_counts?: Record } + | undefined; + return extra?.severity_counts ?? {}; + } + + /** + * Actionable findings = `data_loss + inconsistent`. Those are what + * turn the outer outcome pill amber ("issues") and get the red + * badge on the outer job row. `anomaly` findings are informational + * and render as a blue notice instead — they don't count here. + */ + function actionableFindingCount(job: JobSummary): number { + const s = lastSeverityCounts(job); + return (s.data_loss ?? 0) + (s.inconsistent ?? 0); + } + + function anomalyFindingCount(job: JobSummary): number { + return lastSeverityCounts(job).anomaly ?? 0; + } + + /** + * Pill CSS modifier for a finding's severity — extracted so the + * findings-table cell and any future summary render share one + * source of truth. + * - `data_loss` → red (`err`) + * - `inconsistent` → amber (`paused`) + * - `anomaly` → blue (`notice`) + * - unknown → neutral grey + */ + function severityPillModifier(severity: string): string { + switch (severity) { + case 'data_loss': + return 'err'; + case 'inconsistent': + return 'paused'; + case 'anomaly': + return 'notice'; + default: + return 'neutral'; + } } function outcomeLabel(job: JobSummary): string { if (!job.last_outcome) return t('admin.jobs.never', 'never'); if (job.last_outcome.outcome === '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'); + // `ok` on the wire = dispatch completed. If any actionable + // findings surfaced, we flip to "issues" (amber). If only + // anomalies (informational), we flip to "notices" (blue). + // Clean run stays green. + if (actionableFindingCount(job) > 0) { + return t('admin.jobs.outcome_issues', 'issues'); + } + if (anomalyFindingCount(job) > 0) { + return t('admin.jobs.outcome_notices', 'notices'); + } + return t('admin.jobs.outcome_ok', 'ok'); } return t('admin.jobs.outcome_err', 'err'); } @@ -260,9 +305,13 @@ 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'; + if (actionableFindingCount(job) > 0) { + return 'jobs-panel__pill jobs-panel__pill--paused'; + } + if (anomalyFindingCount(job) > 0) { + return 'jobs-panel__pill jobs-panel__pill--notice'; + } + return 'jobs-panel__pill jobs-panel__pill--ok'; } function statusClass(status: RunStatus): string { @@ -435,8 +484,8 @@
{outcomeLabel(job)} - {#if lastFindingCount(job) > 0} - {@const findings = lastFindingCount(job)} + {#if actionableFindingCount(job) > 0} + {@const findings = actionableFindingCount(job)} {/if} + {#if anomalyFindingCount(job) > 0} + {@const notices = anomalyFindingCount(job)} + + {t('admin.jobs.n_notices', { n: notices }, '{{n}} notices')} + + {/if}
@@ -727,10 +788,9 @@ {f.kind} {f.severity} @@ -961,6 +1021,11 @@ color: var(--color-warning-text); } + .jobs-panel__pill--notice { + background: var(--color-info-bg); + color: var(--color-info-text); + } + .jobs-panel__pill--neutral { background: var(--color-bg-subtle); color: var(--color-text-muted); diff --git a/frontend/static/locales/en.json b/frontend/static/locales/en.json index ab4ad51e..17e588d5 100644 --- a/frontend/static/locales/en.json +++ b/frontend/static/locales/en.json @@ -1237,7 +1237,10 @@ "outcome_ok": "ok", "outcome_err": "err", "outcome_issues": "issues", + "outcome_notices": "notices", "n_findings": "{{n}} findings", + "n_notices": "{{n}} notices", + "notices_present_tooltip": "Informational findings — no action required. Expand for detail.", "state_running": "running", "never": "never", "just_now": "just now", diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 2542545d..cd1dbef2 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -409,6 +409,28 @@ impl JobStoreProvider for PgJobStoreProvider { .collect()) } + async fn finding_severity_counts( + &self, + run_id: Uuid, + ) -> Result, DomainError> { + let rows: Vec<(String, i64)> = sqlx::query_as( + r#" + SELECT severity, COUNT(*)::bigint + FROM jobs.run_findings + WHERE run_id = $1 + GROUP BY severity + "#, + ) + .bind(run_id) + .fetch_all(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("finding_severity_counts", e))?; + Ok(rows + .into_iter() + .map(|(sev, count)| (sev, count.max(0) as u64)) + .collect()) + } + async fn request_cancel(&self, job_name: &str) -> Result, DomainError> { // Only Running → CancelRequested flips. `Paused` can be // cancelled by not resuming — no need for a state change. diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index ce493401..896802d9 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -358,6 +358,19 @@ pub trait JobStoreProvider: Send + Sync { limit: u32, offset: u32, ) -> Result, DomainError>; + + /// Aggregate finding count grouped by severity for a specific + /// run. Used by [`run_or_resume`] to fold per-severity counts + /// into the outer `JobOutcome::extra` so the admin UI can + /// distinguish `data_loss`/`inconsistent` findings (which turn + /// the outer outcome pill amber/red — actionable) from + /// `anomaly` findings (which render as a neutral notice — + /// informational). Runs one grouped SQL query; O(number of + /// distinct severities on the run) rows returned. + async fn finding_severity_counts( + &self, + run_id: Uuid, + ) -> Result, DomainError>; } /// How a `RunProgress` fraction was derived. Lets the UI communicate @@ -573,18 +586,19 @@ pub async fn run_or_resume( // as "has findings" without also fetching the run history. // Called AFTER the handler returns but BEFORE the terminal write, // so stats are the ones accumulated during the run. - let (finding_count, scanned_count) = fetch_outcome_stats(&*provider, run_id).await; + let stats = fetch_outcome_stats(&*provider, run_id).await; match outcome { RunOutcome::Completed => { log_terminal_write_err("mark_completed", run_id, store.mark_completed().await); JobOutcome::ok_with( - finding_count, + stats.finding_count, serde_json::json!({ - "completed": true, - "run_id": run_id.to_string(), - "finding_count": finding_count, - "scanned_count": scanned_count, + "completed": true, + "run_id": run_id.to_string(), + "finding_count": stats.finding_count, + "scanned_count": stats.scanned_count, + "severity_counts": stats.by_severity, }), ) } @@ -592,13 +606,14 @@ 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( - finding_count, + stats.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, + "paused": true, + "run_id": run_id.to_string(), + "cursor_hex": cursor_hex, + "finding_count": stats.finding_count, + "scanned_count": stats.scanned_count, + "severity_counts": stats.by_severity, }), ) } @@ -609,25 +624,58 @@ 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 +/// Aggregate summary of a just-completed run, folded into the +/// outer `JobOutcome::extra`. Missing / failed queries default to +/// zeros so the outer outcome stays quiet instead of erroring. +struct OutcomeStats { + finding_count: u64, + scanned_count: u64, + /// Per-severity counts as a JSON map (`{"data_loss": N, + /// "inconsistent": M, "anomaly": K}`). The frontend uses this + /// to render the outer outcome pill: amber/red when + /// `data_loss + inconsistent > 0` (actionable), neutral notice + /// when only `anomaly > 0` (informational). + by_severity: serde_json::Value, +} + +async fn fetch_outcome_stats(provider: &dyn JobStoreProvider, run_id: Uuid) -> OutcomeStats { + let (finding_count, scanned_count) = match provider.get_run_by_id(run_id).await { + Ok(Some(summary)) => ( + summary .stats .get("finding_count") .and_then(|v| v.as_u64()) - .unwrap_or(0); - let scanned_count = summary + .unwrap_or(0), + summary .stats .get("scanned_count") .and_then(|v| v.as_u64()) - .unwrap_or(0); - (finding_count, scanned_count) - } + .unwrap_or(0), + ), _ => (0, 0), + }; + + // Per-severity breakdown. Only queried when there are findings + // to break down — a clean run doesn't need the extra round-trip. + let by_severity = if finding_count > 0 { + match provider.finding_severity_counts(run_id).await { + Ok(rows) => { + let mut map = serde_json::Map::new(); + for (severity, count) in rows { + map.insert(severity, serde_json::Value::Number(count.into())); + } + serde_json::Value::Object(map) + } + Err(_) => serde_json::Value::Object(Default::default()), + } + } else { + serde_json::Value::Object(Default::default()) + }; + + OutcomeStats { + finding_count, + scanned_count, + by_severity, } } @@ -1029,6 +1077,23 @@ mod tests { .collect()) } + async fn finding_severity_counts( + &self, + run_id: Uuid, + ) -> Result, DomainError> { + let stores = self.stores.lock().unwrap(); + let Some(store) = stores.iter().find(|s| s.run_id == run_id) else { + return Ok(Vec::new()); + }; + let state = store.state.lock().unwrap(); + let mut counts: std::collections::HashMap = + std::collections::HashMap::new(); + for f in state.findings.iter() { + *counts.entry(f.severity.clone()).or_default() += 1; + } + Ok(counts.into_iter().collect()) + } + async fn request_cancel(&self, _job_name: &str) -> Result, DomainError> { let stores = self.stores.lock().unwrap(); if let Some(s) = stores.last() { diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index fd2332e9..342d643f 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -374,10 +374,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { let last_hash = rows.last().map(|r| r.hash.clone()).expect("non-empty rows"); cursor = Some(last_hash.clone()); let batch_len = rows.len() as u64; - if let Err(e) = store - .checkpoint(last_hash.into_bytes(), batch_len) - .await - { + if let Err(e) = store.checkpoint(last_hash.into_bytes(), batch_len).await { return RunOutcome::Failed { message: format!("checkpoint: {e}"), }; diff --git a/src/infrastructure/services/drives_consistency_service.rs b/src/infrastructure/services/drives_consistency_service.rs index 071d61b4..a484f585 100644 --- a/src/infrastructure/services/drives_consistency_service.rs +++ b/src/infrastructure/services/drives_consistency_service.rs @@ -150,19 +150,27 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { // the `files` row is already visible — that would false- // positive as `stale_used_bytes`. 1h matches the window // `blobs_consistency` uses; same rationale (writes-in-flight). + // NOTE: `storage.drives` has no `name` column. The drive's + // display name lives on its root folder (see the schema + // comment on `drives.root_folder_id` — "The display name + // lives here"). LEFT JOIN storage.folders ON id = + // drive.root_folder_id and read `folders.name` as the + // drive's human identifier. `COALESCE` handles the + // (bug-only) case where root_folder_id is NULL. let rows: Vec<(Uuid, String, i64, i64)> = match sqlx::query_as( r#" SELECT - d.id, - d.name, - d.used_bytes, + d.id AS id, + COALESCE(rf.name, '?') AS name, + d.used_bytes AS used_bytes, COALESCE(( SELECT SUM(size)::bigint FROM storage.files WHERE drive_id = d.id AND NOT is_trashed - ), 0) AS actual_bytes + ), 0) AS actual_bytes FROM storage.drives d + LEFT JOIN storage.folders rf ON rf.id = d.root_folder_id WHERE ($1::uuid IS NULL OR d.id > $1) AND d.created_at < NOW() - INTERVAL '1 hour' ORDER BY d.id diff --git a/src/infrastructure/services/files_consistency_service.rs b/src/infrastructure/services/files_consistency_service.rs index 20dc866b..a0d9a791 100644 --- a/src/infrastructure/services/files_consistency_service.rs +++ b/src/infrastructure/services/files_consistency_service.rs @@ -435,6 +435,36 @@ impl RecoverableJobHandler for FilesConsistencyCheck { ) .await; } + + // (4) legacy_uncdc_file: informational — this file + // is served via the pre-CDC whole-file blob fallback, + // not the modern chunk-manifest path. Not broken; + // just misses out on sub-file dedup benefits + never + // shares chunks with newer uploads. Severity + // `anomaly` (surprising state, no known impact) so + // the UI renders it as a notice, not a warning. + // Fires when the blob registry has a whole-file row + // for this hash but there is no manifest. Recovery + // path: `ReingestLegacy` (deferred, see + // `docs/plan/recovery.md`). + if row.blob_size.is_some() && row.manifest_size.is_none() { + finding_count += 1; + record_or_log( + store, + FILES_CONSISTENCY_JOB_NAME, + "legacy_uncdc_file", + "anomaly", + Some(row.id), + serde_json::json!({ + "name": row.name, + "path": path, + "blob_hash": row.blob_hash, + "size": row.size, + "note": "pre-CDC whole-file blob; re-ingest for sub-file dedup", + }), + ) + .await; + } } // Advance cursor + checkpoint. `batch_len` feeds