diff --git a/src/control_plane.rs b/src/control_plane.rs index 9597d43..e43a7ac 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -42,11 +42,14 @@ use crate::drivers::{ ChallengeFailure, CultCacheTopologyDriver, CultCacheWriteLeaseDriver, DockerRunnerDriver, FrozenSourceReceipt, GitSourceDriver, InstalledReleaseObservation, IsolationEvidence, NginxRouteDriver, LockContended, ProcessIdentity, RouteActuation, RouteActuationGate, RouteActuationRefused, RouteActuators, - RouteObservation, RoutePreflightReceipt, RunnerPort, SourcePort, + RouteObservation, RoutePreflightReceipt, RunnerPort, RunnerStepFailure, SourcePort, SystemdTransientWorkloadDriver, TopologyPort, WorkloadObservation, WorkloadPort, WriteLeasePort, }; 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::{ HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver, HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service, @@ -4877,6 +4880,13 @@ fn status( } if let Some(abort) = &transaction.pre_fencing_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 { writeln!(out, " post-fencing abort {abort:?}")?; @@ -5374,6 +5384,32 @@ impl Engine { 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 { + let failure = error.downcast_ref::()?; + 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 /// 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), @@ -6315,6 +6351,9 @@ impl Engine { if latest.value.pre_fencing_abort.is_none() && error.downcast_ref::().is_none() { + if let Some(line) = self.keep_runner_failure(&latest.value, &error) { + eprintln!("{line}"); + } self.begin_pre_fencing_abort(latest, error)?; } else { 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 /// 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. +/// 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, +) -> 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 { let value = value.replace('', ""); if value.is_empty() { @@ -24657,4 +24733,435 @@ GAMECULT_RUNTIME_PRESENCE_IDENTITY = "/etc/gamecult/service/runtime-presence-ide 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 { + 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 { + 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 { + 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 { + 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 { + 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 = 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::(); + 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 } diff --git a/src/drivers.rs b/src/drivers.rs index 7f87896..382bd23 100644 --- a/src/drivers.rs +++ b/src/drivers.rs @@ -157,6 +157,24 @@ pub struct FrozenSourceReceipt { 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, + root: PathBuf, + ) -> Self { + Self { + receipt, + facts, + recipe_bytes, + root, + } + } +} + impl FrozenSourceReceipt { pub fn validate_against(&self, plan: &CompiledDeploymentPlan) -> Result<()> { plan.validate()?; @@ -2152,6 +2170,115 @@ pub(crate) fn docker_run_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, + pub(crate) signal: Option, + pub(crate) stdout_tail: Vec, + pub(crate) stdout_total_bytes: u64, + pub(crate) stderr_tail: Vec, + 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, signal: Option) -> 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, 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 /// operator binding selects the exact image/network/mount affordances, and the /// driver returns complete materialization receipts rather than build truth by @@ -2170,7 +2297,7 @@ impl Default for DockerRunnerDriver { } impl DockerRunnerDriver { - fn docker(&self, args: I) -> Result + fn docker(&self, subject: RunnerSubject, runner_id: &str, args: I) -> Result<()> where I: IntoIterator, S: AsRef, @@ -2179,23 +2306,50 @@ impl DockerRunnerDriver { self.docker_program.is_absolute(), "Docker runner program is not absolute" ); - let output = Command::new(&self.docker_program) + let mut child = Command::new(&self.docker_program) .args(args) .stdin(Stdio::null()) - .output() + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() .with_context(|| format!("starting Docker runner {}", self.docker_program.display()))?; - if !output.status.success() { - bail!( - "Docker runner exited with {}: {}", - output.status, - String::from_utf8_lossy(&output.stderr).trim() - ); + let stdout = child.stdout.take().context("Docker runner has no stdout pipe")?; + let stderr = child.stderr.take().context("Docker runner has no stderr pipe")?; + let stdout = thread::spawn(move || drain_tail(stdout, RUNNER_OUTPUT_TAIL_BYTES)); + let stderr = thread::spawn(move || drain_tail(stderr, RUNNER_OUTPUT_TAIL_BYTES)); + 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( &self, + step_id: &str, + runner_id: &str, runner: &DockerRunnerBinding, workspace: &Path, working_directory: &Path, @@ -2212,8 +2366,7 @@ impl DockerRunnerDriver { build_machine_id_file(workspace)?; let spec = ContainerSpec::for_step(runner, required_environment, source_stamp)?; let args = docker_run_args(&spec, workspace, working_directory, argv)?; - self.docker(args)?; - Ok(()) + self.docker(RunnerSubject::Step(step_id.to_owned()), runner_id, args) } fn materialize_external_input( @@ -2255,7 +2408,11 @@ impl DockerRunnerDriver { input.url.clone(), ]; 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(|| { format!( "reading materialized external input {}", @@ -2377,6 +2534,8 @@ impl RunnerPort for DockerRunnerDriver { ); } self.run_in_workspace( + &step.id, + &step.runner, runner, workspace, &step.working_directory, @@ -16231,4 +16390,292 @@ processes 9 ); 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 { + 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 { + let error = result.expect_err("the step did not fail"); + error + .downcast_ref::() + .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::() + .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>>); + + impl Read for ScriptedReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + 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 { + 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 = (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::().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::() + .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 } diff --git a/src/lib.rs b/src/lib.rs index e71a17f..4f365a3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,4 +7,5 @@ pub mod host_actuator; #[cfg(unix)] pub mod launcher; mod incident; +mod runner_evidence; pub mod provisioning; diff --git a/src/runner_evidence.rs b/src/runner_evidence.rs new file mode 100644 index 0000000..9e9e5ee --- /dev/null +++ b/src/runner_evidence.rs @@ -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, + #[cultcache(key = 8)] + pub(crate) signal: Option, + #[cultcache(key = 9)] + pub(crate) stdout_tail: Vec, + #[cultcache(key = 10)] + pub(crate) stdout_total_bytes: u64, + #[cultcache(key = 11)] + pub(crate) stderr_tail: Vec, + #[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 { + 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, Vec)> { + 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 }, +} + +/// 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> { + 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>; + +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 { + record_reading(path, record, &|path| std::fs::read(path)) +} + +fn record_reading(path: &Path, record: &RunnerFailureRecord, read: ReadStore) -> Result { + 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> { + 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> { + let noun = match record.subject_kind { + SubjectKind::Step => "step", + SubjectKind::ExternalInput => "external input", + }; + let mut lines: Vec> = 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]) -> 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 { + 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> { + let mut names: Vec = std::fs::read_dir(path.parent().context("no parent")?)? + .map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned())) + .collect::>()?; + 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> { + 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(()) + } +}