From 202e5a2e109bde2bcb5da2076ebc0e5010e6ee1a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 16:21:02 +0200 Subject: [PATCH] Idunn records continuity exhaustion as a durable incident in incidents.cc Idunn's daemon, at the continuity decision site, writes one idunn.operator_incident.v1 record per exhaustion into its own published incidents.cc (--incident-store), never control.cc. The record opens once per (target, condition), closes on an observed running workload (recovered) or on the target leaving the admitted set, and retires to incident-history.cc seven days after closing. A store failure is traced once and never returned into the tick. idunn status prints each target's incidents. ReportOnce is demoted to a stderr trace. Does not build the fixture test until tests/fixtures/idunn.operator_incident.v1.cc lands in the next commit. Co-Authored-By: Claude Sonnet 5.5 --- deploy/idunn-yggdrasil.service | 1 + docs/guide.md | 3 +- src/control_plane.rs | 428 ++++++++++++++++++++++- src/drivers.rs | 2 +- src/incident.rs | 619 +++++++++++++++++++++++++++++++++ src/lib.rs | 1 + 6 files changed, 1038 insertions(+), 16 deletions(-) create mode 100644 src/incident.rs diff --git a/deploy/idunn-yggdrasil.service b/deploy/idunn-yggdrasil.service index 5474c67..9758127 100644 --- a/deploy/idunn-yggdrasil.service +++ b/deploy/idunn-yggdrasil.service @@ -41,6 +41,7 @@ ExecStart=/usr/local/bin/idunn serve \ --source-root /var/lib/gamecult/idunn/sources \ --staging-root /var/lib/gamecult/idunn/staging \ --topology-store /var/lib/gamecult/idunn-projection/topology.cc \ + --incident-store /var/lib/gamecult/idunn-projection/incidents.cc \ --idunn-identity-store /var/lib/gamecult/idunn/idunn-service-identity.cc \ --odin-correlation-store /var/lib/gamecult/odin/topology.cc \ --odin-trust-anchor /etc/gamecult/idunn/odin-topology-anchor.cc \ diff --git a/docs/guide.md b/docs/guide.md index f018670..e48bc85 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -44,7 +44,8 @@ Idunn centralizes that lifecycle work: - record every deployment/restart request and result as typed state; - record release targets, deployment artifacts, state migrations, and rollout results as typed state; -- escalate to an operator through Bifrost when human action is needed. +- record operator incidents in its published `incidents.cc`, which Bifrost reads + and delivers. Idunn never waits on delivery. The desired shape is simple: diff --git a/src/control_plane.rs b/src/control_plane.rs index 28185cc..db1e6b4 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -44,6 +44,7 @@ use crate::drivers::{ SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, WriteLeasePort, }; +use crate::incident::{CloseReason, IncidentCondition, IncidentStore, render_incidents}; use crate::host_actuator::{ HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver, HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service, @@ -2380,6 +2381,9 @@ struct RuntimeOptions { source_root: PathBuf, staging_root: PathBuf, topology_store: PathBuf, + /// Operator incidents, published world-readable beside the topology + /// projection. Not part of the control store: see `incident`. + incident_store: PathBuf, odin_correlation_store: PathBuf, odin_trust_anchor: PathBuf, idunn_identity_store: PathBuf, @@ -2405,6 +2409,7 @@ impl Default for RuntimeOptions { source_root: PathBuf::from("/var/lib/gamecult/idunn/sources"), staging_root: PathBuf::from("/var/lib/gamecult/idunn/staging"), topology_store: PathBuf::from("/var/lib/gamecult/idunn/topology.cc"), + incident_store: PathBuf::from("/var/lib/gamecult/idunn-projection/incidents.cc"), odin_correlation_store: PathBuf::from( "/var/lib/gamecult/odin/idunn-runtime-topology.cc", ), @@ -2437,6 +2442,7 @@ enum Command { }, Status { state_store: PathBuf, + incident_store: PathBuf, command_id: Option, }, Cancel { @@ -2468,8 +2474,9 @@ pub fn run(args: impl Iterator) -> Result<()> { ), Command::Status { state_store, + incident_store, command_id, - } => status(&state_store, command_id.as_deref()), + } => status(&state_store, &incident_store, command_id.as_deref()), Command::Cancel { state_store, command_id, @@ -2543,6 +2550,7 @@ fn parse_serve(mut args: impl Iterator) -> Result { "--source-root" => options.source_root = path_value(&mut args, &argument)?, "--staging-root" => options.staging_root = path_value(&mut args, &argument)?, "--topology-store" => options.topology_store = path_value(&mut args, &argument)?, + "--incident-store" => options.incident_store = path_value(&mut args, &argument)?, "--odin-correlation-store" => { options.odin_correlation_store = path_value(&mut args, &argument)? } @@ -2626,11 +2634,14 @@ fn parse_up(mut args: impl Iterator) -> Result { } fn parse_status(mut args: impl Iterator) -> Result { - let mut state_store = RuntimeOptions::default().state_store; + let defaults = RuntimeOptions::default(); + let mut state_store = defaults.state_store; + let mut incident_store = defaults.incident_store; let mut command_id = None; while let Some(argument) = args.next() { match argument.as_str() { "--state-store" => state_store = path_value(&mut args, &argument)?, + "--incident-store" => incident_store = path_value(&mut args, &argument)?, "--command" => command_id = Some(string_value(&mut args, &argument)?), "--help" | "-h" => bail!(usage()), _ => bail!("unknown Idunn status option {argument:?}"), @@ -2638,6 +2649,7 @@ fn parse_status(mut args: impl Iterator) -> Result { } Ok(Command::Status { state_store, + incident_store, command_id, }) } @@ -3091,7 +3103,7 @@ impl ControlSnapshot { } } -fn decode_record(envelope: &CultCacheEnvelope) -> Result +pub(crate) fn decode_record(envelope: &CultCacheEnvelope) -> Result where T: for<'de> Deserialize<'de> + Serialize, { @@ -3634,7 +3646,7 @@ fn target_supervision_envelope(value: &TargetSupervision, now: u64) -> Result( +pub(crate) fn typed_envelope( key: &str, record_type: &str, schema: &str, @@ -3792,8 +3804,12 @@ fn state_advanced(before: &DeploymentTransaction, after: &DeploymentTransaction) &after != before } -/// Says a fault once while it lasts. A scheduler tick that hits the same +/// Says a fault once on stderr while it lasts. A scheduler tick that hits the same /// unreadable file every half second must not print it every half second. +/// +/// It is a trace and owns nothing else: it is in memory, keyed by free-form +/// strings, and an Idunn restart forgets it. Whether the operator is told +/// belongs to the incident record (`incident`), never to this map. #[derive(Default)] struct ReportOnce { last: Mutex>, @@ -4096,7 +4112,7 @@ fn submit( } } -fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { +fn status(store_path: &Path, incident_store: &Path, command_id: Option<&str>) -> Result<()> { let snapshot = ControlSnapshot::read(store_path)?; // Commands stay resident; their finished transactions do not. Without // history a completed command would report as though it had never run, @@ -4185,18 +4201,32 @@ fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { // What supervision holds for each target: the union of the admitted // generations and the metered targets, so a target that has only been // deployed to (no generation yet) still shows what it has used. - let now = now_millis()?; - for target in supervised_targets(&snapshot) { - for line in render_supervision( + for line in render_targets(&snapshot, incident_store, now_millis()?) { + println!("{line}"); + } + Ok(()) +} + +/// Every supervised target with its meters and its incidents. An incident +/// store that cannot be read costs the incident lines and nothing else. +fn render_targets(snapshot: &ControlSnapshot, incident_store: &Path, now: u64) -> Vec { + let incidents = IncidentStore::new(incident_store).read(); + let mut lines = Vec::new(); + for target in supervised_targets(snapshot) { + lines.extend(render_supervision( target, snapshot.admitted_for(target).map(|stored| &stored.value), snapshot.supervision_for(target).map(|stored| &stored.value), now, - ) { - println!("{line}"); + )); + if let Ok(records) = &incidents { + lines.extend(render_incidents(records, target, now)); } } - Ok(()) + if let Err(error) = &incidents { + lines.push(format!("incidents unreadable: {}", error.root_cause())); + } + lines } /// Every target supervision holds something for: the admitted generations @@ -4609,6 +4639,57 @@ impl Engine { } } + fn incidents(&self) -> IncidentStore { + IncidentStore::new(&self.options.incident_store) + } + + /// Settle the outcome of one incident operation. A failure never leaves + /// this function: it is traced once, under its operation, and continuity + /// decides exactly as it would have with a healthy store. It names the + /// operation and the root cause and never a path or a value. Incident + /// writes are never progress. + fn record_incident(&self, operation: &str, outcome: Result) { + let key = format!("incident-store:{operation}"); + match outcome { + Ok(_) => self.clear_fault(&key), + Err(error) => self.report_once( + &key, + format!( + "Idunn could not {operation} an operator incident: {}", + error.root_cause() + ), + ), + } + } + + /// Close every open incident whose target is no longer admitted, then + /// retire what has been closed past retention. + fn reconcile_incidents(&self, snapshot: &ControlSnapshot) { + let store = self.incidents(); + let close_unadmitted = || -> Result { + let now = now_millis()?; + let mut closed = false; + for record in store.read()? { + if record.is_open() + && !snapshot + .admitted + .iter() + .any(|stored| stored.value.target == record.subject) + { + closed |= store.close( + record.condition, + &record.subject, + CloseReason::NoLongerAdmitted, + now, + )?; + } + } + Ok(closed) + }; + self.record_incident("close", close_unadmitted()); + self.record_incident("retire", now_millis().and_then(|now| store.retire_closed(now))); + } + fn host_access(&self) -> Result> { let hub = self .host_actuators @@ -5463,6 +5544,7 @@ impl Engine { .lock() .expect("challenge wait mutex") .retain(|target, wait| snapshot.target_stamp(target) == wait.stamp); + self.reconcile_incidents(&snapshot); let mut progressed = false; let mut admitted = snapshot.admitted.iter().collect::>(); admitted.sort_by_key(|stored| stored.value.target.as_str()); @@ -5631,6 +5713,20 @@ impl Engine { } }; if observation.is_some() { + // The admitted workload was seen running: the only evidence + // that ends an exhaustion incident. The restart window sliding + // is not health and never closes one. + self.record_incident( + "close", + now_millis().and_then(|now| { + self.incidents().close( + IncidentCondition::ContinuityExhausted, + ¤t.value.target, + CloseReason::Recovered, + now, + ) + }), + ); continue; } let workload_error = operational_error @@ -5707,6 +5803,16 @@ impl Engine { continue; } if supervision.restarts_exhausted(now) { + // The record comes first and the trace after it: the trace is + // this record's shadow, never its substitute. + self.record_incident( + "open", + self.incidents().open( + IncidentCondition::ContinuityExhausted, + ¤t.value.target, + now, + ), + ); self.report_once( &continuity_key, format!( @@ -9013,7 +9119,7 @@ fn require_selector(value: &str) -> Result<()> { require_id(value, "deployment selector") } -fn require_id(value: &str, label: &str) -> Result<()> { +pub(crate) fn require_id(value: &str, label: &str) -> Result<()> { ensure!( !value.is_empty() && value.len() <= 256 @@ -9070,7 +9176,7 @@ fn usage() -> &'static str { "Idunn deployment, admission, and continuity control plane\n\n\ idunn serve [runtime options] [--host-actuator-bind ADDR]\n\ idunn up [--state-store PATH] [--no-wait]\n\ - idunn status [--state-store PATH] [--command ID]\n\ + idunn status [--state-store PATH] [--incident-store PATH] [--command ID]\n\ idunn cancel [--state-store PATH]\n\ idunn validate --recipe PATH [--binding PATH]\n\n\ Recipes describe capability and process requirements. Idunn seals exact\n\ @@ -9089,6 +9195,7 @@ mod tests { use super::*; use crate::drivers::{SystemdWorkloadObservation, incarnation_key}; + use crate::incident::INCIDENT_RETENTION_MILLIS; /// Fixed clock for the signing fixture; correlations must land inside /// DEFAULT_TOPOLOGY_MAXIMUM_AGE_MILLIS of it to authenticate. @@ -9303,6 +9410,7 @@ mod tests { source_root: root.join("sources"), staging_root: root.join("staging"), topology_store: root.join("topology.cc"), + incident_store: root.join("incidents.cc"), odin_correlation_store: root.join("odin-correlation.cc"), odin_trust_anchor: odin_anchor, idunn_identity_store: idunn, @@ -12760,6 +12868,11 @@ mod tests { }) } + fn revive(&self) { + self.alive + .store(true, std::sync::atomic::Ordering::SeqCst); + } + fn kill(&self) { self.alive .store(false, std::sync::atomic::Ordering::SeqCst); @@ -14498,6 +14611,293 @@ mod tests { Ok(()) } + // ---- operator incidents ---------------------------------------------- + + fn incident_path(world: &EngineFixture) -> PathBuf { + world.root.join("incidents.cc") + } + + fn incidents_of(world: &EngineFixture) -> Result> { + IncidentStore::new(&incident_path(world)).read() + } + + /// The control store never holds an incident: it would stop every older + /// binary that reads it. + fn assert_control_store_holds_no_incident(world: &EngineFixture) -> Result<()> { + ControlSnapshot::read(&world.state_store)?; + for envelope in SingleFileMessagePackBackingStore::new(&world.state_store) + .pull_all_read_only_snapshot()? + { + assert_ne!(envelope.r#type, "idunn.operator_incident"); + } + Ok(()) + } + + /// Six restarts inside the window, the last long enough ago that its wait is + /// over: only the ceiling holds. + fn spend_the_window(world: &EngineFixture) -> Result<()> { + set_meters( + world, + &restarted_ago(&[3_000_000, 2_400_000, 1_800_000, 1_200_000, 600_000, 200_000])?, + ) + } + + /// One entry has left the window, and the newest restart is recent enough + /// that continuity is still waiting: exhaustion is over, nothing restarts. + fn slide_the_window(world: &EngineFixture) -> Result<()> { + set_meters( + world, + &restarted_ago(&[3_700_000, 2_400_000, 1_800_000, 1_200_000, 600_000, 10_000])?, + ) + } + + #[test] + fn an_exhausted_target_opens_one_incident_across_ticks_and_restarts() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + workload.kill(); + spend_the_window(&world)?; + assert!(!incident_path(&world).exists()); + + assert!(!world.engine.supervise_one_admitted_generation()?); + let opened = incidents_of(&world)?; + assert_eq!(opened.len(), 1); + assert_eq!(opened[0].subject, "service"); + assert_eq!(opened[0].condition, IncidentCondition::ContinuityExhausted); + assert!(opened[0].is_open()); + let written = std::fs::read(incident_path(&world))?; + + for _ in 0..3 { + assert!(!world.engine.supervise_one_admitted_generation()?); + } + // An Idunn restart forgets its in-memory trace; the durable record + // keeps the decision from being written again. + let restarted = + Engine::open_with_systemd_workload(world.engine.options.clone(), workload.clone())?; + assert!(restarted.fault_reports.lock().unwrap().is_empty()); + assert!(!restarted.supervise_one_admitted_generation()?); + assert_eq!(std::fs::read(incident_path(&world))?, written); + assert_eq!(incidents_of(&world)?, opened); + assert!(ControlSnapshot::read(&world.state_store)?.transactions.is_empty()); + assert_control_store_holds_no_incident(&world) + } + + #[test] + fn an_exhaustion_incident_closes_on_observed_health_not_on_the_window() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + workload.kill(); + spend_the_window(&world)?; + world.engine.supervise_one_admitted_generation()?; + let first = incidents_of(&world)?; + assert_eq!(first.len(), 1); + + // The window slides while the workload is still dead: exhaustion is + // over, the target is not healthy, and the incident stays open. + slide_the_window(&world)?; + assert!(!world.engine.supervise_one_admitted_generation()?); + assert_eq!(incidents_of(&world)?, first); + + // Seen running, it closes as recovered. + workload.revive(); + assert!(!world.engine.supervise_one_admitted_generation()?); + let closed = incidents_of(&world)?; + assert_eq!(closed.len(), 1); + assert_eq!(closed[0].incident_key, first[0].incident_key); + assert!( + closed[0] + .closed_at_unix_millis + .is_some_and(|at| at >= closed[0].opened_at_unix_millis) + ); + assert_eq!(closed[0].close_reason, Some(CloseReason::Recovered)); + + // A later exhaustion is a new record with a new opening time. + std::thread::sleep(Duration::from_millis(5)); + workload.kill(); + spend_the_window(&world)?; + world.engine.supervise_one_admitted_generation()?; + let again = incidents_of(&world)?; + assert_eq!(again.len(), 2); + let reopened = again + .iter() + .find(|record| record.is_open()) + .context("no new incident")?; + assert!(reopened.opened_at_unix_millis > first[0].opened_at_unix_millis); + assert_ne!(reopened.incident_key, first[0].incident_key); + assert_control_store_holds_no_incident(&world) + } + + #[test] + fn an_incident_whose_target_is_no_longer_admitted_closes() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + workload.kill(); + spend_the_window(&world)?; + world.engine.supervise_one_admitted_generation()?; + assert!(incidents_of(&world)?[0].is_open()); + + let admitted = ControlSnapshot::read(&world.state_store)? + .admitted_for("service") + .context("no admitted generation")? + .envelope + .clone(); + assert!( + SingleFileMessagePackBackingStore::new(&world.state_store) + .delete_batch_if_unchanged(&[admitted])? + ); + world.engine.supervise_one_admitted_generation()?; + let records = incidents_of(&world)?; + assert_eq!(records.len(), 1); + assert_eq!(records[0].close_reason, Some(CloseReason::NoLongerAdmitted)); + assert_control_store_holds_no_incident(&world) + } + + #[test] + fn a_tick_retires_incidents_closed_past_retention() -> Result<()> { + let world = EngineFixture::with_workload(SwitchWorkload::new())?; + let store = IncidentStore::new(&incident_path(&world)); + let condition = IncidentCondition::ContinuityExhausted; + let long_ago = now_millis()? - INCIDENT_RETENTION_MILLIS - 10_000; + assert!(store.open(condition, "gone", long_ago)?); + assert!(store.close(condition, "gone", CloseReason::Recovered, long_ago + 1)?); + assert!(store.open(condition, "recent", now_millis()?)?); + assert!(store.close(condition, "recent", CloseReason::Recovered, now_millis()?)?); + + world.engine.supervise_one_admitted_generation()?; + let live = store.read()?; + assert_eq!(live.len(), 1); + assert_eq!(live[0].subject, "recent"); + assert!(world.root.join("incident-history.cc").exists()); + Ok(()) + } + + /// The same exhaustion and recovery sequence, reduced to what continuity + /// decided: each tick's return, and the transactions and restart log it left. + fn continuity_decisions( + world: &EngineFixture, + engine: &Engine, + workload: &SwitchWorkload, + ) -> Result> { + let mut decided = Vec::new(); + let tick = |decided: &mut Vec<(bool, usize, usize)>| -> Result<()> { + let moved = engine.supervise_one_admitted_generation()?; + let snapshot = ControlSnapshot::read(&world.state_store)?; + decided.push(( + moved, + snapshot.transactions.len(), + snapshot.supervision_or_new("service").continuity_restarts.len(), + )); + Ok(()) + }; + workload.kill(); + spend_the_window(world)?; + tick(&mut decided)?; + tick(&mut decided)?; + slide_the_window(world)?; + tick(&mut decided)?; + workload.revive(); + tick(&mut decided)?; + workload.kill(); + spend_the_window(world)?; + tick(&mut decided)?; + Ok(decided) + } + + #[test] + fn an_unwritable_incident_store_changes_nothing_continuity_decides() -> Result<()> { + const CANARY: &str = "CANARY-6b1d-secret-segment"; + let healthy_workload = SwitchWorkload::new(); + let healthy = EngineFixture::with_workload(healthy_workload.clone())?; + admit_incumbent(&healthy)?; + let expected = continuity_decisions(&healthy, &healthy.engine, &healthy_workload)?; + assert!(incidents_of(&healthy)?.iter().any(|record| record.is_open())); + + // Not permission-dependent: a store path beneath a regular file cannot + // be created for any uid, root included. + let broken_workload = SwitchWorkload::new(); + let mut broken = EngineFixture::with_workload(broken_workload.clone())?; + admit_incumbent(&broken)?; + let blocker = broken.root.join(CANARY); + std::fs::write(&blocker, b"a file, not a directory")?; + let mut options = broken.engine.options.clone(); + options.incident_store = blocker.join("incidents.cc"); + broken.engine = Engine::open_with_systemd_workload(options, broken_workload.clone())?; + + let actual = continuity_decisions(&broken, &broken.engine, &broken_workload)?; + assert_eq!(actual, expected, "a broken incident store changed a continuity decision"); + assert!(!blocker.join("incidents.cc").exists()); + + // The fault is traced once, under its operation, and echoes no path. + let reports = broken.engine.fault_reports.lock().unwrap(); + let faults = reports + .iter() + .filter(|(key, _)| key.starts_with("incident-store:")) + .collect::>(); + assert_eq!( + faults.len(), + 1, + "{:?}", + faults.iter().map(|(key, _)| key).collect::>() + ); + let message = faults[0].1.last.lock().unwrap().clone().context("fault cleared")?; + assert!(message.contains("open an operator incident"), "{message}"); + assert!(!message.contains(CANARY), "{message}"); + drop(reports); + assert_control_store_holds_no_incident(&broken) + } + + #[test] + fn status_renders_incidents_and_survives_an_unreadable_store() -> Result<()> { + const CANARY: &str = "CANARY-90c2-secret-segment"; + let temp = TempDir::new()?; + let now = now_millis()?; + let mut snapshot = ControlSnapshot::default(); + let meters = TargetSupervision::new("service"); + snapshot.targets.push(Stored { + envelope: target_supervision_envelope(&meters, now)?, + value: meters, + }); + + let readable = temp.path().join("incidents.cc"); + IncidentStore::new(&readable).open(IncidentCondition::ContinuityExhausted, "service", now)?; + let lines = render_targets(&snapshot, &readable, now); + assert!(lines[0].starts_with("target service "), "{lines:?}"); + assert_eq!( + lines.last().map(String::as_str), + Some(format!(" incident continuity-exhausted opened-at {now}").as_str()) + ); + + let unreadable = temp.path().join(CANARY); + std::fs::write(&unreadable, b"not a cultcache file")?; + let lines = render_targets(&snapshot, &unreadable, now); + assert!(lines[0].starts_with("target service "), "{lines:?}"); + let last = lines.last().context("no lines")?; + assert!(last.starts_with("incidents unreadable: "), "{last}"); + assert!(!lines.iter().any(|line| line.contains(CANARY)), "{lines:?}"); + Ok(()) + } + + #[test] + fn the_incident_store_is_named_by_serve_and_status() { + let Command::Serve(options) = + parse(["serve", "--incident-store", "/tmp/i.cc"].into_iter().map(str::to_owned)) + .unwrap() + else { + panic!("expected serve") + }; + assert_eq!(options.incident_store, PathBuf::from("/tmp/i.cc")); + let Command::Status { incident_store, .. } = + parse(["status", "--incident-store", "/tmp/j.cc"].into_iter().map(str::to_owned)) + .unwrap() + else { + panic!("expected status") + }; + assert_eq!(incident_store, PathBuf::from("/tmp/j.cc")); + } + #[test] fn continuity_stops_when_the_targets_own_log_has_spent_the_window() -> Result<()> { let workload = SwitchWorkload::new(); diff --git a/src/drivers.rs b/src/drivers.rs index 189779b..eb162e9 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -7667,7 +7667,7 @@ fn ensure_bundle_is_reachable_by_workload( /// gets EACCES. A chmod by hand does not hold, because each publish writes a /// new file. Integrity here comes from the signatures over the records, not /// from the mode, so the published copy is readable. -fn publish_projection_mode(path: &Path) -> Result<()> { +pub(crate) fn publish_projection_mode(path: &Path) -> Result<()> { #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; diff --git a/src/incident.rs b/src/incident.rs new file mode 100644 index 0000000..fa6f1a6 --- /dev/null +++ b/src/incident.rs @@ -0,0 +1,619 @@ +//! Operator incidents: the durable record that a condition needing a human +//! opened, and that it closed. +//! +//! One file, `incidents.cc`, holds one type. Idunn's daemon is its only writer, +//! at the site that decides the condition; it is also the file readers such as +//! Bifrost open, so no second copy exists to drift. It is never `control.cc`, +//! which refuses any type it does not know: a type added there would stop +//! continuity for every target on a rollback. +//! +//! A record carries no free text. A workload's error never enters it, so the +//! published file leaks nothing a reader should not have. + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result, ensure}; +use cultcache_rs::{ + CultCacheEnvelope, CultCacheExpectedEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, +}; +use serde::{Deserialize, Serialize}; + +use crate::control_plane::{decode_record, require_id, typed_envelope}; +use crate::drivers::publish_projection_mode; + +const INCIDENT_SCHEMA: &str = "idunn.operator_incident.v1"; + +/// How long a closed incident stays in the published file before it retires to +/// the incident history. +pub(crate) const INCIDENT_RETENTION_MILLIS: u64 = 7 * 24 * 60 * 60 * 1000; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum IncidentCondition { + ContinuityExhausted, +} + +impl IncidentCondition { + /// The spelling in the key and on `idunn status`. Pinned equal to the + /// serialized spelling by `the_incident_fixture_is_what_idunn_reads`. + pub(crate) const fn name(self) -> &'static str { + match self { + Self::ContinuityExhausted => "continuity-exhausted", + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum CloseReason { + Recovered, + NoLongerAdmitted, +} + +impl CloseReason { + pub(crate) const fn name(self) -> &'static str { + match self { + Self::Recovered => "recovered", + Self::NoLongerAdmitted => "no-longer-admitted", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache(type = "idunn.operator_incident", schema = "idunn.operator_incident.v1")] +pub(crate) struct IncidentRecord { + #[cultcache(key = 0)] + pub(crate) schema_version: String, + /// `::`. Injective because at most one + /// incident per `(condition, subject)` is open at a time. + #[cultcache(key = 1)] + pub(crate) incident_key: String, + #[cultcache(key = 2)] + pub(crate) condition: IncidentCondition, + /// The target the condition is about. + #[cultcache(key = 3)] + pub(crate) subject: String, + #[cultcache(key = 4)] + pub(crate) opened_at_unix_millis: u64, + #[cultcache(key = 5)] + pub(crate) closed_at_unix_millis: Option, + #[cultcache(key = 6)] + pub(crate) close_reason: Option, +} + +impl IncidentRecord { + fn opened(condition: IncidentCondition, subject: &str, now: u64) -> Self { + Self { + schema_version: INCIDENT_SCHEMA.to_owned(), + incident_key: format!("{}:{subject}:{now}", condition.name()), + condition, + subject: subject.to_owned(), + opened_at_unix_millis: now, + closed_at_unix_millis: None, + close_reason: None, + } + } + + pub(crate) fn is_open(&self) -> bool { + self.closed_at_unix_millis.is_none() + } + + pub(crate) fn validate(&self) -> Result<()> { + ensure!( + self.schema_version == INCIDENT_SCHEMA, + "incident schema is unsupported" + ); + require_id(&self.subject, "incident subject")?; + ensure!( + self.incident_key + == format!( + "{}:{}:{}", + self.condition.name(), + self.subject, + self.opened_at_unix_millis + ), + "incident key is not its condition, subject and opening time" + ); + ensure!( + self.closed_at_unix_millis.is_some() == self.close_reason.is_some(), + "incident has a closing time or a reason but not both" + ); + if let Some(closed_at) = self.closed_at_unix_millis { + ensure!( + closed_at >= self.opened_at_unix_millis, + "incident closes before it opens" + ); + } + Ok(()) + } + + fn envelope(&self, now: u64) -> Result { + self.validate()?; + typed_envelope( + &self.incident_key, + Self::TYPE, + INCIDENT_SCHEMA, + self, + now, + ) + } +} + +struct Stored { + envelope: CultCacheEnvelope, + record: IncidentRecord, +} + +/// The incident file and its history sibling. Every write is a compare-exchange +/// against exactly what was read, and every write is followed by the +/// world-readable publication mode. +pub(crate) struct IncidentStore { + path: PathBuf, +} + +impl IncidentStore { + pub(crate) fn new(path: &Path) -> Self { + Self { + path: path.to_path_buf(), + } + } + + fn history_path(&self) -> PathBuf { + self.path.with_file_name("incident-history.cc") + } + + fn backing(&self) -> SingleFileMessagePackBackingStore { + SingleFileMessagePackBackingStore::new(&self.path) + } + + /// Every record, oldest key first. An absent file is empty. The file + /// refuses any document that is not an operator incident. + pub(crate) fn read(&self) -> Result> { + Ok(self + .read_stored()? + .into_iter() + .map(|stored| stored.record) + .collect()) + } + + fn read_stored(&self) -> Result> { + if !self.path.exists() { + return Ok(Vec::new()); + } + let mut stored = Vec::new(); + for envelope in self + .backing() + .pull_all_read_only_snapshot() + .context("reading the incident store")? + { + ensure!( + envelope.r#type == IncidentRecord::TYPE + && envelope.schema_id.as_deref() == Some(INCIDENT_SCHEMA), + "the incident store holds a document that is not an operator incident" + ); + let record: IncidentRecord = decode_record(&envelope)?; + record.validate()?; + ensure!( + envelope.key == record.incident_key, + "incident store key differs from the incident's identity" + ); + stored.push(Stored { envelope, record }); + } + stored.sort_by(|left, right| left.record.incident_key.cmp(&right.record.incident_key)); + Ok(stored) + } + + fn open_record<'a>( + stored: &'a [Stored], + condition: IncidentCondition, + subject: &str, + ) -> Option<&'a Stored> { + stored.iter().find(|stored| { + stored.record.is_open() + && stored.record.condition == condition + && stored.record.subject == subject + }) + } + + /// Open an incident unless one is already open for `(condition, subject)`. + /// `Ok(true)` means this call wrote it. + pub(crate) fn open( + &self, + condition: IncidentCondition, + subject: &str, + now: u64, + ) -> Result { + let stored = self.read_stored()?; + if Self::open_record(&stored, condition, subject).is_some() { + return Ok(false); + } + let record = IncidentRecord::opened(condition, subject, now); + let written = self.backing().compare_exchange( + &[CultCacheExpectedEnvelope { + r#type: IncidentRecord::TYPE.into(), + key: record.incident_key.clone(), + current: None, + }], + &[record.envelope(now)?], + )?; + if written { + publish_projection_mode(&self.path)?; + } + Ok(written) + } + + /// Close the open incident for `(condition, subject)`, if there is one. + /// `Ok(true)` means this call wrote the closure. + pub(crate) fn close( + &self, + condition: IncidentCondition, + subject: &str, + reason: CloseReason, + now: u64, + ) -> Result { + let stored = self.read_stored()?; + let Some(open) = Self::open_record(&stored, condition, subject) else { + return Ok(false); + }; + let mut closed = open.record.clone(); + // A clock stepped back must not close an incident before it opened. + closed.closed_at_unix_millis = Some(now.max(closed.opened_at_unix_millis)); + closed.close_reason = Some(reason); + let written = self.backing().compare_exchange( + &[CultCacheExpectedEnvelope { + r#type: IncidentRecord::TYPE.into(), + key: closed.incident_key.clone(), + current: Some(open.envelope.clone()), + }], + &[closed.envelope(now)?], + )?; + if written { + publish_projection_mode(&self.path)?; + } + Ok(written) + } + + /// Move every incident closed more than `INCIDENT_RETENTION_MILLIS` ago to + /// the history file, then delete it here. History first: a crash between + /// the two leaves the record in both, and `insert_entry_if_absent` makes + /// the repeat a no-op. An open incident never moves. + pub(crate) fn retire_closed(&self, now: u64) -> Result { + let retiring = self + .read_stored()? + .into_iter() + .filter(|stored| { + stored.record.closed_at_unix_millis.is_some_and(|closed_at| { + now.saturating_sub(closed_at) > INCIDENT_RETENTION_MILLIS + }) + }) + .map(|stored| stored.envelope) + .collect::>(); + if retiring.is_empty() { + return Ok(false); + } + let history = SingleFileMessagePackBackingStore::new(&self.history_path()); + for envelope in &retiring { + history + .insert_entry_if_absent(envelope.clone()) + .context("archiving a closed incident")?; + } + let deleted = self + .backing() + .delete_batch_if_unchanged(&retiring) + .context("retiring a closed incident")?; + if deleted { + publish_projection_mode(&self.path)?; + } + Ok(deleted) + } +} + +/// The lines `idunn status` prints for one target's incidents: each open one, +/// and each closed one still inside retention. +pub(crate) fn render_incidents(records: &[IncidentRecord], target: &str, now: u64) -> Vec { + records + .iter() + .filter(|record| record.subject == target) + .filter(|record| { + record.closed_at_unix_millis.is_none_or(|closed_at| { + now.saturating_sub(closed_at) <= INCIDENT_RETENTION_MILLIS + }) + }) + .map(|record| { + let mut line = format!( + " incident {} opened-at {}", + record.condition.name(), + record.opened_at_unix_millis + ); + if let (Some(closed_at), Some(reason)) = + (record.closed_at_unix_millis, record.close_reason) + { + line.push_str(&format!(" closed-at {closed_at} {}", reason.name())); + } + line + }) + .collect() +} + +#[cfg(test)] +mod tests { + use tempfile::TempDir; + + use super::*; + + const OPENED: u64 = 1_700_000_000_000; + const CLOSED: u64 = 1_700_000_100_000; + const FIXTURE: &str = "tests/fixtures/idunn.operator_incident.v1.cc"; + + fn fixture_path() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join(FIXTURE) + } + + /// The two records the fixture holds, spelled independently of the store: + /// one open, one closed. + fn fixture_records() -> [IncidentRecord; 2] { + let closed_key = format!("continuity-exhausted:closed-target:{OPENED}"); + let open_key = format!("continuity-exhausted:open-target:{CLOSED}"); + [ + IncidentRecord { + schema_version: "idunn.operator_incident.v1".into(), + incident_key: closed_key, + condition: IncidentCondition::ContinuityExhausted, + subject: "closed-target".into(), + opened_at_unix_millis: OPENED, + closed_at_unix_millis: Some(CLOSED), + close_reason: Some(CloseReason::Recovered), + }, + IncidentRecord { + schema_version: "idunn.operator_incident.v1".into(), + incident_key: open_key, + condition: IncidentCondition::ContinuityExhausted, + subject: "open-target".into(), + opened_at_unix_millis: CLOSED, + closed_at_unix_millis: None, + close_reason: None, + }, + ] + } + + #[test] + fn incident_records_validate_their_key_and_closure() { + let [closed, open] = fixture_records(); + closed.validate().unwrap(); + open.validate().unwrap(); + + let mut wrong_key = open.clone(); + wrong_key.incident_key.push('x'); + assert!(wrong_key.validate().is_err()); + + let mut reason_missing = open.clone(); + reason_missing.closed_at_unix_millis = Some(CLOSED + 1); + assert!(reason_missing.validate().is_err()); + + let mut closing_missing = open.clone(); + closing_missing.close_reason = Some(CloseReason::Recovered); + assert!(closing_missing.validate().is_err()); + + let mut before_open = closed.clone(); + before_open.closed_at_unix_millis = Some(OPENED - 1); + assert!(before_open.validate().is_err()); + + let mut empty_subject = open.clone(); + empty_subject.subject = String::new(); + empty_subject.incident_key = format!("continuity-exhausted::{CLOSED}"); + assert!(empty_subject.validate().is_err()); + + let mut wrong_schema = open; + wrong_schema.schema_version = "idunn.operator_incident.v0".into(); + assert!(wrong_schema.validate().is_err()); + } + + #[test] + fn the_incident_fixture_is_what_idunn_reads() { + if std::env::var_os("IDUNN_WRITE_FIXTURES").is_some() { + let temp = TempDir::new().unwrap(); + let store = IncidentStore::new(&temp.path().join("incidents.cc")); + assert!( + store + .open(IncidentCondition::ContinuityExhausted, "closed-target", OPENED) + .unwrap() + ); + assert!( + store + .close( + IncidentCondition::ContinuityExhausted, + "closed-target", + CloseReason::Recovered, + CLOSED + ) + .unwrap() + ); + assert!( + store + .open(IncidentCondition::ContinuityExhausted, "open-target", CLOSED) + .unwrap() + ); + std::fs::copy(temp.path().join("incidents.cc"), fixture_path()).unwrap(); + } + // A copy, so reading the committed file can never leave a lock beside it. + let temp = TempDir::new().unwrap(); + let copy = temp.path().join("incidents.cc"); + std::fs::copy(fixture_path(), ©).unwrap(); + assert_eq!( + IncidentStore::new(©).read().unwrap(), + fixture_records().to_vec() + ); + // The wire spelling readers match on is the spelling the key and + // status use. + let payload = rmp_serde::to_vec(&IncidentCondition::ContinuityExhausted).unwrap(); + assert_eq!( + rmp_serde::from_slice::(&payload).unwrap(), + IncidentCondition::ContinuityExhausted.name() + ); + let payload = rmp_serde::to_vec(&CloseReason::NoLongerAdmitted).unwrap(); + assert_eq!( + rmp_serde::from_slice::(&payload).unwrap(), + CloseReason::NoLongerAdmitted.name() + ); + } + + #[test] + fn the_store_opens_once_closes_once_and_reopens_as_a_new_record() { + let temp = TempDir::new().unwrap(); + let store = IncidentStore::new(&temp.path().join("incidents.cc")); + let condition = IncidentCondition::ContinuityExhausted; + assert!(store.read().unwrap().is_empty()); + assert!(!store.close(condition, "service", CloseReason::Recovered, OPENED).unwrap()); + + assert!(store.open(condition, "service", OPENED).unwrap()); + assert!(!store.open(condition, "service", OPENED + 5).unwrap()); + assert!(store.open(condition, "other", OPENED).unwrap()); + + assert!(store.close(condition, "service", CloseReason::Recovered, CLOSED).unwrap()); + assert!(!store.close(condition, "service", CloseReason::Recovered, CLOSED).unwrap()); + + assert!(store.open(condition, "service", CLOSED + 1).unwrap()); + let records = store.read().unwrap(); + assert_eq!(records.len(), 3); + let keys = records.iter().map(|r| r.incident_key.as_str()).collect::>(); + assert!(keys.contains(&format!("continuity-exhausted:service:{OPENED}").as_str())); + assert!(keys.contains(&format!("continuity-exhausted:service:{}", CLOSED + 1).as_str())); + + // A clock stepped back closes at the opening time, never before it. + assert!(store.close(condition, "other", CloseReason::NoLongerAdmitted, OPENED - 9).unwrap()); + let other = store.read().unwrap().into_iter().find(|r| r.subject == "other").unwrap(); + assert_eq!(other.closed_at_unix_millis, Some(OPENED)); + assert_eq!(other.close_reason, Some(CloseReason::NoLongerAdmitted)); + } + + #[test] + fn closed_incidents_retire_to_history_after_retention() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + for subject in ["old", "young", "open"] { + assert!(store.open(condition, subject, OPENED).unwrap()); + } + assert!(store.close(condition, "old", CloseReason::Recovered, OPENED + 1).unwrap()); + let young_closed = OPENED + INCIDENT_RETENTION_MILLIS; + assert!(store.close(condition, "young", CloseReason::Recovered, young_closed).unwrap()); + + // Exactly at the boundary nothing has aged out; one millisecond later + // only the older closure has. + let boundary = OPENED + 1 + INCIDENT_RETENTION_MILLIS; + assert!(!store.retire_closed(boundary).unwrap()); + assert!(store.retire_closed(boundary + 1).unwrap()); + assert!(!store.retire_closed(boundary + 1).unwrap()); + + let live = store.read().unwrap(); + let mut subjects = live.iter().map(|r| r.subject.as_str()).collect::>(); + subjects.sort_unstable(); + assert_eq!(subjects, ["open", "young"]); + + let history = SingleFileMessagePackBackingStore::new(&temp.path().join("incident-history.cc")) + .pull_all_read_only_snapshot() + .unwrap(); + assert_eq!(history.len(), 1); + assert_eq!(history[0].key, format!("continuity-exhausted:old:{OPENED}")); + + // Later the younger one retires too; the open one never does. + assert!(store.retire_closed(young_closed + INCIDENT_RETENTION_MILLIS + 1).unwrap()); + let live = store.read().unwrap(); + assert_eq!(live.len(), 1); + assert_eq!(live[0].subject, "open"); + } + + #[test] + fn the_store_refuses_any_document_that_is_not_an_incident() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let foreign = CultCacheEnvelope { + key: "x".into(), + r#type: "idunn.target_supervision".into(), + payload: vec![0x90], + stored_at: "2026-01-01T00:00:00+00:00".into(), + schema_id: Some("idunn.target_supervision.v1".into()), + }; + SingleFileMessagePackBackingStore::new(&path) + .insert_entry_if_absent(foreign) + .unwrap(); + let store = IncidentStore::new(&path); + assert!(store.read().is_err()); + assert!( + store + .open(IncidentCondition::ContinuityExhausted, "service", OPENED) + .is_err() + ); + } + + #[test] + fn status_renders_open_and_recently_closed_incidents_for_their_target() { + let [closed, open] = fixture_records(); + let records = [closed, open]; + let lines = render_incidents(&records, "closed-target", CLOSED + 1); + assert_eq!( + lines, + [format!( + " incident continuity-exhausted opened-at {OPENED} closed-at {CLOSED} recovered" + )] + ); + assert_eq!( + render_incidents(&records, "open-target", CLOSED + 1), + [format!(" incident continuity-exhausted opened-at {CLOSED}")] + ); + assert!(render_incidents(&records, "unrelated", CLOSED + 1).is_empty()); + // A closed incident past retention is history, not status. + assert!( + render_incidents(&records, "closed-target", CLOSED + INCIDENT_RETENTION_MILLIS + 1) + .is_empty() + ); + assert_eq!( + render_incidents(&records, "closed-target", CLOSED + INCIDENT_RETENTION_MILLIS).len(), + 1 + ); + } + + /// Idunn runs with `UMask=027`, so a file it creates lands 0640 unless the + /// store publishes it. The umask is process-wide, so it is held only around + /// each write and only one test at a time may hold it. + #[cfg(unix)] + fn under_service_umask(write: impl FnOnce() -> T) -> T { + static UMASK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _held = UMASK.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + // SAFETY: umask only swaps the process file-creation mask. + let previous = unsafe { libc::umask(0o027) }; + let result = write(); + unsafe { libc::umask(previous) }; + result + } + + #[cfg(unix)] + #[test] + fn published_incident_store_is_world_readable() { + use std::os::unix::fs::PermissionsExt; + + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let lock = temp.path().join("incidents.cc.lock"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + let mode = |path: &Path| std::fs::metadata(path).unwrap().permissions().mode() & 0o777; + + assert!(under_service_umask(|| store.open(condition, "service", OPENED)).unwrap()); + assert_eq!((mode(&path), mode(&lock)), (0o644, 0o644), "after open"); + + // Each write replaces the file, and the replacement is published too. + assert!( + under_service_umask(|| store.close(condition, "service", CloseReason::Recovered, CLOSED)) + .unwrap() + ); + assert_eq!((mode(&path), mode(&lock)), (0o644, 0o644), "after close"); + + assert!( + under_service_umask(|| store.retire_closed(CLOSED + INCIDENT_RETENTION_MILLIS + 1)) + .unwrap() + ); + assert_eq!(mode(&path), 0o644, "after retire"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 7b3fd60..b96d835 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,4 +4,5 @@ pub mod deployment_plan; pub mod drivers; pub mod host; pub mod host_actuator; +mod incident; pub mod provisioning;