From 0a41f561d0b0facd0b239ebbda83685fcb8ea0e8 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 2 Aug 2026 01:05:02 +0200 Subject: [PATCH] feat(rotate-key): add report + key fingerprint in hexdigit fmt --- .../src/routes/admin/[[tab]]/+page.svelte | 3 + src/infrastructure/scheduler/pg_job_store.rs | 29 +++++ src/infrastructure/scheduler/recoverable.rs | 117 +++++++++++++++++- .../services/backend_consistency_service.rs | 6 +- .../services/blobs_consistency_service.rs | 4 +- .../services/drives_consistency_service.rs | 4 +- .../services/encrypted_blob_backend.rs | 29 +++++ .../services/files_consistency_service.rs | 4 +- .../services/folders_consistency_service.rs | 4 +- .../services/storage_migration_service.rs | 10 +- .../services/storage_rotate_service.rs | 23 +++- 11 files changed, 210 insertions(+), 23 deletions(-) diff --git a/frontend/src/routes/admin/[[tab]]/+page.svelte b/frontend/src/routes/admin/[[tab]]/+page.svelte index d72c59da..561a0d82 100644 --- a/frontend/src/routes/admin/[[tab]]/+page.svelte +++ b/frontend/src/routes/admin/[[tab]]/+page.svelte @@ -2101,6 +2101,7 @@ data-testid={`admin-storage-test-${entry.name}`} onclick={() => doTestEntry(entry.name)} > + {test?.busy ? t('admin.storage_testing', 'Testing…') : t('admin.storage_test', 'Test')} @@ -2135,6 +2136,7 @@ data-testid={`admin-storage-migrate-${entry.name}`} onclick={() => doMigrateActivate(entry.name)} > + {t('admin.storage_migrate_activate', 'Migrate & activate')} {:else} @@ -2145,6 +2147,7 @@ tabindex={-1} disabled > + {t('admin.storage_migrate_activate', 'Migrate & activate')} {/if} diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index a8a3436c..bddaf599 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -229,6 +229,35 @@ impl JobStore for PgJobStore { Ok(row.and_then(|(v,)| v)) } + async fn merge_stats( + &self, + extras: &serde_json::Map, + ) -> Result<(), DomainError> { + if extras.is_empty() { + return Ok(()); + } + // JSONB concat (`||`) is a shallow merge — right side wins on + // key conflict, which matches the "last-write-wins" semantic + // in the trait doc. Existing keys from the engine's own + // `scanned_count` / `finding_count` (written via `checkpoint` + // / `record_finding`) are preserved because handlers never + // emit those keys in their extras map. + sqlx::query( + r#" + UPDATE jobs.recoverable_runs + SET stats = COALESCE(stats, '{}'::jsonb) || $1::jsonb, + last_progress_at = NOW() + WHERE id = $2 + "#, + ) + .bind(serde_json::Value::Object(extras.clone())) + .bind(self.run_id) + .execute(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("merge_stats", e))?; + Ok(()) + } + async fn mark_completed(&self) -> Result<(), DomainError> { sqlx::query( r#" diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index cbf3af73..f47010bb 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -119,9 +119,65 @@ impl RunStatus { /// writes `status = Failed` with the message. #[derive(Debug, Clone)] pub enum RunOutcome { - Completed, - Paused { cursor: Vec }, - Failed { message: String }, + /// The run walked the whole subject space. + /// + /// `extra_stats` is merged into the run row's `stats` JSONB + /// alongside the engine-owned `scanned_count` + `finding_count` + /// / `severity_counts`. Handlers use it to surface per-run + /// summary counters (e.g. `storage_rotate` reports + /// `{"rewritten": N, "skipped": M, "failed": K}`) — the outcome + /// message in `JobOutcome.extra` and every downstream reader + /// of `RunSummary.stats` see the merged fields. + /// + /// Empty map = "no tenant-specific extras" — same shape as the + /// pre-K3 bare `Completed` variant. Handlers that don't + /// summarise their work call [`Self::completed`]. + Completed { + extra_stats: serde_json::Map, + }, + Paused { + cursor: Vec, + }, + Failed { + message: String, + }, +} + +impl RunOutcome { + /// Convenience for the common case: handler has nothing to add + /// to `stats` beyond what the engine already tracks (finding / + /// scanned counters). Equivalent to + /// `Completed { extra_stats: Map::new() }`. + pub fn completed() -> Self { + RunOutcome::Completed { + extra_stats: serde_json::Map::new(), + } + } + + /// Convenience for handlers that want to surface per-run + /// summary counters. Takes any JSON object literal produced by + /// `serde_json::json!({...})`; panics if the top-level value + /// isn't an Object (programmer bug — the contract is + /// object-shaped). + /// + /// Example — a rotate handler at run-complete: + /// + /// ```ignore + /// return RunOutcome::completed_with(serde_json::json!({ + /// "rewritten": rewritten_count, + /// "skipped": skipped_count, + /// "failed": failed_count, + /// })); + /// ``` + pub fn completed_with(extras: serde_json::Value) -> Self { + match extras { + serde_json::Value::Object(map) => RunOutcome::Completed { extra_stats: map }, + other => panic!( + "RunOutcome::completed_with expected a JSON object, got {}", + other + ), + } + } } // ─── Traits — implementor + port ──────────────────────────────────────────── @@ -295,6 +351,24 @@ pub trait JobStore: Send + Sync { detail: serde_json::Value, ) -> Result<(), DomainError>; + /// **Engine-only.** Merge `extras` into the run row's `stats` + /// JSONB (SQL `stats = stats || $1`). Called by [`run_or_resume`] + /// on [`RunOutcome::Completed`] to persist the handler's + /// per-run summary counters alongside the engine-owned + /// `scanned_count` / `finding_count`. Handler code MUST NOT + /// call this directly — return an `extra_stats` map on + /// `Completed` and the engine handles the write. + /// + /// Idempotent: merging the same map twice yields the same row. + /// A stats key that already exists is OVERWRITTEN by the + /// merge (last-write-wins) — a handler that emits e.g. + /// `"rewritten": 300` at run end always displaces any prior + /// per-batch write of the same key. + async fn merge_stats( + &self, + extras: &serde_json::Map, + ) -> Result<(), DomainError>; + // ─── Terminal writes — engine-only. Do not call from handler code. /// Engine-only. Called by [`run_or_resume`] on @@ -630,8 +704,22 @@ pub async fn run_or_resume( let stats = fetch_outcome_stats(&*provider, run_id).await; match outcome { - RunOutcome::Completed => { + RunOutcome::Completed { extra_stats } => { + // Merge tenant-supplied extras into the run's stats + // JSONB BEFORE the terminal mark, so downstream readers + // see the merged view atomically. `fetch_outcome_stats` + // (a few lines up) already ran and reflects the state + // WITHOUT the merge — re-fetch so the outer JobOutcome + // includes the tenant counters too. + if !extra_stats.is_empty() { + log_terminal_write_err( + "merge_stats", + run_id, + store.merge_stats(&extra_stats).await, + ); + } log_terminal_write_err("mark_completed", run_id, store.mark_completed().await); + let stats = fetch_outcome_stats(&*provider, run_id).await; JobOutcome::ok_with( stats.finding_count, serde_json::json!({ @@ -640,6 +728,7 @@ pub async fn run_or_resume( "finding_count": stats.finding_count, "scanned_count": stats.scanned_count, "severity_counts": stats.by_severity, + "extra_stats": serde_json::Value::Object(extra_stats), }), ) } @@ -865,6 +954,10 @@ mod tests { progress_total: Option, progress_kind: Option, string_params: std::collections::HashMap, + /// K3+: extras merged into the run's stats JSONB via + /// `merge_stats` at Completed time. Tests observe the merged + /// view by reading this map alongside `scanned_count`. + extra_stats: serde_json::Map, } #[async_trait] @@ -924,6 +1017,16 @@ mod tests { async fn get_string_param(&self, key: &str) -> Result, DomainError> { Ok(self.state.lock().unwrap().string_params.get(key).cloned()) } + async fn merge_stats( + &self, + extras: &serde_json::Map, + ) -> Result<(), DomainError> { + let mut s = self.state.lock().unwrap(); + for (k, v) in extras { + s.extra_stats.insert(k.clone(), v.clone()); + } + Ok(()) + } async fn mark_completed(&self) -> Result<(), DomainError> { self.state.lock().unwrap().status = RunStatus::Completed; Ok(()) @@ -976,6 +1079,7 @@ mod tests { progress_total: None, progress_kind: None, string_params: std::collections::HashMap::new(), + extra_stats: serde_json::Map::new(), }), }); let id = store.run_id; @@ -1035,6 +1139,7 @@ mod tests { progress_total: None, progress_kind: None, string_params: std::collections::HashMap::new(), + extra_stats: serde_json::Map::new(), }), }); stores.push(store.clone()); @@ -1204,7 +1309,7 @@ mod tests { _resume_cursor: Option>, ) -> RunOutcome { store.checkpoint(vec![1, 2, 3], 5).await.unwrap(); - RunOutcome::Completed + RunOutcome::completed() } } @@ -1259,7 +1364,7 @@ mod tests { resume_cursor: Option>, ) -> RunOutcome { *self.saw_cursor.lock().unwrap() = resume_cursor; - RunOutcome::Completed + RunOutcome::completed() } } diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 67f9b656..ed3a9f70 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -321,7 +321,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { backend = backend.backend_type(), "backend refused enumeration (typical during migration or on backends without list support)" ); - return RunOutcome::Completed; + return RunOutcome::completed(); } return RunOutcome::Failed { message: format!("backend list failed mid-scan: {e}"), @@ -373,7 +373,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } // Batch DB probe: which of these hashes have a @@ -451,7 +451,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } } } diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index ea142bbb..9f379750 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -381,7 +381,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { "blobs_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } let grace_cutoff = Utc::now() - CREATE_GRACE; @@ -528,7 +528,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { "blobs_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } } } diff --git a/src/infrastructure/services/drives_consistency_service.rs b/src/infrastructure/services/drives_consistency_service.rs index c4436d52..78e9e899 100644 --- a/src/infrastructure/services/drives_consistency_service.rs +++ b/src/infrastructure/services/drives_consistency_service.rs @@ -199,7 +199,7 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { "drives_consistency completed with {} drift finding(s)", drift_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } // Per-row check: cached vs actual. This is the ONE check @@ -255,7 +255,7 @@ impl RecoverableJobHandler for DrivesConsistencyCheck { "drives_consistency completed with {} drift finding(s)", drift_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } } } diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index 6643f50a..11de2b4f 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -313,6 +313,35 @@ impl BlobFormat { } } +impl std::fmt::Display for BlobFormat { + /// Human-friendly format for audit logs + finding details. + /// Renders `key_fp` as SSH-style colon-hex (e.g. + /// `83:96:ff:90:94:d7:ef:de`) instead of the raw byte-array Debug + /// shape (`[131, 150, 255, ...]`). Same spelling `xxd` produces + /// when you inspect a blob's on-disk header, so operators can + /// cross-check without a mental conversion. + /// + /// Handlers that render this in tracing macros should use `%` + /// (Display) — `?` (Debug) still gives the raw byte-array shape + /// for programmer-consumers who need the exact bytes. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobFormat::Legacy => write!(f, "legacy"), + BlobFormat::PlaintextV1 => write!(f, "plaintext-v1"), + BlobFormat::EncryptedV1 { key_fp } => { + write!(f, "encrypted-v1 key_fp=")?; + for (i, byte) in key_fp.iter().enumerate() { + if i > 0 { + write!(f, ":")?; + } + write!(f, "{byte:02x}")?; + } + Ok(()) + } + } + } +} + /// Assemble an encrypted-v1 blob: /// `OXCPT | v1 | key_fp | nonce | ciphertext | tag`. /// diff --git a/src/infrastructure/services/files_consistency_service.rs b/src/infrastructure/services/files_consistency_service.rs index a0d9a791..aec953b9 100644 --- a/src/infrastructure/services/files_consistency_service.rs +++ b/src/infrastructure/services/files_consistency_service.rs @@ -312,7 +312,7 @@ impl RecoverableJobHandler for FilesConsistencyCheck { "files_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } for row in &rows { @@ -490,7 +490,7 @@ impl RecoverableJobHandler for FilesConsistencyCheck { "files_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } } } diff --git a/src/infrastructure/services/folders_consistency_service.rs b/src/infrastructure/services/folders_consistency_service.rs index 4a87bafe..55ab99e0 100644 --- a/src/infrastructure/services/folders_consistency_service.rs +++ b/src/infrastructure/services/folders_consistency_service.rs @@ -255,7 +255,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { "folders_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } // Per-row branches. Add new ones here — same pattern as @@ -352,7 +352,7 @@ impl RecoverableJobHandler for FoldersConsistencyCheck { "folders_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::Completed; + return RunOutcome::completed(); } } } diff --git a/src/infrastructure/services/storage_migration_service.rs b/src/infrastructure/services/storage_migration_service.rs index 49390df8..5b6202b5 100644 --- a/src/infrastructure/services/storage_migration_service.rs +++ b/src/infrastructure/services/storage_migration_service.rs @@ -799,7 +799,15 @@ impl StorageMigrationService { "✅ storage_migration completed — hot-swapped runtime backend to `{target_name}`, \ writes resumed. No restart required." ); - RunOutcome::Completed + // Per-run summary counters merged into `stats` for the admin + // UI drawer. Same shape as `storage_rotate`'s extras + one + // extra `source_missing` counter unique to migration. + RunOutcome::completed_with(serde_json::json!({ + "copied": copied, + "skipped": skipped, + "failed": failed, + "source_missing": source_missing, + })) } } diff --git a/src/infrastructure/services/storage_rotate_service.rs b/src/infrastructure/services/storage_rotate_service.rs index 5f0e02b0..2d7b6966 100644 --- a/src/infrastructure/services/storage_rotate_service.rs +++ b/src/infrastructure/services/storage_rotate_service.rs @@ -232,9 +232,13 @@ impl RecoverableJobHandler for StorageRotateService { event = "storage_rotate.run_started", run_id = %store.run_id(), target_name = %target_name, - head_format = ?head_format, + // `%` (Display) → SSH-style `encrypted-v1 key_fp=83:96:...` + // instead of the raw `[131, 150, 255, ...]` byte-array + // shape Debug produces. Matches how `xxd` renders the + // header bytes on disk. + head_format = %head_format, resuming = !is_fresh, - "storage_rotate started on `{target_name}` (head_format = {head_format:?})" + "storage_rotate started on `{target_name}` (head_format = {head_format})" ); // Seed the progress snapshot. Total = count_total's estimate; @@ -411,8 +415,8 @@ impl RecoverableJobHandler for StorageRotateService { serde_json::json!({ "hash": hash, "phase": "write", - "from": format!("{current_format:?}"), - "to": format!("{head_format:?}"), + "from": format!("{current_format}"), + "to": format!("{head_format}"), "error": e.to_string(), }), ) @@ -484,7 +488,16 @@ impl StorageRotateService { failed = failed, "storage_rotate completed on `{target_name}` — {rewritten} rewritten, {skipped} skipped, {failed} failed" ); - RunOutcome::Completed + // Surface the per-run summary counters as extras merged into + // the run row's `stats` JSONB. Frontend renders whatever keys + // are present, so no wire-format bumping is needed — the + // admin UI's run drawer just picks these up alongside the + // engine-owned `finding_count` + `scanned_count`. + RunOutcome::completed_with(serde_json::json!({ + "rewritten": rewritten, + "skipped": skipped, + "failed": failed, + })) } fn clear_progress(&self) {