diff --git a/crates/core/src/oi/handler/backups.rs b/crates/core/src/oi/handler/backups.rs index a4835a52..b5f32628 100644 --- a/crates/core/src/oi/handler/backups.rs +++ b/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, diff --git a/crates/core/src/oi/handler/registries.rs b/crates/core/src/oi/handler/registries.rs index 6544842b..ca77e43f 100644 --- a/crates/core/src/oi/handler/registries.rs +++ b/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, ®istry)) .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 })) } diff --git a/crates/core/src/oi/handler/registries/tests.rs b/crates/core/src/oi/handler/registries/tests.rs index e3021ee7..83d85dfd 100644 --- a/crates/core/src/oi/handler/registries/tests.rs +++ b/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:#?}" + ); +} diff --git a/crates/core/src/runtime/audit.rs b/crates/core/src/runtime/audit.rs index 1ef5eeba..2f5bb6db 100644 --- a/crates/core/src/runtime/audit.rs +++ b/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"), ); }); diff --git a/crates/core/src/runtime/db.rs b/crates/core/src/runtime/db.rs index 5436644c..72b4ce13 100644 --- a/crates/core/src/runtime/db.rs +++ b/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 { diff --git a/crates/core/src/runtime/db/migrations/v55.sql b/crates/core/src/runtime/db/migrations/v55.sql new file mode 100644 index 00000000..091ae731 --- /dev/null +++ b/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; diff --git a/crates/core/src/runtime/db/tests.rs b/crates/core/src/runtime/db/tests.rs index e9b4c9ec..2e03de71 100644 --- a/crates/core/src/runtime/db/tests.rs +++ b/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] diff --git a/crates/core/src/runtime/faults.rs b/crates/core/src/runtime/faults.rs index 97b36c23..f1c7b75b 100644 --- a/crates/core/src/runtime/faults.rs +++ b/crates/core/src/runtime/faults.rs @@ -1,6 +1,7 @@ use std::sync::OnceLock; use jiff::Timestamp; +use rusqlite::OptionalExtension as _; use seedling_protocol::events::EventSender; use seedling_protocol::names::AppName; use serde::Serialize; @@ -46,10 +47,75 @@ pub struct FaultRecord { pub resource_name: Option, pub instance_id: Option, pub kind: String, + /// The faulty thing; see [`FaultKey`]. Empty for app-wide faults. + pub subject: String, pub timestamp: Timestamp, pub description: String, } +/// A fault's identity: the thing that is faulty, and in what way. +/// +/// Before this existed, every site invented its own notion of sameness — +/// `(kind, instance_id)` here, `(kind, resource_name)` there, `(kind, +/// description)` elsewhere, bare `kind` in one place, and nothing at all in +/// another. Clears were then either too broad (a successful backup of one +/// volume cleared every volume's `backup_failed`) or too fragile (matching on +/// a `host:port` substring of the description). +/// +/// `resource_type`/`resource_name`/`instance_id` remain display metadata. +/// Matching and clearing use this key alone. +// r[impl fault.lifecycle] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct FaultKey { + pub app: AppName, + pub kind: String, + /// The faulty thing: a volume id, an image ref, a `host:port`, an + /// instance hex. Empty when the fault is about the app as a whole. + pub subject: String, +} + +impl FaultKey { + pub fn new(app: &AppName, kind: impl Into, subject: impl Into) -> Self { + Self { + app: app.clone(), + kind: kind.into(), + subject: subject.into(), + } + } + + /// A fault about the app itself rather than any one thing within it. + pub fn app_wide(app: &AppName, kind: impl Into) -> Self { + Self::new(app, kind, "") + } +} + +/// The display metadata that rides along with a fault but takes no part in +/// its identity. +#[derive(Debug, Clone, Default)] +pub struct FaultMeta { + pub resource_type: Option, + pub resource_name: Option, + pub instance_id: Option, +} + +impl FaultMeta { + pub fn instance(resource_type: &str, resource_name: &str, instance_id: &str) -> Self { + Self { + resource_type: Some(resource_type.to_owned()), + resource_name: Some(resource_name.to_owned()), + instance_id: Some(instance_id.to_owned()), + } + } + + pub fn resource(resource_type: &str, resource_name: &str) -> Self { + Self { + resource_type: Some(resource_type.to_owned()), + resource_name: Some(resource_name.to_owned()), + instance_id: None, + } + } +} + // i[fault.record] pub fn file_fault( db: &crate::runtime::db::Db, @@ -60,14 +126,80 @@ pub fn file_fault( kind: &str, description: &str, ) -> rusqlite::Result { + // Sites that have not yet been given an explicit subject derive one the + // same way migration v53 backfilled the existing rows, so a fault filed + // before the migration matches the key its site computes after it. + let subject = instance_id.or(resource_name).unwrap_or(""); + file_keyed( + db, + &FaultKey::new(app, kind, subject), + &FaultMeta { + resource_type: resource_type.map(str::to_owned), + resource_name: resource_name.map(str::to_owned), + instance_id: instance_id.map(str::to_owned), + }, + description, + ) + .map(|(id, _)| id) +} + +/// File `key` unless an active fault already holds it. +/// +/// Returns the fault's id and whether it was newly filed. This is the dedup +/// that four sites hand-rolled as an `already_filed` scan of +/// `list_active_faults`, and that `audit_lag` omitted entirely — so it +/// duplicated without bound. +// r[impl fault.lifecycle] +pub fn file_once( + db: &crate::runtime::db::Db, + key: &FaultKey, + meta: &FaultMeta, + description: &str, +) -> rusqlite::Result { + file_keyed(db, key, meta, description).map(|(_, filed)| filed) +} + +/// Insert the fault, or return the id of the active one already holding the +/// key. The uniqueness is enforced by a partial unique index rather than by a +/// read-then-write, so concurrent filers cannot both win. +fn file_keyed( + db: &crate::runtime::db::Db, + key: &FaultKey, + meta: &FaultMeta, + description: &str, +) -> rusqlite::Result<(String, bool)> { let id = uuid::Uuid::new_v4().to_string(); let now = Timestamp::now(); let timestamp = now.to_string(); - db.conn.execute( - "INSERT INTO faults (id, app, resource_type, resource_name, instance_id, kind, timestamp, description) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", - rusqlite::params![id, app, resource_type, resource_name, instance_id, kind, timestamp, description], + let app = &key.app; + let kind = key.kind.as_str(); + let resource_type = meta.resource_type.as_deref(); + let resource_name = meta.resource_name.as_deref(); + let instance_id = meta.instance_id.as_deref(); + let inserted = db.conn.execute( + "INSERT INTO faults (id, app, resource_type, resource_name, instance_id, kind, timestamp, description, subject) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) + ON CONFLICT DO NOTHING", + rusqlite::params![id, app, resource_type, resource_name, instance_id, kind, timestamp, description, key.subject], )?; + if inserted == 0 { + // The row that won the conflict can be cleared before this reads it, + // in which case the key is free again and the caller's fault should + // be filed rather than reported as a duplicate. + let existing: Option = db + .conn + .query_row( + "SELECT id FROM faults + WHERE app = ?1 AND kind = ?2 AND subject = ?3 AND cleared_at IS NULL", + rusqlite::params![app, kind, key.subject], + |row| row.get(0), + ) + .optional()?; + match existing { + Some(id) => return Ok((id, false)), + None => return file_keyed(db, key, meta, description), + } + } warn!( app = %app, kind, resource_type, resource_name, instance_id, "fault filed: {description}", @@ -79,11 +211,12 @@ pub fn file_fault( resource_name: resource_name.map(str::to_owned), instance_id: instance_id.map(str::to_owned), kind: kind.to_owned(), + subject: key.subject.clone(), timestamp: now, description: description.to_owned(), }; emit_filed(&record); - Ok(id) + Ok((id, true)) } /// Clear a single fault by ID. The `app` is needed for the event broadcast; @@ -126,7 +259,7 @@ pub fn list_active_faults( match app { Some(app_name) => { let mut stmt = db.conn.prepare( - "SELECT id, app, resource_type, resource_name, instance_id, kind, timestamp, description + "SELECT id, app, resource_type, resource_name, instance_id, kind, timestamp, description, subject FROM faults WHERE cleared_at IS NULL AND app = ?1 ORDER BY timestamp", )?; @@ -137,7 +270,7 @@ pub fn list_active_faults( } None => { let mut stmt = db.conn.prepare( - "SELECT id, app, resource_type, resource_name, instance_id, kind, timestamp, description + "SELECT id, app, resource_type, resource_name, instance_id, kind, timestamp, description, subject FROM faults WHERE cleared_at IS NULL ORDER BY timestamp", )?; @@ -162,6 +295,7 @@ fn row_to_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { resource_name: row.get(3)?, instance_id: row.get(4)?, kind: row.get(5)?, + subject: row.get(8)?, timestamp, description: row.get(7)?, }) @@ -242,5 +376,128 @@ pub fn count_active_faults(db: &crate::runtime::db::Db) -> rusqlite::Result ) } +/// Which active faults a [`sync_faults`] call owns, and may therefore clear. +/// +/// Scoping is what keeps a converge call from clearing kinds it knows nothing +/// about: a sweep that computes every currently-conflicting `(host, port)` +/// must not treat the absence of a `backup_failed` key as a reason to clear +/// one. +// r[impl fault.lifecycle] +#[derive(Debug, Clone)] +pub enum FaultScope { + /// Every active fault of this kind, across all apps. For conditions + /// computed globally each tick, such as ingress conflicts. + Kind(String), + /// Every active fault of this kind belonging to one app. + AppKind(AppName, String), +} + +impl FaultScope { + fn owns(&self, record: &FaultRecord) -> bool { + match self { + Self::Kind(kind) => record.kind == *kind, + Self::AppKind(app, kind) => record.app == *app && record.kind == *kind, + } + } + + fn kind(&self) -> &str { + match self { + Self::Kind(kind) | Self::AppKind(_, kind) => kind, + } + } +} + +/// The active faults a scope owns, filtered in SQL. +/// +/// A global sweep such as ingress conflicts runs every tick; reading every +/// active fault in the database and discarding all but one kind would make +/// that cost grow with the total fault count rather than with the kind's. +fn list_active_faults_in_scope( + db: &crate::runtime::db::Db, + scope: &FaultScope, +) -> rusqlite::Result> { + const COLUMNS: &str = + "id, app, resource_type, resource_name, instance_id, kind, timestamp, description, subject"; + match scope { + FaultScope::Kind(kind) => { + let mut stmt = db.conn.prepare(&format!( + "SELECT {COLUMNS} FROM faults + WHERE cleared_at IS NULL AND kind = ?1 + ORDER BY timestamp" + ))?; + stmt.query_map([kind], row_to_record)?.collect() + } + FaultScope::AppKind(app, kind) => { + let mut stmt = db.conn.prepare(&format!( + "SELECT {COLUMNS} FROM faults + WHERE cleared_at IS NULL AND app = ?1 AND kind = ?2 + ORDER BY timestamp" + ))?; + stmt.query_map(rusqlite::params![app, kind], row_to_record)? + .collect() + } + } +} + +/// What a [`sync_faults`] call did. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct SyncOutcome { + pub filed: usize, + pub cleared: usize, +} + +/// Converge the active faults within `scope` to exactly `current`. +/// +/// Files the keys in `current` that are not active, clears the active ones +/// that are no longer in `current`. `describe` supplies a description for each +/// newly-filed key. +/// +/// This replaces the two in-memory prev-set diffs in the reconciler, which +/// cleared only `prior \ current` where `prior` was a `Reconciler` field that +/// starts empty on every daemon start — the faults are in the database, so one +/// filed before a restart could never clear. Comparing against the persisted +/// active set instead is restart-safe by construction: there is no warm-up +/// state to reconstruct and none to forget to persist. +// r[impl fault.lifecycle] +pub fn sync_faults( + db: &crate::runtime::db::Db, + scope: &FaultScope, + current: &std::collections::BTreeMap, +) -> rusqlite::Result { + let active = list_active_faults_in_scope(db, scope)?; + + let mut outcome = SyncOutcome::default(); + + for record in &active { + let key = FaultKey::new(&record.app, record.kind.clone(), record.subject.clone()); + if !current.contains_key(&key) { + clear_fault(db, &record.id, &record.app)?; + outcome.cleared += 1; + } + } + + for (key, (meta, description)) in current { + debug_assert!( + scope.owns(&FaultRecord { + id: String::new(), + app: key.app.clone(), + resource_type: None, + resource_name: None, + instance_id: None, + kind: key.kind.clone(), + subject: key.subject.clone(), + timestamp: Timestamp::now(), + description: String::new(), + }), + "sync_faults given a key outside its own scope: {key:?}" + ); + if file_once(db, key, meta, description)? { + outcome.filed += 1; + } + } + + Ok(outcome) +} + #[cfg(test)] mod tests; diff --git a/crates/core/src/runtime/faults/tests.rs b/crates/core/src/runtime/faults/tests.rs index 53bcc3da..6a406c11 100644 --- a/crates/core/src/runtime/faults/tests.rs +++ b/crates/core/src/runtime/faults/tests.rs @@ -82,7 +82,18 @@ fn clear_faults_by_kind_clears_matching() { let db = Db::open_in_memory().expect("open"); init_test_events(); file_fault(&db, &app("myapp"), None, None, None, "script_error", "err1").expect("file1"); - file_fault(&db, &app("myapp"), None, None, None, "script_error", "err2").expect("file2"); + // Distinct subjects: one active fault per (app, kind, subject) now, so + // two same-subject script errors would be one fault, not two. + file_fault( + &db, + &app("myapp"), + None, + Some("second"), + None, + "script_error", + "err2", + ) + .expect("file2"); file_fault( &db, &app("myapp"), @@ -227,8 +238,8 @@ fn count_active_faults_for_app_counts_only_uncleared() { 0 ); - let id1 = file_fault(&db, &app("myapp"), None, None, None, "err", "1").expect("1"); - file_fault(&db, &app("myapp"), None, None, None, "err", "2").expect("2"); + let id1 = file_fault(&db, &app("myapp"), None, Some("one"), None, "err", "1").expect("1"); + file_fault(&db, &app("myapp"), None, Some("two"), None, "err", "2").expect("2"); file_fault(&db, &app("other"), None, None, None, "err", "3").expect("3"); assert_eq!( count_active_faults_for_app(&db, &app("myapp")).expect("count"), @@ -379,3 +390,138 @@ fn clear_faults_for_instance_only_removes_matching_instance() { assert!(kinds.contains(&"container_start_failed".to_string())); assert!(kinds.contains(&"operation_failed".to_string())); } + +fn meta() -> FaultMeta { + FaultMeta::default() +} + +fn desc(key: &FaultKey) -> std::collections::BTreeMap { + let mut map = std::collections::BTreeMap::new(); + map.insert(key.clone(), (meta(), format!("{} is faulty", key.subject))); + map +} + +// r[verify fault.lifecycle] +// At most one active fault per key: `audit_lag` filed one per lag event +// without bound, and GC prunes only cleared faults. +#[test] +fn file_once_does_not_duplicate() { + let db = Db::open_in_memory().expect("open"); + init_test_events(); + let key = FaultKey::app_wide(&app("seedling"), "audit_lag"); + + assert!(file_once(&db, &key, &meta(), "42 events dropped").unwrap()); + assert!(!file_once(&db, &key, &meta(), "7 more events dropped").unwrap()); + + let active = list_active_faults(&db, None).unwrap(); + assert_eq!(active.len(), 1, "{active:#?}"); +} + +// r[verify fault.lifecycle] +// Clearing keyed no more broadly than filing: this is H8 — a successful +// backup of one volume cleared every other volume's failure. +#[test] +fn faults_for_different_subjects_are_independent() { + let db = Db::open_in_memory().expect("open"); + init_test_events(); + let app = app("backups"); + let ok = FaultKey::new(&app, "backup_failed", "site/data"); + let broken = FaultKey::new(&app, "backup_failed", "site/archive"); + + file_once(&db, &ok, &meta(), "data failed").unwrap(); + file_once(&db, &broken, &meta(), "archive failed").unwrap(); + + // "data" succeeds on a later run and clears only its own key. + let to_clear: Vec<_> = list_active_faults(&db, Some(&app)) + .unwrap() + .into_iter() + .filter(|f| f.kind == "backup_failed" && f.subject == "site/data") + .collect(); + for fault in &to_clear { + clear_fault(&db, &fault.id, &app).unwrap(); + } + + let active = list_active_faults(&db, Some(&app)).unwrap(); + assert_eq!(active.len(), 1, "{active:#?}"); + assert_eq!(active[0].subject, "site/archive"); +} + +// r[verify fault.lifecycle] +// A condition fault is active exactly while its condition holds — including +// when the condition stopped holding before this process started. The +// reconciler used to diff against an in-memory prior set that empties on +// every daemon start, so a fault filed before a restart could never clear. +#[test] +fn sync_clears_a_fault_filed_by_a_previous_process() { + let db = Db::open_in_memory().expect("open"); + init_test_events(); + let system = AppName::new_unchecked("_system"); + let key = FaultKey::new(&system, "ingress_conflict", "example.com:443"); + + // Pre-seed as if a previous daemon lifetime filed it. + file_once(&db, &key, &meta(), "conflict on example.com:443").unwrap(); + + // This tick sees no conflicts at all, with no memory of the previous one. + let outcome = sync_faults( + &db, + &FaultScope::Kind("ingress_conflict".to_owned()), + &std::collections::BTreeMap::new(), + ) + .unwrap(); + + assert_eq!(outcome.cleared, 1); + assert!(list_active_faults(&db, None).unwrap().is_empty()); +} + +// r[verify fault.lifecycle] +#[test] +fn sync_is_idempotent_and_converges_from_any_state() { + let db = Db::open_in_memory().expect("open"); + init_test_events(); + let system = AppName::new_unchecked("_system"); + let key = FaultKey::new(&system, "ingress_conflict", "a.example:443"); + let scope = FaultScope::Kind("ingress_conflict".to_owned()); + + let first = sync_faults(&db, &scope, &desc(&key)).unwrap(); + assert_eq!( + first, + SyncOutcome { + filed: 1, + cleared: 0 + } + ); + + let second = sync_faults(&db, &scope, &desc(&key)).unwrap(); + assert_eq!( + second, + SyncOutcome { + filed: 0, + cleared: 0 + } + ); + + assert_eq!(list_active_faults(&db, None).unwrap().len(), 1); +} + +// r[verify fault.lifecycle] +// A sweep may only clear within its declared scope; otherwise converging one +// kind to empty would wipe every other kind's faults. +#[test] +fn sync_never_touches_kinds_outside_its_scope() { + let db = Db::open_in_memory().expect("open"); + init_test_events(); + let system = AppName::new_unchecked("_system"); + let other = FaultKey::new(&system, "resolver_failed", ""); + file_once(&db, &other, &meta(), "resolver down").unwrap(); + + sync_faults( + &db, + &FaultScope::Kind("ingress_conflict".to_owned()), + &std::collections::BTreeMap::new(), + ) + .unwrap(); + + let active = list_active_faults(&db, None).unwrap(); + assert_eq!(active.len(), 1); + assert_eq!(active[0].kind, "resolver_failed"); +} diff --git a/crates/core/src/runtime/tailscale.rs b/crates/core/src/runtime/tailscale.rs index d07f7ca9..cee251cb 100644 --- a/crates/core/src/runtime/tailscale.rs +++ b/crates/core/src/runtime/tailscale.rs @@ -164,6 +164,7 @@ impl TailscaleProvider { match self.poll_once().await { Ok(identity) => { consecutive_failures = 0; + self.sync_unreachable_fault(false); self.reconcile_db(identity); } Err(TailscaleError::Unreachable(msg)) => { @@ -192,6 +193,11 @@ impl TailscaleProvider { // Mark any existing discovered row stale so the // reconciler stops emitting routes for it. self.mark_existing_stale(true); + // r[impl fault.lifecycle] — the fault this threshold + // exists for was documented but never actually filed, + // so a tailnet that had been unreachable for hours + // showed up only as a `warn!` in the log. + self.sync_unreachable_fault(true); } } } @@ -205,6 +211,47 @@ impl TailscaleProvider { } } + /// Converge the `tailscale_unreachable` system fault to whether the + /// provider currently considers tailscaled unreachable. + /// + /// A condition fault: true exactly while the condition holds. Converging + /// rather than filing-and-hoping means the clear is the sweep, so the + /// fault cannot outlive the outage — including across a daemon restart, + /// where the failure counter starts at zero but the fault is in the + /// database. + // r[impl fault.lifecycle] + fn sync_unreachable_fault(&self, unreachable: bool) { + let socket = self.config.socket_path.display().to_string(); + self.db.call(move |db| { + let system = seedling_protocol::names::AppName::new_unchecked("_system"); + let key = crate::runtime::faults::FaultKey::app_wide(&system, "tailscale_unreachable"); + let mut current = std::collections::BTreeMap::new(); + if unreachable { + current.insert( + key, + ( + crate::runtime::faults::FaultMeta::default(), + format!( + "tailscaled has been unreachable for {FAULT_AFTER_FAILURES} \ + consecutive polls via {socket}; the discovered site ingress is \ + marked stale and its routes are withdrawn" + ), + ), + ); + } + if let Err(e) = crate::runtime::faults::sync_faults( + db, + &crate::runtime::faults::FaultScope::AppKind( + system.clone(), + "tailscale_unreachable".to_owned(), + ), + ¤t, + ) { + warn!("tailscale: failed to converge tailscale_unreachable fault: {e}"); + } + }); + } + /// Single poll attempt. Returns `Ok(None)` when the backend is reachable /// but reports no identity (e.g. backend not yet running, no Self /// section), `Ok(Some(identity))` on success, and `Err` on transport / diff --git a/crates/core/src/system/reconcile.rs b/crates/core/src/system/reconcile.rs index cc9babb3..bd058c6d 100644 --- a/crates/core/src/system/reconcile.rs +++ b/crates/core/src/system/reconcile.rs @@ -433,12 +433,6 @@ pub struct Reconciler { /// debounce transient probe misses across ticks instead of bouncing /// the container on the first 2-second timeout. resolver_health_fail_count: std::sync::atomic::AtomicU32, - /// Last tick's set of (hostname, port) tuples that were in an - /// app-vs-site-ingress conflict. Used to clear the corresponding - /// `ingress_conflict` faults on the first tick where the conflict - /// no longer appears. - // r[impl ingress.site.conflict] - prev_ingress_conflicts: std::collections::BTreeSet<(String, u16)>, /// DNS resolver feeding `site_service_endpoints.remote_host` lookups /// into the data plane. Optional only because the daemon's startup /// path may have failed to read system DNS config; in that case the @@ -446,12 +440,6 @@ pub struct Reconciler { /// unresolved. // r[impl service.site.address] site_resolver: Option>, - /// Set of `(site_service_name, kind)` faults filed last tick. `kind` - /// is the fault kind string. The reconciler diffs this against the - /// current tick's set to clear faults that no longer apply. - // r[impl service.site.address] - prev_site_service_faults: - std::collections::BTreeSet<(seedling_protocol::names::SiteServiceName, &'static str)>, } impl Reconciler { @@ -522,9 +510,7 @@ impl Reconciler { cert_endpoint_url, tls_coordinator, resolver_health_fail_count: std::sync::atomic::AtomicU32::new(0), - prev_ingress_conflicts: std::collections::BTreeSet::new(), site_resolver, - prev_site_service_faults: std::collections::BTreeSet::new(), } } diff --git a/crates/core/src/system/reconcile/faults.rs b/crates/core/src/system/reconcile/faults.rs index 9178f216..96fed334 100644 --- a/crates/core/src/system/reconcile/faults.rs +++ b/crates/core/src/system/reconcile/faults.rs @@ -1,3 +1,5 @@ +use std::collections::BTreeMap; + use seedling_protocol::names::AppName; use super::{Reconciler, pods, volumes}; @@ -464,11 +466,22 @@ impl Reconciler { .iter() .map(|(i, s)| (i.clone(), s.clone())) .collect(); + // r[impl fault.lifecycle] — the file set and the clear set must be + // disjoint per instance. `stop_sent` is recorded before the stop is + // attempted, so an instance whose stop just failed appears in both: + // its `stop_failed` fault was filed and then cleared in the same tick, + // and the operator never saw a stop that keeps failing. + let failed_stops: std::collections::HashSet = update + .stop_failures + .iter() + .map(|(inst, _)| inst.id.to_hex()) + .collect(); let stopped_instances: Vec = update .observations .iter() .filter(|(_, kind, _)| *kind == "stop_sent") .map(|(inst, _, _)| inst.id.to_hex()) + .filter(|hex| !failed_stops.contains(hex)) .collect(); self.db.call(move |db| { Self::file_instance_faults(db, &app, &start_failures, "start_failed"); @@ -558,25 +571,28 @@ impl Reconciler { }); } - /// Reconcile `ingress_conflict` faults against the current - /// `(hostname, port)` collision set. New conflicts get one fault per - /// involved app and one against `_system` for each site ingress; - /// resolved conflicts (in the prior set but not the current) are - /// auto-cleared. + /// Converge `ingress_conflict` faults to the current `(hostname, port)` + /// collision set: one fault per involved app ingress and one against + /// `_system` per involved site ingress, and nothing else. + /// + /// This used to clear only `prior \\ current`, where `prior` was a + /// `Reconciler` field that starts empty on every daemon start — so a + /// conflict fault filed before a restart could never clear, however long + /// the conflict had been resolved. Comparing against the persisted active + /// set instead needs no warm-up state and is restart-safe by construction. // r[impl ingress.site.conflict] + // r[impl fault.lifecycle] pub(super) fn reconcile_ingress_conflicts( &mut self, report: &super::site_proxy::ConflictReport, ) { - // Snapshot the parties for the closure. let parties = report.parties.clone(); - let current = report.conflicts.clone(); - let prior = std::mem::take(&mut self.prev_ingress_conflicts); - - // 1. File faults for current conflicts (idempotent). - let parties_for_file = parties.clone(); self.db.call(move |db| { - for party in &parties_for_file { + let system = AppName::new_unchecked("_system"); + let mut current: BTreeMap = + BTreeMap::new(); + + for party in &parties { let app_summary: Vec = party .apps .iter() @@ -585,104 +601,45 @@ impl Reconciler { let site_summary: Vec = party.site.clone(); let host_port = format!("{}:{}", party.hostname, party.port); - // App side: one fault per involved app+ingress. for (app, ingress_name) in &party.apps { - let already = faults::list_active_faults(db, Some(app)) - .unwrap_or_default() - .iter() - .any(|f| { - f.kind == "ingress_conflict" - && f.resource_name.as_deref() == Some(ingress_name.as_str()) - }); - if already { - continue; - } - let desc = format!( - "ingress conflict on ({host_port}) with site ingress(es) {site_summary:?}; \ - both sides are dropped from the proxy until resolved" + // Subject is the party's own ingress, so one app's fault + // clearing cannot depend on another app's still standing. + current.insert( + faults::FaultKey::new(app, "ingress_conflict", ingress_name.as_str()), + ( + faults::FaultMeta::resource("ingress", ingress_name.as_str()), + format!( + "ingress conflict on ({host_port}) with site ingress(es) \ + {site_summary:?}; both sides are dropped from the proxy until \ + resolved" + ), + ), ); - if let Err(e) = faults::file_fault( - db, - app, - Some("ingress"), - Some(ingress_name.as_str()), - None, - "ingress_conflict", - &desc, - ) { - tracing::warn!(app = %app, ingress = %ingress_name, "failed to file ingress_conflict fault: {e}"); - } } - // Site side: one fault per involved site ingress, scoped - // to the `_system` sentinel app per the existing system - // fault pattern. - let system = AppName::new_unchecked("_system"); for site_name in &party.site { - let already = faults::list_active_faults(db, Some(&system)) - .unwrap_or_default() - .iter() - .any(|f| { - f.kind == "ingress_conflict" - && f.resource_type.as_deref() == Some("site_ingress") - && f.resource_name.as_deref() == Some(site_name.as_str()) - }); - if already { - continue; - } - let desc = format!( - "ingress conflict on ({host_port}) with app ingress(es) {app_summary:?}; \ - both sides are dropped from the proxy until resolved" + current.insert( + faults::FaultKey::new(&system, "ingress_conflict", site_name.as_str()), + ( + faults::FaultMeta::resource("site_ingress", site_name.as_str()), + format!( + "ingress conflict on ({host_port}) with app ingress(es) \ + {app_summary:?}; both sides are dropped from the proxy until \ + resolved" + ), + ), ); - if let Err(e) = faults::file_fault( - db, - &system, - Some("site_ingress"), - Some(site_name.as_str()), - None, - "ingress_conflict", - &desc, - ) { - tracing::warn!(site_ingress = %site_name, "failed to file ingress_conflict fault: {e}"); - } } } - }); - - // 2. Clear faults for resolved conflicts (in prior \ current). - let resolved: Vec<(String, u16)> = prior.difference(¤t).cloned().collect(); - if !resolved.is_empty() { - self.db.call(move |db| { - let system = AppName::new_unchecked("_system"); - for (host, port) in &resolved { - let host_port = format!("{host}:{port}"); - // Sweep both `_system` and any other app whose - // ingress_conflict fault description references the - // resolved tuple. Description match keeps the surface - // narrow without needing to remember which apps were - // involved last tick. - let active = faults::list_active_faults(db, None).unwrap_or_default(); - for f in active { - if f.kind != "ingress_conflict" { - continue; - } - if !f.description.contains(&host_port) { - continue; - } - let owner = if f.resource_type.as_deref() == Some("site_ingress") { - system.clone() - } else { - f.app.clone() - }; - if let Err(e) = faults::clear_fault(db, &f.id, &owner) { - tracing::warn!(fault_id = %f.id, "failed to clear ingress_conflict fault: {e}"); - } - } - } - }); - } - self.prev_ingress_conflicts = current; + if let Err(e) = faults::sync_faults( + db, + &faults::FaultScope::Kind("ingress_conflict".to_owned()), + ¤t, + ) { + tracing::warn!("failed to converge ingress_conflict faults: {e}"); + } + }); } /// File a `site_ingress_target_missing` fault for each unresolved @@ -767,12 +724,15 @@ impl Reconciler { }); } - /// Reconcile `site_service_endpoint_unresolvable` and - /// `site_service_endpoint_unroutable` faults against the current - /// per-service classification. New faults are filed with the failing - /// hosts in the description; stale faults (for services that no - /// longer have any failing endpoint of that kind) auto-clear. + /// Converge `site_service_endpoint_unresolvable` and + /// `site_service_endpoint_unroutable` faults to the current per-service + /// classification, with the failing hosts in the description. + /// + /// Like the ingress-conflict sweep, this used to clear only + /// `prior \\ current` from an in-memory field that empties on restart, so + /// a fault outlived the condition across a daemon restart. // r[impl service.site.address] + // r[impl fault.lifecycle] pub(super) fn reconcile_site_service_faults( &mut self, set: super::phases::SiteServiceFaultSet, @@ -780,111 +740,53 @@ impl Reconciler { const KIND_UNRESOLVABLE: &str = "site_service_endpoint_unresolvable"; const KIND_UNROUTABLE: &str = "site_service_endpoint_unroutable"; - let mut current: std::collections::BTreeSet<( - seedling_protocol::names::SiteServiceName, - &'static str, - )> = std::collections::BTreeSet::new(); - for name in set.unresolvable.keys() { - current.insert((name.clone(), KIND_UNRESOLVABLE)); - } - for name in set.unroutable.keys() { - current.insert((name.clone(), KIND_UNROUTABLE)); - } - - let prior = std::mem::take(&mut self.prev_site_service_faults); - - // 1. File faults for current entries (idempotent). let unresolvable = set.unresolvable.clone(); let unroutable = set.unroutable.clone(); self.db.call(move |db| { let system = AppName::new_unchecked("_system"); + let mut by_kind: BTreeMap<&'static str, BTreeMap> = + BTreeMap::new(); + for (name, hosts) in &unresolvable { - let already = faults::list_active_faults(db, Some(&system)) - .unwrap_or_default() - .iter() - .any(|f| { - f.kind == KIND_UNRESOLVABLE - && f.resource_type.as_deref() == Some("site_service") - && f.resource_name.as_deref() == Some(name.as_str()) - }); - if already { - continue; - } - let desc = format!( - "site service {:?} has DNS-named endpoint(s) that failed to resolve: {}", - name.as_str(), - hosts.join(", "), + by_kind.entry(KIND_UNRESOLVABLE).or_default().insert( + faults::FaultKey::new(&system, KIND_UNRESOLVABLE, name.as_str()), + ( + faults::FaultMeta::resource("site_service", name.as_str()), + format!( + "site service {:?} has DNS-named endpoint(s) that failed to resolve: {}", + name.as_str(), + hosts.join(", "), + ), + ), ); - if let Err(e) = faults::file_fault( - db, - &system, - Some("site_service"), - Some(name.as_str()), - None, - KIND_UNRESOLVABLE, - &desc, - ) { - tracing::warn!(site_service = %name.as_str(), "failed to file site_service_endpoint_unresolvable: {e}"); - } } for (name, hosts) in &unroutable { - let already = faults::list_active_faults(db, Some(&system)) - .unwrap_or_default() - .iter() - .any(|f| { - f.kind == KIND_UNROUTABLE - && f.resource_type.as_deref() == Some("site_service") - && f.resource_name.as_deref() == Some(name.as_str()) - }); - if already { - continue; - } - let desc = format!( - "site service {:?} has endpoint(s) that require NAT64 but NAT64 is not active: {}", - name.as_str(), - hosts.join(", "), + by_kind.entry(KIND_UNROUTABLE).or_default().insert( + faults::FaultKey::new(&system, KIND_UNROUTABLE, name.as_str()), + ( + faults::FaultMeta::resource("site_service", name.as_str()), + format!( + "site service {:?} has endpoint(s) that require NAT64 but NAT64 is not active: {}", + name.as_str(), + hosts.join(", "), + ), + ), ); - if let Err(e) = faults::file_fault( + } + + // Each kind converges within its own scope, so an empty set for + // one kind clears that kind without touching the other. + for kind in [KIND_UNRESOLVABLE, KIND_UNROUTABLE] { + let current = by_kind.remove(kind).unwrap_or_default(); + if let Err(e) = faults::sync_faults( db, - &system, - Some("site_service"), - Some(name.as_str()), - None, - KIND_UNROUTABLE, - &desc, + &faults::FaultScope::AppKind(system.clone(), kind.to_owned()), + ¤t, ) { - tracing::warn!(site_service = %name.as_str(), "failed to file site_service_endpoint_unroutable: {e}"); + tracing::warn!("failed to converge {kind} faults: {e}"); } } }); - - // 2. Clear faults that were in the prior set but not the current. - let resolved: Vec<(seedling_protocol::names::SiteServiceName, &'static str)> = - prior.difference(¤t).cloned().collect(); - if !resolved.is_empty() { - self.db.call(move |db| { - let system = AppName::new_unchecked("_system"); - let active = faults::list_active_faults(db, Some(&system)).unwrap_or_default(); - for (name, kind) in &resolved { - for f in &active { - if f.kind != *kind { - continue; - } - if f.resource_type.as_deref() != Some("site_service") { - continue; - } - if f.resource_name.as_deref() != Some(name.as_str()) { - continue; - } - if let Err(e) = faults::clear_fault(db, &f.id, &system) { - tracing::warn!(fault_id = %f.id, "failed to clear site_service fault: {e}"); - } - } - } - }); - } - - self.prev_site_service_faults = current; } pub(super) fn clear_system_fault(&self, fault_kind: &str) { diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index 63500ca7..bb00aa1e 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -1169,6 +1169,13 @@ Some internal operations (for example [backup.list](#r--backup.list), [backup.re > When a Service's routing pool contains only unhealthy backends (i.e. the prefer-healthy rule has fallen back to "anything running" per [lifecycle.service.routing-pool](#r--lifecycle.service.routing-pool)), the runtime must file a fault of kind `service_degraded` associated with that Service. > The fault is cleared automatically when at least one backend in the pool becomes healthy or when the Service is unscheduled. +> r[fault.lifecycle] +> Every fault kind defines both the condition under which it is filed and the condition under which it clears; a kind with no clearing condition is not permitted. +> A fault identifies the thing that is faulty, not merely the app it belongs to, and at most one fault is active for a given (app, kind, subject) at a time. +> Clearing is keyed no more broadly than filing: an event affecting one subject must not clear a fault held by another. +> A fault whose filing condition is a state of the world ("this is true right now") is active exactly while that condition holds, including across daemon restarts — its clearing must not depend on state held only in memory. +> A fault that is deliberately retained after its trigger has passed must name the lifecycle event that clears it. + > r[fault.surfacing] > Faults must be surfaced to operators through the operator interface (defined in a separate spec). > The runtime must not silently discard faults.