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

Merge idunn/runner-failure-evidence: a failed runner keeps its evidence

Browse source 
Cut idunn-deploy-provenance:cut_spec:cut-runner-failure-evidence.r1,
reports h1-h3, Soul s1-s3. The docker runner kept only stderr, so a step
reporting on stdout failed with an empty message (Vili's node --test). A
failed step's name and bounded stdout/stderr tails go to a root-only typed
store read by idunn status; errors and the journal carry no output; one
allowlist escape at the status sink.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-25 08:32:18 +01:00
commit 8400d0712f
4 changed files with 1774 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

509
src/control_plane.rs
View file

@ -42,11 +42,14 @@ use crate::drivers::{
ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt, ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt,
GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver, GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver,
LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators, LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators,
RouteObservation, RoutePreflightReceipt, RunnerPort, SourcePort, RouteObservation, RoutePreflightReceipt, RunnerPort, RunnerStepFailure, SourcePort,
SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort,
WriteLeasePort, WriteLeasePort,
}; };
use crate::incident::{CloseReason, IncidentCondition, IncidentStore, render_incidents}; use crate::incident::{CloseReason, IncidentCondition, IncidentStore, render_incidents};
use crate::runner_evidence::{
self, Kept, RunnerFailureRecord, runner_evidence_store_path, unreadable_store_path,
};
use crate::host_actuator::{ use crate::host_actuator::{
HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver, HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver,
HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service, HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service,
@ -4877,6 +4880,13 @@ fn status(
} }
if let Some(abort) = &transaction.pre_fencing_abort { if let Some(abort) = &transaction.pre_fencing_abort {
writeln!(out, " pre-fencing abort {abort:?}")?; writeln!(out, " pre-fencing abort {abort:?}")?;
// The step that failed and its output tail live in their
// own private store. An unreadable store costs its lines.
runner_evidence::show_status(
out,
&runner_evidence_store_path(store_path),
&transaction.transaction_id,
)?;
} }
if let Some(abort) = &transaction.post_fencing_abort { if let Some(abort) = &transaction.post_fencing_abort {
writeln!(out, " post-fencing abort {abort:?}")?; writeln!(out, " post-fencing abort {abort:?}")?;
@ -5374,6 +5384,32 @@ impl Engine {
IncidentStore::new(&self.options.incident_store) IncidentStore::new(&self.options.incident_store)
} }
/// A failed runner step keeps its output tail in the runner evidence
/// store and says so on the journal, naming no output. The store write is
/// best effort: its failure is named in the line and never stops the
/// abort that follows.
fn keep_runner_failure(
&self,
transaction: &DeploymentTransaction,
error: &anyhow::Error,
) -> Option<String> {
let failure = error.downcast_ref::<RunnerStepFailure>()?;
let path = runner_evidence_store_path(&self.options.state_store);
let recorded = now_millis().and_then(|now| {
runner_evidence::record(
&path,
&RunnerFailureRecord::from_failure(
&transaction.transaction_id,
&transaction.command_id,
&transaction.target,
failure,
now,
),
)
});
Some(runner_failure_journal_line(transaction, failure, &path, &recorded))
}
/// Settle the outcome of one incident operation. A failure never leaves /// Settle the outcome of one incident operation. A failure never leaves
/// this function: it is traced once per episode, under the operation and /// 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), /// the subject it was decided for (`*` for a pass over the whole store),
@ -6315,6 +6351,9 @@ impl Engine {
if latest.value.pre_fencing_abort.is_none() if latest.value.pre_fencing_abort.is_none()
&& error.downcast_ref::<LockContended>().is_none() && error.downcast_ref::<LockContended>().is_none()
{ {
if let Some(line) = self.keep_runner_failure(&latest.value, &error) {
eprintln!("{line}");
}
self.begin_pre_fencing_abort(latest, error)?; self.begin_pre_fencing_abort(latest, error)?;
} else { } else {
self.record_resumable_error(latest, &error)?; self.record_resumable_error(latest, &error)?;
@ -10641,6 +10680,43 @@ const DETAIL_EMPTY: &str = "(no detail)";
/// only judge, and this output always satisfies it. NUL is dropped, empty /// only judge, and this output always satisfies it. NUL is dropped, empty
/// input gets a fixed placeholder, and an over-long value keeps its head and /// input gets a fixed placeholder, and an over-long value keeps its head and
/// its tail (where a build log's error is) cut on char boundaries. /// its tail (where a build log's error is) cut on char boundaries.
/// The journal line for a failed runner step: the transaction, the subject and
/// status (the failure's `Display`), and where the evidence is. Never output.
fn runner_failure_journal_line(
transaction: &DeploymentTransaction,
failure: &RunnerStepFailure,
store_path: &Path,
recorded: &Result<Kept>,
) -> String {
let head = format!(
"Idunn transaction {} (command {}, target {}) {failure}",
transaction.transaction_id, transaction.command_id, transaction.target
);
match recorded {
Ok(kept) => {
let aside = match kept {
Kept::Stored => String::new(),
Kept::SetAside { replaced } => format!(
"; an undecodable evidence store was set aside as {}{}",
unreadable_store_path(store_path).display(),
match replaced {
Some(last_written) => format!(
", replacing an earlier set-aside store last written at unix millis {last_written}"
),
None => String::new(),
}
),
};
format!(
"{head}; evidence in {}{aside}; read it with idunn status --command {}",
store_path.display(),
transaction.command_id
)
}
Err(error) => format!("{head}; evidence was not kept: {error:#}"),
}
}
fn bounded_detail(value: &str) -> String { fn bounded_detail(value: &str) -> String {
let value = value.replace('', ""); let value = value.replace('', "");
if value.is_empty() { if value.is_empty() {
@ -24657,4 +24733,435 @@ GAMECULT_RUNTIME_PRESENCE_IDENTITY = "/etc/gamecult/service/runtime-presence-ide
Ok(()) Ok(())
} }
} }
// BEGIN runner-failure-evidence tests
// A failed runner step is named, its output tail is kept in
// runner-evidence.cc, and no output reaches an error, control.cc or the
// journal line. The runner is the real DockerRunnerDriver driving a fake
// docker program, through the engine's Sealing path.
const RUNNER_CANARY: &str = "CANARY-7c41e0d9";
/// A source that freezes a fixed tree, so a deployment reaches the runner.
struct TreeSource {
recipe: String,
root: PathBuf,
}
impl SourcePort for TreeSource {
fn resolve(
&self,
_: &OperatorBinding,
_: &str,
_: u64,
) -> Result<crate::drivers::ResolvedSource> {
Ok(crate::drivers::ResolvedSource {
facts: crate::deployment_plan::tests::source(&self.recipe),
recipe_bytes: self.recipe.clone().into_bytes(),
})
}
fn freeze(&self, transaction_id: &str, plan: &CompiledDeploymentPlan) -> Result<FrozenSourceReceipt> {
Ok(FrozenSourceReceipt {
transaction_id: transaction_id.to_owned(),
plan_id: plan.plan_id.clone(),
snapshot_sha256: format!("sha256-{}", "a".repeat(64)),
})
}
fn observe_frozen(
&self,
plan: &CompiledDeploymentPlan,
receipt: &FrozenSourceReceipt,
) -> Result<crate::drivers::FrozenSource> {
Ok(crate::drivers::FrozenSource::fixture(
receipt.clone(),
plan.source.clone(),
plan.recipe_blob.clone(),
self.root.clone(),
))
}
fn cleanup(&self, _: &str, _: Option<&FrozenSourceReceipt>) -> Result<()> {
Ok(())
}
}
/// The unrouted recipe with two steps ahead of its build, and no external
/// input: the second step, `test-daemon`, is the one the fake docker fails.
fn two_step_recipe() -> String {
let recipe = unrouted_recipe().replace(
"capability = \"service.runtime\"",
&format!("capability = \"{ODIN_RENDEZVOUS_CAPABILITY}\""),
);
let (head, rest) = recipe
.split_once("[[external_inputs]]")
.expect("the recipe declares an external input");
let (_, tail) = rest.split_once("[[artifacts]]").expect("the recipe declares artifacts");
let head = head.replacen(
"[[steps]]\nid = \"build\"",
"[[steps]]\nid = \"install\"\nphase = \"prepare\"\nrunner = \"rust\"\nargv = [\"cargo\", \"fetch\"]\n\n\
[[steps]]\nid = \"test-daemon\"\nphase = \"test\"\nrunner = \"rust\"\nargv = [\"cargo\", \"test\"]\n\n\
[[steps]]\nid = \"build\"",
1,
);
format!("{head}[[artifacts]]{tail}")
}
/// A Sealing world whose runner is the real driver over a fake docker: the
/// first call succeeds, the second prints the canary and an escape
/// sequence on both streams and exits 3.
#[cfg(unix)]
fn failing_runner_world() -> Result<(EngineFixture, TempDir)> {
use std::os::unix::fs::PermissionsExt;
let tools = TempDir::new()?;
let docker = tools.path().join("docker");
std::fs::write(
&docker,
format!(
"#!/bin/sh\nn=$(cat \"$0.count\" 2>/dev/null || echo 0)\nn=$((n+1))\necho $n > \"$0.count\"\n\
[ \"$n\" -lt 2 ] && exit 0\n\
printf 'out {RUNNER_CANARY} \\033[2J\\n'\nprintf 'err {RUNNER_CANARY}\\n' >&2\nexit 3\n"
),
)?;
std::fs::set_permissions(&docker, std::fs::Permissions::from_mode(0o755))?;
let recipe = two_step_recipe();
let mut world = sealing_world(&recipe)?;
let tree = tools.path().join("tree");
std::fs::create_dir_all(&tree)?;
std::fs::write(tree.join("README"), b"source")?;
world.engine.source = Arc::new(TreeSource { recipe, root: tree });
world.engine.docker_runner = DockerRunnerDriver { docker_program: docker };
Ok((world, tools))
}
/// Scheduler ticks until the deployment ends (or the budget is spent).
fn tick_until_ended(world: &EngineFixture) -> Result<DeploymentTransaction> {
for _ in 0..24 {
world.engine.resume_backoff.lock().unwrap().clear();
let _ = world.engine.run_scheduler_tick();
if latest(world)?.completion.is_some() {
break;
}
}
latest(world)
}
fn printed_status(world: &EngineFixture) -> Result<String> {
let mut printed = Vec::new();
status(&mut printed, &world.state_store, &world.engine.options.incident_store, Some("up-service"))?;
Ok(String::from_utf8(printed)?)
}
fn canary_failure(stdout: &[u8]) -> RunnerStepFailure {
RunnerStepFailure {
subject: crate::drivers::RunnerSubject::Step("test-daemon".into()),
runner: "rust".into(),
exit_code: Some(3),
signal: None,
stdout_tail: stdout.to_vec(),
stdout_total_bytes: stdout.len() as u64,
stderr_tail: stdout.to_vec(),
stderr_total_bytes: stdout.len() as u64,
}
}
#[cfg(unix)]
#[test]
fn a_failed_runner_step_is_named_kept_and_never_echoed() -> Result<()> {
use std::os::unix::fs::PermissionsExt;
let (world, _tools) = failing_runner_world()?;
let ended = tick_until_ended(&world)?;
let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
bail!("the deployment did not fail before fencing: {:?} {:?}", ended.phase, ended.last_error);
};
// Named: the second step, on its runner, with the exit status.
assert!(
error.contains("runner step test-daemon on runner rust exited with exit status 3"),
"{error}"
);
// Never echoed: no error text, no control.cc byte, carries the output.
let abort = ended.pre_fencing_abort.as_ref().context("no pre-fencing abort")?;
for text in [error.as_str(), abort.error.as_str(), &format!("{ended:?}")] {
assert!(!text.contains(RUNNER_CANARY), "runner output reached a record: {text}");
}
// The terminal transaction has moved to history.cc: both stores.
for store in [world.state_store.clone(), history_store_path(&world.state_store)] {
let bytes = std::fs::read(&store)?;
assert!(!bytes.is_empty(), "{} is empty", store.display());
assert!(
!bytes.windows(RUNNER_CANARY.len()).any(|window| window == RUNNER_CANARY.as_bytes()),
"runner output reached {}",
store.display()
);
}
// Kept: in the private store, tails of both streams, and on status.
let evidence = runner_evidence_store_path(&world.state_store);
assert_eq!(std::fs::metadata(&evidence)?.permissions().mode() & 0o777, 0o600);
let record = runner_evidence::for_transaction(&evidence, &ended.transaction_id)?
.context("no evidence for the failed transaction")?;
assert_eq!(record.subject_id, "test-daemon");
assert_eq!(record.exit_code, Some(3));
assert!(String::from_utf8_lossy(&record.stdout_tail).contains(RUNNER_CANARY));
assert!(String::from_utf8_lossy(&record.stderr_tail).contains(RUNNER_CANARY));
let printed = printed_status(&world)?;
assert!(printed.contains("failed step test-daemon"), "{printed}");
assert!(printed.contains(RUNNER_CANARY), "{printed}");
assert!(!printed.contains('\u{1b}'), "status printed a raw escape byte");
Ok(())
}
#[test]
fn the_journal_line_names_the_store_and_never_the_output() -> Result<()> {
let world = EngineFixture::new()?;
let (transaction, _) = terminal_transaction_with_command("ghostlight")?;
let failure = canary_failure(RUNNER_CANARY.as_bytes());
let line = world
.engine
.keep_runner_failure(&transaction, &anyhow::Error::new(failure.clone()).context("sealing"))
.context("a runner failure produced no journal line")?;
assert!(!line.contains(RUNNER_CANARY), "{line}");
for needle in [
transaction.transaction_id.as_str(),
"up-ghostlight",
"ghostlight",
"runner step test-daemon on runner rust exited with exit status 3",
"runner-evidence.cc",
"idunn status --command up-ghostlight",
] {
assert!(line.contains(needle), "{needle} missing from {line}");
}
// Not a runner failure: nothing is kept and nothing is said.
assert!(
world
.engine
.keep_runner_failure(&transaction, &anyhow::anyhow!("a different failure"))
.is_none()
);
// The line a failed write produces says so, with no output.
let unwritable = runner_failure_journal_line(
&transaction,
&failure,
Path::new("/var/lib/gamecult/idunn/runner-evidence.cc"),
&Err(anyhow::anyhow!("disk full")),
);
assert!(unwritable.contains("evidence was not kept: disk full"), "{unwritable}");
assert!(!unwritable.contains(RUNNER_CANARY));
Ok(())
}
#[cfg(unix)]
#[test]
fn a_failed_evidence_write_does_not_stop_the_abort() -> Result<()> {
let (world, _tools) = failing_runner_world()?;
// A directory where the store should be: every write to it fails.
std::fs::create_dir(runner_evidence_store_path(&world.state_store))?;
let ended = tick_until_ended(&world)?;
let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
bail!("the unwritable store stopped the abort: {:?} {:?}", ended.phase, ended.last_error);
};
assert!(error.contains("runner step test-daemon on runner rust"), "{error}");
let line = world
.engine
.keep_runner_failure(&ended, &anyhow::Error::new(canary_failure(b"")))
.context("no journal line")?;
assert!(line.contains("evidence was not kept"), "{line}");
Ok(())
}
#[cfg(unix)]
#[test]
fn an_undecodable_evidence_store_does_not_stop_the_next_failure_keeping_its_evidence() -> Result<()> {
let damage = format!("\u{c1} not a store {RUNNER_CANARY}-in-the-damage");
let (world, _tools) = failing_runner_world()?;
let evidence = runner_evidence_store_path(&world.state_store);
std::fs::write(&evidence, damage.as_bytes())?;
let ended = tick_until_ended(&world)?;
// The transaction's own outcome is the runner failure, not the store.
let Some(TransactionCompletion::FailedBeforeFencing { error }) = &ended.completion else {
bail!("the damaged store changed the outcome: {:?} {:?}", ended.phase, ended.last_error);
};
assert!(error.contains("runner step test-daemon on runner rust"), "{error}");
assert!(!error.contains("damage"), "{error}");
// The damage is kept aside; the failure's evidence is in a fresh store.
assert_eq!(std::fs::read(unreadable_store_path(&evidence))?, damage.as_bytes());
let record = runner_evidence::for_transaction(&evidence, &ended.transaction_id)?
.context("the failure kept no evidence beside a damaged store")?;
assert_eq!(record.subject_id, "test-daemon");
assert!(printed_status(&world)?.contains("failed step test-daemon"));
// The journal says it, naming the file and none of its content.
std::fs::write(&evidence, damage.as_bytes())?;
let line = world
.engine
.keep_runner_failure(&ended, &anyhow::Error::new(canary_failure(b"x")))
.context("no journal line")?;
assert!(line.contains("an undecodable evidence store was set aside as"), "{line}");
assert!(line.contains("runner-evidence.unreadable.cc"), "{line}");
assert!(!line.contains("damage") && !line.contains("not a store"), "{line}");
Ok(())
}
#[cfg(unix)]
#[test]
fn an_unreadable_evidence_store_costs_only_its_lines() -> Result<()> {
let (world, _tools) = failing_runner_world()?;
let ended = tick_until_ended(&world)?;
std::fs::write(runner_evidence_store_path(&world.state_store), b"\xc1not a store")?;
let printed = printed_status(&world)?;
assert!(printed.contains(&format!("transaction {}", ended.transaction_id)), "{printed}");
assert!(printed.contains("target service phase"), "{printed}");
assert!(printed.contains("pre-fencing abort"), "{printed}");
assert_eq!(printed.matches("runner evidence unreadable").count(), 1, "{printed}");
Ok(())
}
/// Everything status prints from the evidence store goes through one
/// allowlist, so a hostile build log reaches the terminal as visible
/// escapes whatever its class.
#[cfg(unix)]
#[test]
fn status_prints_hostile_runner_output_as_visible_escapes_only() -> Result<()> {
let (world, _tools) = failing_runner_world()?;
let ended = tick_until_ended(&world)?;
let hostile: Vec<(&str, &str)> = vec![
("\u{1b}[2J", "\\u{1b}[2J"),
("\u{7}", "\\u{7}"),
("\r", "\\u{d}"),
("\u{202e}", "\\u{202e}"),
("\u{2066}", "\\u{2066}"),
("\u{85}", "\\u{85}"),
("\u{e0041}", "\\u{e0041}"),
("\u{200b}", "\\u{200b}"),
("\u{feff}", "\\u{feff}"),
];
let mut tail: Vec<u8> = Vec::new();
for &(raw, _) in &hostile {
tail.extend_from_slice(b"a");
tail.extend_from_slice(raw.as_bytes());
tail.extend_from_slice(b"b\n");
}
tail.extend_from_slice(b"bad \xff\xc1 bytes\n");
runner_evidence::record(
&runner_evidence_store_path(&world.state_store),
&RunnerFailureRecord::from_failure(
&ended.transaction_id,
&ended.command_id,
&ended.target,
&canary_failure(&tail),
9,
),
)?;
let printed = printed_status(&world)?;
for &(raw, escaped) in &hostile {
assert!(!printed.contains(raw), "{escaped} reached the terminal raw");
assert!(printed.contains(&format!("a{escaped}b")), "{escaped} missing: {printed}");
}
assert!(printed.contains("bad \\xff\\xc1 bytes"), "{printed}");
let section = printed.find(" failed step").context("no evidence section")?;
assert!(printed[section..].is_ascii(), "non-ASCII reached the terminal: {printed}");
Ok(())
}
#[cfg(unix)]
#[test]
fn status_prints_the_unreadable_store_line_escaped() -> Result<()> {
let (world, _tools) = failing_runner_world()?;
tick_until_ended(&world)?;
let evidence = runner_evidence_store_path(&world.state_store);
// The decode error of this store quotes the stored subject kind.
let mut bytes = std::fs::read(&evidence)?;
let at = bytes
.windows(4)
.position(|window| window == b"step")
.context("no step marker in the evidence store")?;
bytes[at..at + 4].copy_from_slice(b"\x1b[2J");
std::fs::write(&evidence, bytes)?;
let printed = printed_status(&world)?;
assert_eq!(printed.matches("runner evidence unreadable").count(), 1, "{printed}");
assert!(!printed.contains('\u{1b}'), "an escape byte reached the terminal");
assert!(printed.contains("\\u{1b}[2J"), "{printed}");
Ok(())
}
#[test]
fn the_journal_line_names_a_displaced_set_aside_store_by_time_only() -> Result<()> {
let (transaction, _) = terminal_transaction_with_command("ghostlight")?;
let failure = canary_failure(RUNNER_CANARY.as_bytes());
let path = Path::new("/var/lib/gamecult/idunn/runner-evidence.cc");
let first = runner_failure_journal_line(
&transaction,
&failure,
path,
&Ok(Kept::SetAside { replaced: None }),
);
assert!(first.contains("was set aside as"), "{first}");
assert!(!first.contains("replacing"), "{first}");
let second = runner_failure_journal_line(
&transaction,
&failure,
path,
&Ok(Kept::SetAside { replaced: Some(1234) }),
);
assert!(
second.contains("replacing an earlier set-aside store last written at unix millis 1234"),
"{second}"
);
assert!(!second.contains(RUNNER_CANARY));
Ok(())
}
#[cfg(unix)]
#[test]
fn control_store_types_are_unchanged_by_a_runner_failure() -> Result<()> {
let (world, _tools) = failing_runner_world()?;
tick_until_ended(&world)?;
// The store a rolled-back Idunn reads: it refuses any foreign type.
ControlSnapshot::read(&world.state_store)?;
let mut transactions = 0;
for store in [world.state_store.clone(), history_store_path(&world.state_store)] {
for envelope in SingleFileMessagePackBackingStore::new(&store).pull_all_read_only_snapshot()? {
assert!(!envelope.r#type.contains("runner_failure"), "{}", envelope.r#type);
if envelope.r#type == DeploymentTransaction::TYPE {
transactions += 1;
assert_eq!(envelope.schema_id.as_deref(), Some("idunn.deployment_transaction.v4"));
}
}
}
assert_eq!(transactions, 1, "the failed transaction is in neither store");
Ok(())
}
/// The records control.cc holds re-encode byte for byte. The hex was
/// captured at base 804ed75, before this cut touched anything: a runner
/// failure adds no field, no type and no schema to the control store.
#[test]
fn control_records_encode_exactly_as_they_did_at_base() -> Result<()> {
let command = DeploymentCommand {
schema_version: DEPLOYMENT_COMMAND_SCHEMA.into(),
command_id: "up-service".into(),
kind: CommandKind::Deploy,
selector: "service".into(),
requested_by: "test".into(),
requested_at_unix_millis: 100,
};
let mut transaction = DeploymentTransaction::new(&command, "service".into(), 0, None, 100)?;
transaction.transaction_id = "tx-pinned".into();
transaction.last_error = Some("runner step test-daemon on runner rust exited with exit status 3".into());
transaction.pre_fencing_abort = Some(PreFencingAbort {
error: "runner step test-daemon on runner rust exited with exit status 3".into(),
candidate_cleanup: CleanupEvidence::Skipped,
topology_reconciliation: CleanupEvidence::Skipped,
source_cleanup: CleanupEvidence::Complete,
});
let hex = |bytes: &[u8]| bytes.iter().map(|byte| format!("{byte:02x}")).collect::<String>();
let command_hex = hex(&command_envelope(&command, 100)?.payload);
let transaction_hex = hex(&transaction_envelope(&transaction, 100)?.payload);
assert_eq!(command_hex, PINNED_COMMAND_HEX, "command bytes changed");
assert_eq!(transaction_hex, PINNED_TRANSACTION_HEX, "transaction bytes changed");
Ok(())
}
const PINNED_COMMAND_HEX: &str = "96bb6964756e6e2e6465706c6f796d656e745f636f6d6d616e642e7632aa75702d73657276696365a66465706c6f79a773657276696365a47465737464";
const PINNED_TRANSACTION_HEX: &str = "dc0025bf6964756e6e2e6465706c6f796d656e745f7472616e73616374696f6e2e7634a974782d70696e6e6564aa75702d73657276696365a66465706c6f79a77365727669636500a77365616c696e676464c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c000d94072756e6e6572207374657020746573742d6461656d6f6e206f6e2072756e6e657220727573742065786974656420776974682065786974207374617475732033c094d94072756e6e6572207374657020746573742d6461656d6f6e206f6e2072756e6e657220727573742065786974656420776974682065786974207374617475732033a7736b6970706564a7736b6970706564a8636f6d706c657465c0c0c0c0";
// END runner-failure-evidence tests
} }

473
src/drivers.rs
View file

@ -157,6 +157,24 @@ pub struct FrozenSourceReceipt {
pub snapshot_sha256: String, pub snapshot_sha256: String,
} }
#[cfg(test)]
impl FrozenSource {
/// A frozen source over an existing tree, for tests that drive a runner.
pub(crate) fn fixture(
receipt: FrozenSourceReceipt,
facts: SourceSelectionFacts,
recipe_bytes: Vec<u8>,
root: PathBuf,
) -> Self {
Self {
receipt,
facts,
recipe_bytes,
root,
}
}
}
impl FrozenSourceReceipt { impl FrozenSourceReceipt {
pub fn validate_against(&self, plan: &CompiledDeploymentPlan) -> Result<()> { pub fn validate_against(&self, plan: &CompiledDeploymentPlan) -> Result<()> {
plan.validate()?; plan.validate()?;
@ -2152,6 +2170,115 @@ pub(crate) fn docker_run_args(
Ok(args) Ok(args)
} }
/// Bytes of each output stream that a failed runner step keeps. The daemon
/// never holds more than this per stream while the step runs.
pub(crate) const RUNNER_OUTPUT_TAIL_BYTES: usize = 8192;
/// What a runner container was running: a recipe step or an external input,
/// by the id the recipe gave it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum RunnerSubject {
Step(String),
ExternalInput(String),
}
impl RunnerSubject {
pub(crate) fn id(&self) -> &str {
match self {
Self::Step(id) | Self::ExternalInput(id) => id,
}
}
fn noun(&self) -> &'static str {
match self {
Self::Step(_) => "step",
Self::ExternalInput(_) => "external input",
}
}
}
/// A runner container that did not exit successfully: what failed, how, and
/// the last `RUNNER_OUTPUT_TAIL_BYTES` of each stream with its total size.
/// The tails are evidence for `runner-evidence.cc`. `Display` and `Debug`
/// carry none of them, so no error chain, record or log line renders output.
#[derive(Clone, PartialEq, Eq)]
pub(crate) struct RunnerStepFailure {
pub(crate) subject: RunnerSubject,
pub(crate) runner: String,
pub(crate) exit_code: Option<i32>,
pub(crate) signal: Option<i32>,
pub(crate) stdout_tail: Vec<u8>,
pub(crate) stdout_total_bytes: u64,
pub(crate) stderr_tail: Vec<u8>,
pub(crate) stderr_total_bytes: u64,
}
impl RunnerStepFailure {
/// The status as an operator reads it.
pub(crate) fn status(&self) -> String {
runner_status(self.exit_code, self.signal)
}
}
/// How a container ended, as an operator reads it.
pub(crate) fn runner_status(exit_code: Option<i32>, signal: Option<i32>) -> String {
match (exit_code, signal) {
(Some(code), _) => format!("exit status {code}"),
(None, Some(signal)) => format!("signal {signal}"),
(None, None) => "an unknown status".to_owned(),
}
}
impl std::fmt::Display for RunnerStepFailure {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
formatter,
"runner {} {} on runner {} exited with {}; its output tail is kept in runner-evidence.cc",
self.subject.noun(),
self.subject.id(),
self.runner,
self.status()
)
}
}
impl std::fmt::Debug for RunnerStepFailure {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RunnerStepFailure")
.field("subject", &self.subject)
.field("runner", &self.runner)
.field("exit_code", &self.exit_code)
.field("signal", &self.signal)
.field("stdout_total_bytes", &self.stdout_total_bytes)
.field("stderr_total_bytes", &self.stderr_total_bytes)
.finish_non_exhaustive()
}
}
impl std::error::Error for RunnerStepFailure {}
/// Read `reader` to its end keeping only the last `cap` bytes and counting
/// every byte, so a verbose step never buffers whole in the daemon.
fn drain_tail(mut reader: impl Read, cap: usize) -> std::io::Result<(Vec<u8>, u64)> {
let mut tail = Vec::new();
let mut total = 0u64;
let mut chunk = [0u8; 8192];
loop {
let read = match reader.read(&mut chunk) {
Ok(0) => return Ok((tail, total)),
Ok(read) => read,
Err(error) if error.kind() == ErrorKind::Interrupted => continue,
Err(error) => return Err(error),
};
total += read as u64;
tail.extend_from_slice(&chunk[..read]);
if tail.len() > cap {
tail.drain(..tail.len() - cap);
}
}
}
/// Docker is only a runner substrate. Recipe argv stays an argv vector, the /// Docker is only a runner substrate. Recipe argv stays an argv vector, the
/// operator binding selects the exact image/network/mount affordances, and the /// operator binding selects the exact image/network/mount affordances, and the
/// driver returns complete materialization receipts rather than build truth by /// driver returns complete materialization receipts rather than build truth by
@ -2170,7 +2297,7 @@ impl Default for DockerRunnerDriver {
} }
impl DockerRunnerDriver { impl DockerRunnerDriver {
fn docker<I, S>(&self, args: I) -> Result<Output> fn docker<I, S>(&self, subject: RunnerSubject, runner_id: &str, args: I) -> Result<()>
where where
I: IntoIterator<Item = S>, I: IntoIterator<Item = S>,
S: AsRef<OsStr>, S: AsRef<OsStr>,
@ -2179,23 +2306,50 @@ impl DockerRunnerDriver {
self.docker_program.is_absolute(), self.docker_program.is_absolute(),
"Docker runner program is not absolute" "Docker runner program is not absolute"
); );
let output = Command::new(&self.docker_program) let mut child = Command::new(&self.docker_program)
.args(args) .args(args)
.stdin(Stdio::null()) .stdin(Stdio::null())
.output() .stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.with_context(|| format!("starting Docker runner {}", self.docker_program.display()))?; .with_context(|| format!("starting Docker runner {}", self.docker_program.display()))?;
if !output.status.success() { let stdout = child.stdout.take().context("Docker runner has no stdout pipe")?;
bail!( let stderr = child.stderr.take().context("Docker runner has no stderr pipe")?;
"Docker runner exited with {}: {}", let stdout = thread::spawn(move || drain_tail(stdout, RUNNER_OUTPUT_TAIL_BYTES));
output.status, let stderr = thread::spawn(move || drain_tail(stderr, RUNNER_OUTPUT_TAIL_BYTES));
String::from_utf8_lossy(&output.stderr).trim() let status = child.wait().context("waiting for the Docker runner")?;
); let (stdout_tail, stdout_total_bytes) = stdout
.join()
.map_err(|_| anyhow!("the runner stdout reader panicked"))?
.context("reading the runner stdout")?;
let (stderr_tail, stderr_total_bytes) = stderr
.join()
.map_err(|_| anyhow!("the runner stderr reader panicked"))?
.context("reading the runner stderr")?;
if status.success() {
return Ok(());
} }
Ok(output) #[cfg(unix)]
let signal = std::os::unix::process::ExitStatusExt::signal(&status);
#[cfg(not(unix))]
let signal = None;
Err(RunnerStepFailure {
subject,
runner: runner_id.to_owned(),
exit_code: status.code(),
signal,
stdout_tail,
stdout_total_bytes,
stderr_tail,
stderr_total_bytes,
}
.into())
} }
fn run_in_workspace( fn run_in_workspace(
&self, &self,
step_id: &str,
runner_id: &str,
runner: &DockerRunnerBinding, runner: &DockerRunnerBinding,
workspace: &Path, workspace: &Path,
working_directory: &Path, working_directory: &Path,
@ -2212,8 +2366,7 @@ impl DockerRunnerDriver {
build_machine_id_file(workspace)?; build_machine_id_file(workspace)?;
let spec = ContainerSpec::for_step(runner, required_environment, source_stamp)?; let spec = ContainerSpec::for_step(runner, required_environment, source_stamp)?;
let args = docker_run_args(&spec, workspace, working_directory, argv)?; let args = docker_run_args(&spec, workspace, working_directory, argv)?;
self.docker(args)?; self.docker(RunnerSubject::Step(step_id.to_owned()), runner_id, args)
Ok(())
} }
fn materialize_external_input( fn materialize_external_input(
@ -2255,7 +2408,11 @@ impl DockerRunnerDriver {
input.url.clone(), input.url.clone(),
]; ];
let args = docker_run_args(&spec, workspace, Path::new("."), &curl_argv)?; let args = docker_run_args(&spec, workspace, Path::new("."), &curl_argv)?;
self.docker(args)?; self.docker(
RunnerSubject::ExternalInput(input.id.clone()),
&input.runner,
args,
)?;
let bytes = fs::read(&destination).with_context(|| { let bytes = fs::read(&destination).with_context(|| {
format!( format!(
"reading materialized external input {}", "reading materialized external input {}",
@ -2377,6 +2534,8 @@ impl RunnerPort for DockerRunnerDriver {
); );
} }
self.run_in_workspace( self.run_in_workspace(
&step.id,
&step.runner,
runner, runner,
workspace, workspace,
&step.working_directory, &step.working_directory,
@ -16231,4 +16390,292 @@ processes 9
); );
Ok(()) Ok(())
} }
// BEGIN runner-failure-evidence tests
// The runner decides what failed and what it kept: these drive the real
// DockerRunnerDriver over a fake docker program, never the capture alone.
const RUNNER_CANARY: &str = "CANARY-5d2e91b7";
#[cfg(unix)]
fn fake_docker(dir: &Path, body: &str) -> Result<DockerRunnerDriver> {
use std::os::unix::fs::PermissionsExt;
let program = dir.join("docker");
fs::write(&program, format!("#!/bin/sh\n{body}\n"))?;
fs::set_permissions(&program, fs::Permissions::from_mode(0o755))?;
Ok(DockerRunnerDriver {
docker_program: program,
})
}
/// One recipe step through the real `run_in_workspace` path.
#[cfg(unix)]
fn run_step(driver: &DockerRunnerDriver, dir: &Path, step_id: &str) -> Result<()> {
let workspace = dir.join("workspace");
fs::create_dir_all(&workspace)?;
driver.run_in_workspace(
step_id,
"rust",
&fixture_docker_runner_binding(),
&workspace,
Path::new("."),
&["cargo".to_owned(), "test".to_owned()],
&std::collections::BTreeSet::new(),
("IDUNN_STAMP", "rev"),
)
}
#[cfg(unix)]
fn failure_of(result: Result<()>) -> Result<RunnerStepFailure> {
let error = result.expect_err("the step did not fail");
error
.downcast_ref::<RunnerStepFailure>()
.cloned()
.with_context(|| format!("the failure is not typed: {error:#}"))
}
#[cfg(unix)]
#[test]
fn a_failed_step_is_named_with_both_tails() -> Result<()> {
let temp = tempfile::tempdir()?;
// The first step passes, the second fails: the count is the fake's.
let driver = fake_docker(
temp.path(),
&format!(
"n=$(cat \"$0.count\" 2>/dev/null || echo 0); n=$((n+1)); echo $n > \"$0.count\"\n\
[ \"$n\" -lt 2 ] && exit 0\n\
echo 'stdout line one'; echo 'stdout report {RUNNER_CANARY}'\n\
echo 'stderr report' >&2\nexit 3"
),
)?;
run_step(&driver, temp.path(), "first")?;
let failure = failure_of(run_step(&driver, temp.path(), "second"))?;
assert_eq!(failure.subject, RunnerSubject::Step("second".into()));
assert_eq!(failure.runner, "rust");
assert_eq!(failure.exit_code, Some(3));
assert_eq!(failure.signal, None);
assert!(
failure
.stdout_tail
.ends_with(format!("stdout report {RUNNER_CANARY}\n").as_bytes()),
"stdout was dropped: {:?}",
String::from_utf8_lossy(&failure.stdout_tail)
);
assert_eq!(failure.stderr_tail, b"stderr report\n");
assert_eq!(failure.stdout_total_bytes, failure.stdout_tail.len() as u64);
Ok(())
}
#[cfg(unix)]
#[test]
fn a_step_killed_by_a_signal_is_named_by_its_signal() -> Result<()> {
let temp = tempfile::tempdir()?;
let driver = fake_docker(temp.path(), "kill -9 $$")?;
let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
assert_eq!((failure.exit_code, failure.signal), (None, Some(9)));
assert!(failure.to_string().contains("exited with signal 9"), "{failure}");
Ok(())
}
#[cfg(unix)]
#[test]
fn a_failed_external_input_is_named_as_one() -> Result<()> {
let temp = tempfile::tempdir()?;
let driver = fake_docker(temp.path(), "echo curl failed >&2\nexit 22")?;
let declaration = TargetDeclaration::parse(crate::deployment_plan::tests::RECIPE)?;
let binding = OperatorBinding::parse(crate::deployment_plan::tests::BINDING)?;
let input_id = declaration.external_inputs[0].id.clone();
let workspaces = BTreeMap::from([("rust".to_owned(), temp.path().join("workspace"))]);
fs::create_dir_all(&workspaces["rust"])?;
let error = driver
.materialize_external_input(&declaration, &binding, &workspaces, &input_id)
.expect_err("the download did not fail");
let failure = error
.downcast_ref::<RunnerStepFailure>()
.with_context(|| format!("the failure is not typed: {error:#}"))?;
assert_eq!(failure.subject, RunnerSubject::ExternalInput(input_id.clone()));
assert_eq!(failure.exit_code, Some(22));
assert!(
failure.to_string().contains(&format!("runner external input {input_id} on runner rust")),
"{failure}"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn runner_output_never_reaches_the_error_text() -> Result<()> {
let temp = tempfile::tempdir()?;
let driver = fake_docker(
temp.path(),
&format!("echo 'out {RUNNER_CANARY}'\necho 'err {RUNNER_CANARY}' >&2\nexit 1"),
)?;
let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
assert!(String::from_utf8_lossy(&failure.stdout_tail).contains(RUNNER_CANARY));
assert!(String::from_utf8_lossy(&failure.stderr_tail).contains(RUNNER_CANARY));
let error = anyhow::Error::new(failure.clone()).context("sealing");
for text in [
format!("{error:#}"),
format!("{error:?}"),
format!("{failure}"),
format!("{failure:?}"),
] {
assert!(!text.contains(RUNNER_CANARY), "runner output reached text: {text}");
}
Ok(())
}
/// A reader that answers from a script, so a read error the OS rarely
/// produces (an interrupted call) is produced on demand.
struct ScriptedReader(Vec<std::io::Result<Vec<u8>>>);
impl Read for ScriptedReader {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
if self.0.is_empty() {
return Ok(0);
}
let bytes = self.0.remove(0)?;
buffer[..bytes.len()].copy_from_slice(&bytes);
Ok(bytes.len())
}
}
struct BrokenReader;
impl Read for BrokenReader {
fn read(&mut self, _: &mut [u8]) -> std::io::Result<usize> {
Err(std::io::Error::other("persistent read failure"))
}
}
#[test]
fn an_interrupted_read_is_retried_and_loses_no_byte() {
let reader = ScriptedReader(vec![
Ok(b"ab".to_vec()),
Err(std::io::Error::from(ErrorKind::Interrupted)),
Ok(b"cd".to_vec()),
Err(std::io::Error::from(ErrorKind::Interrupted)),
]);
let drained = drain_tail(reader, 3).expect("an interrupted read ended the drain");
assert_eq!(drained, (b"bcd".to_vec(), 4));
}
#[test]
fn a_persistent_read_error_ends_the_drain_instead_of_spinning() {
let (sender, receiver) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = sender.send(drain_tail(BrokenReader, 8).is_err());
});
assert_eq!(
receiver.recv_timeout(Duration::from_secs(5)),
Ok(true),
"a persistent read error spun or was swallowed"
);
}
#[test]
fn a_non_interrupted_read_error_after_data_is_returned_not_retried() {
let reader = ScriptedReader(vec![
Ok(b"ab".to_vec()),
Err(std::io::Error::from(ErrorKind::BrokenPipe)),
Ok(b"cd".to_vec()),
]);
let error = drain_tail(reader, 8).expect_err("a broken pipe was retried past");
assert_eq!(error.kind(), ErrorKind::BrokenPipe);
}
#[cfg(unix)]
#[test]
fn each_tail_is_the_last_bound_bytes_with_the_total() -> Result<()> {
let temp = tempfile::tempdir()?;
let data = temp.path().join("data");
for size in [
0,
1,
RUNNER_OUTPUT_TAIL_BYTES - 1,
RUNNER_OUTPUT_TAIL_BYTES,
RUNNER_OUTPUT_TAIL_BYTES + 1,
1024 * 1024,
] {
// Distinct neighbours, so a shifted or head-keeping tail differs.
let bytes: Vec<u8> = (0..size).map(|index| (index % 251) as u8).collect();
fs::write(&data, &bytes)?;
let driver = fake_docker(
temp.path(),
&format!("cat '{}'\ncat '{}' >&2\nexit 1", data.display(), data.display()),
)?;
let failure = failure_of(run_step(&driver, temp.path(), "build"))?;
let expected = &bytes[size.saturating_sub(RUNNER_OUTPUT_TAIL_BYTES)..];
assert_eq!(failure.stdout_tail, expected, "stdout tail at {size} bytes");
assert_eq!(failure.stderr_tail, expected, "stderr tail at {size} bytes");
assert_eq!(failure.stdout_total_bytes, size as u64);
assert_eq!(failure.stderr_total_bytes, size as u64);
}
Ok(())
}
#[cfg(unix)]
#[test]
fn a_successful_step_and_a_missing_program_keep_their_outcomes() -> Result<()> {
let temp = tempfile::tempdir()?;
run_step(&fake_docker(temp.path(), "echo fine")?, temp.path(), "build")?;
let missing = DockerRunnerDriver {
docker_program: temp.path().join("no-such-docker"),
};
let error = run_step(&missing, temp.path(), "build").expect_err("a missing program ran");
assert!(error.downcast_ref::<RunnerStepFailure>().is_none());
assert!(format!("{error:#}").contains("starting Docker runner"), "{error:#}");
Ok(())
}
/// The 2026-10-25 vili failure: `node --test` reported on stdout alone,
/// stderr stayed empty (the runner's /tmp is a noexec tmpfs) and the step
/// exited 1. The old capture dropped stdout and printed an empty message.
#[cfg(unix)]
#[test]
fn a_step_that_reports_only_on_stdout_keeps_that_report_and_its_name() -> Result<()> {
let temp = tempfile::tempdir()?;
let driver = fake_docker(
temp.path(),
&format!("echo 'not ok 1 - daemon {RUNNER_CANARY}'\necho '# fail 1'\nexit 1"),
)?;
let error = run_step(&driver, temp.path(), "test-daemon").expect_err("the step passed");
// The error names the step and carries no output.
let text = format!("{error:#}");
assert!(text.contains("runner step test-daemon on runner rust exited with exit status 1"), "{text}");
assert!(!text.contains(RUNNER_CANARY) && !text.contains("not ok"), "{text}");
// The evidence keeps the stdout tail; the empty stderr is a recorded 0.
let failure = error
.downcast_ref::<RunnerStepFailure>()
.context("the failure is not typed")?;
let store = temp.path().join("runner-evidence.cc");
let record = crate::runner_evidence::RunnerFailureRecord::from_failure(
"tx-vili",
"up-vili",
"vili",
failure,
1,
);
crate::runner_evidence::record(&store, &record)?;
let kept = crate::runner_evidence::for_transaction(&store, "tx-vili")?.context("no evidence kept")?;
assert_eq!(kept.subject_id, "test-daemon");
assert_eq!(kept.exit_code, Some(1));
assert_eq!(
String::from_utf8_lossy(&kept.stdout_tail),
format!("not ok 1 - daemon {RUNNER_CANARY}\n# fail 1\n")
);
assert_eq!(kept.stdout_total_bytes, kept.stdout_tail.len() as u64);
assert!(kept.stderr_tail.is_empty());
assert_eq!(kept.stderr_total_bytes, 0);
let mut shown = Vec::new();
crate::runner_evidence::show_status(&mut shown, &store, "tx-vili")?;
let rendered = String::from_utf8(shown)?;
assert!(rendered.contains("stderr (last 0 of 0 bytes)"), "{rendered}");
assert!(rendered.contains(&format!("| not ok 1 - daemon {RUNNER_CANARY}")), "{rendered}");
// Debug names the failure and prints no tail.
let debug = format!("{failure:?}");
assert!(debug.starts_with("RunnerStepFailure") && !debug.contains("not ok"), "{debug}");
Ok(())
}
// END runner-failure-evidence tests
} }

1
src/lib.rs
View file

@ -7,4 +7,5 @@ pub mod host_actuator;
#[cfg(unix)] #[cfg(unix)]
pub mod launcher; pub mod launcher;
mod incident; mod incident;
mod runner_evidence;
pub mod provisioning; pub mod provisioning;

805
src/runner_evidence.rs Normal file
View file

@ -0,0 +1,805 @@
//! The output tail of a failed runner step.
//!
//! One file, `runner-evidence.cc`, beside `control.cc`, holds one type: the
//! last bytes of a failed runner container's stdout and stderr, one record per
//! transaction. The engine's pre-fence failure path is its only writer, and
//! `idunn status --command` its only reader. It is never `control.cc`, which
//! refuses any type it does not know, and never `incidents.cc`, which is
//! published and carries no free text.
//!
//! The tail is raw runner output. It is kept unredacted and readable by root
//! alone (mode 0600): root can already read every secret the runner was given.
//! It never enters an error, a transaction record or the journal. The status
//! reader is a terminal, so everything it prints from this store goes through
//! one allowlist sink (`emit`): printable ASCII passes, all else is escaped.
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::{
LockContended, RUNNER_OUTPUT_TAIL_BYTES, RunnerStepFailure, RunnerSubject,
exchange_behind_private_lock, runner_status,
};
const RUNNER_FAILURE_SCHEMA: &str = "idunn.runner_failure.v1";
/// Records kept; the writer drops the oldest past this, so the store needs no
/// other owner.
pub(crate) const RUNNER_EVIDENCE_RECORDS: usize = 32;
/// `runner-evidence.cc`, beside the control store.
pub(crate) fn runner_evidence_store_path(state_store: &Path) -> PathBuf {
state_store.with_file_name("runner-evidence.cc")
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum SubjectKind {
Step,
ExternalInput,
}
#[derive(Clone, PartialEq, Eq, DatabaseEntry)]
#[cultcache(type = "idunn.runner_failure", schema = "idunn.runner_failure.v1")]
pub(crate) struct RunnerFailureRecord {
#[cultcache(key = 0)]
pub(crate) schema_version: String,
#[cultcache(key = 1)]
pub(crate) transaction_id: String,
#[cultcache(key = 2)]
pub(crate) command_id: String,
#[cultcache(key = 3)]
pub(crate) target: String,
#[cultcache(key = 4)]
pub(crate) subject_kind: SubjectKind,
#[cultcache(key = 5)]
pub(crate) subject_id: String,
#[cultcache(key = 6)]
pub(crate) runner: String,
#[cultcache(key = 7)]
pub(crate) exit_code: Option<i32>,
#[cultcache(key = 8)]
pub(crate) signal: Option<i32>,
#[cultcache(key = 9)]
pub(crate) stdout_tail: Vec<u8>,
#[cultcache(key = 10)]
pub(crate) stdout_total_bytes: u64,
#[cultcache(key = 11)]
pub(crate) stderr_tail: Vec<u8>,
#[cultcache(key = 12)]
pub(crate) stderr_total_bytes: u64,
#[cultcache(key = 13)]
pub(crate) recorded_at_unix_millis: u64,
}
/// Tails stay out of `Debug`, so a failed assertion or a logged record never
/// prints runner output.
impl std::fmt::Debug for RunnerFailureRecord {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RunnerFailureRecord")
.field("transaction_id", &self.transaction_id)
.field("subject_id", &self.subject_id)
.finish_non_exhaustive()
}
}
impl RunnerFailureRecord {
pub(crate) fn from_failure(
transaction_id: &str,
command_id: &str,
target: &str,
failure: &RunnerStepFailure,
now: u64,
) -> Self {
let (subject_kind, subject_id) = match &failure.subject {
RunnerSubject::Step(id) => (SubjectKind::Step, id),
RunnerSubject::ExternalInput(id) => (SubjectKind::ExternalInput, id),
};
Self {
schema_version: RUNNER_FAILURE_SCHEMA.to_owned(),
transaction_id: transaction_id.to_owned(),
command_id: command_id.to_owned(),
target: target.to_owned(),
subject_kind,
subject_id: subject_id.clone(),
runner: failure.runner.clone(),
exit_code: failure.exit_code,
signal: failure.signal,
stdout_tail: failure.stdout_tail.clone(),
stdout_total_bytes: failure.stdout_total_bytes,
stderr_tail: failure.stderr_tail.clone(),
stderr_total_bytes: failure.stderr_total_bytes,
recorded_at_unix_millis: now,
}
}
pub(crate) fn validate(&self) -> Result<()> {
ensure!(
self.schema_version == RUNNER_FAILURE_SCHEMA,
"runner failure schema is unsupported"
);
require_id(&self.transaction_id, "runner failure transaction")?;
require_id(&self.command_id, "runner failure command")?;
require_id(&self.target, "runner failure target")?;
require_id(&self.subject_id, "runner failure subject")?;
require_id(&self.runner, "runner failure runner")?;
for (tail, total, stream) in [
(&self.stdout_tail, self.stdout_total_bytes, "stdout"),
(&self.stderr_tail, self.stderr_total_bytes, "stderr"),
] {
ensure!(
tail.len() <= RUNNER_OUTPUT_TAIL_BYTES,
"runner failure {stream} tail exceeds its bound"
);
ensure!(
tail.len() as u64 <= total,
"runner failure {stream} tail exceeds its stream"
);
}
Ok(())
}
fn envelope(&self) -> Result<CultCacheEnvelope> {
self.validate()?;
typed_envelope(
&self.transaction_id,
Self::TYPE,
RUNNER_FAILURE_SCHEMA,
self,
self.recorded_at_unix_millis,
)
}
}
/// One read of the store: the envelopes exactly as stored, and the records
/// decoded from them in the same order.
fn snapshot(path: &Path) -> Result<(Vec<CultCacheEnvelope>, Vec<RunnerFailureRecord>)> {
let envelopes = SingleFileMessagePackBackingStore::new(path)
.pull_all_read_only_snapshot()
.context("reading the runner evidence store")?;
let mut records = Vec::new();
for envelope in &envelopes {
ensure!(
envelope.r#type == RunnerFailureRecord::TYPE
&& envelope.schema_id.as_deref() == Some(RUNNER_FAILURE_SCHEMA),
"the runner evidence store holds a document that is not a runner failure"
);
let record: RunnerFailureRecord = decode_record(envelope)?;
record.validate()?;
ensure!(
envelope.key == record.transaction_id,
"runner evidence store key differs from the record's transaction"
);
records.push(record);
}
Ok((envelopes, records))
}
/// How a write left the store.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Kept {
Stored,
/// The store held bytes that do not decode; it was moved to
/// `unreadable_store_path` and the record opened a fresh one. `replaced`
/// is the last-written time (unix milliseconds) of an earlier set-aside
/// store this one displaced, which is gone: one slot, and it says so.
SetAside { replaced: Option<u64> },
}
/// Where an undecodable store is kept. One slot: the next one replaces it, so
/// the damage a bad store can cost is one extra file.
pub(crate) fn unreadable_store_path(path: &Path) -> PathBuf {
path.with_file_name("runner-evidence.unreadable.cc")
}
/// Moves the store over the previous set-aside one, root-only, and reports
/// what it displaced. Nothing of the store's content is named: a decode error
/// can quote stored values, and none is carried here.
fn set_aside_unreadable(path: &Path) -> Result<Option<u64>> {
let aside = unreadable_store_path(path);
let replaced = std::fs::symlink_metadata(&aside).ok().map(|earlier| {
earlier
.modified()
.ok()
.and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
.map_or(0, |since| since.as_millis() as u64)
});
std::fs::rename(path, &aside)
.context("the runner evidence store is undecodable and could not be set aside")?;
keep_private(&aside)?;
Ok(replaced)
}
/// How the writer reads the store's bytes; tests inject an I/O error here.
type ReadStore<'a> = &'a dyn Fn(&Path) -> std::io::Result<Vec<u8>>;
fn unreadable_fault(error: &std::io::Error) -> anyhow::Error {
anyhow::anyhow!("the runner evidence store could not be read: {}", error.kind())
}
/// The one writer. Upsert by transaction id; past `RUNNER_EVIDENCE_RECORDS` the
/// oldest records go in the same exchange. The file ends at mode 0600.
///
/// A store is set aside only when bytes that were read do not decode. An I/O
/// error leaves the store in place, keeps no evidence for that failure, and
/// the error names its kind for the journal.
pub(crate) fn record(path: &Path, record: &RunnerFailureRecord) -> Result<Kept> {
record_reading(path, record, &|path| std::fs::read(path))
}
fn record_reading(path: &Path, record: &RunnerFailureRecord, read: ReadStore) -> Result<Kept> {
let first = match read(path) {
Ok(bytes) => Some(bytes),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
Err(error) => return Err(unreadable_fault(&error)),
};
let (envelopes, records, how) = match snapshot(path) {
Ok((envelopes, records)) => (envelopes, records, Kept::Stored),
Err(_) => {
let Some(first) = first else {
bail!("the runner evidence store could not be decoded");
};
// Undecodable only if the same bytes are still there to read: a
// second read that fails, or finds other bytes, is not a verdict.
match read(path) {
Ok(second) if second == first => {}
Ok(_) => bail!("the runner evidence store changed while it was being read"),
Err(error) => return Err(unreadable_fault(&error)),
}
let replaced = set_aside_unreadable(path)?;
(Vec::new(), Vec::new(), Kept::SetAside { replaced })
}
};
let mut kept: Vec<(u64, CultCacheEnvelope)> = envelopes
.iter()
.zip(&records)
.filter(|(_, stored)| stored.transaction_id != record.transaction_id)
.map(|(envelope, stored)| (stored.recorded_at_unix_millis, envelope.clone()))
.collect();
kept.push((record.recorded_at_unix_millis, record.envelope()?));
// Stable: records of one instant keep their stored order, so the record
// being written survives a tie.
kept.sort_by_key(|(recorded_at, _)| *recorded_at);
let excess = kept.len().saturating_sub(RUNNER_EVIDENCE_RECORDS);
let replacements: Vec<_> = kept
.into_iter()
.skip(excess)
.map(|(_, envelope)| envelope)
.collect();
match exchange_behind_private_lock(path, &envelopes, &replacements)? {
TryCompareExchangeSnapshotOutcome::Exchanged => {}
TryCompareExchangeSnapshotOutcome::Mismatch => {
bail!("the runner evidence store changed while it was being written")
}
TryCompareExchangeSnapshotOutcome::LockContended => {
return Err(LockContended {
store: "runner evidence store",
}
.into());
}
}
keep_private(path)?;
Ok(how)
}
#[cfg(unix)]
fn keep_private(path: &Path) -> Result<()> {
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
.context("making the runner evidence store private")
}
#[cfg(not(unix))]
fn keep_private(_path: &Path) -> Result<()> {
Ok(())
}
/// The one reader. An absent store holds nothing.
pub(crate) fn for_transaction(
path: &Path,
transaction_id: &str,
) -> Result<Option<RunnerFailureRecord>> {
Ok(snapshot(path)?
.1
.into_iter()
.find(|stored| stored.transaction_id == transaction_id))
}
/// The record as `idunn status --command` shows it, as raw lines. They are
/// not text yet: `show_status` is the only caller, and `emit` the only
/// writer.
fn render(record: &RunnerFailureRecord) -> Vec<Vec<u8>> {
let noun = match record.subject_kind {
SubjectKind::Step => "step",
SubjectKind::ExternalInput => "external input",
};
let mut lines: Vec<Vec<u8>> = vec![
format!(" failed {noun} {}", record.subject_id).into_bytes(),
format!(" runner {}", record.runner).into_bytes(),
format!(" status {}", runner_status(record.exit_code, record.signal)).into_bytes(),
];
for (stream, tail, total) in [
("stdout", &record.stdout_tail, record.stdout_total_bytes),
("stderr", &record.stderr_tail, record.stderr_total_bytes),
] {
lines.push(format!(" {stream} (last {} of {total} bytes)", tail.len()).into_bytes());
let text = tail.strip_suffix(b"\n").unwrap_or(tail);
if !text.is_empty() {
lines.extend(text.split(|byte| *byte == b'\n').map(|line| {
let mut shown = b" | ".to_vec();
shown.extend_from_slice(line);
shown
}));
}
}
lines
}
/// What `idunn status --command` prints for a transaction's runner evidence:
/// its record, or one line naming why the store is unreadable. Nothing is
/// written when the transaction has no record.
pub(crate) fn show_status(
out: &mut impl std::io::Write,
store: &Path,
transaction_id: &str,
) -> std::io::Result<()> {
let lines = match for_transaction(store, transaction_id) {
Ok(Some(record)) => render(&record),
Ok(None) => return Ok(()),
Err(error) => {
vec![format!(" runner evidence unreadable: {}", error.root_cause()).into_bytes()]
}
};
emit(out, &lines)
}
/// The one sink: every byte the status read prints from the evidence store,
/// or from an error about it, is written here and nowhere else. An allowlist:
/// printable ASCII and tab pass; every other code point is a visible
/// `\u{..}` escape and every invalid UTF-8 byte a `\x..` escape. A newline
/// inside a line is escaped too: this function writes the only ones.
fn emit(out: &mut impl std::io::Write, lines: &[Vec<u8>]) -> std::io::Result<()> {
for line in lines {
out.write_all(escape(line).as_bytes())?;
out.write_all(b"\n")?;
}
Ok(())
}
fn escape(bytes: &[u8]) -> String {
let mut shown = String::with_capacity(bytes.len());
for chunk in bytes.utf8_chunks() {
for character in chunk.valid().chars() {
match character {
' '..='~' | '\t' => shown.push(character),
_ => shown.push_str(&format!("\\u{{{:x}}}", u32::from(character))),
}
}
for byte in chunk.invalid() {
shown.push_str(&format!("\\x{byte:02x}"));
}
}
shown
}
#[cfg(all(test, unix))]
mod tests {
use super::*;
use std::os::unix::fs::PermissionsExt;
fn failure(subject: RunnerSubject, stdout: &[u8], stderr: &[u8]) -> RunnerStepFailure {
RunnerStepFailure {
subject,
runner: "rust".into(),
exit_code: Some(3),
signal: None,
stdout_tail: stdout.to_vec(),
stdout_total_bytes: stdout.len() as u64,
stderr_tail: stderr.to_vec(),
stderr_total_bytes: stderr.len() as u64,
}
}
/// What status writes for a record, through the one sink.
fn shown(record: &RunnerFailureRecord) -> String {
let mut written = Vec::new();
emit(&mut written, &render(record)).expect("a Vec takes any write");
String::from_utf8(written).expect("the sink writes ASCII only")
}
fn shown_lines(record: &RunnerFailureRecord) -> Vec<String> {
shown(record).lines().map(str::to_owned).collect()
}
fn record_at(transaction: &str, now: u64) -> RunnerFailureRecord {
RunnerFailureRecord::from_failure(
transaction,
"up-service",
"service",
&failure(RunnerSubject::Step("build".into()), b"out\n", b"err\n"),
now,
)
}
#[test]
fn the_evidence_store_keeps_the_newest_32() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
// The literal, not the constant: the spec's bound is 32.
for index in 0..=32 {
record(&path, &record_at(&format!("tx-{index:02}"), 1000 + index as u64))?;
}
let kept: Vec<_> = snapshot(&path)?.1.into_iter().map(|stored| stored.transaction_id).collect();
assert_eq!(kept.len(), 32);
assert!(!kept.contains(&"tx-00".to_owned()), "the oldest survived");
assert!(kept.contains(&"tx-32".to_owned()), "the newest was evicted");
assert!(kept.contains(&"tx-01".to_owned()), "the second oldest was evicted early");
// Rewriting a resident transaction replaces it and drops nothing.
let mut again = record_at("tx-05", 5000);
again.stdout_tail = b"rewritten\n".to_vec();
again.stdout_total_bytes = 10;
record(&path, &again)?;
let kept = snapshot(&path)?.1;
assert_eq!(kept.len(), 32);
assert!(kept.iter().any(|stored| stored.transaction_id == "tx-01"));
let stored = for_transaction(&path, "tx-05")?.context("rewritten record is gone")?;
assert_eq!(stored.stdout_tail, b"rewritten\n");
assert_eq!(for_transaction(&path, "tx-00")?, None);
Ok(())
}
#[test]
fn a_record_round_trips_byte_for_byte() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
let written = RunnerFailureRecord::from_failure(
"tx-a",
"up-service",
"service",
&failure(RunnerSubject::ExternalInput("toolchain".into()), &[0, 1, 255, 10], b""),
77,
);
record(&path, &written)?;
assert_eq!(for_transaction(&path, "tx-a")?, Some(written));
assert_eq!(for_transaction(&path, "tx-b")?, None);
Ok(())
}
#[test]
fn the_evidence_store_is_private() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
// Whatever the umask, the mode is the store's own rule, on every write.
for index in 0..2 {
record(&path, &record_at(&format!("tx-{index}"), 10 + index))?;
assert_eq!(std::fs::metadata(&path)?.permissions().mode() & 0o777, 0o600);
}
Ok(())
}
#[test]
fn a_tail_over_its_bound_or_its_stream_is_refused() {
let mut oversize = record_at("tx-a", 1);
oversize.stdout_tail = vec![b'x'; RUNNER_OUTPUT_TAIL_BYTES + 1];
oversize.stdout_total_bytes = u64::MAX;
assert!(oversize.validate().is_err());
let mut impossible = record_at("tx-a", 1);
impossible.stderr_total_bytes = 1;
assert!(impossible.validate().is_err());
let mut exact = record_at("tx-a", 1);
exact.stdout_tail = vec![b'x'; RUNNER_OUTPUT_TAIL_BYTES];
exact.stdout_total_bytes = RUNNER_OUTPUT_TAIL_BYTES as u64;
assert!(exact.validate().is_ok());
}
#[test]
fn status_renders_the_tail_with_control_bytes_escaped() {
let mut stored = record_at("tx-a", 1);
stored.stdout_tail = b"first\n\x1b[2Jsecond\x00end\tTab\nlast\n".to_vec();
stored.stdout_total_bytes = 5000;
stored.stderr_tail = b"bad\r\x07".to_vec();
stored.stderr_total_bytes = 6;
let lines = shown_lines(&stored);
let text = lines.join("\n");
assert!(!text.contains('\u{1b}') && !text.contains('\0') && !text.contains('\r'));
assert!(!text.contains('\u{7}'));
assert_eq!(
lines,
vec![
" failed step build",
" runner rust",
" status exit status 3",
" stdout (last 30 of 5000 bytes)",
" | first",
" | \\u{1b}[2Jsecond\\u{0}end\tTab",
" | last",
" stderr (last 5 of 6 bytes)",
" | bad\\u{d}\\u{7}",
]
);
}
#[test]
fn debug_names_the_record_and_prints_no_tail() {
let mut stored = record_at("tx-a", 1);
stored.stdout_tail = b"SECRET-OUTPUT".to_vec();
stored.stdout_total_bytes = 13;
let debug = format!("{stored:?}");
assert!(debug.contains("tx-a") && !debug.contains("SECRET"), "{debug}");
}
#[test]
fn an_empty_tail_renders_only_its_header() {
let mut stored = record_at("tx-a", 1);
stored.stdout_tail = Vec::new();
stored.stdout_total_bytes = 0;
let lines = shown_lines(&stored);
assert!(lines.contains(&" stdout (last 0 of 0 bytes)".to_owned()));
assert!(!lines.iter().any(|line| line == " | "));
}
#[test]
fn the_reader_names_an_unreadable_store_as_an_error() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
std::fs::write(&path, b"\xc1 not a store")?;
assert!(for_transaction(&path, "tx-a").is_err());
Ok(())
}
fn files_beside(path: &Path) -> Result<Vec<String>> {
let mut names: Vec<String> = std::fs::read_dir(path.parent().context("no parent")?)?
.map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned()))
.collect::<Result<_>>()?;
names.retain(|name| name.starts_with("runner-evidence") && !name.contains("lock"));
names.sort();
Ok(names)
}
/// A store a newer Idunn wrote: well-formed, but a schema this one does not know.
fn write_future_store(path: &Path) -> Result<()> {
let mut future = record_at("tx-future", 1);
future.schema_version = "idunn.runner_failure.v2".into();
let envelope = typed_envelope(
&future.transaction_id,
RunnerFailureRecord::TYPE,
"idunn.runner_failure.v2",
&future,
1,
)?;
let outcome = exchange_behind_private_lock(path, &[], &[envelope])?;
ensure!(
matches!(outcome, TryCompareExchangeSnapshotOutcome::Exchanged),
"the future store was not written"
);
Ok(())
}
#[test]
fn a_corrupt_store_is_set_aside_and_the_next_failure_keeps_its_evidence() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
let damage = b"\xc1 not a store \x00 damage";
std::fs::write(&path, damage)?;
let kept = record(&path, &record_at("tx-a", 5))?;
assert_eq!(kept, Kept::SetAside { replaced: None });
// Kept, not deleted: the damage is where the journal says, root-only.
let aside = unreadable_store_path(&path);
assert_eq!(std::fs::read(&aside)?, damage);
assert_eq!(std::fs::metadata(&aside)?.permissions().mode() & 0o777, 0o600);
// The failure's evidence is in a fresh, private store.
assert_eq!(
for_transaction(&path, "tx-a")?.map(|stored| stored.transaction_id),
Some("tx-a".to_owned())
);
assert_eq!(std::fs::metadata(&path)?.permissions().mode() & 0o777, 0o600);
// A healthy store is not set aside again.
assert_eq!(record(&path, &record_at("tx-b", 6))?, Kept::Stored);
assert_eq!(std::fs::read(&aside)?, damage);
Ok(())
}
#[test]
fn a_future_version_store_is_set_aside_and_the_next_failure_keeps_its_evidence() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
write_future_store(&path)?;
let before = std::fs::read(&path)?;
assert!(for_transaction(&path, "tx-future").is_err());
assert_eq!(
record(&path, &record_at("tx-a", 5))?,
Kept::SetAside { replaced: None }
);
assert_eq!(std::fs::read(unreadable_store_path(&path))?, before);
assert!(for_transaction(&path, "tx-a")?.is_some());
assert_eq!(for_transaction(&path, "tx-future")?, None);
Ok(())
}
#[test]
fn at_most_one_unreadable_store_is_kept() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
let damages: [&[u8]; 3] = [b"\xc1 first", b"\xc1 second", b"\xc1 third"];
for (index, damage) in damages.into_iter().enumerate() {
std::fs::write(&path, damage)?;
record(&path, &record_at(&format!("tx-{index}"), 5 + index as u64))?;
assert_eq!(
std::fs::read(unreadable_store_path(&path))?,
damage,
"the newest damage is the one kept"
);
}
assert_eq!(
files_beside(&path)?,
vec![
"runner-evidence.cc".to_owned(),
"runner-evidence.unreadable.cc".to_owned()
]
);
Ok(())
}
#[test]
fn a_store_that_cannot_be_read_is_left_in_place_and_the_write_fails() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
// A directory where the store should be: unreadable, not undecodable.
std::fs::create_dir(&path)?;
assert!(record(&path, &record_at("tx-a", 1)).is_err());
assert!(path.is_dir());
assert!(!unreadable_store_path(&path).exists());
Ok(())
}
#[test]
fn every_code_point_but_printable_ascii_is_a_visible_escape() {
// The allowlist, exhaustively: no class list decides anything.
for code in 0..=0x10ffffu32 {
let Some(character) = char::from_u32(code) else {
continue;
};
let text = character.to_string();
let shown = escape(text.as_bytes());
if matches!(character, ' '..='~' | '\t') {
assert_eq!(shown, text, "U+{code:04X} should pass");
} else {
assert_eq!(shown, format!("\\u{{{code:x}}}"), "U+{code:04X}");
}
}
// Invalid UTF-8 is escaped byte by byte; valid text round it passes.
assert_eq!(escape(b"a\xc1\xffb\xe2\x82"), "a\\xc1\\xffb\\xe2\\x82");
assert_eq!(escape(b"ok \xe2\x82\xac"), "ok \\u{20ac}");
}
#[test]
fn a_newline_inside_a_line_cannot_forge_a_line() {
let mut stored = record_at("tx-a", 1);
stored.subject_id = "build\n status exit status 0".into();
let lines = shown_lines(&stored);
assert_eq!(lines[0], " failed step build\\u{a} status exit status 0");
assert_eq!(
lines.iter().filter(|line| line.starts_with(" status ")).count(),
1
);
}
/// A store whose first record is well-formed except that its subject kind
/// holds `marker`: the decode error quotes it.
fn store_with_hostile_subject_kind(path: &Path, marker: &[u8; 4]) -> Result<()> {
record(path, &record_at("tx-a", 1))?;
let mut bytes = std::fs::read(path)?;
let at = bytes
.windows(4)
.position(|window| window == b"step")
.context("no step marker")?;
bytes[at..at + 4].copy_from_slice(marker);
std::fs::write(path, bytes)?;
Ok(())
}
#[test]
fn the_unreadable_store_line_is_escaped_with_the_rest() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
store_with_hostile_subject_kind(&path, b"\x1b[2J")?;
let mut written = Vec::new();
show_status(&mut written, &path, "tx-a")?;
assert!(!written.contains(&0x1b), "an escape byte reached the terminal");
assert!(written.iter().all(|byte| matches!(byte, b' '..=b'~' | b'\t' | b'\n')));
let text = String::from_utf8(written)?;
assert!(text.starts_with(" runner evidence unreadable: "), "{text}");
assert!(text.contains("\\u{1b}[2J"), "the quoted value was not shown escaped: {text}");
Ok(())
}
#[test]
fn a_failed_set_aside_names_itself_keeps_the_store_and_echoes_nothing() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
store_with_hostile_subject_kind(&path, b"CNRY")?;
let damage = std::fs::read(&path)?;
// Sanity: this damage decodes to an error that quotes the canary.
assert!(format!("{:#}", snapshot(&path).unwrap_err()).contains("CNRY"));
// A non-empty directory where the aside goes: rename fails even as root.
let aside = unreadable_store_path(&path);
std::fs::create_dir(&aside)?;
std::fs::write(aside.join("occupied"), b"x")?;
let error = record(&path, &record_at("tx-b", 2)).unwrap_err();
for text in [format!("{error:#}"), format!("{error:?}")] {
assert!(text.contains("could not be set aside"), "{text}");
assert!(!text.contains("CNRY"), "stored content reached the error: {text}");
}
// The failure cost nothing: the store is as it was, with no evidence
// written over it.
assert_eq!(std::fs::read(&path)?, damage);
assert!(aside.join("occupied").exists());
Ok(())
}
#[test]
fn an_io_error_leaves_the_store_in_place_and_keeps_no_evidence() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
record(&path, &record_at("tx-a", 1))?;
let before = std::fs::read(&path)?;
let denied = || std::io::Error::from(std::io::ErrorKind::PermissionDenied);
// The first read fails on a healthy store.
let error = record_reading(&path, &record_at("tx-b", 2), &|_| Err(denied())).unwrap_err();
assert!(format!("{error:#}").contains("permission denied"), "{error:#}");
assert_eq!(std::fs::read(&path)?, before, "the store changed");
assert!(!unreadable_store_path(&path).exists());
assert!(for_transaction(&path, "tx-b")?.is_none());
// The second read, the one that confirms the bytes do not decode, fails.
std::fs::write(&path, b"\xc1 not a store")?;
let calls = std::cell::Cell::new(0usize);
let read = |path: &Path| {
calls.set(calls.get() + 1);
if calls.get() == 2 {
return Err(denied());
}
std::fs::read(path)
};
let error = record_reading(&path, &record_at("tx-b", 2), &read).unwrap_err();
assert!(format!("{error:#}").contains("permission denied"), "{error:#}");
assert_eq!(std::fs::read(&path)?, b"\xc1 not a store");
assert!(!unreadable_store_path(&path).exists());
// Bytes that changed between the reads are no verdict either.
let calls = std::cell::Cell::new(0usize);
let read = |path: &Path| -> std::io::Result<Vec<u8>> {
calls.set(calls.get() + 1);
let mut bytes = std::fs::read(path)?;
bytes.push(calls.get() as u8);
Ok(bytes)
};
assert!(record_reading(&path, &record_at("tx-b", 2), &read).is_err());
assert_eq!(std::fs::read(&path)?, b"\xc1 not a store");
assert!(!unreadable_store_path(&path).exists());
Ok(())
}
#[test]
fn a_second_set_aside_says_it_replaced_the_first() -> Result<()> {
let temp = tempfile::tempdir()?;
let path = runner_evidence_store_path(&temp.path().join("control.cc"));
std::fs::write(&path, b"\xc1 first")?;
assert_eq!(
record(&path, &record_at("tx-a", 1))?,
Kept::SetAside { replaced: None }
);
std::fs::write(&path, b"\xc1 second")?;
let Kept::SetAside { replaced: Some(_) } = record(&path, &record_at("tx-b", 2))? else {
bail!("a displaced set-aside store went unreported");
};
assert_eq!(std::fs::read(unreadable_store_path(&path))?, b"\xc1 second");
Ok(())
}
}