From efa89952a1b5abb2bb992f91a1de0712de89cb8c Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:15:26 -0700 Subject: [PATCH 01/15] Fix: bound external digest execution --- CHANGELOG.md | 6 + docs/conformance/golden-file-worldline.md | 8 +- docs/dependencies/blake3-1.8.5.md | 12 +- .../cap-std-and-cap-fs-ext-4.0.2.md | 17 +- xtask/Cargo.toml | 4 +- xtask/src/bounded_process.rs | 3 +- xtask/src/bounded_process/capture.rs | 40 +++ xtask/src/bounded_process/input.rs | 101 +++++++ xtask/src/external_digest.rs | 162 +++++++++++ xtask/src/external_digest/tests.rs | 104 +++++++ .../src/golden_file_worldline/b3sum_oracle.rs | 220 +------------- xtask/src/golden_file_worldline/error.rs | 56 +--- xtask/src/main.rs | 5 + xtask/src/protocol_conformance/error.rs | 119 +------- .../protocol_conformance/external_digest.rs | 268 +----------------- .../external_digest/tests.rs | 62 ---- xtask/tests/source_policy_contract.rs | 44 ++- 17 files changed, 527 insertions(+), 704 deletions(-) create mode 100644 xtask/src/bounded_process/input.rs create mode 100644 xtask/src/external_digest.rs create mode 100644 xtask/src/external_digest/tests.rs delete mode 100644 xtask/src/protocol_conformance/external_digest/tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e9029eb..60316e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,6 +96,12 @@ after its public API and format compatibility policies are established. witness, reproducible Gear-table recipe, scalar and streaming FastCDC laws, source mutations, and exact boundary corpus; the three superseded Python programs have been removed. +- Golden File Worldline and protocol-conformance `b3sum` witnesses now share + one external-digest process boundary. Its deadline begins before process + spawn and stdin transfer, stdin is streamed without a combined preimage + allocation, stdout and stderr have independent limits, and every timeout or + collection failure kills and reaps the process group while retaining typed + failure context. - Fuzz policy admission, target reconciliation, bounded campaign execution, minimization deadlines, retained-corpus admission, and workflow contract tests now run through the repository's Rust `xtask`; the superseded Python diff --git a/docs/conformance/golden-file-worldline.md b/docs/conformance/golden-file-worldline.md index 0058b29..42da58a 100644 --- a/docs/conformance/golden-file-worldline.md +++ b/docs/conformance/golden-file-worldline.md @@ -158,7 +158,13 @@ test-only oracle is not a production streaming API and provides no: The model computes identity through Keep's public `BlobId` API, while the checked-in vectors were generated independently with `b3sum`. Agreement is -therefore not a self-authored golden. +therefore not a self-authored golden. The repository checker streams each +canonical preimage through the same deadline-bounded external-digest adapter +used by protocol conformance. The deadline covers process spawn and stdin +transfer as well as output collection; stdout and stderr have independent byte +limits; and timeout or collection failure kills and reaps the complete child +process group. Digest-specific preimage construction remains owned by this +checker rather than the process adapter. ## Exact M1 acceptance boundary diff --git a/docs/dependencies/blake3-1.8.5.md b/docs/dependencies/blake3-1.8.5.md index a5ca41e..d455c4b 100644 --- a/docs/dependencies/blake3-1.8.5.md +++ b/docs/dependencies/blake3-1.8.5.md @@ -22,10 +22,14 @@ types or hashing wrappers. This separation independently checks preimage framing, length encoding, canonical text and binary encodings, mutation semantics, and committed witness bytes. For every identity and content mutation, the repository checker also streams the canonical preimage through external -`b3sum`; a mismatch with the in-process result is a refusal. The checked-in -vectors and runtime cross-check therefore cover the algorithm boundary without -claiming that the Rust path independently implements the BLAKE3 compression -function. +`b3sum`; a mismatch with the in-process result is a refusal. Golden File +Worldline and protocol conformance share one deadline-bounded process adapter, +but retain separate preimage construction. The adapter streams caller-owned +slices without a combined allocation, bounds stdout and stderr independently, +and kills and reaps the child process group on timeout or collection failure. +The checked-in vectors and runtime cross-check therefore cover the algorithm +boundary without claiming that the Rust path independently implements the +BLAKE3 compression function. The manifest disables default features and enables exactly: diff --git a/docs/dependencies/cap-std-and-cap-fs-ext-4.0.2.md b/docs/dependencies/cap-std-and-cap-fs-ext-4.0.2.md index ffc681b..b4ab0a5 100644 --- a/docs/dependencies/cap-std-and-cap-fs-ext-4.0.2.md +++ b/docs/dependencies/cap-std-and-cap-fs-ext-4.0.2.md @@ -25,10 +25,12 @@ These packages are absent from Keep's published library graph, public API, content identities, durable formats, and production behavior. No dependency-owned type crosses out of the private repository-task adapter. -The bounded subprocess adapter uses Rustix's safe process API to send -`SIGKILL` to a dedicated child process group after a subprocess deadline or -collection failure. This prevents descendants that inherited an output pipe -from surviving the failed repository task. +The bounded subprocess adapter uses Rustix's safe filesystem API to mark child +stdin nonblocking before deadline-bounded input transfer. It uses Rustix's safe +process API to send `SIGKILL` to a dedicated child process group after a +subprocess deadline or collection failure. This prevents a non-reading child +from blocking its parent indefinitely and prevents descendants that inherited +an output pipe from surviving the failed repository task. The [signal-hook dependency admission](signal-hook-0.4.4.md) records the terminal-signal guard that routes interruption through the same @@ -61,7 +63,7 @@ work, and require unsafe code that Keep otherwise forbids. ## Features and resolved graph All three direct dependencies disable default features. Keep enables only -`cap-fs-ext`'s `std` feature and Rustix's `process` and `std` features; +`cap-fs-ext`'s `std` feature and Rustix's `fs`, `process`, and `std` features; `cap-std` has no enabled feature. All declarations are optional and are activated solely by `repository-tasks`. @@ -104,8 +106,9 @@ dependencies. ## Failure and recovery boundaries -An open, metadata, read, descriptor-duplication, child-directory setup, or -child-spawn failure is a typed refusal. The task never repairs, rewrites, or +An open, metadata, read, descriptor-duplication, descriptor-flag, +child-directory setup, child-spawn, stdin-write, output-collection, deadline, +or cleanup failure is a typed refusal. The task never repairs, rewrites, or substitutes repository data. Retained handles exist only for one verification process and carry no durability or recovery semantics. diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml index fee2de4..b5bad5c 100644 --- a/xtask/Cargo.toml +++ b/xtask/Cargo.toml @@ -32,8 +32,8 @@ cap-std = { version = "=4.0.2", default-features = false, optional = true } md-5 = { version = "=0.11.0", default-features = false, optional = true } # Dedicated unsafe boundary sets exact child working directories by descriptor. repository-process-spawn = { path = "../repository-process-spawn", optional = true } -# Safe POSIX process-group signaling bounds failed repository-tool subprocesses. -rustix = { version = "=1.1.4", default-features = false, features = ["process", "std"], optional = true } +# Safe POSIX descriptor flags and process-group signaling bound repository tools. +rustix = { version = "=1.1.4", default-features = false, features = ["fs", "process", "std"], optional = true } # Serde drives duplicate-refusing repository JSON admission; no types escape xtask. serde = { version = "=1.0.229", default-features = false, features = ["std"], optional = true } # Typed JSON admission checks the committed documentation-tool lock graph. diff --git a/xtask/src/bounded_process.rs b/xtask/src/bounded_process.rs index 25e8ea7..628b3c5 100644 --- a/xtask/src/bounded_process.rs +++ b/xtask/src/bounded_process.rs @@ -5,6 +5,7 @@ mod capture_limit; mod cleanup; mod deadline; mod error; +mod input; mod interrupt; mod process_group; mod reader; @@ -14,7 +15,7 @@ use std::process::Command; use std::time::Duration; use capture::wait_for_child; -pub(crate) use capture::{capture, capture_with, capture_with_limits}; +pub(crate) use capture::{capture, capture_with, capture_with_input_limits, capture_with_limits}; pub(crate) use capture_limit::CaptureLimits; use deadline::ProcessDeadline; pub(crate) use error::ProcessError; diff --git a/xtask/src/bounded_process/capture.rs b/xtask/src/bounded_process/capture.rs index a3f7ff5..4866d10 100644 --- a/xtask/src/bounded_process/capture.rs +++ b/xtask/src/bounded_process/capture.rs @@ -6,6 +6,7 @@ use std::thread; use std::time::Duration; use super::cleanup::{cleanup_process, join_after_cleanup, join_readers}; +use super::input::write_input; use super::{ CaptureLimits, InterruptGuard, ProcessDeadline, ProcessError, ProcessOutput, ReaderWorker, }; @@ -57,6 +58,25 @@ pub(crate) fn capture_with_limits( CapturedProcess::start(program, command, spawn, interrupts, limits)?.finish(program, &deadline) } +/// Runs one captured child with bounded streaming input and exact stream limits. +/// +/// The complete deadline starts before spawn and covers every nonblocking stdin +/// write, both output readers, child execution, and cleanup. Input slices are +/// streamed directly without constructing a combined preimage allocation. +pub(crate) fn capture_with_input_limits( + program: &'static str, + command: &mut Command, + input: &[&[u8]], + deadline: Option, + limits: CaptureLimits, +) -> Result { + let deadline = ProcessDeadline::new(program, deadline)?; + let interrupts = InterruptGuard::begin(program)?; + command.stdin(Stdio::piped()); + CapturedProcess::start(program, command, Command::spawn, interrupts, limits)? + .finish_with_input(program, input, &deadline) +} + struct CapturedProcess { child: Child, interrupts: InterruptGuard, @@ -151,6 +171,26 @@ impl CapturedProcess { }) } + fn finish_with_input( + mut self, + program: &'static str, + input: &[&[u8]], + deadline: &ProcessDeadline, + ) -> Result { + let Some(mut stdin) = self.child.stdin.take() else { + let error = ProcessError::MissingStream { + program, + stream: "stdin", + }; + return Err(self.cleanup_readers(error)); + }; + if let Err(error) = write_input(program, &mut stdin, input, deadline, &self.interrupts) { + return Err(self.cleanup_readers(error)); + } + drop(stdin); + self.finish(program, deadline) + } + fn cleanup_readers(self, error: ProcessError) -> ProcessError { let mut child = self.child; let error = cleanup_process(&mut child, error); diff --git a/xtask/src/bounded_process/input.rs b/xtask/src/bounded_process/input.rs new file mode 100644 index 0000000..025f85d --- /dev/null +++ b/xtask/src/bounded_process/input.rs @@ -0,0 +1,101 @@ +//! This module owns deadline-bounded child-process input streaming. + +use std::io::{self, Write}; +use std::process::ChildStdin; +use std::thread; +use std::time::Duration; + +use rustix::fs::{OFlags, fcntl_getfl, fcntl_setfl}; + +use super::{InterruptGuard, ProcessDeadline, ProcessError}; + +const INPUT_POLL_INTERVAL: Duration = Duration::from_millis(10); + +/// Streams admitted slices to child stdin without a combined allocation. +/// +/// The pipe is placed in nonblocking mode before the first write. Every retry +/// observes the complete process deadline and terminal-signal guard. +pub(super) fn write_input( + program: &'static str, + stdin: &mut ChildStdin, + parts: &[&[u8]], + deadline: &ProcessDeadline, + interrupts: &InterruptGuard, +) -> Result<(), ProcessError> { + let flags = fcntl_getfl(&*stdin) + .map_err(|source| process_io(program, "read child input flags", source.into()))?; + fcntl_setfl(&*stdin, flags | OFlags::NONBLOCK).map_err(|source| { + process_io(program, "configure nonblocking child input", source.into()) + })?; + for part in parts { + write_part(program, stdin, part, deadline, interrupts)?; + } + Ok(()) +} + +fn write_part( + program: &'static str, + stdin: &mut ChildStdin, + part: &[u8], + deadline: &ProcessDeadline, + interrupts: &InterruptGuard, +) -> Result<(), ProcessError> { + let mut remaining = part; + while !remaining.is_empty() { + match stdin.write(remaining) { + Ok(0) => { + return Err(process_io( + program, + "write child input", + io::Error::new(io::ErrorKind::WriteZero, "child input made no progress"), + )); + } + Ok(written) => { + remaining = remaining.get(written..).ok_or_else(|| { + process_io( + program, + "account child input", + io::Error::new( + io::ErrorKind::InvalidData, + "child input write exceeded the admitted slice", + ), + ) + })?; + } + Err(source) if source.kind() == io::ErrorKind::Interrupted => {} + Err(source) if source.kind() == io::ErrorKind::WouldBlock => { + wait_for_input(program, deadline, interrupts)?; + } + Err(source) => return Err(process_io(program, "write child input", source)), + } + } + Ok(()) +} + +fn wait_for_input( + program: &'static str, + deadline: &ProcessDeadline, + interrupts: &InterruptGuard, +) -> Result<(), ProcessError> { + if let Some(error) = interrupts.refusal(program) { + return Err(error); + } + let interval = match deadline.remaining(program)? { + Some((remaining, _duration)) => INPUT_POLL_INTERVAL.min(remaining), + None => INPUT_POLL_INTERVAL, + }; + thread::sleep(interval); + Ok(()) +} + +const fn process_io( + program: &'static str, + action: &'static str, + source: io::Error, +) -> ProcessError { + ProcessError::Io { + program, + action, + source, + } +} diff --git a/xtask/src/external_digest.rs b/xtask/src/external_digest.rs new file mode 100644 index 0000000..0405f9d --- /dev/null +++ b/xtask/src/external_digest.rs @@ -0,0 +1,162 @@ +//! This module owns the bounded external digest witness. + +use std::env; +use std::error::Error; +use std::fmt; +use std::process::Command; +use std::string::FromUtf8Error; +use std::time::Duration; + +use crate::bounded_process::{ + CaptureLimits, ProcessError, ProcessOutput, capture_with_input_limits, +}; +use crate::diagnostic::escaped_controls; + +const B3SUM: &str = "b3sum"; +const DIAGNOSTIC_LIMIT_BYTES: usize = 65_536; +const DIGEST_BYTES: usize = 32; +const PROCESS_DEADLINE: Duration = Duration::from_secs(10); +const PROCESS_LIMITS: CaptureLimits = CaptureLimits::new(DIGEST_BYTES, DIAGNOSTIC_LIMIT_BYTES); + +/// A typed refusal from the external digest witness. +pub(crate) enum ExternalDigestError { + /// The executable search path required for the hermetic child is absent. + Environment { + /// The missing admitted environment variable. + variable: &'static str, + }, + /// A bounded process operation failed. + Process { + /// The precise process-layer refusal. + source: ProcessError, + }, + /// A failed child emitted diagnostics that were not UTF-8. + DiagnosticEncoding { + /// The platform exit code, or `None` after signal termination. + code: Option, + /// The original diagnostic decoding failure. + source: FromUtf8Error, + }, + /// The child returned a non-success status with admitted diagnostics. + Failed { + /// The platform exit code, or `None` after signal termination. + code: Option, + /// The bounded UTF-8 diagnostic. + stderr: String, + }, + /// A successful child emitted an unexpected diagnostic. + UnexpectedDiagnostic, + /// A successful child returned a digest with the wrong width. + Width { + /// The number of bytes returned by the child. + observed: usize, + }, +} + +/// Computes one raw BLAKE3 digest through the independent `b3sum` witness. +/// +/// Input parts are streamed without concatenation. The child receives only the +/// admitted executable search path and `C` locale, and the complete operation +/// is bounded by one ten-second deadline and independent output limits. +pub(crate) fn b3sum(parts: &[&[u8]]) -> Result<[u8; DIGEST_BYTES], ExternalDigestError> { + let mut command = b3sum_command()?; + execute(&mut command, parts, PROCESS_DEADLINE) +} + +fn b3sum_command() -> Result { + let path = env::var_os("PATH").ok_or(ExternalDigestError::Environment { variable: "PATH" })?; + let mut command = Command::new(B3SUM); + command + .args(["--raw", "--no-mmap", "--num-threads", "1"]) + .env_clear() + .env("PATH", path) + .env("LC_ALL", "C"); + Ok(command) +} + +fn execute( + command: &mut Command, + parts: &[&[u8]], + deadline: Duration, +) -> Result<[u8; DIGEST_BYTES], ExternalDigestError> { + let output = capture_with_input_limits(B3SUM, command, parts, Some(deadline), PROCESS_LIMITS) + .map_err(|source| ExternalDigestError::Process { source })?; + validate(output) +} + +fn validate(output: ProcessOutput) -> Result<[u8; DIGEST_BYTES], ExternalDigestError> { + if !output.succeeded { + return match String::from_utf8(output.stderr) { + Ok(stderr) => Err(ExternalDigestError::Failed { + code: output.code, + stderr, + }), + Err(source) => Err(ExternalDigestError::DiagnosticEncoding { + code: output.code, + source, + }), + }; + } + if !output.stderr.is_empty() { + return Err(ExternalDigestError::UnexpectedDiagnostic); + } + let observed = output.stdout.len(); + output + .stdout + .try_into() + .map_err(|_output| ExternalDigestError::Width { observed }) +} + +impl fmt::Debug for ExternalDigestError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, formatter) + } +} + +impl fmt::Display for ExternalDigestError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Environment { variable } => { + write!( + formatter, + "external digest requires the {variable} environment variable" + ) + } + Self::Process { source } => fmt::Display::fmt(source, formatter), + Self::DiagnosticEncoding { code, .. } => write!( + formatter, + "{B3SUM} failed with status {code:?} and non-UTF-8 diagnostics" + ), + Self::Failed { code, stderr } => { + write!(formatter, "{B3SUM} failed with status {code:?}: ")?; + escaped_controls(formatter, stderr) + } + Self::UnexpectedDiagnostic => { + write!( + formatter, + "{B3SUM} wrote diagnostics despite successful exit" + ) + } + Self::Width { observed } => write!( + formatter, + "{B3SUM} returned {observed} digest bytes instead of {DIGEST_BYTES}" + ), + } + } +} + +impl Error for ExternalDigestError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::Process { source } => Some(source), + Self::DiagnosticEncoding { source, .. } => Some(source), + Self::Environment { .. } + | Self::Failed { .. } + | Self::UnexpectedDiagnostic + | Self::Width { .. } => None, + } + } +} + +#[cfg(test)] +mod tests; diff --git a/xtask/src/external_digest/tests.rs b/xtask/src/external_digest/tests.rs new file mode 100644 index 0000000..d28995e --- /dev/null +++ b/xtask/src/external_digest/tests.rs @@ -0,0 +1,104 @@ +//! This module owns bounded external digest regression evidence. + +use std::env; +use std::process::Command; +use std::sync::mpsc::{self, RecvTimeoutError}; +use std::time::Duration; + +use crate::bounded_process::{ProcessError, ProcessOutput}; + +use super::{B3SUM, ExternalDigestError, execute, validate}; + +const BLOCKING_CHILD: &str = "KEEP_XTASK_DIGEST_BLOCKING_CHILD"; + +#[test] +fn child_that_never_reads_digest_input_obeys_the_complete_deadline() +-> Result<(), Box> { + let executable = env::current_exe()?; + let mut command = Command::new(executable); + command + .args([ + "--exact", + "external_digest::tests::digest_child_does_not_read_stdin", + ]) + .env(BLOCKING_CHILD, "1"); + let input = vec![0_u8; 1_048_576]; + let duration = Duration::from_millis(50); + + assert!(matches!( + execute(&mut command, &[&input], duration), + Err(ExternalDigestError::Process { + source: ProcessError::Timeout { + program: B3SUM, + duration: observed, + }, + }) if observed == duration + )); + Ok(()) +} + +#[test] +fn digest_child_does_not_read_stdin() { + if env::var_os(BLOCKING_CHILD).is_none() { + return; + } + let (sender, receiver) = mpsc::channel::<()>(); + assert!(matches!( + receiver.recv_timeout(Duration::from_secs(1)), + Err(RecvTimeoutError::Timeout) + )); + drop(sender); +} + +#[test] +fn failed_digest_preserves_utf8_and_non_utf8_diagnostics() { + let utf8 = validate(ProcessOutput { + code: Some(7), + succeeded: false, + stdout: Vec::new(), + stderr: b"refused".to_vec(), + }); + assert!(matches!( + utf8, + Err(ExternalDigestError::Failed { + code: Some(7), + ref stderr, + }) if stderr == "refused" + )); + + let non_utf8 = validate(ProcessOutput { + code: Some(9), + succeeded: false, + stdout: Vec::new(), + stderr: vec![0xff], + }); + assert!(matches!( + non_utf8, + Err(ExternalDigestError::DiagnosticEncoding { code: Some(9), .. }) + )); +} + +#[test] +fn successful_digest_requires_silent_exact_width_output() { + let diagnostic = validate(ProcessOutput { + code: Some(0), + succeeded: true, + stdout: vec![0_u8; 32], + stderr: b"unexpected".to_vec(), + }); + assert!(matches!( + diagnostic, + Err(ExternalDigestError::UnexpectedDiagnostic) + )); + + let width = validate(ProcessOutput { + code: Some(0), + succeeded: true, + stdout: vec![0_u8; 31], + stderr: Vec::new(), + }); + assert!(matches!( + width, + Err(ExternalDigestError::Width { observed: 31 }) + )); +} diff --git a/xtask/src/golden_file_worldline/b3sum_oracle.rs b/xtask/src/golden_file_worldline/b3sum_oracle.rs index b0b8c26..218ad9d 100644 --- a/xtask/src/golden_file_worldline/b3sum_oracle.rs +++ b/xtask/src/golden_file_worldline/b3sum_oracle.rs @@ -1,227 +1,33 @@ -//! This module owns the bounded `b3sum` adapter for identity digest evidence. +//! This module owns the independent digest adapter for identity evidence. -use std::io::{self, Write}; -use std::path::Path; -use std::process::{Child, ChildStdin, Command, Stdio}; -use std::thread::{self, JoinHandle}; - -use crate::process_output::{BoundedBytes, bounded_bytes}; +use crate::external_digest; use super::GoldenError; use super::digest_port::IdentityDigestOracle; const ALGORITHM: u8 = 1; -const B3SUM: &str = "b3sum"; const DATA_MAGIC: [u8; 16] = *b"KEEP:BLOB:DATA\0\0"; -const DIAGNOSTIC_LIMIT_BYTES: usize = 65_536; -const DIGEST_BYTES: usize = 32; -const OUTPUT_LIMIT_BYTES: usize = DIGEST_BYTES + 1; const VERSION: u16 = 1; pub(super) struct B3sumOracle; -struct B3sumProcess { - child: Child, - stdin: ChildStdin, - stdout_worker: JoinHandle>, - diagnostic_worker: JoinHandle>, -} - impl IdentityDigestOracle for B3sumOracle { fn identity_digest(&self, payload: &[u8]) -> Result<[u8; 32], GoldenError> { let length = u64::try_from(payload.len()).map_err(|source| { GoldenError::violation(format!("payload length cannot be represented: {source}")) })?; - let process = start_b3sum()?; - collect_b3sum(process, payload, length) - } -} - -fn start_b3sum() -> Result { - let mut child = Command::new(B3SUM) - .args(["--raw", "--no-mmap", "--num-threads", "1"]) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .map_err(|source| GoldenError::io("start b3sum", Path::new(B3SUM), source))?; - let stdin = take_stdin(&mut child)?; - let stdout = take_stdout(&mut child)?; - let stderr = take_stderr(&mut child)?; - let stdout_worker = match start_reader("xtask-b3sum-output", stdout, OUTPUT_LIMIT_BYTES) { - Ok(worker) => worker, - Err(error) => { - cleanup_child(&mut child)?; - return Err(error); - } - }; - let diagnostic_worker = - match start_reader("xtask-b3sum-diagnostic", stderr, DIAGNOSTIC_LIMIT_BYTES) { - Ok(worker) => worker, - Err(error) => { - let cleanup = cleanup_child(&mut child); - let join = join_reader(stdout_worker, "output"); - cleanup?; - join?; - return Err(error); - } - }; - Ok(B3sumProcess { - child, - stdin, - stdout_worker, - diagnostic_worker, - }) -} - -fn take_stdin(child: &mut Child) -> Result { - let stdin = child.stdin.take(); - stdin.map_or_else(|| missing_pipe(child, "stdin"), Ok) -} - -fn take_stdout(child: &mut Child) -> Result { - let stdout = child.stdout.take(); - stdout.map_or_else(|| missing_pipe(child, "stdout"), Ok) -} - -fn take_stderr(child: &mut Child) -> Result { - let stderr = child.stderr.take(); - stderr.map_or_else(|| missing_pipe(child, "stderr"), Ok) -} - -fn missing_pipe(child: &mut Child, stream: &'static str) -> Result { - cleanup_child(child)?; - Err(GoldenError::violation(format!( - "b3sum {stream} pipe is absent" - ))) -} - -fn start_reader( - name: &str, - reader: impl io::Read + Send + 'static, - maximum: usize, -) -> Result>, GoldenError> { - thread::Builder::new() - .name(String::from(name)) - .spawn(move || bounded_bytes(reader, maximum)) - .map_err(|source| GoldenError::io("start b3sum output reader", Path::new(B3SUM), source)) -} - -fn collect_b3sum( - process: B3sumProcess, - payload: &[u8], - length: u64, -) -> Result<[u8; 32], GoldenError> { - let B3sumProcess { - mut child, - stdin, - stdout_worker, - diagnostic_worker, - } = process; - if let Err(write_error) = write_preimage(stdin, payload, length) { - cleanup_child(&mut child)?; - join_reader(stdout_worker, "output")?; - join_reader(diagnostic_worker, "diagnostic")?; - return Err(write_error); + let version = VERSION.to_be_bytes(); + let algorithm = [ALGORITHM]; + let length = length.to_be_bytes(); + external_digest::b3sum(&[ + DATA_MAGIC.as_slice(), + version.as_slice(), + algorithm.as_slice(), + payload, + length.as_slice(), + ]) + .map_err(GoldenError::external_digest) } - let status = child - .wait() - .map_err(|source| GoldenError::io("wait for b3sum", Path::new(B3SUM), source))?; - let output = join_reader(stdout_worker, "output")?; - let diagnostic = join_reader(diagnostic_worker, "diagnostic")?; - validate_b3sum(status.code(), status.success(), output, diagnostic) -} - -fn write_preimage(mut stdin: ChildStdin, payload: &[u8], length: u64) -> Result<(), GoldenError> { - for bytes in [ - DATA_MAGIC.as_slice(), - VERSION.to_be_bytes().as_slice(), - [ALGORITHM].as_slice(), - payload, - length.to_be_bytes().as_slice(), - ] { - stdin - .write_all(bytes) - .map_err(|source| GoldenError::io("write b3sum preimage", Path::new(B3SUM), source))?; - } - Ok(()) -} - -fn join_reader( - worker: JoinHandle>, - stream: &'static str, -) -> Result { - worker - .join() - .map_err(|_| GoldenError::violation(format!("b3sum {stream} reader panicked")))? - .map_err(|source| GoldenError::io("read b3sum output", Path::new(B3SUM), source)) -} - -fn validate_b3sum( - code: Option, - success: bool, - output: BoundedBytes, - diagnostic: BoundedBytes, -) -> Result<[u8; 32], GoldenError> { - refuse_exceeded("output", &output, OUTPUT_LIMIT_BYTES)?; - refuse_exceeded("diagnostic", &diagnostic, DIAGNOSTIC_LIMIT_BYTES)?; - if !success { - return Err(process_failure(code, diagnostic.bytes)); - } - if !diagnostic.bytes.is_empty() { - return Err(GoldenError::violation( - "b3sum wrote diagnostics despite successful exit", - )); - } - output - .bytes - .try_into() - .map_err(|_| GoldenError::violation("b3sum returned a noncanonical digest width")) -} - -const fn refuse_exceeded( - stream: &'static str, - output: &BoundedBytes, - maximum: usize, -) -> Result<(), GoldenError> { - if output.exceeded { - Err(GoldenError::ProcessOutputBound { - program: B3SUM, - stream, - maximum, - }) - } else { - Ok(()) - } -} - -fn process_failure(code: Option, diagnostic: Vec) -> GoldenError { - match String::from_utf8(diagnostic) { - Ok(stderr) => GoldenError::ProcessFailed { - program: B3SUM, - code, - stderr, - }, - Err(source) => GoldenError::ProcessDiagnosticEncoding { - program: B3SUM, - code, - source, - }, - } -} - -fn cleanup_child(child: &mut Child) -> Result<(), GoldenError> { - let stop = child.kill().or_else(|source| { - if source.kind() == io::ErrorKind::InvalidInput { - Ok(()) - } else { - Err(source) - } - }); - let wait = child.wait(); - stop.map_err(|source| GoldenError::io("stop b3sum", Path::new(B3SUM), source))?; - wait.map_err(|source| GoldenError::io("wait for b3sum", Path::new(B3SUM), source))?; - Ok(()) } #[cfg(test)] diff --git a/xtask/src/golden_file_worldline/error.rs b/xtask/src/golden_file_worldline/error.rs index 9f418e9..1de9076 100644 --- a/xtask/src/golden_file_worldline/error.rs +++ b/xtask/src/golden_file_worldline/error.rs @@ -11,6 +11,9 @@ use crate::diagnostic::{escaped_controls, escaped_path}; use xtask::protocol_admission::RelativePathError; pub(crate) enum GoldenError { + ExternalDigest { + source: Box, + }, Integer { field: String, source: ParseIntError, @@ -24,21 +27,6 @@ pub(crate) enum GoldenError { path: PathBuf, source: FromUtf8Error, }, - ProcessDiagnosticEncoding { - program: &'static str, - code: Option, - source: FromUtf8Error, - }, - ProcessFailed { - program: &'static str, - code: Option, - stderr: String, - }, - ProcessOutputBound { - program: &'static str, - stream: &'static str, - maximum: usize, - }, Path { parameter: String, source: RelativePathError, @@ -47,6 +35,12 @@ pub(crate) enum GoldenError { } impl GoldenError { + pub(super) fn external_digest(source: impl Error + Send + Sync + 'static) -> Self { + Self::ExternalDigest { + source: Box::new(source), + } + } + pub(super) fn io(action: &'static str, path: impl Into, source: io::Error) -> Self { Self::Io { action, @@ -70,6 +64,7 @@ impl fmt::Display for GoldenError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str("golden corpus check failed: ")?; match self { + Self::ExternalDigest { source } => fmt::Display::fmt(source, formatter), Self::Integer { field, .. } => { formatter.write_str("cannot parse canonical ")?; escaped_controls(formatter, field) @@ -83,28 +78,6 @@ impl fmt::Display for GoldenError { escaped_path(formatter, path)?; formatter.write_str(": protocol is not UTF-8") } - Self::ProcessDiagnosticEncoding { program, code, .. } => { - write!( - formatter, - "{program} failed with status {code:?} and non-UTF-8 diagnostics" - ) - } - Self::ProcessFailed { - program, - code, - stderr, - } => { - write!(formatter, "{program} failed with status {code:?}: ")?; - escaped_controls(formatter, stderr) - } - Self::ProcessOutputBound { - program, - stream, - maximum, - } => write!( - formatter, - "{program} {stream} exceeded the {maximum}-byte bound" - ), Self::Path { parameter, source } => { formatter.write_str("unsafe source path: ")?; escaped_controls(formatter, parameter)?; @@ -118,15 +91,12 @@ impl fmt::Display for GoldenError { impl Error for GoldenError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { + Self::ExternalDigest { source } => Some(source.as_ref()), Self::Integer { source, .. } => Some(source), Self::Io { source, .. } => Some(source), - Self::Utf8 { source, .. } | Self::ProcessDiagnosticEncoding { source, .. } => { - Some(source) - } + Self::Utf8 { source, .. } => Some(source), Self::Path { source, .. } => Some(source), - Self::ProcessFailed { .. } | Self::ProcessOutputBound { .. } | Self::Violation(_) => { - None - } + Self::Violation(_) => None, } } } diff --git a/xtask/src/main.rs b/xtask/src/main.rs index a6d4358..cfa6908 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -19,6 +19,11 @@ mod diagnostic; reason = "the command and task-error boundaries are sibling consumers" )] mod documentation_integrity; +#[allow( + clippy::redundant_pub_crate, + reason = "the external digest witness is shared by sibling repository tasks" +)] +mod external_digest; #[allow( clippy::redundant_pub_crate, reason = "the command and task-error boundaries are sibling consumers" diff --git a/xtask/src/protocol_conformance/error.rs b/xtask/src/protocol_conformance/error.rs index cef88c2..fabfa26 100644 --- a/xtask/src/protocol_conformance/error.rs +++ b/xtask/src/protocol_conformance/error.rs @@ -6,16 +6,14 @@ use std::io; use std::num::ParseIntError; use std::path::PathBuf; use std::string::FromUtf8Error; -use std::time::Duration; use crate::diagnostic::{escaped_controls, escaped_path}; +use crate::external_digest::ExternalDigestError; use xtask::protocol_admission::RelativePathError; pub(crate) enum ConformanceError { - Cleanup { - primary: Box, - action: &'static str, - source: io::Error, + ExternalDigest { + source: ExternalDigestError, }, Integer { field: String, @@ -30,32 +28,6 @@ pub(crate) enum ConformanceError { parameter: String, source: RelativePathError, }, - ProcessDiagnosticEncoding { - program: &'static str, - code: Option, - source: FromUtf8Error, - }, - ProcessFailed { - program: &'static str, - code: Option, - stderr: String, - }, - ProcessOutputBound { - program: &'static str, - stream: &'static str, - maximum: usize, - }, - ProcessTimeout { - program: &'static str, - duration: Duration, - }, - ReaderPanic { - program: &'static str, - stream: &'static str, - }, - WriterPanic { - program: &'static str, - }, Utf8 { path: PathBuf, source: FromUtf8Error, @@ -64,12 +36,8 @@ pub(crate) enum ConformanceError { } impl ConformanceError { - pub(super) fn cleanup(primary: Self, action: &'static str, source: io::Error) -> Self { - Self::Cleanup { - primary: Box::new(primary), - action, - source, - } + pub(super) const fn external_digest(source: ExternalDigestError) -> Self { + Self::ExternalDigest { source } } pub(super) fn io(action: &'static str, path: impl Into, source: io::Error) -> Self { @@ -94,14 +62,7 @@ impl fmt::Debug for ConformanceError { impl ConformanceError { fn fmt_body(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - Self::Cleanup { - primary, - action, - source, - } => { - primary.fmt_body(formatter)?; - write!(formatter, "; cleanup could not {action}: {source}") - } + Self::ExternalDigest { source } => fmt::Display::fmt(source, formatter), Self::Integer { field, .. } => { formatter.write_str("cannot parse canonical ")?; escaped_controls(formatter, field) @@ -116,33 +77,6 @@ impl ConformanceError { escaped_controls(formatter, parameter)?; write!(formatter, ": {source}") } - Self::ProcessDiagnosticEncoding { program, code, .. } => write!( - formatter, - "{program} failed with status {code:?} and non-UTF-8 diagnostics" - ), - Self::ProcessFailed { - program, - code, - stderr, - } => { - write!(formatter, "{program} failed with status {code:?}: ")?; - escaped_controls(formatter, stderr) - } - Self::ProcessOutputBound { - program, - stream, - maximum, - } => write!( - formatter, - "{program} {stream} exceeded the {maximum}-byte bound" - ), - Self::ProcessTimeout { program, duration } => { - write!(formatter, "{program} exceeded {duration:?}") - } - Self::ReaderPanic { program, stream } => { - write!(formatter, "{program} {stream} reader panicked") - } - Self::WriterPanic { program } => write!(formatter, "{program} stdin writer panicked"), Self::Utf8 { path, .. } => { escaped_path(formatter, path)?; formatter.write_str(": protocol is not UTF-8") @@ -162,47 +96,12 @@ impl fmt::Display for ConformanceError { impl Error for ConformanceError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { - Self::Cleanup { primary, .. } => Some(primary.as_ref()), + Self::ExternalDigest { source } => Some(source), Self::Io { source, .. } => Some(source), Self::Integer { source, .. } => Some(source), Self::Path { source, .. } => Some(source), - Self::ProcessDiagnosticEncoding { source, .. } | Self::Utf8 { source, .. } => { - Some(source) - } - Self::ProcessFailed { .. } - | Self::ProcessOutputBound { .. } - | Self::ProcessTimeout { .. } - | Self::ReaderPanic { .. } - | Self::WriterPanic { .. } - | Self::Violation(_) => None, + Self::Utf8 { source, .. } => Some(source), + Self::Violation(_) => None, } } } - -#[cfg(test)] -mod tests { - use std::error::Error; - use std::io; - - use super::ConformanceError; - - #[test] - fn cleanup_preserves_one_prefix_and_the_primary_source() { - let error = ConformanceError::cleanup( - ConformanceError::violation("primary refusal"), - "close witness", - io::Error::other("cleanup refusal"), - ); - assert_eq!( - error.to_string(), - "protocol conformance check failed: primary refusal; \ - cleanup could not close witness: cleanup refusal" - ); - assert!(matches!( - error.source(), - Some(source) - if source.to_string() - == "protocol conformance check failed: primary refusal" - )); - } -} diff --git a/xtask/src/protocol_conformance/external_digest.rs b/xtask/src/protocol_conformance/external_digest.rs index 0c8d738..e736757 100644 --- a/xtask/src/protocol_conformance/external_digest.rs +++ b/xtask/src/protocol_conformance/external_digest.rs @@ -1,273 +1,11 @@ -//! This module owns the bounded external `b3sum` conformance witness. +//! This module owns the conformance port to the shared external digest witness. -use std::io::{self, Write}; -use std::path::Path; -use std::process::{Child, ChildStdin, Command, ExitStatus, Stdio}; -use std::thread::{self, JoinHandle}; -use std::time::{Duration, Instant}; - -use crate::process_output::{BoundedBytes, bounded_bytes}; +use crate::external_digest; use super::ConformanceError; -const B3SUM: &str = "b3sum"; -const DIAGNOSTIC_LIMIT: usize = 65_536; const DIGEST_BYTES: usize = 32; -const TIMEOUT: Duration = Duration::from_secs(10); - -struct B3sumProcess { - child: Child, - stdin: ChildStdin, - stdout_worker: JoinHandle>, - stderr_worker: JoinHandle>, -} pub(super) fn digest(parts: &[&[u8]]) -> Result<[u8; DIGEST_BYTES], ConformanceError> { - let process = start()?; - collect_with_timeout(process, parts, TIMEOUT) -} - -fn start() -> Result { - let mut child = Command::new(B3SUM) - .args(["--raw", "--no-mmap", "--num-threads", "1"]) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .map_err(|source| ConformanceError::io("start b3sum", Path::new(B3SUM), source))?; - let stdin = take_stdin(&mut child)?; - let stdout = take_stdout(&mut child)?; - let stderr = take_stderr(&mut child)?; - let stdout_worker = match start_reader("output", stdout, DIGEST_BYTES) { - Ok(worker) => worker, - Err(error) => return Err(cleanup(&mut child, error)), - }; - let stderr_worker = match start_reader("diagnostic", stderr, DIAGNOSTIC_LIMIT) { - Ok(worker) => worker, - Err(error) => { - let error = cleanup(&mut child, error); - drop(join_reader(stdout_worker, "output")); - return Err(error); - } - }; - Ok(B3sumProcess { - child, - stdin, - stdout_worker, - stderr_worker, - }) -} - -fn take_stdin(child: &mut Child) -> Result { - child.stdin.take().ok_or_else(|| { - cleanup( - child, - ConformanceError::violation("b3sum stdin pipe is absent"), - ) - }) -} - -fn take_stdout(child: &mut Child) -> Result { - child.stdout.take().ok_or_else(|| { - cleanup( - child, - ConformanceError::violation("b3sum stdout pipe is absent"), - ) - }) -} - -fn take_stderr(child: &mut Child) -> Result { - child.stderr.take().ok_or_else(|| { - cleanup( - child, - ConformanceError::violation("b3sum stderr pipe is absent"), - ) - }) -} - -fn start_reader( - stream: &'static str, - reader: impl io::Read + Send + 'static, - maximum: usize, -) -> Result>, ConformanceError> { - thread::Builder::new() - .name(format!("conformance-b3sum-{stream}")) - .spawn(move || bounded_bytes(reader, maximum)) - .map_err(|source| ConformanceError::io("start b3sum reader", Path::new(B3SUM), source)) -} - -fn collect_with_timeout( - process: B3sumProcess, - parts: &[&[u8]], - duration: Duration, -) -> Result<[u8; DIGEST_BYTES], ConformanceError> { - let B3sumProcess { - mut child, - stdin, - stdout_worker, - stderr_worker, - } = process; - let status = write_and_wait(&mut child, stdin, parts, duration); - let output = join_reader(stdout_worker, "output"); - let diagnostic = join_reader(stderr_worker, "diagnostic"); - let status = status?; - validate(status, output?, diagnostic?) -} - -fn write_and_wait( - child: &mut Child, - stdin: ChildStdin, - parts: &[&[u8]], - duration: Duration, -) -> Result { - let expires = Instant::now() - .checked_add(duration) - .ok_or_else(|| cleanup(child, timeout(duration)))?; - thread::scope(|scope| { - let writer = thread::Builder::new() - .name(String::from("conformance-b3sum-input")) - .spawn_scoped(scope, move || { - let mut stdin = stdin; - write_parts(&mut stdin, parts) - }) - .map_err(|source| ConformanceError::io("start b3sum writer", Path::new(B3SUM), source)); - let writer = match writer { - Ok(worker) => worker, - Err(error) => return Err(cleanup(child, error)), - }; - let status = wait_until(child, expires, duration); - let written = writer - .join() - .map_err(|_| ConformanceError::WriterPanic { program: B3SUM }); - match status { - Ok(status) => { - written??; - Ok(status) - } - Err(primary) => { - drop(written); - Err(primary) - } - } - }) + external_digest::b3sum(parts).map_err(ConformanceError::external_digest) } - -fn write_parts(stdin: &mut ChildStdin, parts: &[&[u8]]) -> Result<(), ConformanceError> { - for part in parts { - stdin - .write_all(part) - .map_err(|source| ConformanceError::io("write b3sum preimage", B3SUM, source))?; - } - Ok(()) -} - -fn wait_until( - child: &mut Child, - expires: Instant, - duration: Duration, -) -> Result { - loop { - match child.try_wait() { - Ok(Some(status)) => return Ok(status), - Ok(None) if Instant::now() >= expires => { - return Err(cleanup(child, timeout(duration))); - } - Ok(None) => thread::sleep(Duration::from_millis(10)), - Err(source) => { - return Err(cleanup( - child, - ConformanceError::io("poll b3sum", B3SUM, source), - )); - } - } - } -} - -const fn timeout(duration: Duration) -> ConformanceError { - ConformanceError::ProcessTimeout { - program: B3SUM, - duration, - } -} - -fn join_reader( - worker: JoinHandle>, - stream: &'static str, -) -> Result { - worker - .join() - .map_err(|_| ConformanceError::ReaderPanic { - program: B3SUM, - stream, - })? - .map_err(|source| ConformanceError::io("read b3sum output", B3SUM, source)) -} - -fn validate( - status: ExitStatus, - output: BoundedBytes, - diagnostic: BoundedBytes, -) -> Result<[u8; DIGEST_BYTES], ConformanceError> { - refuse_exceeded("output", &output, DIGEST_BYTES)?; - refuse_exceeded("diagnostic", &diagnostic, DIAGNOSTIC_LIMIT)?; - if !status.success() { - return Err(process_failure(status.code(), diagnostic.bytes)); - } - if !diagnostic.bytes.is_empty() { - return Err(ConformanceError::violation( - "b3sum wrote diagnostics despite successful exit", - )); - } - output - .bytes - .try_into() - .map_err(|_| ConformanceError::violation("b3sum returned a noncanonical digest width")) -} - -const fn refuse_exceeded( - stream: &'static str, - output: &BoundedBytes, - maximum: usize, -) -> Result<(), ConformanceError> { - if output.exceeded { - Err(ConformanceError::ProcessOutputBound { - program: B3SUM, - stream, - maximum, - }) - } else { - Ok(()) - } -} - -fn process_failure(code: Option, diagnostic: Vec) -> ConformanceError { - match String::from_utf8(diagnostic) { - Ok(stderr) => ConformanceError::ProcessFailed { - program: B3SUM, - code, - stderr, - }, - Err(source) => ConformanceError::ProcessDiagnosticEncoding { - program: B3SUM, - code, - source, - }, - } -} - -fn cleanup(child: &mut Child, primary: ConformanceError) -> ConformanceError { - let kill = child.kill(); - let wait = child.wait(); - if let Err(source) = kill - && source.kind() != io::ErrorKind::InvalidInput - { - return ConformanceError::cleanup(primary, "kill b3sum", source); - } - if let Err(source) = wait { - return ConformanceError::cleanup(primary, "wait for b3sum", source); - } - primary -} - -#[cfg(test)] -mod tests; diff --git a/xtask/src/protocol_conformance/external_digest/tests.rs b/xtask/src/protocol_conformance/external_digest/tests.rs deleted file mode 100644 index 5cac4d3..0000000 --- a/xtask/src/protocol_conformance/external_digest/tests.rs +++ /dev/null @@ -1,62 +0,0 @@ -//! This module owns external-digest process deadline regression evidence. - -use std::env; -use std::process::{Command, Stdio}; -use std::sync::mpsc::{self, RecvTimeoutError}; -use std::time::Duration; - -use super::{ - B3SUM, B3sumProcess, ConformanceError, collect_with_timeout, start_reader, take_stderr, - take_stdin, take_stdout, -}; - -const BLOCKING_CHILD: &str = "KEEP_XTASK_B3SUM_BLOCKING_CHILD"; - -#[test] -fn blocked_stdin_write_obeys_process_deadline() -> Result<(), ConformanceError> { - let executable = env::current_exe().map_err(|source| { - ConformanceError::io("locate digest test executable", "current", source) - })?; - let mut child = Command::new(executable) - .args([ - "--exact", - "protocol_conformance::external_digest::tests::digest_child_does_not_read_stdin", - ]) - .env(BLOCKING_CHILD, "1") - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .map_err(|source| { - ConformanceError::io("start digest test executable", "current", source) - })?; - let stdin = take_stdin(&mut child)?; - let stdout = take_stdout(&mut child)?; - let stderr = take_stderr(&mut child)?; - let process = B3sumProcess { - child, - stdin, - stdout_worker: start_reader("output", stdout, 32)?, - stderr_worker: start_reader("diagnostic", stderr, 65_536)?, - }; - let preimage = vec![0_u8; 1_048_576]; - assert!(matches!( - collect_with_timeout(process, &[&preimage], Duration::from_millis(50)), - Err(ConformanceError::ProcessTimeout { program, duration }) - if program == B3SUM && duration == Duration::from_millis(50) - )); - Ok(()) -} - -#[test] -fn digest_child_does_not_read_stdin() { - if env::var_os(BLOCKING_CHILD).is_none() { - return; - } - let (sender, receiver) = mpsc::channel::<()>(); - assert!(matches!( - receiver.recv_timeout(Duration::from_secs(1)), - Err(RecvTimeoutError::Timeout) - )); - drop(sender); -} diff --git a/xtask/tests/source_policy_contract.rs b/xtask/tests/source_policy_contract.rs index f1d9efb..b239c2a 100644 --- a/xtask/tests/source_policy_contract.rs +++ b/xtask/tests/source_policy_contract.rs @@ -5,12 +5,16 @@ const BOUNDED_PROCESS: &str = include_str!("../src/bounded_process.rs"); const BOUNDED_PROCESS_CAPTURE: &str = include_str!("../src/bounded_process/capture.rs"); const BOUNDED_PROCESS_CAPTURE_LIMIT: &str = include_str!("../src/bounded_process/capture_limit.rs"); const BOUNDED_PROCESS_ERROR: &str = include_str!("../src/bounded_process/error.rs"); +const BOUNDED_PROCESS_INPUT: &str = include_str!("../src/bounded_process/input.rs"); const BOUNDED_PROCESS_INTERRUPT: &str = include_str!("../src/bounded_process/interrupt.rs"); const BOUNDED_PROCESS_GROUP: &str = include_str!("../src/bounded_process/process_group.rs"); const BOUNDED_PROCESS_GROUP_TESTS: &str = include_str!("../src/bounded_process/process_group/tests.rs"); const BOUNDED_PROCESS_READER: &str = include_str!("../src/bounded_process/reader.rs"); const BOUNDED_PROCESS_TESTS: &str = include_str!("../src/bounded_process/tests.rs"); +const CONFORMANCE_B3SUM: &str = include_str!("../src/protocol_conformance/external_digest.rs"); +const EXTERNAL_DIGEST: &str = include_str!("../src/external_digest.rs"); +const GOLDEN_B3SUM: &str = include_str!("../src/golden_file_worldline/b3sum_oracle.rs"); const REPOSITORY_FIXTURE: &str = include_str!("../src/repository_fixture.rs"); const GIT_INVENTORY_ERROR: &str = include_str!("../src/git_inventory/error.rs"); const GIT_PATH_STREAM: &str = include_str!("../src/git_inventory/path_stream.rs"); @@ -73,6 +77,19 @@ fn sanitized_git_fixture_has_one_process_authority() { assert_eq!(definitions, 1); } +#[test] +fn external_digest_has_one_process_authority() { + let process_authorities = CONFORMANCE_B3SUM.matches("Command::new(").count() + + EXTERNAL_DIGEST.matches("Command::new(").count() + + GOLDEN_B3SUM.matches("Command::new(").count(); + assert_eq!(process_authorities, 1); + assert!(EXTERNAL_DIGEST.contains("capture_with_input_limits(")); + assert!(!CONFORMANCE_B3SUM.contains("Command::new(")); + assert!(!GOLDEN_B3SUM.contains("Command::new(")); + assert!(!GOLDEN_B3SUM.contains(".wait()")); + assert!(!GOLDEN_B3SUM.contains(".write_all(")); +} + #[test] fn git_inventory_uses_the_deadline_bounded_process_layer() { assert!(GIT_PROCESS.contains("const GIT_DEADLINE: Duration")); @@ -157,7 +174,7 @@ fn process_fixtures_isolate_git_from_host_configuration() { } #[test] -fn repository_process_boundaries_document_every_exported_contract() -> Result<(), String> { +fn bounded_process_capture_documents_every_exported_contract() -> Result<(), String> { require_docs( BOUNDED_PROCESS, &["pub(crate) struct ProcessOutput", "pub(crate) fn status("], @@ -167,6 +184,7 @@ fn repository_process_boundaries_document_every_exported_contract() -> Result<() &[ "pub(crate) fn capture(", "pub(crate) fn capture_with(", + "pub(crate) fn capture_with_input_limits(", "pub(crate) fn capture_with_limits(", ], )?; @@ -191,7 +209,12 @@ fn repository_process_boundaries_document_every_exported_contract() -> Result<() " Timeout {", " pub(crate) fn is_not_found(", ], - )?; + ) +} + +#[test] +fn bounded_process_support_documents_every_exported_contract() -> Result<(), String> { + require_docs(BOUNDED_PROCESS_INPUT, &["pub(super) fn write_input("])?; require_docs( BOUNDED_PROCESS_INTERRUPT, &[ @@ -216,6 +239,23 @@ fn repository_process_boundaries_document_every_exported_contract() -> Result<() " pub(super) fn receive(", " pub(super) fn join(", ], + ) +} + +#[test] +fn repository_process_adapters_document_every_exported_contract() -> Result<(), String> { + require_docs( + EXTERNAL_DIGEST, + &[ + "pub(crate) enum ExternalDigestError", + " Environment {", + " Process {", + " DiagnosticEncoding {", + " Failed {", + " UnexpectedDiagnostic,", + " Width {", + "pub(crate) fn b3sum(", + ], )?; require_docs(REPOSITORY_FIXTURE, &["pub(crate) fn run_git("])?; require_docs( From f146939919154c006ba78b94e5db38f73201bae3 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:16:49 -0700 Subject: [PATCH 02/15] Fix: observe deadlines during input progress --- xtask/src/bounded_process/input.rs | 58 ++++++++++++++++++++++++++++-- 1 file changed, 56 insertions(+), 2 deletions(-) diff --git a/xtask/src/bounded_process/input.rs b/xtask/src/bounded_process/input.rs index 025f85d..37123ae 100644 --- a/xtask/src/bounded_process/input.rs +++ b/xtask/src/bounded_process/input.rs @@ -35,14 +35,15 @@ pub(super) fn write_input( fn write_part( program: &'static str, - stdin: &mut ChildStdin, + writer: &mut impl Write, part: &[u8], deadline: &ProcessDeadline, interrupts: &InterruptGuard, ) -> Result<(), ProcessError> { let mut remaining = part; while !remaining.is_empty() { - match stdin.write(remaining) { + observe_input_boundary(program, deadline, interrupts)?; + match writer.write(remaining) { Ok(0) => { return Err(process_io( program, @@ -72,6 +73,18 @@ fn write_part( Ok(()) } +fn observe_input_boundary( + program: &'static str, + deadline: &ProcessDeadline, + interrupts: &InterruptGuard, +) -> Result<(), ProcessError> { + if let Some(error) = interrupts.refusal(program) { + return Err(error); + } + deadline.remaining(program)?; + Ok(()) +} + fn wait_for_input( program: &'static str, deadline: &ProcessDeadline, @@ -99,3 +112,44 @@ const fn process_io( source, } } + +#[cfg(test)] +mod tests { + use super::*; + + const TEST_PROGRAM: &str = "input-progress-test"; + + struct OneByteWriter; + + impl Write for OneByteWriter { + fn write(&mut self, bytes: &[u8]) -> Result { + Ok(usize::from(!bytes.is_empty())) + } + + fn flush(&mut self) -> Result<(), io::Error> { + Ok(()) + } + } + + #[test] + fn continuous_input_progress_still_obeys_the_complete_deadline() -> Result<(), ProcessError> { + let deadline = ProcessDeadline::new(TEST_PROGRAM, Some(Duration::ZERO))?; + let interrupts = InterruptGuard::begin(TEST_PROGRAM)?; + let result = write_part( + TEST_PROGRAM, + &mut OneByteWriter, + b"deadline", + &deadline, + &interrupts, + ); + + assert!(matches!( + result, + Err(ProcessError::Timeout { + program: TEST_PROGRAM, + duration: Duration::ZERO, + }) + )); + Ok(()) + } +} From 2cd86a228049ef5a1558721f2e1eb33fe8a843c9 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:19:51 -0700 Subject: [PATCH 03/15] Refactor: separate process policy contracts --- xtask/tests/process_policy_contract.rs | 255 +++++++++++++++++++++++++ xtask/tests/source_policy_contract.rs | 252 ------------------------ 2 files changed, 255 insertions(+), 252 deletions(-) create mode 100644 xtask/tests/process_policy_contract.rs diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs new file mode 100644 index 0000000..016fd7e --- /dev/null +++ b/xtask/tests/process_policy_contract.rs @@ -0,0 +1,255 @@ +//! Written-policy regression evidence for bounded repository processes. + +const BOUNDED_PROCESS: &str = include_str!("../src/bounded_process.rs"); +const BOUNDED_PROCESS_CAPTURE: &str = include_str!("../src/bounded_process/capture.rs"); +const BOUNDED_PROCESS_CAPTURE_LIMIT: &str = include_str!("../src/bounded_process/capture_limit.rs"); +const BOUNDED_PROCESS_ERROR: &str = include_str!("../src/bounded_process/error.rs"); +const BOUNDED_PROCESS_INPUT: &str = include_str!("../src/bounded_process/input.rs"); +const BOUNDED_PROCESS_INTERRUPT: &str = include_str!("../src/bounded_process/interrupt.rs"); +const BOUNDED_PROCESS_GROUP: &str = include_str!("../src/bounded_process/process_group.rs"); +const BOUNDED_PROCESS_GROUP_TESTS: &str = + include_str!("../src/bounded_process/process_group/tests.rs"); +const BOUNDED_PROCESS_READER: &str = include_str!("../src/bounded_process/reader.rs"); +const BOUNDED_PROCESS_TESTS: &str = include_str!("../src/bounded_process/tests.rs"); +const CONFORMANCE_B3SUM: &str = include_str!("../src/protocol_conformance/external_digest.rs"); +const EXTERNAL_DIGEST: &str = include_str!("../src/external_digest.rs"); +const GOLDEN_B3SUM: &str = include_str!("../src/golden_file_worldline/b3sum_oracle.rs"); +const REPOSITORY_FIXTURE: &str = include_str!("../src/repository_fixture.rs"); +const GIT_INVENTORY_ERROR: &str = include_str!("../src/git_inventory/error.rs"); +const GIT_PATH_STREAM: &str = include_str!("../src/git_inventory/path_stream.rs"); +const GIT_PROCESS: &str = include_str!("../src/git_inventory/process.rs"); +const SOURCE_PURE_RUST_TESTS: &str = include_str!("../src/source_structure/pure_rust_tests.rs"); + +#[test] +fn sanitized_git_fixture_has_one_process_authority() { + let definitions = REPOSITORY_FIXTURE.matches("fn run_git(").count() + + SOURCE_PURE_RUST_TESTS.matches("fn run_git(").count(); + assert_eq!(definitions, 1); +} + +#[test] +fn external_digest_has_one_process_authority() { + let process_authorities = CONFORMANCE_B3SUM.matches("Command::new(").count() + + EXTERNAL_DIGEST.matches("Command::new(").count() + + GOLDEN_B3SUM.matches("Command::new(").count(); + assert_eq!(process_authorities, 1); + assert!(EXTERNAL_DIGEST.contains("capture_with_input_limits(")); + assert!(!CONFORMANCE_B3SUM.contains("Command::new(")); + assert!(!GOLDEN_B3SUM.contains("Command::new(")); + assert!(!GOLDEN_B3SUM.contains(".wait()")); + assert!(!GOLDEN_B3SUM.contains(".write_all(")); +} + +#[test] +fn git_inventory_uses_the_deadline_bounded_process_layer() { + assert!(GIT_PROCESS.contains("const GIT_DEADLINE: Duration")); + assert!(GIT_PROCESS.contains("bounded_process::capture_with_limits(")); +} + +#[test] +fn captured_process_keeps_the_group_leader_until_reader_collection_finishes() +-> Result<(), &'static str> { + let (_, after_signature) = BOUNDED_PROCESS_CAPTURE + .split_once(" fn finish(") + .ok_or("captured process must retain its finish boundary")?; + let (body, _) = after_signature + .split_once("\n fn cleanup_readers") + .ok_or("captured process finish boundary must remain inspectable")?; + let wait = body + .find("wait_for_child") + .ok_or("captured process must reap its child")?; + + for operation in [ + "self.stdout.receive", + "self.stderr.receive", + "self.stdout.join", + "self.stderr.join", + ] { + let position = body + .find(operation) + .ok_or("captured process must collect and join both output streams")?; + assert!(position < wait, "child wait precedes {operation}"); + } + assert!( + !body + .get(wait..) + .unwrap_or_default() + .contains("cleanup_process"), + "cleanup may group-kill after the child ownership lifetime ends" + ); + Ok(()) +} + +#[test] +fn descendant_cleanup_uses_disconnect_evidence_instead_of_elapsed_time() { + assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("descendant_survived_cleanup")); + assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("Duration::from_millis(500)")); + assert!(BOUNDED_PROCESS_GROUP_TESTS.contains("require_descendant_disconnect")); +} + +#[test] +fn repository_git_fixtures_use_the_bounded_process_layer() { + assert!(REPOSITORY_FIXTURE.contains("bounded_process::status(")); + assert!(REPOSITORY_FIXTURE.contains("GIT_FIXTURE_DEADLINE")); + assert!(!REPOSITORY_FIXTURE.contains(".output()")); +} + +#[test] +fn repository_git_fixtures_clear_the_ambient_environment() { + assert!(REPOSITORY_FIXTURE.contains(".env_clear()")); + assert!(REPOSITORY_FIXTURE.contains("env::var_os(\"PATH\")")); + assert!(REPOSITORY_FIXTURE.contains(".env(\"PATH\"")); + assert!(REPOSITORY_FIXTURE.contains(".env(\"LC_ALL\", \"C\")")); +} + +#[test] +fn process_fixtures_do_not_write_to_rust_stdout() { + assert!(!BOUNDED_PROCESS_TESTS.contains("io::stdout()")); +} + +#[test] +fn process_fixtures_isolate_git_from_host_configuration() { + assert_eq!( + BOUNDED_PROCESS_TESTS + .matches("Command::new(\"git\")") + .count(), + 1 + ); + assert!(BOUNDED_PROCESS_TESTS.contains(".env_clear()")); + assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"PATH\"")); + assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"GIT_CONFIG_NOSYSTEM\", \"1\")")); + assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"GIT_CONFIG_GLOBAL\"")); + assert!(BOUNDED_PROCESS_TESTS.contains("--template=")); + assert!(!BOUNDED_PROCESS_TESTS.contains("template.display()")); +} + +#[test] +fn bounded_process_capture_documents_every_exported_contract() -> Result<(), String> { + require_docs( + BOUNDED_PROCESS, + &["pub(crate) struct ProcessOutput", "pub(crate) fn status("], + )?; + require_docs( + BOUNDED_PROCESS_CAPTURE, + &[ + "pub(crate) fn capture(", + "pub(crate) fn capture_with(", + "pub(crate) fn capture_with_input_limits(", + "pub(crate) fn capture_with_limits(", + ], + )?; + require_docs( + BOUNDED_PROCESS_CAPTURE_LIMIT, + &[ + "pub(crate) struct CaptureLimits", + " pub(crate) const fn new(", + ], + )?; + require_docs( + BOUNDED_PROCESS_ERROR, + &[ + "pub(crate) enum ProcessError", + " Additional {", + " Cleanup {", + " Io {", + " Interrupted {", + " MissingStream {", + " OutputLimit {", + " ReaderPanic {", + " Timeout {", + " pub(crate) fn is_not_found(", + ], + ) +} + +#[test] +fn bounded_process_support_documents_every_exported_contract() -> Result<(), String> { + require_docs(BOUNDED_PROCESS_INPUT, &["pub(super) fn write_input("])?; + require_docs( + BOUNDED_PROCESS_INTERRUPT, + &[ + "pub(super) struct InterruptGuard", + " pub(super) fn begin(", + " pub(super) fn refusal(", + ], + )?; + require_docs( + BOUNDED_PROCESS_GROUP, + &[ + "pub(super) struct ProcessGroup", + " pub(super) fn for_child(", + " pub(super) fn terminate(", + ], + )?; + require_docs( + BOUNDED_PROCESS_READER, + &[ + "pub(super) struct ReaderWorker", + " pub(super) fn start(", + " pub(super) fn receive(", + " pub(super) fn join(", + ], + ) +} + +#[test] +fn repository_process_adapters_document_every_exported_contract() -> Result<(), String> { + require_docs( + EXTERNAL_DIGEST, + &[ + "pub(crate) enum ExternalDigestError", + " Environment {", + " Process {", + " DiagnosticEncoding {", + " Failed {", + " UnexpectedDiagnostic,", + " Width {", + "pub(crate) fn b3sum(", + ], + )?; + require_docs(REPOSITORY_FIXTURE, &["pub(crate) fn run_git("])?; + require_docs( + GIT_INVENTORY_ERROR, + &[ + "pub(crate) enum GitOutputUnit", + " Bytes,", + " Items,", + "pub(crate) enum GitInventoryError", + " DuplicatePath(", + " EmptyPath {", + " Failed {", + " DiagnosticEncoding {", + " OutputBound {", + " OutputFraming {", + " Process {", + " Run {", + ], + )?; + require_docs( + GIT_PATH_STREAM, + &[ + "pub(super) fn read_paths(", + "pub(crate) struct GitPath(", + " pub(crate) const fn new(", + " pub(crate) fn as_bytes(", + ], + )?; + require_docs(GIT_PROCESS, &["pub(crate) fn paths_with("]) +} + +fn require_docs(source: &str, declarations: &[&str]) -> Result<(), String> { + for declaration in declarations { + let (before, _) = source + .split_once(declaration) + .ok_or_else(|| format!("missing declaration `{declaration}`"))?; + let documented = before + .lines() + .rev() + .find(|line| !line.trim_start().starts_with("#[")) + .is_some_and(|line| line.trim_start().starts_with("///")); + if !documented { + return Err(format!("missing rustdoc for `{declaration}`")); + } + } + Ok(()) +} diff --git a/xtask/tests/source_policy_contract.rs b/xtask/tests/source_policy_contract.rs index b239c2a..34ab456 100644 --- a/xtask/tests/source_policy_contract.rs +++ b/xtask/tests/source_policy_contract.rs @@ -1,26 +1,8 @@ //! Written-policy regression evidence for the executable source-size law. const RUST_STANDARDS: &str = include_str!("../../docs/Rust Standards.md"); -const BOUNDED_PROCESS: &str = include_str!("../src/bounded_process.rs"); -const BOUNDED_PROCESS_CAPTURE: &str = include_str!("../src/bounded_process/capture.rs"); -const BOUNDED_PROCESS_CAPTURE_LIMIT: &str = include_str!("../src/bounded_process/capture_limit.rs"); -const BOUNDED_PROCESS_ERROR: &str = include_str!("../src/bounded_process/error.rs"); -const BOUNDED_PROCESS_INPUT: &str = include_str!("../src/bounded_process/input.rs"); -const BOUNDED_PROCESS_INTERRUPT: &str = include_str!("../src/bounded_process/interrupt.rs"); -const BOUNDED_PROCESS_GROUP: &str = include_str!("../src/bounded_process/process_group.rs"); -const BOUNDED_PROCESS_GROUP_TESTS: &str = - include_str!("../src/bounded_process/process_group/tests.rs"); -const BOUNDED_PROCESS_READER: &str = include_str!("../src/bounded_process/reader.rs"); -const BOUNDED_PROCESS_TESTS: &str = include_str!("../src/bounded_process/tests.rs"); -const CONFORMANCE_B3SUM: &str = include_str!("../src/protocol_conformance/external_digest.rs"); -const EXTERNAL_DIGEST: &str = include_str!("../src/external_digest.rs"); -const GOLDEN_B3SUM: &str = include_str!("../src/golden_file_worldline/b3sum_oracle.rs"); -const REPOSITORY_FIXTURE: &str = include_str!("../src/repository_fixture.rs"); -const GIT_INVENTORY_ERROR: &str = include_str!("../src/git_inventory/error.rs"); -const GIT_PATH_STREAM: &str = include_str!("../src/git_inventory/path_stream.rs"); const GIT_PROCESS: &str = include_str!("../src/git_inventory/process.rs"); const REPOSITORY_FILE: &str = include_str!("../src/repository_file.rs"); -const SOURCE_PURE_RUST_TESTS: &str = include_str!("../src/source_structure/pure_rust_tests.rs"); const SOURCE_STRUCTURE: &str = include_str!("../src/source_structure.rs"); const SOURCE_INVENTORY: &str = include_str!("../src/source_structure/source_inventory.rs"); @@ -69,237 +51,3 @@ fn regular_source_admission_has_one_refusal_boundary() { ); assert_eq!(SOURCE_STRUCTURE.matches("fn admit_regular(").count(), 1); } - -#[test] -fn sanitized_git_fixture_has_one_process_authority() { - let definitions = REPOSITORY_FIXTURE.matches("fn run_git(").count() - + SOURCE_PURE_RUST_TESTS.matches("fn run_git(").count(); - assert_eq!(definitions, 1); -} - -#[test] -fn external_digest_has_one_process_authority() { - let process_authorities = CONFORMANCE_B3SUM.matches("Command::new(").count() - + EXTERNAL_DIGEST.matches("Command::new(").count() - + GOLDEN_B3SUM.matches("Command::new(").count(); - assert_eq!(process_authorities, 1); - assert!(EXTERNAL_DIGEST.contains("capture_with_input_limits(")); - assert!(!CONFORMANCE_B3SUM.contains("Command::new(")); - assert!(!GOLDEN_B3SUM.contains("Command::new(")); - assert!(!GOLDEN_B3SUM.contains(".wait()")); - assert!(!GOLDEN_B3SUM.contains(".write_all(")); -} - -#[test] -fn git_inventory_uses_the_deadline_bounded_process_layer() { - assert!(GIT_PROCESS.contains("const GIT_DEADLINE: Duration")); - assert!(GIT_PROCESS.contains("bounded_process::capture_with_limits(")); -} - -#[test] -fn captured_process_keeps_the_group_leader_until_reader_collection_finishes() --> Result<(), &'static str> { - let (_, after_signature) = BOUNDED_PROCESS_CAPTURE - .split_once(" fn finish(") - .ok_or("captured process must retain its finish boundary")?; - let (body, _) = after_signature - .split_once("\n fn cleanup_readers") - .ok_or("captured process finish boundary must remain inspectable")?; - let wait = body - .find("wait_for_child") - .ok_or("captured process must reap its child")?; - - for operation in [ - "self.stdout.receive", - "self.stderr.receive", - "self.stdout.join", - "self.stderr.join", - ] { - let position = body - .find(operation) - .ok_or("captured process must collect and join both output streams")?; - assert!(position < wait, "child wait precedes {operation}"); - } - assert!( - !body - .get(wait..) - .unwrap_or_default() - .contains("cleanup_process"), - "cleanup may group-kill after the child ownership lifetime ends" - ); - Ok(()) -} - -#[test] -fn descendant_cleanup_uses_disconnect_evidence_instead_of_elapsed_time() { - assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("descendant_survived_cleanup")); - assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("Duration::from_millis(500)")); - assert!(BOUNDED_PROCESS_GROUP_TESTS.contains("require_descendant_disconnect")); -} - -#[test] -fn repository_git_fixtures_use_the_bounded_process_layer() { - assert!(REPOSITORY_FIXTURE.contains("bounded_process::status(")); - assert!(REPOSITORY_FIXTURE.contains("GIT_FIXTURE_DEADLINE")); - assert!(!REPOSITORY_FIXTURE.contains(".output()")); -} - -#[test] -fn repository_git_fixtures_clear_the_ambient_environment() { - assert!(REPOSITORY_FIXTURE.contains(".env_clear()")); - assert!(REPOSITORY_FIXTURE.contains("env::var_os(\"PATH\")")); - assert!(REPOSITORY_FIXTURE.contains(".env(\"PATH\"")); - assert!(REPOSITORY_FIXTURE.contains(".env(\"LC_ALL\", \"C\")")); -} - -#[test] -fn process_fixtures_do_not_write_to_rust_stdout() { - assert!(!BOUNDED_PROCESS_TESTS.contains("io::stdout()")); -} - -#[test] -fn process_fixtures_isolate_git_from_host_configuration() { - assert_eq!( - BOUNDED_PROCESS_TESTS - .matches("Command::new(\"git\")") - .count(), - 1 - ); - assert!(BOUNDED_PROCESS_TESTS.contains(".env_clear()")); - assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"PATH\"")); - assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"GIT_CONFIG_NOSYSTEM\", \"1\")")); - assert!(BOUNDED_PROCESS_TESTS.contains(".env(\"GIT_CONFIG_GLOBAL\"")); - assert!(BOUNDED_PROCESS_TESTS.contains("--template=")); - assert!(!BOUNDED_PROCESS_TESTS.contains("template.display()")); -} - -#[test] -fn bounded_process_capture_documents_every_exported_contract() -> Result<(), String> { - require_docs( - BOUNDED_PROCESS, - &["pub(crate) struct ProcessOutput", "pub(crate) fn status("], - )?; - require_docs( - BOUNDED_PROCESS_CAPTURE, - &[ - "pub(crate) fn capture(", - "pub(crate) fn capture_with(", - "pub(crate) fn capture_with_input_limits(", - "pub(crate) fn capture_with_limits(", - ], - )?; - require_docs( - BOUNDED_PROCESS_CAPTURE_LIMIT, - &[ - "pub(crate) struct CaptureLimits", - " pub(crate) const fn new(", - ], - )?; - require_docs( - BOUNDED_PROCESS_ERROR, - &[ - "pub(crate) enum ProcessError", - " Additional {", - " Cleanup {", - " Io {", - " Interrupted {", - " MissingStream {", - " OutputLimit {", - " ReaderPanic {", - " Timeout {", - " pub(crate) fn is_not_found(", - ], - ) -} - -#[test] -fn bounded_process_support_documents_every_exported_contract() -> Result<(), String> { - require_docs(BOUNDED_PROCESS_INPUT, &["pub(super) fn write_input("])?; - require_docs( - BOUNDED_PROCESS_INTERRUPT, - &[ - "pub(super) struct InterruptGuard", - " pub(super) fn begin(", - " pub(super) fn refusal(", - ], - )?; - require_docs( - BOUNDED_PROCESS_GROUP, - &[ - "pub(super) struct ProcessGroup", - " pub(super) fn for_child(", - " pub(super) fn terminate(", - ], - )?; - require_docs( - BOUNDED_PROCESS_READER, - &[ - "pub(super) struct ReaderWorker", - " pub(super) fn start(", - " pub(super) fn receive(", - " pub(super) fn join(", - ], - ) -} - -#[test] -fn repository_process_adapters_document_every_exported_contract() -> Result<(), String> { - require_docs( - EXTERNAL_DIGEST, - &[ - "pub(crate) enum ExternalDigestError", - " Environment {", - " Process {", - " DiagnosticEncoding {", - " Failed {", - " UnexpectedDiagnostic,", - " Width {", - "pub(crate) fn b3sum(", - ], - )?; - require_docs(REPOSITORY_FIXTURE, &["pub(crate) fn run_git("])?; - require_docs( - GIT_INVENTORY_ERROR, - &[ - "pub(crate) enum GitOutputUnit", - " Bytes,", - " Items,", - "pub(crate) enum GitInventoryError", - " DuplicatePath(", - " EmptyPath {", - " Failed {", - " DiagnosticEncoding {", - " OutputBound {", - " OutputFraming {", - " Process {", - " Run {", - ], - )?; - require_docs( - GIT_PATH_STREAM, - &[ - "pub(super) fn read_paths(", - "pub(crate) struct GitPath(", - " pub(crate) const fn new(", - " pub(crate) fn as_bytes(", - ], - )?; - require_docs(GIT_PROCESS, &["pub(crate) fn paths_with("]) -} - -fn require_docs(source: &str, declarations: &[&str]) -> Result<(), String> { - for declaration in declarations { - let (before, _) = source - .split_once(declaration) - .ok_or_else(|| format!("missing declaration `{declaration}`"))?; - let documented = before - .lines() - .rev() - .find(|line| !line.trim_start().starts_with("#[")) - .is_some_and(|line| line.trim_start().starts_with("///")); - if !documented { - return Err(format!("missing rustdoc for `{declaration}`")); - } - } - Ok(()) -} From 61d280e3659f95b04e976228328805a59d36edcf Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:20:38 -0700 Subject: [PATCH 04/15] Docs: delimit synchronous spawn deadlines --- docs/conformance/golden-file-worldline.md | 11 ++++++----- xtask/src/bounded_process/capture.rs | 7 ++++--- xtask/src/external_digest.rs | 6 ++++-- 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/docs/conformance/golden-file-worldline.md b/docs/conformance/golden-file-worldline.md index 42da58a..99cf271 100644 --- a/docs/conformance/golden-file-worldline.md +++ b/docs/conformance/golden-file-worldline.md @@ -160,11 +160,12 @@ The model computes identity through Keep's public `BlobId` API, while the checked-in vectors were generated independently with `b3sum`. Agreement is therefore not a self-authored golden. The repository checker streams each canonical preimage through the same deadline-bounded external-digest adapter -used by protocol conformance. The deadline covers process spawn and stdin -transfer as well as output collection; stdout and stderr have independent byte -limits; and timeout or collection failure kills and reaps the complete child -process group. Digest-specific preimage construction remains owned by this -checker rather than the process adapter. +used by protocol conformance. The deadline clock starts before synchronous +spawn. After spawn returns, its remaining time bounds stdin transfer and output +collection; stdout and stderr have independent byte limits; and timeout or +collection failure kills and reaps the complete child process group. +Digest-specific preimage construction remains owned by this checker rather than +the process adapter. ## Exact M1 acceptance boundary diff --git a/xtask/src/bounded_process/capture.rs b/xtask/src/bounded_process/capture.rs index 4866d10..9d814b5 100644 --- a/xtask/src/bounded_process/capture.rs +++ b/xtask/src/bounded_process/capture.rs @@ -60,9 +60,10 @@ pub(crate) fn capture_with_limits( /// Runs one captured child with bounded streaming input and exact stream limits. /// -/// The complete deadline starts before spawn and covers every nonblocking stdin -/// write, both output readers, child execution, and cleanup. Input slices are -/// streamed directly without constructing a combined preimage allocation. +/// The deadline clock starts before synchronous spawn. After spawn returns, its +/// remaining time bounds every nonblocking stdin write, both output readers, +/// child execution, and cleanup. Input slices are streamed directly without +/// constructing a combined preimage allocation. pub(crate) fn capture_with_input_limits( program: &'static str, command: &mut Command, diff --git a/xtask/src/external_digest.rs b/xtask/src/external_digest.rs index 0405f9d..85b394a 100644 --- a/xtask/src/external_digest.rs +++ b/xtask/src/external_digest.rs @@ -56,8 +56,10 @@ pub(crate) enum ExternalDigestError { /// Computes one raw BLAKE3 digest through the independent `b3sum` witness. /// /// Input parts are streamed without concatenation. The child receives only the -/// admitted executable search path and `C` locale, and the complete operation -/// is bounded by one ten-second deadline and independent output limits. +/// admitted executable search path and `C` locale. One ten-second deadline +/// clock starts before synchronous spawn; after spawn returns, its remaining +/// time bounds input transfer, child execution, and independently limited +/// output collection. pub(crate) fn b3sum(parts: &[&[u8]]) -> Result<[u8; DIGEST_BYTES], ExternalDigestError> { let mut command = b3sum_command()?; execute(&mut command, parts, PROCESS_DEADLINE) From 380bee8a19b003da464deb017abe20d06367e600 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:25:52 -0700 Subject: [PATCH 05/15] Fix: bound failed process cleanup --- CHANGELOG.md | 3 +- docs/conformance/golden-file-worldline.md | 7 +- xtask/src/bounded_process/capture.rs | 27 +++-- xtask/src/bounded_process/cleanup.rs | 121 ++++++++++++++++------ xtask/src/bounded_process/reader.rs | 82 +++++++++++++-- xtask/tests/process_policy_contract.rs | 13 ++- 6 files changed, 190 insertions(+), 63 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 60316e1..12d091e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -101,7 +101,8 @@ after its public API and format compatibility policies are established. spawn and stdin transfer, stdin is streamed without a combined preimage allocation, stdout and stderr have independent limits, and every timeout or collection failure kills and reaps the process group while retaining typed - failure context. + failure context. Child reaping and stalled-reader retirement use a separate + fixed cleanup deadline instead of blocking without limit. - Fuzz policy admission, target reconciliation, bounded campaign execution, minimization deadlines, retained-corpus admission, and workflow contract tests now run through the repository's Rust `xtask`; the superseded Python diff --git a/docs/conformance/golden-file-worldline.md b/docs/conformance/golden-file-worldline.md index 99cf271..64fc85c 100644 --- a/docs/conformance/golden-file-worldline.md +++ b/docs/conformance/golden-file-worldline.md @@ -163,9 +163,10 @@ canonical preimage through the same deadline-bounded external-digest adapter used by protocol conformance. The deadline clock starts before synchronous spawn. After spawn returns, its remaining time bounds stdin transfer and output collection; stdout and stderr have independent byte limits; and timeout or -collection failure kills and reaps the complete child process group. -Digest-specific preimage construction remains owned by this checker rather than -the process adapter. +collection failure kills the complete child process group. Child reaping and +reader retirement use a separate fixed cleanup deadline and remain typed if +teardown cannot be proved. Digest-specific preimage construction remains owned +by this checker rather than the process adapter. ## Exact M1 acceptance boundary diff --git a/xtask/src/bounded_process/capture.rs b/xtask/src/bounded_process/capture.rs index 9d814b5..f2ecef4 100644 --- a/xtask/src/bounded_process/capture.rs +++ b/xtask/src/bounded_process/capture.rs @@ -5,7 +5,7 @@ use std::process::{Child, Command, ExitStatus, Stdio}; use std::thread; use std::time::Duration; -use super::cleanup::{cleanup_process, join_after_cleanup, join_readers}; +use super::cleanup::{cleanup_process, retire_after_cleanup, retire_readers}; use super::input::write_input; use super::{ CaptureLimits, InterruptGuard, ProcessDeadline, ProcessError, ProcessOutput, ReaderWorker, @@ -19,8 +19,9 @@ const PROCESS_POLL_INTERVAL: Duration = Duration::from_millis(10); /// /// Each stream is drained concurrently and retains at most one mebibyte. The /// optional deadline covers child execution and reader collection. Failures -/// terminate the child's dedicated process group, join both readers, and retain -/// the primary and cleanup errors in [`ProcessError`]. +/// terminate the child's dedicated process group, retire both readers within a +/// fixed cleanup deadline, and retain primary and cleanup errors in +/// [`ProcessError`]. pub(crate) fn capture( program: &'static str, command: &mut Command, @@ -45,7 +46,7 @@ pub(crate) fn capture_with( /// Runs one captured child with exact independent stream limits. /// /// The deadline covers child execution and both reader workers. Every failure -/// terminates the dedicated process group and joins the workers before return. +/// terminates the dedicated process group and bounds worker retirement. pub(crate) fn capture_with_limits( program: &'static str, command: &mut Command, @@ -62,8 +63,9 @@ pub(crate) fn capture_with_limits( /// /// The deadline clock starts before synchronous spawn. After spawn returns, its /// remaining time bounds every nonblocking stdin write, both output readers, -/// child execution, and cleanup. Input slices are streamed directly without -/// constructing a combined preimage allocation. +/// and child execution. Failed-operation teardown uses a separate fixed +/// deadline for child reaping and reader retirement. Input slices are streamed +/// directly without constructing a combined preimage allocation. pub(crate) fn capture_with_input_limits( program: &'static str, command: &mut Command, @@ -125,7 +127,7 @@ impl CapturedProcess { Ok(reader) => reader, Err(error) => { let error = cleanup_process(&mut child, error); - return Err(join_after_cleanup(stdout, error)); + return Err(retire_after_cleanup(stdout, error)); } }; Ok(Self { @@ -150,13 +152,8 @@ impl CapturedProcess { Ok(output) => output, Err(error) => return Err(self.cleanup_readers(error)), }; - if let Err(error) = self.stdout.join() { - let error = cleanup_process(&mut self.child, error); - return Err(join_after_cleanup(self.stderr, error)); - } - if let Err(error) = self.stderr.join() { - return Err(cleanup_process(&mut self.child, error)); - } + drop(self.stdout); + drop(self.stderr); let status = wait_for_child(program, &mut self.child, deadline, &self.interrupts)?; refuse_exceeded(program, "stdout", self.limits.stdout_bytes(), &stdout).and_then(|()| { refuse_exceeded(program, "stderr", self.limits.stderr_bytes(), &stderr) @@ -195,7 +192,7 @@ impl CapturedProcess { fn cleanup_readers(self, error: ProcessError) -> ProcessError { let mut child = self.child; let error = cleanup_process(&mut child, error); - join_readers(self.stdout, self.stderr, error) + retire_readers(self.stdout, self.stderr, error) } } diff --git a/xtask/src/bounded_process/cleanup.rs b/xtask/src/bounded_process/cleanup.rs index df05746..6feb84f 100644 --- a/xtask/src/bounded_process/cleanup.rs +++ b/xtask/src/bounded_process/cleanup.rs @@ -2,51 +2,39 @@ use std::io; use std::process::Child; +use std::thread; +use std::time::{Duration, Instant}; use super::process_group::ProcessGroup; use super::{ProcessError, ReaderWorker}; +const CLEANUP_DEADLINE: Duration = Duration::from_secs(2); +const CLEANUP_POLL_INTERVAL: Duration = Duration::from_millis(10); + pub(super) fn cleanup_process(child: &mut Child, primary: ProcessError) -> ProcessError { + drop(child.stdin.take()); let process_group = ProcessGroup::for_child(child).and_then(ProcessGroup::terminate); - let kill = child.kill(); - let wait = child.wait(); - if let Err(source) = process_group { - return ProcessError::Cleanup { - primary: Box::new(primary), - action: "terminate child process group", - source, - }; - } - if let Err(source) = kill - && source.kind() != io::ErrorKind::InvalidInput - { - return ProcessError::Cleanup { - primary: Box::new(primary), - action: "kill child process", - source, - }; - } - if let Err(source) = wait { - return ProcessError::Cleanup { - primary: Box::new(primary), - action: "reap child process", - source, - }; - } - primary + let kill = match child.kill() { + Err(source) if source.kind() == io::ErrorKind::InvalidInput => Ok(()), + result => result, + }; + let reap = reap_child(child); + let primary = with_cleanup(primary, "terminate child process group", process_group); + let primary = with_cleanup(primary, "kill child process", kill); + with_cleanup(primary, "reap child process", reap) } -pub(super) fn join_readers( +pub(super) fn retire_readers( stdout: ReaderWorker, stderr: ReaderWorker, primary: ProcessError, ) -> ProcessError { - let primary = join_after_cleanup(stdout, primary); - join_after_cleanup(stderr, primary) + let primary = retire_after_cleanup(stdout, primary); + retire_after_cleanup(stderr, primary) } -pub(super) fn join_after_cleanup(reader: ReaderWorker, primary: ProcessError) -> ProcessError { - match reader.join() { +pub(super) fn retire_after_cleanup(reader: ReaderWorker, primary: ProcessError) -> ProcessError { + match reader.retire(CLEANUP_DEADLINE) { Ok(()) => primary, Err(additional) => ProcessError::Additional { primary: Box::new(primary), @@ -54,3 +42,74 @@ pub(super) fn join_after_cleanup(reader: ReaderWorker, primary: ProcessError) -> }, } } + +fn reap_child(child: &mut Child) -> Result<(), io::Error> { + wait_until_reaped(CLEANUP_DEADLINE, || { + child.try_wait().map(|status| status.is_some()) + }) +} + +fn wait_until_reaped( + deadline: Duration, + mut poll: impl FnMut() -> Result, +) -> Result<(), io::Error> { + let expires = Instant::now() + .checked_add(deadline) + .ok_or_else(|| cleanup_timeout(deadline))?; + loop { + if poll()? { + return Ok(()); + } + let remaining = expires + .checked_duration_since(Instant::now()) + .ok_or_else(|| cleanup_timeout(deadline))?; + if remaining.is_zero() { + return Err(cleanup_timeout(deadline)); + } + thread::sleep(CLEANUP_POLL_INTERVAL.min(remaining)); + } +} + +fn with_cleanup( + primary: ProcessError, + action: &'static str, + result: Result<(), io::Error>, +) -> ProcessError { + match result { + Ok(()) => primary, + Err(source) => ProcessError::Cleanup { + primary: Box::new(primary), + action, + source, + }, + } +} + +fn cleanup_timeout(deadline: Duration) -> io::Error { + io::Error::new( + io::ErrorKind::TimedOut, + format!("child was not reaped within {deadline:?}"), + ) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + #[test] + fn stalled_child_reap_obeys_the_cleanup_deadline() -> Result<(), String> { + let mut polls = 0_u8; + let error = wait_until_reaped(Duration::ZERO, || { + polls = polls.saturating_add(1); + Ok(false) + }) + .err() + .ok_or_else(|| String::from("an unreaped child outlived the cleanup deadline"))?; + + assert_eq!(polls, 1); + assert_eq!(error.kind(), io::ErrorKind::TimedOut); + Ok(()) + } +} diff --git a/xtask/src/bounded_process/reader.rs b/xtask/src/bounded_process/reader.rs index c1f423d..9f9f461 100644 --- a/xtask/src/bounded_process/reader.rs +++ b/xtask/src/bounded_process/reader.rs @@ -3,6 +3,7 @@ use std::io; use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; use std::thread::{self, JoinHandle}; +use std::time::Duration; use crate::process_output::{BoundedBytes, bounded_bytes}; @@ -12,6 +13,7 @@ const INTERRUPT_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_m /// Owns one bounded stream reader and its single-result channel. pub(super) struct ReaderWorker { + completed: bool, handle: JoinHandle<()>, program: &'static str, receiver: Receiver>, @@ -41,6 +43,7 @@ impl ReaderWorker { source, })?; Ok(Self { + completed: false, handle, program, receiver, @@ -53,7 +56,7 @@ impl ReaderWorker { /// This call blocks only for the smaller of the remaining deadline and the /// fixed interrupt interval, so timeout and interruption remain observable. pub(super) fn receive( - &self, + &mut self, deadline: &ProcessDeadline, interrupts: &InterruptGuard, ) -> Result { @@ -64,6 +67,7 @@ impl ReaderWorker { let (wait, duration) = receive_wait(deadline, self.program)?; match self.receiver.recv_timeout(wait) { Ok(result) => { + self.completed = true; return result.map_err(|source| ProcessError::Io { program: self.program, action: "read child output", @@ -75,17 +79,42 @@ impl ReaderWorker { deadline.remaining(self.program)?; } Err(RecvTimeoutError::Disconnected) => { + self.completed = true; return Err(reader_panic(self.program, self.stream)); } } } } - /// Joins the completed reader and maps a worker panic to a typed failure. - pub(super) fn join(self) -> Result<(), ProcessError> { - self.handle - .join() - .map_err(|_panic| reader_panic(self.program, self.stream)) + /// Retires one reader without an unbounded thread join. + /// + /// A result already received proves the worker crossed its only fallible + /// read boundary. Otherwise retirement waits only for `deadline`; a stalled + /// reader becomes a typed timeout and its thread handle is detached. + pub(super) fn retire(self, deadline: Duration) -> Result<(), ProcessError> { + let result = if self.completed { + Ok(()) + } else { + match self.receiver.recv_timeout(deadline) { + Ok(Ok(_output)) => Ok(()), + Ok(Err(source)) => Err(ProcessError::Io { + program: self.program, + action: "read child output", + source, + }), + Err(RecvTimeoutError::Disconnected) => Err(reader_panic(self.program, self.stream)), + Err(RecvTimeoutError::Timeout) => Err(ProcessError::Io { + program: self.program, + action: "retire output reader", + source: io::Error::new( + io::ErrorKind::TimedOut, + format!("{} reader did not retire", self.stream), + ), + }), + } + }; + drop(self.handle); + result } } @@ -102,3 +131,44 @@ fn receive_wait( const fn reader_panic(program: &'static str, stream: &'static str) -> ProcessError { ProcessError::ReaderPanic { program, stream } } + +#[cfg(test)] +mod tests { + use std::sync::mpsc; + use std::time::Duration; + + use super::*; + + struct BlockingReader(mpsc::Receiver<()>); + + impl io::Read for BlockingReader { + fn read(&mut self, _buffer: &mut [u8]) -> Result { + self.0 + .recv() + .map_err(|source| io::Error::new(io::ErrorKind::BrokenPipe, source))?; + Ok(0) + } + } + + #[test] + fn stalled_reader_retirement_obeys_the_cleanup_deadline() + -> Result<(), Box> { + let (release, blocked) = mpsc::channel(); + let worker = ReaderWorker::start("reader-test", "stdout", BlockingReader(blocked), 1)?; + let error = worker + .retire(Duration::ZERO) + .err() + .ok_or_else(|| io::Error::other("a stalled reader retired successfully"))?; + + assert!(matches!( + error, + ProcessError::Io { + program: "reader-test", + action: "retire output reader", + ref source, + } if source.kind() == io::ErrorKind::TimedOut + )); + release.send(())?; + Ok(()) + } +} diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index 016fd7e..517e162 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -59,12 +59,7 @@ fn captured_process_keeps_the_group_leader_until_reader_collection_finishes() .find("wait_for_child") .ok_or("captured process must reap its child")?; - for operation in [ - "self.stdout.receive", - "self.stderr.receive", - "self.stdout.join", - "self.stderr.join", - ] { + for operation in ["self.stdout.receive", "self.stderr.receive"] { let position = body .find(operation) .ok_or("captured process must collect and join both output streams")?; @@ -77,6 +72,10 @@ fn captured_process_keeps_the_group_leader_until_reader_collection_finishes() .contains("cleanup_process"), "cleanup may group-kill after the child ownership lifetime ends" ); + assert!( + !body.contains(".join()"), + "reader collection may wait forever" + ); Ok(()) } @@ -187,7 +186,7 @@ fn bounded_process_support_documents_every_exported_contract() -> Result<(), Str "pub(super) struct ReaderWorker", " pub(super) fn start(", " pub(super) fn receive(", - " pub(super) fn join(", + " pub(super) fn retire(", ], ) } From 36041311ecb37e6c69da8d11ce03473771424c98 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:27:18 -0700 Subject: [PATCH 06/15] Fix: retain primary cleanup sources --- xtask/src/bounded_process/error.rs | 47 +++++++++++++++++++++++++++--- 1 file changed, 43 insertions(+), 4 deletions(-) diff --git a/xtask/src/bounded_process/error.rs b/xtask/src/bounded_process/error.rs index cc46e8e..c305b63 100644 --- a/xtask/src/bounded_process/error.rs +++ b/xtask/src/bounded_process/error.rs @@ -83,8 +83,13 @@ impl fmt::Display for ProcessError { additional, } => write!(formatter, "{primary}; additionally {additional}"), Self::Cleanup { - primary, action, .. - } => write!(formatter, "{primary}; additionally failed to {action}"), + primary, + action, + source, + } => write!( + formatter, + "{primary}; additionally failed to {action}: {source}" + ), Self::Io { program, action, .. } => write!(formatter, "cannot {action} {program} process"), @@ -117,8 +122,10 @@ impl fmt::Display for ProcessError { impl Error for ProcessError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { - Self::Additional { primary, .. } => Some(primary), - Self::Cleanup { source, .. } | Self::Io { source, .. } => Some(source), + Self::Additional { primary, .. } | Self::Cleanup { primary, .. } => { + Some(primary.as_ref()) + } + Self::Io { source, .. } => Some(source), Self::Interrupted { .. } | Self::MissingStream { .. } | Self::OutputLimit { .. } @@ -127,3 +134,35 @@ impl Error for ProcessError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cleanup_failure_retains_the_primary_source_and_reports_cleanup_context() { + let error = ProcessError::Cleanup { + primary: Box::new(ProcessError::Timeout { + program: "source-test", + duration: Duration::from_secs(1), + }), + action: "reap child process", + source: io::Error::other("cleanup refused"), + }; + + assert!(matches!( + error + .source() + .and_then(|source| source.downcast_ref::()), + Some(ProcessError::Timeout { + program: "source-test", + duration, + }) if *duration == Duration::from_secs(1) + )); + assert_eq!( + error.to_string(), + "source-test process exceeded its 1-second deadline; additionally failed to \ + reap child process: cleanup refused" + ); + } +} From 4dbc9825b602d19ea5b62739bcc7bbd0e89fe8fc Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:30:07 -0700 Subject: [PATCH 07/15] Fix: preserve subsecond timeout diagnostics --- xtask/src/bounded_process/error.rs | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/xtask/src/bounded_process/error.rs b/xtask/src/bounded_process/error.rs index c305b63..6fddea4 100644 --- a/xtask/src/bounded_process/error.rs +++ b/xtask/src/bounded_process/error.rs @@ -112,8 +112,7 @@ impl fmt::Display for ProcessError { } Self::Timeout { program, duration } => write!( formatter, - "{program} process exceeded its {}-second deadline", - duration.as_secs() + "{program} process exceeded its {duration:?} deadline" ), } } @@ -161,8 +160,21 @@ mod tests { )); assert_eq!( error.to_string(), - "source-test process exceeded its 1-second deadline; additionally failed to \ + "source-test process exceeded its 1s deadline; additionally failed to \ reap child process: cleanup refused" ); } + + #[test] + fn timeout_diagnostic_preserves_subsecond_duration() { + let error = ProcessError::Timeout { + program: "duration-test", + duration: Duration::from_millis(50), + }; + + assert_eq!( + error.to_string(), + "duration-test process exceeded its 50ms deadline" + ); + } } From c11c7df1934e9a886b0ab7b9892eef254cbde8ec Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:31:04 -0700 Subject: [PATCH 08/15] Docs: name per-step cleanup grace --- CHANGELOG.md | 4 ++-- docs/conformance/golden-file-worldline.md | 2 +- xtask/src/bounded_process/capture.rs | 6 +++--- xtask/src/bounded_process/cleanup.rs | 20 ++++++++++---------- xtask/src/bounded_process/error.rs | 6 +++--- xtask/src/bounded_process/reader.rs | 6 +++--- 6 files changed, 22 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12d091e..63d3a14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -101,8 +101,8 @@ after its public API and format compatibility policies are established. spawn and stdin transfer, stdin is streamed without a combined preimage allocation, stdout and stderr have independent limits, and every timeout or collection failure kills and reaps the process group while retaining typed - failure context. Child reaping and stalled-reader retirement use a separate - fixed cleanup deadline instead of blocking without limit. + failure context. Child reaping and stalled-reader retirement use fixed + per-step cleanup grace periods instead of blocking without limit. - Fuzz policy admission, target reconciliation, bounded campaign execution, minimization deadlines, retained-corpus admission, and workflow contract tests now run through the repository's Rust `xtask`; the superseded Python diff --git a/docs/conformance/golden-file-worldline.md b/docs/conformance/golden-file-worldline.md index 64fc85c..a0f7fb2 100644 --- a/docs/conformance/golden-file-worldline.md +++ b/docs/conformance/golden-file-worldline.md @@ -164,7 +164,7 @@ used by protocol conformance. The deadline clock starts before synchronous spawn. After spawn returns, its remaining time bounds stdin transfer and output collection; stdout and stderr have independent byte limits; and timeout or collection failure kills the complete child process group. Child reaping and -reader retirement use a separate fixed cleanup deadline and remain typed if +reader retirement use fixed per-step cleanup grace periods and remain typed if teardown cannot be proved. Digest-specific preimage construction remains owned by this checker rather than the process adapter. diff --git a/xtask/src/bounded_process/capture.rs b/xtask/src/bounded_process/capture.rs index f2ecef4..d3637f8 100644 --- a/xtask/src/bounded_process/capture.rs +++ b/xtask/src/bounded_process/capture.rs @@ -20,7 +20,7 @@ const PROCESS_POLL_INTERVAL: Duration = Duration::from_millis(10); /// Each stream is drained concurrently and retains at most one mebibyte. The /// optional deadline covers child execution and reader collection. Failures /// terminate the child's dedicated process group, retire both readers within a -/// fixed cleanup deadline, and retain primary and cleanup errors in +/// fixed per-step cleanup grace, and retain primary and cleanup errors in /// [`ProcessError`]. pub(crate) fn capture( program: &'static str, @@ -63,8 +63,8 @@ pub(crate) fn capture_with_limits( /// /// The deadline clock starts before synchronous spawn. After spawn returns, its /// remaining time bounds every nonblocking stdin write, both output readers, -/// and child execution. Failed-operation teardown uses a separate fixed -/// deadline for child reaping and reader retirement. Input slices are streamed +/// and child execution. Failed-operation teardown uses a fixed per-step cleanup +/// grace for child reaping and reader retirement. Input slices are streamed /// directly without constructing a combined preimage allocation. pub(crate) fn capture_with_input_limits( program: &'static str, diff --git a/xtask/src/bounded_process/cleanup.rs b/xtask/src/bounded_process/cleanup.rs index 6feb84f..928f032 100644 --- a/xtask/src/bounded_process/cleanup.rs +++ b/xtask/src/bounded_process/cleanup.rs @@ -8,7 +8,7 @@ use std::time::{Duration, Instant}; use super::process_group::ProcessGroup; use super::{ProcessError, ReaderWorker}; -const CLEANUP_DEADLINE: Duration = Duration::from_secs(2); +const CLEANUP_STEP_GRACE: Duration = Duration::from_secs(2); const CLEANUP_POLL_INTERVAL: Duration = Duration::from_millis(10); pub(super) fn cleanup_process(child: &mut Child, primary: ProcessError) -> ProcessError { @@ -34,7 +34,7 @@ pub(super) fn retire_readers( } pub(super) fn retire_after_cleanup(reader: ReaderWorker, primary: ProcessError) -> ProcessError { - match reader.retire(CLEANUP_DEADLINE) { + match reader.retire(CLEANUP_STEP_GRACE) { Ok(()) => primary, Err(additional) => ProcessError::Additional { primary: Box::new(primary), @@ -44,27 +44,27 @@ pub(super) fn retire_after_cleanup(reader: ReaderWorker, primary: ProcessError) } fn reap_child(child: &mut Child) -> Result<(), io::Error> { - wait_until_reaped(CLEANUP_DEADLINE, || { + wait_until_reaped(CLEANUP_STEP_GRACE, || { child.try_wait().map(|status| status.is_some()) }) } fn wait_until_reaped( - deadline: Duration, + grace: Duration, mut poll: impl FnMut() -> Result, ) -> Result<(), io::Error> { let expires = Instant::now() - .checked_add(deadline) - .ok_or_else(|| cleanup_timeout(deadline))?; + .checked_add(grace) + .ok_or_else(|| cleanup_timeout(grace))?; loop { if poll()? { return Ok(()); } let remaining = expires .checked_duration_since(Instant::now()) - .ok_or_else(|| cleanup_timeout(deadline))?; + .ok_or_else(|| cleanup_timeout(grace))?; if remaining.is_zero() { - return Err(cleanup_timeout(deadline)); + return Err(cleanup_timeout(grace)); } thread::sleep(CLEANUP_POLL_INTERVAL.min(remaining)); } @@ -85,10 +85,10 @@ fn with_cleanup( } } -fn cleanup_timeout(deadline: Duration) -> io::Error { +fn cleanup_timeout(grace: Duration) -> io::Error { io::Error::new( io::ErrorKind::TimedOut, - format!("child was not reaped within {deadline:?}"), + format!("child was not reaped within the {grace:?} cleanup grace"), ) } diff --git a/xtask/src/bounded_process/error.rs b/xtask/src/bounded_process/error.rs index 6fddea4..f7c9724 100644 --- a/xtask/src/bounded_process/error.rs +++ b/xtask/src/bounded_process/error.rs @@ -5,14 +5,14 @@ use std::fmt; use std::io; use std::time::Duration; -/// A typed failure from synchronous, bounded child-process execution. +/// A typed failure from deadline-accounted child-process execution. pub(crate) enum ProcessError { /// Reader or cleanup collection found another failure after the primary one. Additional { primary: Box, additional: Box, }, - /// Process-group termination or child reaping failed after a primary error. + /// Process-group termination, child killing, or bounded reaping failed. Cleanup { primary: Box, action: &'static str, @@ -45,7 +45,7 @@ pub(crate) enum ProcessError { program: &'static str, stream: &'static str, }, - /// The complete child operation exceeded its admitted duration. + /// The admitted child-operation deadline elapsed. Timeout { program: &'static str, duration: Duration, diff --git a/xtask/src/bounded_process/reader.rs b/xtask/src/bounded_process/reader.rs index 9f9f461..b2edcd2 100644 --- a/xtask/src/bounded_process/reader.rs +++ b/xtask/src/bounded_process/reader.rs @@ -89,13 +89,13 @@ impl ReaderWorker { /// Retires one reader without an unbounded thread join. /// /// A result already received proves the worker crossed its only fallible - /// read boundary. Otherwise retirement waits only for `deadline`; a stalled + /// read boundary. Otherwise retirement waits only for `grace`; a stalled /// reader becomes a typed timeout and its thread handle is detached. - pub(super) fn retire(self, deadline: Duration) -> Result<(), ProcessError> { + pub(super) fn retire(self, grace: Duration) -> Result<(), ProcessError> { let result = if self.completed { Ok(()) } else { - match self.receiver.recv_timeout(deadline) { + match self.receiver.recv_timeout(grace) { Ok(Ok(_output)) => Ok(()), Ok(Err(source)) => Err(ProcessError::Io { program: self.program, From 8b79ffb4aa57b8ce6c9cbf150a21c08eb402c630 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:48:34 -0700 Subject: [PATCH 09/15] Docs: record bounded reader retirement --- ...007-terminal-signal-process-group-guard.md | 9 +- ...0008-deadline-bounded-reader-retirement.md | 89 +++++++++++++++++++ docs/adr/README.md | 1 + xtask/tests/process_policy_contract.rs | 17 ++++ 4 files changed, 114 insertions(+), 2 deletions(-) create mode 100644 docs/adr/0008-deadline-bounded-reader-retirement.md diff --git a/docs/adr/0007-terminal-signal-process-group-guard.md b/docs/adr/0007-terminal-signal-process-group-guard.md index 1b684ad..46d59f7 100644 --- a/docs/adr/0007-terminal-signal-process-group-guard.md +++ b/docs/adr/0007-terminal-signal-process-group-guard.md @@ -36,8 +36,8 @@ per-stream capture limits. An observed terminal signal becomes a typed `ProcessError::Interrupted` refusal. The normal failure path then sends `SIGKILL` to the dedicated child -process group, kills and reaps the child, and joins captured-output readers -before returning. Captured processes collect and join both readers before the +process group, kills and reaps the child, and retires captured-output readers +before returning. Captured processes collect both reader results before the successful child wait consumes the process-group leader's waitable identity. Every reader failure that can invoke group cleanup therefore occurs while that identity is still owned, and no cleanup path addresses a numeric process-group @@ -45,6 +45,11 @@ ID after the child has been reaped. Output-limit and late-interrupt refusals found after a successful wait require no cleanup because the child and readers have already terminated. +[ADR-0008](0008-deadline-bounded-reader-retirement.md) governs failed-operation +reader retirement. It permits deadline-bounded retirement and explicit +detachment when a reader cannot prove completion after process cleanup; it does +not permit an unbounded join. + The terminal signal is not sent directly to the child group; one cleanup authority avoids races between signal delivery and mandatory process-group termination. diff --git a/docs/adr/0008-deadline-bounded-reader-retirement.md b/docs/adr/0008-deadline-bounded-reader-retirement.md new file mode 100644 index 0000000..326a93a --- /dev/null +++ b/docs/adr/0008-deadline-bounded-reader-retirement.md @@ -0,0 +1,89 @@ +# ADR-0008: Deadline-Bounded Reader Retirement + +- Status: Accepted +- Date: 2026-07-28 +- Owners: Keep repository verification +- Related issue: #59 +- Depends on: ADR-0007 + +## Context + +Repository verification tasks run external tools synchronously and drain +standard output and standard error on dedicated reader threads. Concurrent +draining prevents a child from blocking when either pipe fills, and per-stream +capture limits prevent unbounded retained output. + +A failed child does not guarantee that a reader thread can be joined. A +descendant may retain an inherited pipe, an injected reader may stop making +progress, or an operating-system read may remain blocked after the primary +operation has failed. An unbounded join would turn a typed timeout, +interruption, or collection refusal into an unbounded caller hang. + +Stable Rust does not provide a safe operation that cancels an arbitrary blocked +thread. Keep forbids unsafe code, and asynchronous process I/O would introduce a +runtime and a second process-lifecycle model without a demonstrated consumer +need. + +## Decision + +Failed captured-process operations perform cleanup in this order: + +1. Drop the parent's child-standard-input handle. +2. Terminate the child's dedicated process group. +3. Kill the direct child as an idempotent fallback. +4. Poll direct-child reaping for one fixed per-step cleanup grace. +5. Give each reader worker one fixed per-step cleanup grace to publish its + bounded result. + +Reader retirement is successful when the caller already received the worker's +only result or the result arrives within the cleanup grace. If a worker remains +blocked, its join handle is detached. This bounded retirement is permitted only +on an already-failed operation; it can never convert a failure into success. + +The returned error retains the primary failure as its source. A reader +retirement timeout is attached as an additional typed failure, so bounded +retirement does not hide either the cause of the operation failure or the +incomplete cleanup observation. + +Process-group termination and direct-child reaping precede reader retirement. +For real child pipes, terminating every process that could retain the pipe is +the operation that makes reader completion reachable. The retirement grace is +not a durability claim and does not prove that an arbitrary injected reader has +stopped; it bounds how long repository verification waits for that evidence. + +## Alternatives considered + +- Joining every reader without a deadline was rejected because one inherited or + stalled pipe could hang repository verification forever. +- Detaching every reader immediately was rejected because it would discard + available read failures and make ordinary cleanup nondeterministic. +- Cancelling blocked threads through platform-specific or unsafe APIs was + rejected because Rust cannot make arbitrary cancellation memory-safe and + Keep forbids unsafe code. +- Moving subprocess capture to an asynchronous runtime was rejected because the + synchronous core already has explicit deadlines and no consumer requires an + async process boundary. +- Reading the streams sequentially was rejected because either child pipe can + fill while the other stream is being drained. + +## Consequences + +Every failed captured-process operation returns after bounded cleanup steps. +The caller receives the primary typed failure plus any observed cleanup +failure. Successful operations still require both bounded reader results and +the direct child's exit status. + +A detached reader thread may remain alive until its underlying read completes +or the process exits. It owns only its reader, bounded output accumulator, and +single-result sender. It owns no Keep data, repository lock, mutation +authority, or caller reference. This residual lifetime is explicit uncertainty, +not a claim that cleanup completed. + +The two-second cleanup grace is a per-step bound. It is separate from the +operation deadline and may be consumed once for child reaping and once for each +reader. Tests use zero-duration injected retirement and reap boundaries; they +do not classify behavior from scheduler timing. + +This decision changes no Keep library API, content identity, durable format, or +recovery protocol. It governs only private repository-process orchestration in +`xtask`. diff --git a/docs/adr/README.md b/docs/adr/README.md index 0521dfb..cd77ea1 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -59,3 +59,4 @@ encryption, concurrency, or public-API surface it governs. - [ADR-0005: Durable segment store protocol](0005-durable-segment-store-protocol.md) - [ADR-0006: Descriptor-bound child working directory](0006-descriptor-bound-child-working-directory.md) - [ADR-0007: Terminal signal process-group guard](0007-terminal-signal-process-group-guard.md) +- [ADR-0008: Deadline-bounded reader retirement](0008-deadline-bounded-reader-retirement.md) diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index 517e162..17f9d19 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -19,6 +19,8 @@ const GIT_INVENTORY_ERROR: &str = include_str!("../src/git_inventory/error.rs"); const GIT_PATH_STREAM: &str = include_str!("../src/git_inventory/path_stream.rs"); const GIT_PROCESS: &str = include_str!("../src/git_inventory/process.rs"); const SOURCE_PURE_RUST_TESTS: &str = include_str!("../src/source_structure/pure_rust_tests.rs"); +const READER_RETIREMENT_ADR: &str = + include_str!("../../docs/adr/0008-deadline-bounded-reader-retirement.md"); #[test] fn sanitized_git_fixture_has_one_process_authority() { @@ -79,6 +81,21 @@ fn captured_process_keeps_the_group_leader_until_reader_collection_finishes() Ok(()) } +#[test] +fn reader_detachment_has_a_concurrency_decision_record() { + for decision in [ + "bounded retirement", + "detached", + "primary failure", + "unbounded join", + ] { + assert!( + READER_RETIREMENT_ADR.contains(decision), + "reader-retirement ADR omits `{decision}`" + ); + } +} + #[test] fn descendant_cleanup_uses_disconnect_evidence_instead_of_elapsed_time() { assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("descendant_survived_cleanup")); From 4439dd0ad9fb151f56c225a7f5a1bd544acb9270 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:49:48 -0700 Subject: [PATCH 10/15] Fix: remove digest fixture clock escape --- xtask/src/external_digest/tests.rs | 10 +++------- xtask/tests/process_policy_contract.rs | 8 ++++++++ 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/xtask/src/external_digest/tests.rs b/xtask/src/external_digest/tests.rs index d28995e..0f994ea 100644 --- a/xtask/src/external_digest/tests.rs +++ b/xtask/src/external_digest/tests.rs @@ -2,7 +2,6 @@ use std::env; use std::process::Command; -use std::sync::mpsc::{self, RecvTimeoutError}; use std::time::Duration; use crate::bounded_process::{ProcessError, ProcessOutput}; @@ -42,12 +41,9 @@ fn digest_child_does_not_read_stdin() { if env::var_os(BLOCKING_CHILD).is_none() { return; } - let (sender, receiver) = mpsc::channel::<()>(); - assert!(matches!( - receiver.recv_timeout(Duration::from_secs(1)), - Err(RecvTimeoutError::Timeout) - )); - drop(sender); + loop { + std::thread::park(); + } } #[test] diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index 17f9d19..ea4ef36 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -13,6 +13,7 @@ const BOUNDED_PROCESS_READER: &str = include_str!("../src/bounded_process/reader const BOUNDED_PROCESS_TESTS: &str = include_str!("../src/bounded_process/tests.rs"); const CONFORMANCE_B3SUM: &str = include_str!("../src/protocol_conformance/external_digest.rs"); const EXTERNAL_DIGEST: &str = include_str!("../src/external_digest.rs"); +const EXTERNAL_DIGEST_TESTS: &str = include_str!("../src/external_digest/tests.rs"); const GOLDEN_B3SUM: &str = include_str!("../src/golden_file_worldline/b3sum_oracle.rs"); const REPOSITORY_FIXTURE: &str = include_str!("../src/repository_fixture.rs"); const GIT_INVENTORY_ERROR: &str = include_str!("../src/git_inventory/error.rs"); @@ -96,6 +97,13 @@ fn reader_detachment_has_a_concurrency_decision_record() { } } +#[test] +fn blocking_digest_fixture_has_no_wall_clock_escape() { + assert!(!EXTERNAL_DIGEST_TESTS.contains("recv_timeout(")); + assert!(!EXTERNAL_DIGEST_TESTS.contains("Duration::from_secs(1)")); + assert!(EXTERNAL_DIGEST_TESTS.contains("std::thread::park()")); +} + #[test] fn descendant_cleanup_uses_disconnect_evidence_instead_of_elapsed_time() { assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("descendant_survived_cleanup")); From 6d5cd18f0ca3f5bde3779a1d494d49216ceabedd Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:50:48 -0700 Subject: [PATCH 11/15] Fix: preserve typed Golden digest errors --- xtask/src/golden_file_worldline/error.rs | 29 +++++++++++++++++++----- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/xtask/src/golden_file_worldline/error.rs b/xtask/src/golden_file_worldline/error.rs index 1de9076..b98169d 100644 --- a/xtask/src/golden_file_worldline/error.rs +++ b/xtask/src/golden_file_worldline/error.rs @@ -8,11 +8,12 @@ use std::path::PathBuf; use std::string::FromUtf8Error; use crate::diagnostic::{escaped_controls, escaped_path}; +use crate::external_digest::ExternalDigestError; use xtask::protocol_admission::RelativePathError; pub(crate) enum GoldenError { ExternalDigest { - source: Box, + source: ExternalDigestError, }, Integer { field: String, @@ -35,10 +36,8 @@ pub(crate) enum GoldenError { } impl GoldenError { - pub(super) fn external_digest(source: impl Error + Send + Sync + 'static) -> Self { - Self::ExternalDigest { - source: Box::new(source), - } + pub(super) const fn external_digest(source: ExternalDigestError) -> Self { + Self::ExternalDigest { source } } pub(super) fn io(action: &'static str, path: impl Into, source: io::Error) -> Self { @@ -91,7 +90,7 @@ impl fmt::Display for GoldenError { impl Error for GoldenError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { - Self::ExternalDigest { source } => Some(source.as_ref()), + Self::ExternalDigest { source } => Some(source), Self::Integer { source, .. } => Some(source), Self::Io { source, .. } => Some(source), Self::Utf8 { source, .. } => Some(source), @@ -100,3 +99,21 @@ impl Error for GoldenError { } } } + +#[cfg(test)] +mod tests { + use super::GoldenError; + use crate::external_digest::ExternalDigestError; + + #[test] + fn external_digest_refusal_retains_its_typed_variant() { + let error = GoldenError::external_digest(ExternalDigestError::Width { observed: 31 }); + + assert!(matches!( + error, + GoldenError::ExternalDigest { + source: ExternalDigestError::Width { observed: 31 }, + } + )); + } +} From 716be073f8896dffc0bf142084bf34d5424f954e Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:51:53 -0700 Subject: [PATCH 12/15] Docs: centralize digest process policy --- docs/dependencies/blake3-1.8.5.md | 11 +++++------ xtask/tests/process_policy_contract.rs | 15 +++++++++++++++ 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/docs/dependencies/blake3-1.8.5.md b/docs/dependencies/blake3-1.8.5.md index d455c4b..1866fda 100644 --- a/docs/dependencies/blake3-1.8.5.md +++ b/docs/dependencies/blake3-1.8.5.md @@ -24,12 +24,11 @@ semantics, and committed witness bytes. For every identity and content mutation, the repository checker also streams the canonical preimage through external `b3sum`; a mismatch with the in-process result is a refusal. Golden File Worldline and protocol conformance share one deadline-bounded process adapter, -but retain separate preimage construction. The adapter streams caller-owned -slices without a combined allocation, bounds stdout and stderr independently, -and kills and reaps the child process group on timeout or collection failure. -The checked-in vectors and runtime cross-check therefore cover the algorithm -boundary without claiming that the Rust path independently implements the -BLAKE3 compression function. +but retain separate preimage construction. The +[Golden File Worldline reference model](../conformance/golden-file-worldline.md#reference-model) +records the process-adapter contract. The checked-in vectors and runtime +cross-check therefore cover the algorithm boundary without claiming that the +Rust path independently implements the BLAKE3 compression function. The manifest disables default features and enables exactly: diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index ea4ef36..a26db31 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -11,6 +11,7 @@ const BOUNDED_PROCESS_GROUP_TESTS: &str = include_str!("../src/bounded_process/process_group/tests.rs"); const BOUNDED_PROCESS_READER: &str = include_str!("../src/bounded_process/reader.rs"); const BOUNDED_PROCESS_TESTS: &str = include_str!("../src/bounded_process/tests.rs"); +const BLAKE3_ADMISSION: &str = include_str!("../../docs/dependencies/blake3-1.8.5.md"); const CONFORMANCE_B3SUM: &str = include_str!("../src/protocol_conformance/external_digest.rs"); const EXTERNAL_DIGEST: &str = include_str!("../src/external_digest.rs"); const EXTERNAL_DIGEST_TESTS: &str = include_str!("../src/external_digest/tests.rs"); @@ -104,6 +105,20 @@ fn blocking_digest_fixture_has_no_wall_clock_escape() { assert!(EXTERNAL_DIGEST_TESTS.contains("std::thread::park()")); } +#[test] +fn digest_admission_defers_process_mechanics_to_the_process_contract() { + assert!(BLAKE3_ADMISSION.contains( + "[Golden File Worldline reference model](../conformance/golden-file-worldline.md#reference-model)" + )); + for duplicated_mechanic in [ + "slices without a combined allocation", + "bounds stdout and stderr independently", + "kills and reaps the child process group", + ] { + assert!(!BLAKE3_ADMISSION.contains(duplicated_mechanic)); + } +} + #[test] fn descendant_cleanup_uses_disconnect_evidence_instead_of_elapsed_time() { assert!(!BOUNDED_PROCESS_GROUP_TESTS.contains("descendant_survived_cleanup")); From 01f43d1214db0df0d503c5ffbb784c56242fb8fb Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:52:53 -0700 Subject: [PATCH 13/15] Refactor: centralize external digest width --- xtask/src/external_digest.rs | 3 ++- xtask/src/protocol_conformance/external_digest.rs | 4 +--- xtask/tests/process_policy_contract.rs | 2 ++ 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/xtask/src/external_digest.rs b/xtask/src/external_digest.rs index 85b394a..359b699 100644 --- a/xtask/src/external_digest.rs +++ b/xtask/src/external_digest.rs @@ -14,7 +14,8 @@ use crate::diagnostic::escaped_controls; const B3SUM: &str = "b3sum"; const DIAGNOSTIC_LIMIT_BYTES: usize = 65_536; -const DIGEST_BYTES: usize = 32; +/// Width of an admitted raw BLAKE3 digest. +pub(crate) const DIGEST_BYTES: usize = 32; const PROCESS_DEADLINE: Duration = Duration::from_secs(10); const PROCESS_LIMITS: CaptureLimits = CaptureLimits::new(DIGEST_BYTES, DIAGNOSTIC_LIMIT_BYTES); diff --git a/xtask/src/protocol_conformance/external_digest.rs b/xtask/src/protocol_conformance/external_digest.rs index e736757..5116413 100644 --- a/xtask/src/protocol_conformance/external_digest.rs +++ b/xtask/src/protocol_conformance/external_digest.rs @@ -1,11 +1,9 @@ //! This module owns the conformance port to the shared external digest witness. -use crate::external_digest; +use crate::external_digest::{self, DIGEST_BYTES}; use super::ConformanceError; -const DIGEST_BYTES: usize = 32; - pub(super) fn digest(parts: &[&[u8]]) -> Result<[u8; DIGEST_BYTES], ConformanceError> { external_digest::b3sum(parts).map_err(ConformanceError::external_digest) } diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index a26db31..56f657c 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -38,6 +38,8 @@ fn external_digest_has_one_process_authority() { + GOLDEN_B3SUM.matches("Command::new(").count(); assert_eq!(process_authorities, 1); assert!(EXTERNAL_DIGEST.contains("capture_with_input_limits(")); + assert!(EXTERNAL_DIGEST.contains("pub(crate) const DIGEST_BYTES")); + assert!(!CONFORMANCE_B3SUM.contains("const DIGEST_BYTES")); assert!(!CONFORMANCE_B3SUM.contains("Command::new(")); assert!(!GOLDEN_B3SUM.contains("Command::new(")); assert!(!GOLDEN_B3SUM.contains(".wait()")); From affd8bc442c7e7128f44ee30e4ef6a4fd66fe8f6 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:53:59 -0700 Subject: [PATCH 14/15] Test: enforce digest caller delegation --- xtask/tests/process_policy_contract.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/xtask/tests/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs index 56f657c..f99edbc 100644 --- a/xtask/tests/process_policy_contract.rs +++ b/xtask/tests/process_policy_contract.rs @@ -40,10 +40,11 @@ fn external_digest_has_one_process_authority() { assert!(EXTERNAL_DIGEST.contains("capture_with_input_limits(")); assert!(EXTERNAL_DIGEST.contains("pub(crate) const DIGEST_BYTES")); assert!(!CONFORMANCE_B3SUM.contains("const DIGEST_BYTES")); - assert!(!CONFORMANCE_B3SUM.contains("Command::new(")); - assert!(!GOLDEN_B3SUM.contains("Command::new(")); - assert!(!GOLDEN_B3SUM.contains(".wait()")); - assert!(!GOLDEN_B3SUM.contains(".write_all(")); + for caller in [CONFORMANCE_B3SUM, GOLDEN_B3SUM] { + assert!(!caller.contains("Command::new(")); + assert!(!caller.contains(".wait()")); + assert!(!caller.contains(".write_all(")); + } } #[test] From 46fb8d3ed8d1c37a3732f26494d87e682ef6078f Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 28 Jul 2026 22:56:56 -0700 Subject: [PATCH 15/15] Fix: preserve fuzz facade isolation --- .../src/golden_file_worldline/b3sum_oracle.rs | 14 ++++++++ xtask/src/golden_file_worldline/error.rs | 33 ++++--------------- xtask/src/golden_file_worldline/mod.rs | 2 +- xtask/src/golden_protocol_fuzz.rs | 2 +- 4 files changed, 23 insertions(+), 28 deletions(-) diff --git a/xtask/src/golden_file_worldline/b3sum_oracle.rs b/xtask/src/golden_file_worldline/b3sum_oracle.rs index 218ad9d..36ae83d 100644 --- a/xtask/src/golden_file_worldline/b3sum_oracle.rs +++ b/xtask/src/golden_file_worldline/b3sum_oracle.rs @@ -33,8 +33,22 @@ impl IdentityDigestOracle for B3sumOracle { #[cfg(test)] mod tests { use super::{B3sumOracle, IdentityDigestOracle}; + use crate::external_digest::ExternalDigestError; + use crate::golden_file_worldline::GoldenError; use crate::golden_file_worldline::identity_oracle::digest; + #[test] + fn external_digest_refusal_retains_its_typed_variant() { + let error = GoldenError::external_digest(ExternalDigestError::Width { observed: 31 }); + + assert!(matches!( + error, + GoldenError::ExternalDigest { + source: ExternalDigestError::Width { observed: 31 }, + } + )); + } + #[test] fn external_oracle_agrees_on_the_identity_preimage() { let payload = b"independent digest boundary"; diff --git a/xtask/src/golden_file_worldline/error.rs b/xtask/src/golden_file_worldline/error.rs index b98169d..76066d4 100644 --- a/xtask/src/golden_file_worldline/error.rs +++ b/xtask/src/golden_file_worldline/error.rs @@ -8,12 +8,11 @@ use std::path::PathBuf; use std::string::FromUtf8Error; use crate::diagnostic::{escaped_controls, escaped_path}; -use crate::external_digest::ExternalDigestError; use xtask::protocol_admission::RelativePathError; -pub(crate) enum GoldenError { +pub(crate) enum GoldenError { ExternalDigest { - source: ExternalDigestError, + source: ExternalDigestSource, }, Integer { field: String, @@ -35,8 +34,8 @@ pub(crate) enum GoldenError { Violation(String), } -impl GoldenError { - pub(super) const fn external_digest(source: ExternalDigestError) -> Self { +impl GoldenError { + pub(super) const fn external_digest(source: ExternalDigestSource) -> Self { Self::ExternalDigest { source } } @@ -53,13 +52,13 @@ impl GoldenError { } } -impl fmt::Debug for GoldenError { +impl fmt::Debug for GoldenError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Display::fmt(self, formatter) } } -impl fmt::Display for GoldenError { +impl fmt::Display for GoldenError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str("golden corpus check failed: ")?; match self { @@ -87,7 +86,7 @@ impl fmt::Display for GoldenError { } } -impl Error for GoldenError { +impl Error for GoldenError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { Self::ExternalDigest { source } => Some(source), @@ -99,21 +98,3 @@ impl Error for GoldenError { } } } - -#[cfg(test)] -mod tests { - use super::GoldenError; - use crate::external_digest::ExternalDigestError; - - #[test] - fn external_digest_refusal_retains_its_typed_variant() { - let error = GoldenError::external_digest(ExternalDigestError::Width { observed: 31 }); - - assert!(matches!( - error, - GoldenError::ExternalDigest { - source: ExternalDigestError::Width { observed: 31 }, - } - )); - } -} diff --git a/xtask/src/golden_file_worldline/mod.rs b/xtask/src/golden_file_worldline/mod.rs index f598177..d2bfaee 100644 --- a/xtask/src/golden_file_worldline/mod.rs +++ b/xtask/src/golden_file_worldline/mod.rs @@ -17,7 +17,7 @@ use std::path::Path; use b3sum_oracle::B3sumOracle; use corpus_protocol::Corpus; -pub(crate) use error::GoldenError; +pub(crate) type GoldenError = error::GoldenError; pub(super) fn check(repository_root: &Path) -> Result<(), GoldenError> { let corpus = Corpus::open(repository_root.join("conformance/golden-file-worldline/v1"))?; diff --git a/xtask/src/golden_protocol_fuzz.rs b/xtask/src/golden_protocol_fuzz.rs index 164eb3c..3f5633b 100644 --- a/xtask/src/golden_protocol_fuzz.rs +++ b/xtask/src/golden_protocol_fuzz.rs @@ -25,7 +25,7 @@ mod mutation_value; #[cfg(feature = "repository-tasks")] use corpus_protocol::Corpus; -pub(super) use error::GoldenError; +pub(super) type GoldenError = error::GoldenError; pub(super) fn admit(selector: u8, input: &[u8]) -> Result<(), GoldenError> { fuzz_admission::admit(selector, input)