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

Merge watchdog idunn-incident: continuity exhaustion is a durable incident

Browse source 
Eureka idunn-watchdog cut idunn-incident, attempts 1-3. Soul verdict
idunn-watchdog:verdict:cut-idunn-incident.s3 holds. Incidents live in
their own incidents.cc (0644, lock private 0600), written with non-blocking
snapshot exchanges; the tick never waits on them. Close rule: running and
no longer exhausted (operator ruling exhaustion-close-rule-op).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-01 01:15:08 +02:00
commit a909f931ea
7 changed files with 1648 additions and 54 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:

700
src/control_plane.rs
View file

@ -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<String>,
},
Cancel {
@ -2598,8 +2605,14 @@ 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(
&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<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)?
}
@ -2762,11 +2776,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:?}"),
@ -2774,6 +2791,7 @@ fn parse_status(mut args: impl Iterator<Item = String>) -> Result<Command> {
}
Ok(Command::Status {
state_store,
incident_store,
command_id,
})
}
@ -3303,7 +3321,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,
{
@ -3846,7 +3864,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,
@ -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<Option<String>>,
@ -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<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
@ -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<T>(&self, operation: &str, subject: &str, outcome: Result<T>) {
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<HostActuatorAccess<'_>> {
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::<Vec<_>>();
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",
&current.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(&current.value.target)
.restarts_exhausted(now)
{
return Ok(false);
}
self.incidents().close(
IncidentCondition::ContinuityExhausted,
&current.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",
&current.value.target,
self.incidents().open(
IncidentCondition::ContinuityExhausted,
&current.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 <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 expire <command-id> [--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<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"));
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<Option<CloseReason>> {
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::<Vec<_>>();
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::<Vec<_>>();
// 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();

52
src/drivers.rs
View file

@ -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<String> {
@ -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(())
}

945
src/incident.rs Normal file
View file

@ -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,
/// `<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()
}
/// 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<CultCacheEnvelope> {
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<CultCacheEnvelope>,
records: Vec<IncidentRecord>,
}
impl Snapshot {
fn open_record(&self, condition: IncidentCondition, subject: &str) -> Option<usize> {
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<Vec<IncidentRecord>> {
let mut records = self.snapshot()?.records;
records.sort_by(|left, right| left.incident_key.cmp(&right.incident_key));
Ok(records)
}
fn snapshot(&self) -> Result<Snapshot> {
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<bool> {
self.open_on(&self.snapshot()?, condition, subject, now)
}
fn open_on(
&self,
snapshot: &Snapshot,
condition: IncidentCondition,
subject: &str,
now: u64,
) -> Result<bool> {
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<bool> {
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<bool> {
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::<Vec<_>>();
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<String> {
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(), &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() {
// 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::<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`. 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<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
}
/// 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<bool> {
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<T>(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<T>(mut attempt: impl FnMut() -> Result<T>) -> 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);
}
}

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;

BIN
tests/fixtures/idunn.operator_incident.v1.cc vendored Normal file
View file

Binary file not shown.