feat(faults): one fault identity and a shared file/clear discipline (audit theme 4)

crates/core/src/oi/handler/backups.rs

@@ -433,16 +433,14 @@ async fn run_volume_backup(
         Err(e) => {
             tracing::error!(strategy = %strategy.name, vol = %vol_id, "invalid volume id: {e}");
             let app_owned = backing_app_name.clone();
+            let vol_owned = vol_id.to_owned();
             let desc = format!("strategy {:?}: {e}", strategy.name);
             tokio::task::block_in_place(|| {
                 state.db.call(move |db| {
-                    let _ = faults::file_fault(
+                    let _ = faults::file_once(
                         db,
-                        &app_owned,
-                        None,
-                        None,
-                        None,
-                        "backup_source_unavailable",
+                        &faults::FaultKey::new(&app_owned, "backup_source_unavailable", &vol_owned),
+                        &faults::FaultMeta::resource("volume", &vol_owned),
                         &desc,
                     );
                 })
@@ -458,15 +456,13 @@ async fn run_volume_backup(
             "strategy {:?}: volume {vol_id:?} path does not exist",
             strategy.name
         );
+        let vol_owned = vol_id.to_owned();
         tokio::task::block_in_place(|| {
             state.db.call(move |db| {
-                let _ = faults::file_fault(
+                let _ = faults::file_once(
                     db,
-                    &app_owned,
-                    None,
-                    None,
-                    None,
-                    "backup_source_unavailable",
+                    &faults::FaultKey::new(&app_owned, "backup_source_unavailable", &vol_owned),
+                    &faults::FaultMeta::resource("volume", &vol_owned),
                     &desc,
                 );
             })
@@ -498,15 +494,13 @@ async fn run_volume_backup(
                     "strategy {:?}: failed to snapshot volume {vol_id:?}: {e}",
                     strategy.name
                 );
+                let vol_owned = vol_id.to_owned();
                 tokio::task::block_in_place(|| {
                     state.db.call(move |db| {
-                        let _ = faults::file_fault(
+                        let _ = faults::file_once(
                             db,
-                            &app_owned,
-                            None,
-                            None,
-                            None,
-                            "backup_failed",
+                            &faults::FaultKey::new(&app_owned, "backup_failed", &vol_owned),
+                            &faults::FaultMeta::resource("volume", &vol_owned),
                             &desc,
                         );
                     })
@@ -571,11 +565,14 @@ async fn run_volume_backup(
         let _ = vol_store.remove_site(&snapshot_name).await;
 
         if success {
+            // r[impl backup.execution] — clear only this volume's faults. A
+            // kind-wide clear meant the second volume of a strategy erased the
+            // first one's failure, so a partially-failing backup looked clean.
             let app_owned = backing_app_name.clone();
+            let vol_owned = vol_id.to_owned();
             tokio::task::block_in_place(|| {
                 state.db.call(move |db| {
-                    faults::clear_faults_by_kind(db, &app_owned, "backup_failed").ok();
-                    faults::clear_faults_by_kind(db, &app_owned, "backup_source_unavailable").ok();
+                    clear_backup_faults_for_volume(db, &app_owned, &vol_owned);
                 })
             });
             return;
@@ -605,16 +602,43 @@ async fn run_volume_backup(
             "strategy {:?}: save-snapshot failed for volume {vol_id:?}",
             strategy.name
         );
+        let vol_owned = vol_id.to_owned();
         tokio::task::block_in_place(|| {
             state.db.call(move |db| {
-                let _ =
-                    faults::file_fault(db, &app_owned, None, None, None, "backup_failed", &desc);
+                let _ = faults::file_once(
+                    db,
+                    &faults::FaultKey::new(&app_owned, "backup_failed", &vol_owned),
+                    &faults::FaultMeta::resource("volume", &vol_owned),
+                    &desc,
+                );
             })
         });
         return;
     }
 }
 
+/// Clear the backup faults belonging to one volume.
+///
+/// `backup_failed` and `backup_source_unavailable` are condition faults about
+/// a specific volume: true until that volume backs up successfully. Keying
+/// them by volume is what lets a success clear its own without touching the
+/// other volumes in the same strategy.
+// r[impl fault.lifecycle]
+fn clear_backup_faults_for_volume(db: &crate::runtime::db::Db, app: &AppName, vol_id: &str) {
+    for kind in ["backup_failed", "backup_source_unavailable"] {
+        let active: Vec<_> = faults::list_active_faults(db, Some(app))
+            .unwrap_or_default()
+            .into_iter()
+            .filter(|f| f.kind == kind && f.subject == vol_id)
+            .collect();
+        for fault in &active {
+            if let Err(e) = faults::clear_fault(db, &fault.id, app) {
+                tracing::warn!(app = %app, vol = %vol_id, "failed to clear {kind} fault: {e}");
+            }
+        }
+    }
+}
+
 // r[impl backup.execution]
 async fn acquire_scheduler_slot(
     state: &Arc<OiState>,

crates/core/src/oi/handler/registries.rs

@@ -30,6 +30,15 @@ pub(crate) fn add_registry(state: &OiState, params: RegistryParams) -> HandlerRe
         .db
         .call(move |db| registries::add_allowed_registry(db, &registry))
         .map_err(|e| OiError::new(ErrorCode::NotFound, format!("db error: {e}")))?;
+
+    // r[impl fault.lifecycle] — `disallowed_registry` is a condition fault:
+    // true exactly while an app's images reference a registry outside the
+    // allowlist. Adding the registry is the natural remediation, so it has to
+    // re-evaluate for the same reason removing one does — otherwise the fault
+    // the operator just fixed stands until something else happens to reload
+    // the app.
+    re_evaluate_all_apps(state);
+
     Ok(json!({ "ok": true }))
 }
 

crates/core/src/oi/handler/registries/tests.rs

@@ -84,3 +84,42 @@ fn disallowed_registry_files_fault_and_removal_reevaluates_apps() {
     let faults = oi.call("/faults/list", json!({ "app": "demo" })).unwrap();
     assert_eq!(faults.as_array().unwrap().len(), 1);
 }
+
+// r[verify fault.lifecycle]
+// i[verify registry.add]
+// `disallowed_registry` is true exactly while an app's images reference a
+// registry outside the allowlist. Adding the registry is the remediation, so
+// it has to re-evaluate — otherwise the fault the operator just fixed stood
+// until something else happened to reload the app.
+#[test]
+fn adding_a_registry_clears_the_fault_it_resolves() {
+    let oi = TestOi::new();
+    let script = r#"
+        app.deployment("web").image("quay.io/acme/web:1.0");
+    "#;
+    oi.call("/apps/create", json!({ "app": "demo", "script": script }))
+        .unwrap();
+
+    let faults = oi.call("/faults/list", json!({ "app": "demo" })).unwrap();
+    assert!(
+        faults
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|f| f["kind"] == "disallowed_registry"),
+        "precondition: quay.io is not allowlisted: {faults:#?}"
+    );
+
+    oi.call("/registries/add", json!({ "registry": "quay.io" }))
+        .unwrap();
+
+    let faults = oi.call("/faults/list", json!({ "app": "demo" })).unwrap();
+    assert!(
+        !faults
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|f| f["kind"] == "disallowed_registry"),
+        "the fault must clear once the registry is allowed: {faults:#?}"
+    );
+}

crates/core/src/runtime/audit.rs

@@ -95,14 +95,19 @@ pub fn spawn_audit_task(
                     // r[impl audit.log.resilience]
                     warn!(dropped = n, "audit log receiver lagged, events lost");
                     db.call(move |db| {
+                        // r[impl fault.lifecycle] — an event fault: it records
+                        // that lag happened, and there is no "currently
+                        // lagging" set to converge against. Dedup on file, and
+                        // the clear path is the operator's: the fault stands
+                        // until acknowledged, because nothing else can know
+                        // the dropped events have been accounted for. Without
+                        // the dedup, a lagging feed filed one fault per lag
+                        // event without bound and GC prunes only cleared ones.
                         let seedling_app = AppName::new_unchecked("seedling");
-                        let _ = crate::runtime::faults::file_fault(
+                        let _ = crate::runtime::faults::file_once(
                             db,
-                            &seedling_app,
-                            None,
-                            None,
-                            None,
-                            "audit_lag",
+                            &crate::runtime::faults::FaultKey::app_wide(&seedling_app, "audit_lag"),
+                            &crate::runtime::faults::FaultMeta::default(),
                             &format!("audit log receiver lagged, {n} events dropped"),
                         );
                     });

crates/core/src/runtime/db.rs

@@ -132,6 +132,7 @@ const SQL_V53: &str = include_str!("db/migrations/v53.sql");
 // r[impl autonomous.restart.record]
 // r[impl autonomous.restart.rate.settings]
 const SQL_V54: &str = include_str!("db/migrations/v54.sql");
+const SQL_V55: &str = include_str!("db/migrations/v55.sql");
 
 const MIGRATIONS: &[Migration] = &[
     Migration {
@@ -399,6 +400,11 @@ const MIGRATIONS: &[Migration] = &[
         sql: SQL_V54,
         custom_run: None,
     },
+    Migration {
+        version: 55,
+        sql: SQL_V55,
+        custom_run: None,
+    },
 ];
 
 fn migration_hash(sql: &str) -> String {

crates/core/src/runtime/db/migrations/v55.sql

@@ -0,0 +1,49 @@
+-- r[impl fault.lifecycle]
+-- Give faults a first-class subject: the thing that is faulty.
+--
+-- Until now the subject was smeared across three columns and, where none of
+-- them fitted, the description text — image refs, `host:port` tuples, `[key]`
+-- prefixes. Matching by description-substring is fragile, and where no column
+-- fitted at all the subject was simply absent, so clearing had to fall back to
+-- app + kind. That is what made a successful backup of one volume clear every
+-- other volume's `backup_failed` fault.
+--
+-- resource_type / resource_name / instance_id stay as display metadata;
+-- identity is (app, kind, subject).
+ALTER TABLE faults ADD COLUMN subject TEXT NOT NULL DEFAULT '';
+
+-- Backfill from whichever column carried the subject, so a fault filed before
+-- this migration still matches the key its site computes after it and can be
+-- cleared rather than stranded. Cleared rows are backfilled too: this table is
+-- also the fault history the operator interface reads, and leaving historical
+-- rows without a subject would make past faults ambiguous in exactly the way
+-- the column exists to prevent.
+UPDATE faults
+SET subject = COALESCE(instance_id, resource_name, '')
+WHERE subject = '';
+
+-- Duplicate active faults for one key already exist (audit_lag files without
+-- any dedup at all), and the index below would refuse to build over them.
+-- Clear all but the newest of each group rather than deleting: the fault list
+-- is built from this table's history.
+UPDATE faults
+SET cleared_at = timestamp
+WHERE cleared_at IS NULL
+  AND id NOT IN (
+      SELECT id FROM (
+          SELECT id,
+                 ROW_NUMBER() OVER (
+                     PARTITION BY app, kind, subject
+                     ORDER BY timestamp DESC, id DESC
+                 ) AS rn
+          FROM faults
+          WHERE cleared_at IS NULL
+      )
+      WHERE rn = 1
+  );
+
+-- At most one active fault per key. Partial, so cleared rows accumulate freely
+-- for the history the fault list is built from.
+CREATE UNIQUE INDEX IF NOT EXISTS faults_active_key
+    ON faults (app, kind, subject)
+    WHERE cleared_at IS NULL;

crates/core/src/runtime/db/tests.rs

@@ -13,7 +13,7 @@ fn open_in_memory_succeeds() {
             |r| r.get(0),
         )
         .expect("schema_version should exist");
-    assert_eq!(version, 54);
+    assert_eq!(version, 55);
 }
 
 // r[verify history.persistence]
@@ -45,7 +45,7 @@ fn params_table_exists() {
             |r| r.get(0),
         )
         .expect("schema_version should exist");
-    assert_eq!(version, 54);
+    assert_eq!(version, 55);
 }
 
 // i[verify app.persist]