Runner failure: a typed, bounded step failure kept in runner-evidence.cc

DockerRunnerDriver::docker no longer buffers whole streams or interpolates
stderr into the error. It pipes both streams through an 8 KiB tail and returns
RunnerStepFailure (subject, runner, status, tails); the engine's pre-fence
path is the one writer of runner-evidence.cc (0600, newest 32) and journals a
line that names the store and no output. idunn status --command renders the
record with control bytes escaped.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-25 02:11:14 +01:00
commit ef82355bab
4 changed files with 508 additions and 14 deletions

View file

@ -42,11 +42,12 @@ 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, RunnerFailureRecord, runner_evidence_store_path};
use crate::host_actuator::{
HostActuatorAccess, HostActuatorHub, HostActuatorRunnerDriver, HostActuatorWorkloadDriver,
HostUnobservable, IdunnHostActuatorIdentity, SharedHostActuatorHub, spawn_hub_service,
@ -4877,6 +4878,24 @@ 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.
match runner_evidence::for_transaction(
&runner_evidence_store_path(store_path),
&transaction.transaction_id,
) {
Ok(Some(record)) => {
for line in runner_evidence::render(&record) {
writeln!(out, "{line}")?;
}
}
Ok(None) => {}
Err(error) => writeln!(
out,
" runner evidence unreadable: {}",
error.root_cause()
)?,
}
}
if let Some(abort) = &transaction.post_fencing_abort {
writeln!(out, " post-fencing abort {abort:?}")?;
@ -5374,6 +5393,33 @@ 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) {
let Some(failure) = error.downcast_ref::<RunnerStepFailure>() else {
return;
};
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,
),
)
});
eprintln!(
"{}",
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 +6361,7 @@ impl Engine {
if latest.value.pre_fencing_abort.is_none()
&& error.downcast_ref::<LockContended>().is_none()
{
self.keep_runner_failure(&latest.value, &error);
self.begin_pre_fencing_abort(latest, error)?;
} else {
self.record_resumable_error(latest, &error)?;
@ -10641,6 +10688,28 @@ 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(()) => format!(
"{head}; evidence in {}; 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() {

View file

@ -2152,6 +2152,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<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
/// operator binding selects the exact image/network/mount affordances, and the
/// driver returns complete materialization receipts rather than build truth by
@ -2170,7 +2279,7 @@ impl Default for 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
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
@ -2179,23 +2288,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 +2348,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 +2390,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 +2516,8 @@ impl RunnerPort for DockerRunnerDriver {
);
}
self.run_in_workspace(
&step.id,
&step.runner,
runner,
workspace,
&step.working_directory,

View file

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

283
src/runner_evidence.rs Normal file
View file

@ -0,0 +1,283 @@
//! 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 render
//! below escapes control characters because the status reader is a terminal.
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))
}
/// 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.
pub(crate) fn record(path: &Path, record: &RunnerFailureRecord) -> Result<()> {
let (envelopes, records) = snapshot(path)?;
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)
}
#[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` prints it. The store keeps raw
/// bytes; every control character except newline and tab is escaped here, so
/// build output cannot drive the operator's terminal.
pub(crate) fn render(record: &RunnerFailureRecord) -> Vec<String> {
let noun = match record.subject_kind {
SubjectKind::Step => "step",
SubjectKind::ExternalInput => "external input",
};
let mut lines = vec![
format!(" failed {noun} {}", escape(&record.subject_id)),
format!(" runner {}", escape(&record.runner)),
format!(" status {}", runner_status(record.exit_code, record.signal)),
];
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()
));
let text = escape(&String::from_utf8_lossy(tail));
let text = text.strip_suffix('\n').unwrap_or(&text);
if !text.is_empty() {
lines.extend(text.split('\n').map(|line| format!(" | {line}")));
}
}
lines
}
fn escape(text: &str) -> String {
let mut escaped = String::with_capacity(text.len());
for character in text.chars() {
if character.is_control() && character != '\n' && character != '\t' {
escaped.push_str(&format!("\\u{{{:x}}}", u32::from(character)));
} else {
escaped.push(character);
}
}
escaped
}