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 2c5fb3d..d3edbf8 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -1,6 +1,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::env; use std::fs::{self, File, OpenOptions}; +use std::io::Write; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -44,6 +45,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, @@ -2505,6 +2507,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, @@ -2530,6 +2535,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", ), @@ -2562,6 +2568,7 @@ enum Command { }, Status { state_store: PathBuf, + incident_store: PathBuf, command_id: Option, }, Cancel { @@ -2598,8 +2605,14 @@ pub fn run(args: impl Iterator) -> Result<()> { ), Command::Status { state_store, + incident_store, command_id, - } => status(&state_store, command_id.as_deref()), + } => status( + &mut std::io::stdout().lock(), + &state_store, + &incident_store, + command_id.as_deref(), + ), Command::Cancel { state_store, command_id, @@ -2679,6 +2692,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)? } @@ -2762,11 +2776,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:?}"), @@ -2774,6 +2791,7 @@ fn parse_status(mut args: impl Iterator) -> Result { } Ok(Command::Status { state_store, + incident_store, command_id, }) } @@ -3303,7 +3321,7 @@ impl ControlSnapshot { } } -fn decode_record(envelope: &CultCacheEnvelope) -> Result +pub(crate) fn decode_record(envelope: &CultCacheEnvelope) -> Result where T: for<'de> Deserialize<'de> + Serialize, { @@ -3846,7 +3864,7 @@ fn target_supervision_envelope(value: &TargetSupervision, now: u64) -> Result( +pub(crate) fn typed_envelope( key: &str, record_type: &str, schema: &str, @@ -4004,8 +4022,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>, @@ -4308,7 +4330,12 @@ fn submit( } } -fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { +fn status( + out: &mut impl Write, + 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, @@ -4336,10 +4363,11 @@ fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { ); transactions.sort_by_key(|value| value.ordinal); let (state, detail) = derived_command_status(&transactions); - println!( + writeln!( + out, "{} {} {} {}", command.command_id, command.selector, state, detail - ); + )?; // Naming one command asks about that command, so print what a stuck // transaction is actually waiting on. A gate reason lives in // `last_error` and was never rendered anywhere, which left "Sealing" @@ -4348,67 +4376,85 @@ fn status(store_path: &Path, command_id: Option<&str>) -> Result<()> { // ids are here because they are exactly what a brake release must name. if command_id.is_some() { for transaction in transactions { - println!(" transaction {}", transaction.transaction_id); - println!( + writeln!(out, " transaction {}", transaction.transaction_id)?; + writeln!( + out, " target {} phase {:?}", transaction.target, transaction.phase - ); + )?; if let Some(expected) = &transaction.expected { - println!(" runtime {}", expected.runtime_id); - println!(" release {}", expected.sealed_release_id); + writeln!(out, " runtime {}", expected.runtime_id)?; + writeln!(out, " release {}", expected.sealed_release_id)?; } - println!( + writeln!( + out, " odin publisher cursor {}", transaction.odin_publisher_sequence_cursor - ); + )?; if let Some(evidence) = &transaction.latest_odin_observation { - println!( + writeln!( + out, " latest odin observation sequence {}", evidence.publisher_sequence - ); + )?; } if let Some(reason) = &transaction.last_error { - println!(" waiting on {reason}"); + writeln!(out, " waiting on {reason}")?; } // A Complete transaction that is not terminal still owns its // target through unfinished cleanup. Without this an operator // sees "Complete" and a stale reason and cannot tell what is // holding the target. if let Some(completion) = &transaction.completion { - println!(" completion {completion:?}"); + writeln!(out, " completion {completion:?}")?; } if let Some(abort) = &transaction.pre_fencing_abort { - println!(" pre-fencing abort {abort:?}"); + writeln!(out, " pre-fencing abort {abort:?}")?; } if let Some(abort) = &transaction.post_fencing_abort { - println!(" post-fencing abort {abort:?}"); + writeln!(out, " post-fencing abort {abort:?}")?; } if let Some(cleanup) = &transaction.post_commit_cleanup { - println!(" post-commit cleanup {cleanup:?}"); + writeln!(out, " post-commit cleanup {cleanup:?}")?; } - println!( + writeln!( + out, " terminal {} owns-target {}", transaction.is_terminal(), transaction.blocks_new_target_mutation() - ); + )?; } } } // 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()?) { + writeln!(out, "{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 @@ -4821,6 +4867,56 @@ 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 per episode, under the operation and + /// the subject it was decided for (`*` for a pass over the whole store), + /// and continuity decides exactly as it would have with a healthy store. + /// Each decision site has its own key, so one site's success never ends + /// another site's episode. The message 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, subject: &str, outcome: Result) { + let key = format!("incident-store:{operation}:{subject}"); + 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 subject is a target that 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()?; + for record in store.read()? { + if record.is_open() + && record.condition.ends_with_admission() + && snapshot.admitted_for(&record.subject).is_none() + { + store.close( + record.condition, + &record.subject, + CloseReason::NoLongerAdmitted, + now, + )?; + } + } + Ok(()) + }; + 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 @@ -5252,7 +5348,13 @@ fn boot(options: RuntimeOptions) -> Result<(ProcessLock, Engine)> { fn serve(options: RuntimeOptions) -> Result<()> { let (_lock, engine) = boot(options)?; - loop { + serve_while(&engine, || true) +} + +/// The daemon's loop: tick, sleep, and go again while `again` says so. `serve` +/// says so forever; a test says so a few times. +fn serve_while(engine: &Engine, mut again: impl FnMut() -> bool) -> Result<()> { + while again() { match engine.run_scheduler_tick() { Ok(true) => continue, Ok(false) => {} @@ -5267,6 +5369,7 @@ fn serve(options: RuntimeOptions) -> Result<()> { } thread::sleep(Duration::from_millis(engine.options.poll_millis)); } + Ok(()) } impl Engine { @@ -5680,6 +5783,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()); @@ -5848,6 +5952,31 @@ impl Engine { } }; if observation.is_some() { + // The admitted workload was seen running: the only evidence + // that ends an exhaustion incident, once the window no longer holds it. + // The window sliding alone is not health and never closes one. + self.record_incident( + "close", + ¤t.value.target, + now_millis().and_then(|now| { + // Running is not enough while the window still holds + // the restarts that exhausted it: a release that + // survives one poll after each restart is still the + // same incident. + if snapshot + .supervision_or_new(¤t.value.target) + .restarts_exhausted(now) + { + return Ok(false); + } + self.incidents().close( + IncidentCondition::ContinuityExhausted, + ¤t.value.target, + CloseReason::Recovered, + now, + ) + }), + ); continue; } let workload_error = operational_error @@ -5924,6 +6053,17 @@ 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", + ¤t.value.target, + self.incidents().open( + IncidentCondition::ContinuityExhausted, + ¤t.value.target, + now, + ), + ); self.report_once( &continuity_key, format!( @@ -9338,7 +9478,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 @@ -9395,7 +9535,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 expire [--state-store PATH]\n\ idunn validate --recipe PATH [--binding PATH]\n\n\ @@ -9415,6 +9555,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. @@ -9650,6 +9791,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, @@ -13180,6 +13322,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); @@ -14968,6 +15115,493 @@ 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")); + assert!(usage().contains("idunn status [--state-store PATH] [--incident-store PATH]")); + 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")); + } + + /// A standing fault is traced once. A key is cleared only by an operation + /// that succeeded, so another site's short-circuit must not end it: a + /// target that is running but still exhausted closes nothing, and that + /// used to clear the reconcile pass's fault and print it again every pass. + #[test] + fn a_failing_reconcile_is_traced_once_while_a_running_target_stays_exhausted() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + std::fs::write(incident_path(&world), b"not an incident store")?; + spend_the_window(&world)?; + + for pass in 0..6 { + world.engine.supervise_one_admitted_generation()?; + let reports = world.engine.fault_reports.lock().unwrap(); + assert!( + reports.contains_key("incident-store:close:*"), + "pass {pass}: the standing reconcile fault was cleared, so it prints again" + ); + } + Ok(()) + } + + #[test] + fn a_release_that_runs_one_pass_after_each_restart_keeps_one_incident() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + + // Each cycle: the restarts are spent, the release dies, is seen running + // for one pass (its last restart took), and dies again. + for _ in 0..4 { + workload.kill(); + spend_the_window(&world)?; + world.engine.supervise_one_admitted_generation()?; + workload.revive(); + world.engine.supervise_one_admitted_generation()?; + let records = incidents_of(&world)?; + assert_eq!(records.len(), 1, "a flapping release opened another incident"); + assert!(records[0].is_open(), "running while still exhausted closed the incident"); + } + + // Real recovery: running, and the window no longer holds the restarts. + slide_the_window(&world)?; + world.engine.supervise_one_admitted_generation()?; + let records = incidents_of(&world)?; + assert_eq!(records.len(), 1); + assert_eq!(records[0].close_reason, Some(CloseReason::Recovered)); + assert_control_store_holds_no_incident(&world) + } + + #[test] + fn a_target_leaving_admission_closes_only_its_own_incident() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + let service = admit_incumbent(&world)?; + let store = IncidentStore::new(&incident_path(&world)); + let condition = IncidentCondition::ContinuityExhausted; + let now = now_millis()?; + for subject in ["service", "neighbour", "gone"] { + assert!(store.open(condition, subject, now)?); + } + + // Two targets are admitted: service, and a neighbour. Only the target + // that is not admitted has its incident closed. + let mut snapshot = ControlSnapshot::read(&world.state_store)?; + let envelope = snapshot + .admitted_for("service") + .context("no admitted generation")? + .envelope + .clone(); + snapshot.admitted.push(Stored { + envelope, + value: AdmittedGeneration { + target: "neighbour".into(), + ..service + }, + }); + world.engine.reconcile_incidents(&snapshot); + + let closed = |subject: &str| -> Result> { + Ok(incidents_of(&world)? + .into_iter() + .find(|record| record.subject == subject) + .context("no such incident")? + .close_reason) + }; + assert_eq!(closed("service")?, None); + assert_eq!(closed("neighbour")?, None); + assert_eq!(closed("gone")?, Some(CloseReason::NoLongerAdmitted)); + Ok(()) + } + + #[cfg(unix)] + #[test] + fn a_held_incident_lock_changes_nothing_continuity_decides() -> Result<()> { + use crate::incident::tests::{STUCK_AFTER, hold_lock_of, run_while_held}; + + 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)?; + + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + let path = incident_path(&world); + let actual = run_while_held(hold_lock_of(&path), || { + continuity_decisions(&world, &world.engine, &workload) + })?; + assert_eq!(actual, expected, "a held incident lock changed a continuity decision"); + assert!(!path.exists(), "a contended write landed"); + + // Every refused write is a traced fault naming no path. + { + let reports = world.engine.fault_reports.lock().unwrap(); + let faults = reports + .iter() + .filter(|(key, _)| key.starts_with("incident-store:")) + .map(|(_, report)| report.last.lock().unwrap().clone().unwrap_or_default()) + .collect::>(); + assert!(!faults.is_empty()); + let root = world.root.display().to_string(); + assert!(faults.iter().all(|fault| !fault.contains(&root)), "{faults:?}"); + } + + // The lock is free again: the next pass retries and the incident lands. + // (A child another test forked while it was held can keep it a moment.) + let deadline = std::time::Instant::now() + STUCK_AFTER; + while !incidents_of(&world)?.iter().any(|record| record.is_open()) { + assert!(std::time::Instant::now() < deadline, "the incident never landed"); + world.engine.supervise_one_admitted_generation()?; + std::thread::sleep(Duration::from_millis(20)); + } + assert_control_store_holds_no_incident(&world) + } + + #[test] + fn status_prints_an_open_incident_under_its_target() -> 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 opened = incidents_of(&world)?[0].opened_at_unix_millis; + + let mut printed = Vec::new(); + status(&mut printed, &world.state_store, &incident_path(&world), None)?; + let printed = String::from_utf8(printed)?; + let target = printed.find("target service ").context("no target line")?; + let incident = printed + .find(&format!(" incident continuity-exhausted opened-at {opened}")) + .context("status printed no incident")?; + assert!(target < incident, "{printed}"); + Ok(()) + } + + #[test] + fn serve_ticks_until_told_to_stop() -> Result<()> { + let workload = SwitchWorkload::new(); + let world = EngineFixture::with_workload(workload.clone())?; + admit_incumbent(&world)?; + workload.kill(); + spend_the_window(&world)?; + let mut options = world.engine.options.clone(); + options.poll_millis = 1; + let engine = Engine::open_with_systemd_workload(options, workload.clone())?; + + let mut asked = 0; + serve_while(&engine, || { + asked += 1; + asked <= 3 + })?; + assert_eq!(asked, 4); + assert!(incidents_of(&world)?.iter().any(|record| record.is_open()), "no tick ran"); + Ok(()) + } + + #[test] + fn run_dispatches_serve_and_status() { + let temp = TempDir::new().unwrap(); + let args = |words: &[&str]| words.iter().map(|word| (*word).to_owned()).collect::>(); + // A control store beneath a regular file: serve's boot cannot make it, + // and a serve that returned at once would say Ok instead. + let file = temp.path().join("not-a-directory"); + std::fs::write(&file, b"a file").unwrap(); + let store = file.join("control.cc"); + assert!( + run(args(&["serve", "--state-store", store.to_str().unwrap()]).into_iter()).is_err() + ); + // A control store that is not one: status reads it and refuses. + let junk = temp.path().join("junk.cc"); + std::fs::write(&junk, b"not a cultcache file").unwrap(); + assert!( + run(args(&["status", "--state-store", junk.to_str().unwrap()]).into_iter()).is_err() + ); + } + #[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 072d49d..8fd3192 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -7695,29 +7695,34 @@ fn ensure_bundle_is_reachable_by_workload( Ok(()) } +/// One published file: `0644`, because Idunn's `UMask=027` would land it +/// `0640 root:root` and deny its readers. A chmod by hand does not hold, since +/// each publish writes a new file. +pub(crate) fn publish_file_mode(path: &Path) -> Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if path.exists() { + fs::set_permissions(path, fs::Permissions::from_mode(0o644)) + .with_context(|| format!("publishing {}", path.display()))?; + } + } + Ok(()) +} + /// The topology store is Idunn's *published* surface: every managed target /// reads it to verify its own Expected incarnation against the Idunn anchor. /// /// Idunn runs with `UMask=027`, which is right for its private state and wrong /// for this one file -- it lands `0640 root:root`, and a `DynamicUser` 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. +/// gets EACCES. Integrity here comes from the signatures over the records, not +/// from the mode, so the published copy is readable. The lock sibling too: +/// CultCache opens it alongside the store, so a 0640 lock denies the read just +/// as surely as a 0640 store, and it is created fresh under Idunn's umask on +/// every publish. fn publish_projection_mode(path: &Path) -> Result<()> { - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - // The lock sibling too: CultCache opens it alongside the store, so a - // 0640 lock denies the read just as surely as a 0640 store, and it is - // created fresh under Idunn's umask on every publish. - for path in [path.to_path_buf(), authority_lock_path(path)] { - if path.exists() { - fs::set_permissions(&path, fs::Permissions::from_mode(0o644)) - .with_context(|| format!("publishing {}", path.display()))?; - } - } - } - Ok(()) + publish_file_mode(path)?; + publish_file_mode(&authority_lock_path(path)) } fn build_machine_id(workspace: &Path) -> Result { @@ -9358,18 +9363,25 @@ Content-Le".to_vec()), |_| {}), let address = reservation.local_addr()?; drop(reservation); + // The listener never accepts: the kernel completes a handshake from + // its backlog, so the client connects as soon as it is bound. Nothing + // here can block on a connection that never arrives; the listener only + // has to outlive the connect, and gives up waiting after a bound. + let (connected, connect_done) = std::sync::mpsc::channel::<()>(); let server = thread::spawn(move || -> std::io::Result<()> { thread::sleep(Duration::from_millis(75)); let listener = std::net::TcpListener::bind(address)?; - let (_stream, _) = listener.accept()?; + let _ = connect_done.recv_timeout(Duration::from_secs(30)); + drop(listener); Ok(()) }); - let stream = connect_route_socket(address)?; - drop(stream); + let stream = connect_route_socket(address); + let _ = connected.send(()); server .join() .map_err(|_| anyhow!("route readiness test server panicked"))??; + drop(stream?); Ok(()) } diff --git a/src/incident.rs b/src/incident.rs new file mode 100644 index 0000000..4aacc10 --- /dev/null +++ b/src/incident.rs @@ -0,0 +1,945 @@ +//! 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 external +//! readers 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, bail, ensure}; +use cultcache_rs::{ + CultCacheEnvelope, DatabaseEntry, SingleFileMessagePackBackingStore, + TryCompareExchangeSnapshotOutcome, +}; +use serde::{Deserialize, Serialize}; + +use crate::control_plane::{decode_record, require_id, typed_envelope}; +use crate::drivers::publish_file_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", + } + } + + /// Whether the subject is an admitted target, so that the target leaving + /// admission ends the condition. A condition about anything else (a + /// transaction, a store) answers false and is closed by its own owner. + pub(crate) const fn ends_with_admission(self) -> bool { + match self { + Self::ContinuityExhausted => true, + } + } +} + +#[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() + } + + /// Whether the record still belongs in the published file: open, or closed + /// no more than `INCIDENT_RETENTION_MILLIS` ago. The one predicate behind + /// both retirement and `idunn status`. + pub(crate) fn inside_retention(&self, now: u64) -> bool { + self.closed_at_unix_millis + .is_none_or(|closed_at| now.saturating_sub(closed_at) <= INCIDENT_RETENTION_MILLIS) + } + + 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, + ) + } +} + +/// One read of the incident file: its envelopes exactly as stored, and the +/// records decoded from them, in the same order. Every write expects exactly +/// these envelopes, so a write only lands on the file it was decided against. +struct Snapshot { + envelopes: Vec, + records: Vec, +} + +impl Snapshot { + fn open_record(&self, condition: IncidentCondition, subject: &str) -> Option { + self.records.iter().position(|record| { + record.is_open() && record.condition == condition && record.subject == subject + }) + } +} + +/// The incident file and its history sibling. Incidents are written from the +/// scheduler tick, so no write waits: each is one nonblocking compare-exchange +/// of the whole file against the snapshot it was decided on. A lost race or a +/// held lock is an error and a no-op, retried on a later pass. The lock beside +/// the file is not published and is created 0600 (`secure_lock`): the readers +/// of `incidents.cc` take no lock, so only Idunn can ever hold it. +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") + } + + /// 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> { + let mut records = self.snapshot()?.records; + records.sort_by(|left, right| left.incident_key.cmp(&right.incident_key)); + Ok(records) + } + + fn snapshot(&self) -> Result { + let envelopes = SingleFileMessagePackBackingStore::new(&self.path) + .pull_all_read_only_snapshot() + .context("reading the incident store")?; + let mut records = Vec::new(); + for envelope in &envelopes { + 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" + ); + records.push(record); + } + Ok(Snapshot { envelopes, records }) + } + + /// Replace `path`'s whole content with `replacements` if it is still + /// exactly `expected`. Never waits for the lock. + fn exchange( + path: &Path, + expected: &[CultCacheEnvelope], + replacements: &[CultCacheEnvelope], + ) -> Result<()> { + secure_lock(path)?; + match SingleFileMessagePackBackingStore::new(path) + .try_compare_exchange_snapshot(expected, replacements)? + { + TryCompareExchangeSnapshotOutcome::Exchanged => Ok(()), + TryCompareExchangeSnapshotOutcome::Mismatch => { + bail!("the incident store changed while it was being written") + } + TryCompareExchangeSnapshotOutcome::LockContended => { + bail!("the incident store is locked by another holder") + } + } + } + + /// 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 { + self.open_on(&self.snapshot()?, condition, subject, now) + } + + fn open_on( + &self, + snapshot: &Snapshot, + condition: IncidentCondition, + subject: &str, + now: u64, + ) -> Result { + if snapshot.open_record(condition, subject).is_some() { + return Ok(false); + } + let mut replacements = snapshot.envelopes.clone(); + replacements.push(IncidentRecord::opened(condition, subject, now).envelope(now)?); + Self::exchange(&self.path, &snapshot.envelopes, &replacements)?; + publish_file_mode(&self.path)?; + Ok(true) + } + + /// 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 snapshot = self.snapshot()?; + let Some(index) = snapshot.open_record(condition, subject) else { + return Ok(false); + }; + let mut closed = snapshot.records[index].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 mut replacements = snapshot.envelopes.clone(); + replacements[index] = closed.envelope(now)?; + Self::exchange(&self.path, &snapshot.envelopes, &replacements)?; + publish_file_mode(&self.path)?; + Ok(true) + } + + /// Move every incident that has left retention to the history file, then + /// drop it here. History first: a crash between the two leaves the record + /// in both, and the repeat finds it already archived. An open incident + /// never moves. + pub(crate) fn retire_closed(&self, now: u64) -> Result { + let snapshot = self.snapshot()?; + let (kept, retiring): (Vec<_>, Vec<_>) = snapshot + .envelopes + .iter() + .zip(&snapshot.records) + .partition(|(_, record)| record.inside_retention(now)); + if retiring.is_empty() { + return Ok(false); + } + let history_path = self.history_path(); + let archived = SingleFileMessagePackBackingStore::new(&history_path) + .pull_all_read_only_snapshot() + .context("reading the incident history")?; + let mut history = archived.clone(); + history.extend( + retiring + .iter() + .filter(|(envelope, _)| !archived.iter().any(|held| held.key == envelope.key)) + .map(|(envelope, _)| (*envelope).clone()), + ); + if history.len() != archived.len() { + Self::exchange(&history_path, &archived, &history)?; + } + let kept = kept + .into_iter() + .map(|(envelope, _)| envelope.clone()) + .collect::>(); + Self::exchange(&self.path, &snapshot.envelopes, &kept)?; + publish_file_mode(&self.path)?; + Ok(true) + } +} + +/// Make the lock beside `store` Idunn's alone, before CultCache opens it. +/// +/// CultCache creates a lock with mode 0666, so the published directory decides +/// who may open it: under Yggdrasil's default ACL (`u::rw g::r o::r`) that is +/// 0644, and any uid could hold the lock and keep every incident from being +/// recorded. Created here with 0600 the create mode is intersected with the +/// ACL and still holds, and no wider file ever exists. A lock left wider by an +/// earlier run is tightened on the descriptor just opened (a descriptor some +/// other uid already holds is not revoked). Readers of the store take no lock, +/// so nothing but Idunn needs to open this file. +fn secure_lock(store: &Path) -> Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; + + let mut name = store + .file_name() + .context("the incident store has no file name")? + .to_owned(); + name.push(".lock"); + let lock = store.with_file_name(name); + if let Some(parent) = lock.parent() { + std::fs::create_dir_all(parent).context("creating the incident store directory")?; + } + std::fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .mode(0o600) + .open(&lock) + .context("opening the incident lock")? + .set_permissions(std::fs::Permissions::from_mode(0o600)) + .context("making the incident lock private")?; + } + #[cfg(not(unix))] + let _ = store; + Ok(()) +} + +/// 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 && record.inside_retention(now)) + .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)] +pub(crate) mod tests { + #[cfg(unix)] + use std::time::{Duration, Instant}; + + 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() { + // The ruled retention: seven days. + assert_eq!(INCIDENT_RETENTION_MILLIS, 604_800_000); + 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`. The umask is process-wide, so it is held + /// only around each call and only one test at a time may hold it. + #[cfg(target_os = "linux")] + 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 + } + + /// The body's condition: a directory carrying the default ACL + /// `u::rw g::r o::r` that Yggdrasil's idunn-projection directory has. Under + /// it the umask is ignored and a file created with mode 0666 lands 0644. + #[cfg(target_os = "linux")] + fn give_default_acl(dir: &Path) { + use std::os::unix::ffi::OsStrExt; + + // posix_acl_xattr: version 2, then (tag, perm, id) entries. The tags + // are ACL_USER_OBJ, ACL_GROUP_OBJ and ACL_OTHER; the id is undefined. + let mut value = 2u32.to_le_bytes().to_vec(); + for (tag, perm) in [(0x01u16, 6u16), (0x04, 4), (0x20, 4)] { + value.extend(tag.to_le_bytes()); + value.extend(perm.to_le_bytes()); + value.extend(u32::MAX.to_le_bytes()); + } + let dir = std::ffi::CString::new(dir.as_os_str().as_bytes()).unwrap(); + // SAFETY: both strings are NUL-terminated and the value is a live buffer. + let rc = unsafe { + libc::setxattr( + dir.as_ptr(), + c"system.posix_acl_default".as_ptr(), + value.as_ptr().cast(), + value.len(), + 0, + ) + }; + assert_eq!( + rc, + 0, + "the test filesystem must accept a default POSIX ACL: {}", + std::io::Error::last_os_error() + ); + } + + /// Whether uid 65534 can read `path`; `None` when the uid cannot be + /// switched (not root, or no setpriv), which leaves the question unproven. + #[cfg(target_os = "linux")] + fn readable_by_another_uid(path: &Path) -> Option { + std::process::Command::new("setpriv") + .args(["--reuid=65534", "--regid=65534", "--clear-groups", "cat"]) + .arg(path) + .output() + .ok() + .map(|output| output.status.success()) + } + + #[cfg(target_os = "linux")] + #[test] + fn under_a_default_acl_the_store_is_published_and_its_locks_are_private() { + use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; + + let temp = TempDir::new().unwrap(); + let dir = temp.path(); + std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o755)).unwrap(); + give_default_acl(dir); + let path = dir.join("incidents.cc"); + let lock = dir.join("incidents.cc.lock"); + let history_lock = dir.join("incident-history.cc.lock"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + let mode = |path: &Path| std::fs::metadata(path).unwrap().permissions().mode() & 0o777; + + // Control: the lock CultCache would have made here, open(2) mode 0666 + // under Idunn's umask. Without the ACL in force this reads 0640 and the + // test below proves nothing about the body. + let control = dir.join("control.lock"); + under_service_umask(|| { + std::fs::OpenOptions::new().create(true).write(true).mode(0o666).open(&control) + }) + .unwrap(); + assert_eq!(mode(&control), 0o644, "the default ACL is not in force"); + + assert!(under_service_umask(|| store.open(condition, "service", OPENED)).unwrap()); + assert_eq!(mode(&path), 0o644, "after open"); + assert_eq!(mode(&lock), 0o600, "the lock after open"); + + assert!( + under_service_umask(|| store.close(condition, "service", CloseReason::Recovered, CLOSED)) + .unwrap() + ); + assert_eq!(mode(&path), 0o644, "after close"); + assert_eq!(mode(&lock), 0o600, "the lock after close"); + + assert!( + under_service_umask(|| store.retire_closed(CLOSED + INCIDENT_RETENTION_MILLIS + 1)) + .unwrap() + ); + assert_eq!(mode(&path), 0o644, "after retire"); + assert_eq!(mode(&lock), 0o600, "the lock after retire"); + assert_eq!(mode(&history_lock), 0o600, "the history lock after retire"); + + // The consequence: another uid cannot open the lock, so it cannot hold + // it. Provable only as root with setpriv, and only if the same uid can + // read the control file, which shows the switch itself works. + if readable_by_another_uid(&control) == Some(true) { + assert_eq!(readable_by_another_uid(&lock), Some(false)); + assert_eq!(readable_by_another_uid(&history_lock), Some(false)); + } else { + eprintln!("UNPROVEN: no uid switch here; the lock modes above stand alone"); + } + } + + #[cfg(unix)] + #[test] + fn a_lock_left_wider_by_an_earlier_run_is_tightened_before_it_is_used() { + 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"); + std::fs::write(&lock, b"").unwrap(); + std::fs::set_permissions(&lock, std::fs::Permissions::from_mode(0o666)).unwrap(); + + assert!( + IncidentStore::new(&path) + .open(IncidentCondition::ContinuityExhausted, "service", OPENED) + .unwrap() + ); + let mode = std::fs::metadata(&lock).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o600); + } + + /// An exclusive lock on a store's lock file, held through another file + /// description. flock conflicts between descriptions, so the holder need + /// not be another process or another uid: the test does not depend on who + /// it runs as. + #[cfg(unix)] + pub(crate) struct HeldLock(#[allow(dead_code)] std::fs::File); + + #[cfg(unix)] + pub(crate) fn hold_lock_of(store: &Path) -> HeldLock { + use std::os::unix::io::AsRawFd; + + let mut name = store.file_name().unwrap().to_owned(); + name.push(".lock"); + let file = std::fs::OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(store.with_file_name(name)) + .unwrap(); + // SAFETY: flock on a descriptor this function owns. + assert_eq!(unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }, 0); + HeldLock(file) + } + + /// How long a test lets a held lock, or a released lock's last straggler, + /// stand in its way before it calls the store stuck. Nothing correct ever + /// spends it: the code under test finishes in milliseconds, and the limit + /// is only what stops a blocked write from hanging the suite. No + /// assertion compares elapsed time to a bound the machine's load can move. + #[cfg(unix)] + pub(crate) const STUCK_AFTER: Duration = Duration::from_secs(30); + + /// Run `work` while `held` is held, and fail if it waited on it. The holder + /// lets go only when `work` has returned, or after `STUCK_AFTER`: work that + /// waits on the lock is freed by that second release and is then seen to + /// have waited, so the test fails instead of hanging. + #[cfg(unix)] + pub(crate) fn run_while_held(held: HeldLock, work: impl FnOnce() -> T) -> T { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, mpsc}; + + let (finished, work_done) = mpsc::channel::<()>(); + let gave_up = Arc::new(AtomicBool::new(false)); + let holder = { + let gave_up = Arc::clone(&gave_up); + std::thread::spawn(move || { + if work_done.recv_timeout(STUCK_AFTER) == Err(mpsc::RecvTimeoutError::Timeout) { + gave_up.store(true, Ordering::SeqCst); + } + drop(held); + }) + }; + let result = work(); + let _ = finished.send(()); + holder.join().unwrap(); + assert!(!gave_up.load(Ordering::SeqCst), "the work waited on a held lock"); + result + } + + /// Retry `attempt` until it lands. A released lock can stay held for a + /// moment by a child another test forked while it was held. + #[cfg(unix)] + fn eventually(mut attempt: impl FnMut() -> Result) -> T { + let deadline = Instant::now() + STUCK_AFTER; + loop { + match attempt() { + Ok(value) => return value, + Err(error) if Instant::now() >= deadline => panic!("never landed: {error:#}"), + Err(_) => std::thread::sleep(Duration::from_millis(20)), + } + } + } + + #[cfg(unix)] + #[test] + fn no_incident_write_waits_on_a_held_lock_and_each_lands_when_it_is_freed() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + assert!(store.open(condition, "closing", OPENED).unwrap()); + assert!(store.close(condition, "closing", CloseReason::Recovered, OPENED + 1).unwrap()); + assert!(store.open(condition, "open", OPENED).unwrap()); + let before = std::fs::read(&path).unwrap(); + let retired = OPENED + 1 + INCIDENT_RETENTION_MILLIS + 1; + + let outcomes = run_while_held(hold_lock_of(&path), || { + ( + store.open(condition, "new", CLOSED), + store.close(condition, "open", CloseReason::Recovered, CLOSED), + store.retire_closed(retired), + ) + }); + for outcome in [outcomes.0, outcomes.1, outcomes.2] { + assert!(outcome.is_err()); + } + assert_eq!(std::fs::read(&path).unwrap(), before, "a contended write changed the file"); + + // Nothing was lost by the refusal: the same calls land once it is free. + // (Retirement archives first, so its refused delete left the record in + // history too; the repeat archives nothing twice.) + assert!(eventually(|| store.open(condition, "new", CLOSED))); + assert!(eventually(|| store.close(condition, "open", CloseReason::Recovered, CLOSED))); + assert!(eventually(|| store.retire_closed(retired))); + let history = SingleFileMessagePackBackingStore::new(&temp.path().join("incident-history.cc")); + assert_eq!(history.pull_all_read_only_snapshot().unwrap().len(), 1); + } + + #[cfg(unix)] + #[test] + fn a_held_history_lock_keeps_the_incident_where_it_is() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let history = temp.path().join("incident-history.cc"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + assert!(store.open(condition, "old", OPENED).unwrap()); + assert!(store.close(condition, "old", CloseReason::Recovered, OPENED + 1).unwrap()); + let before = std::fs::read(&path).unwrap(); + let retired = OPENED + 1 + INCIDENT_RETENTION_MILLIS + 1; + + let outcome = run_while_held(hold_lock_of(&history), || store.retire_closed(retired)); + assert!(outcome.is_err()); + assert_eq!(std::fs::read(&path).unwrap(), before); + assert!(eventually(|| store.retire_closed(retired))); + } + + #[test] + fn a_write_decided_on_a_stale_snapshot_does_not_land() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + + // Two writers read the file with no incident open for the subject. + let stale = store.snapshot().unwrap(); + assert!(store.open(condition, "service", OPENED).unwrap()); + + // The slower one must lose: one open incident per subject. + let error = store + .open_on(&stale, condition, "service", OPENED + 1) + .unwrap_err(); + assert!(error.to_string().contains("changed"), "{error}"); + let records = store.read().unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].opened_at_unix_millis, OPENED); + } + + #[test] + fn retiring_after_a_crash_between_history_and_delete_archives_once() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("incidents.cc"); + let store = IncidentStore::new(&path); + let condition = IncidentCondition::ContinuityExhausted; + assert!(store.open(condition, "old", OPENED).unwrap()); + assert!(store.close(condition, "old", CloseReason::Recovered, OPENED + 1).unwrap()); + // History already holds the record: the delete never ran. + let envelope = store.snapshot().unwrap().envelopes[0].clone(); + let history = SingleFileMessagePackBackingStore::new(&temp.path().join("incident-history.cc")); + assert!(history.insert_entry_if_absent(envelope).unwrap()); + + assert!(store.retire_closed(OPENED + 1 + INCIDENT_RETENTION_MILLIS + 1).unwrap()); + assert!(store.read().unwrap().is_empty()); + assert_eq!(history.pull_all_read_only_snapshot().unwrap().len(), 1); + } +} 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; diff --git a/tests/fixtures/idunn.operator_incident.v1.cc b/tests/fixtures/idunn.operator_incident.v1.cc new file mode 100644 index 0000000..180c6f0 Binary files /dev/null and b/tests/fixtures/idunn.operator_incident.v1.cc differ