GameCult/Idunn
Watch
9
0
Fork
You've already forked Idunn
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Idunn records continuity exhaustion as a durable incident in incidents.cc

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 16:21:02 +02:00
commit 202e5a2e10
6 changed files with 1038 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

1
deploy/idunn-yggdrasil.service
View file

@ -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 \

3
docs/guide.md
View file

@ -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:

428
src/control_plane.rs
View file

@ -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<String>,
},
Cancel {
@ -2468,8 +2474,9 @@ pub fn run(args: impl Iterator<Item = String>) -> 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<Item = String>) -> Result<Command> {
"--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<Item = String>) -> Result<Command> {
}
fn parse_status(mut args: impl Iterator<Item = String>) -> Result<Command> {
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<Item = String>) -> Result<Command> {
}
Ok(Command::Status {
state_store,
incident_store,
command_id,
})
}
@ -3091,7 +3103,7 @@ impl ControlSnapshot {
}
}
fn decode_record<T>(envelope: &CultCacheEnvelope) -> Result<T>
pub(crate) fn decode_record<T>(envelope: &CultCacheEnvelope) -> Result<T>
where
T: for<'de> Deserialize<'de> + Serialize,
{
@ -3634,7 +3646,7 @@ fn target_supervision_envelope(value: &TargetSupervision, now: u64) -> Result<Cu
)
}
fn typed_envelope<T: Serialize>(
pub(crate) fn typed_envelope<T: Serialize>(
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<Option<String>>,
@ -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<String> {
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<bool>) {
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<bool> {
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<HostActuatorAccess<'_>> {
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::<Vec<_>>();
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,
&current.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,
&current.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 <service|profile:name> [--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 <command-id> [--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<Vec<crate::incident::IncidentRecord>> {
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<Vec<(bool, usize, usize)>> {
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::<Vec<_>>();
assert_eq!(
faults.len(),
1,
"{:?}",
faults.iter().map(|(key, _)| key).collect::<Vec<_>>()
);
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();

2
src/drivers.rs
View file

@ -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;

619
src/incident.rs Normal file
View file

@ -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,
/// `<condition>:<subject>:<opened_at>`. 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<u64>,
#[cultcache(key = 6)]
pub(crate) close_reason: Option<CloseReason>,
}
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<CultCacheEnvelope> {
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<Vec<IncidentRecord>> {
Ok(self
.read_stored()?
.into_iter()
.map(|stored| stored.record)
.collect())
}
fn read_stored(&self) -> Result<Vec<Stored>> {
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<bool> {
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<bool> {
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<bool> {
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::<Vec<_>>();
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<String> {
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(), &copy).unwrap();
assert_eq!(
IncidentStore::new(&copy).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::<String>(&payload).unwrap(),
IncidentCondition::ContinuityExhausted.name()
);
let payload = rmp_serde::to_vec(&CloseReason::NoLongerAdmitted).unwrap();
assert_eq!(
rmp_serde::from_slice::<String>(&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::<Vec<_>>();
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::<Vec<_>>();
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<T>(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");
}
}

1
src/lib.rs
View file

@ -4,4 +4,5 @@ pub mod deployment_plan;
pub mod drivers;
pub mod host;
pub mod host_actuator;
mod incident;
pub mod provisioning;