diff --git a/CHANGELOG.md b/CHANGELOG.md index e9029eb..63d3a14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,6 +96,13 @@ 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. 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/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/docs/conformance/golden-file-worldline.md b/docs/conformance/golden-file-worldline.md index 0058b29..a0f7fb2 100644 --- a/docs/conformance/golden-file-worldline.md +++ b/docs/conformance/golden-file-worldline.md @@ -158,7 +158,15 @@ 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 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 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. ## Exact M1 acceptance boundary diff --git a/docs/dependencies/blake3-1.8.5.md b/docs/dependencies/blake3-1.8.5.md index a5ca41e..1866fda 100644 --- a/docs/dependencies/blake3-1.8.5.md +++ b/docs/dependencies/blake3-1.8.5.md @@ -22,10 +22,13 @@ 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 +[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/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..d3637f8 100644 --- a/xtask/src/bounded_process/capture.rs +++ b/xtask/src/bounded_process/capture.rs @@ -5,7 +5,8 @@ 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, }; @@ -18,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 per-step cleanup grace, and retain primary and cleanup errors in +/// [`ProcessError`]. pub(crate) fn capture( program: &'static str, command: &mut Command, @@ -44,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, @@ -57,6 +59,27 @@ 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 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 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, + 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, @@ -104,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 { @@ -129,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) @@ -151,10 +169,30 @@ 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); - 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..928f032 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_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 { + 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_STEP_GRACE) { 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_STEP_GRACE, || { + child.try_wait().map(|status| status.is_some()) + }) +} + +fn wait_until_reaped( + grace: Duration, + mut poll: impl FnMut() -> Result, +) -> Result<(), io::Error> { + let expires = Instant::now() + .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(grace))?; + if remaining.is_zero() { + return Err(cleanup_timeout(grace)); + } + 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(grace: Duration) -> io::Error { + io::Error::new( + io::ErrorKind::TimedOut, + format!("child was not reaped within the {grace:?} cleanup grace"), + ) +} + +#[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/error.rs b/xtask/src/bounded_process/error.rs index cc46e8e..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, @@ -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"), @@ -107,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" ), } } @@ -117,8 +121,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 +133,48 @@ 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 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" + ); + } +} diff --git a/xtask/src/bounded_process/input.rs b/xtask/src/bounded_process/input.rs new file mode 100644 index 0000000..37123ae --- /dev/null +++ b/xtask/src/bounded_process/input.rs @@ -0,0 +1,155 @@ +//! 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, + writer: &mut impl Write, + part: &[u8], + deadline: &ProcessDeadline, + interrupts: &InterruptGuard, +) -> Result<(), ProcessError> { + let mut remaining = part; + while !remaining.is_empty() { + observe_input_boundary(program, deadline, interrupts)?; + match writer.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 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, + 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, + } +} + +#[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(()) + } +} diff --git a/xtask/src/bounded_process/reader.rs b/xtask/src/bounded_process/reader.rs index c1f423d..b2edcd2 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 `grace`; a stalled + /// reader becomes a typed timeout and its thread handle is detached. + pub(super) fn retire(self, grace: Duration) -> Result<(), ProcessError> { + let result = if self.completed { + Ok(()) + } else { + match self.receiver.recv_timeout(grace) { + 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/src/external_digest.rs b/xtask/src/external_digest.rs new file mode 100644 index 0000000..359b699 --- /dev/null +++ b/xtask/src/external_digest.rs @@ -0,0 +1,165 @@ +//! 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; +/// 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); + +/// 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. 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) +} + +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..0f994ea --- /dev/null +++ b/xtask/src/external_digest/tests.rs @@ -0,0 +1,100 @@ +//! This module owns bounded external digest regression evidence. + +use std::env; +use std::process::Command; +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; + } + loop { + std::thread::park(); + } +} + +#[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..36ae83d 100644 --- a/xtask/src/golden_file_worldline/b3sum_oracle.rs +++ b/xtask/src/golden_file_worldline/b3sum_oracle.rs @@ -1,234 +1,54 @@ -//! 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 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))?; + 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) } - 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)] 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 9f418e9..76066d4 100644 --- a/xtask/src/golden_file_worldline/error.rs +++ b/xtask/src/golden_file_worldline/error.rs @@ -10,7 +10,10 @@ use std::string::FromUtf8Error; use crate::diagnostic::{escaped_controls, escaped_path}; use xtask::protocol_admission::RelativePathError; -pub(crate) enum GoldenError { +pub(crate) enum GoldenError { + ExternalDigest { + source: ExternalDigestSource, + }, 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, @@ -46,7 +34,11 @@ pub(crate) enum GoldenError { Violation(String), } -impl GoldenError { +impl GoldenError { + pub(super) const fn external_digest(source: ExternalDigestSource) -> Self { + Self::ExternalDigest { source } + } + pub(super) fn io(action: &'static str, path: impl Into, source: io::Error) -> Self { Self::Io { action, @@ -60,16 +52,17 @@ 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 { + Self::ExternalDigest { source } => fmt::Display::fmt(source, formatter), Self::Integer { field, .. } => { formatter.write_str("cannot parse canonical ")?; escaped_controls(formatter, field) @@ -83,28 +76,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)?; @@ -115,18 +86,15 @@ 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), 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/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) 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..5116413 100644 --- a/xtask/src/protocol_conformance/external_digest.rs +++ b/xtask/src/protocol_conformance/external_digest.rs @@ -1,273 +1,9 @@ -//! 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::{self, DIGEST_BYTES}; 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/process_policy_contract.rs b/xtask/tests/process_policy_contract.rs new file mode 100644 index 0000000..f99edbc --- /dev/null +++ b/xtask/tests/process_policy_contract.rs @@ -0,0 +1,297 @@ +//! 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 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"); +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"); +const READER_RETIREMENT_ADR: &str = + include_str!("../../docs/adr/0008-deadline-bounded-reader-retirement.md"); + +#[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!(EXTERNAL_DIGEST.contains("pub(crate) const DIGEST_BYTES")); + assert!(!CONFORMANCE_B3SUM.contains("const DIGEST_BYTES")); + for caller in [CONFORMANCE_B3SUM, GOLDEN_B3SUM] { + assert!(!caller.contains("Command::new(")); + assert!(!caller.contains(".wait()")); + assert!(!caller.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"] { + 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" + ); + assert!( + !body.contains(".join()"), + "reader collection may wait forever" + ); + 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 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 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")); + 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 retire(", + ], + ) +} + +#[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 f1d9efb..34ab456 100644 --- a/xtask/tests/source_policy_contract.rs +++ b/xtask/tests/source_policy_contract.rs @@ -1,22 +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_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 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"); @@ -65,201 +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 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 repository_process_boundaries_document_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_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(", - ], - )?; - 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(", - ], - )?; - 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(()) -}