use std::io::{ErrorKind, Read, Write};
use std::path::Path;
use std::process::{Command, ExitStatus, Stdio};
use spec_spine_core::verify;
use spec_spine_types::{
Error, Severity, Verdict, VerifyFailure, VerifyOutcome, VerifyReport, Violation, verdict::verb,
};
use crate::load_repo_config;
use crate::out;
const STACK_VAR: &str = "SPEC_SPINE_VERIFY_STACK";
const RE_ENTRY_CODE: &str = "R-001";
fn transcript(json: bool, args: std::fmt::Arguments<'_>) {
if json {
diagnostic(args);
} else {
out::line(args);
}
}
fn diagnostic(args: std::fmt::Arguments<'_>) {
let stderr = std::io::stderr();
let mut handle = stderr.lock();
let _ = writeln!(handle, "{args}");
}
const DRAIN_BUF: usize = 16 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Drained {
Delivered,
Discarded,
InputFailed,
}
fn drain_to_stderr<R: Read>(src: &mut R) -> Drained {
drain_lines(src, |bytes| {
let stderr = std::io::stderr();
let mut handle = stderr.lock();
handle.write_all(bytes)
})
}
const LINE_HOLD: usize = DRAIN_BUF;
fn drain_lines<R: Read>(
src: &mut R,
mut deliver: impl FnMut(&[u8]) -> std::io::Result<()>,
) -> Drained {
let mut buf = [0u8; DRAIN_BUF];
let mut held: Vec<u8> = Vec::with_capacity(LINE_HOLD + DRAIN_BUF);
let mut outcome = Drained::Delivered;
let mut send = |bytes: &[u8], outcome: &mut Drained| {
if *outcome == Drained::Delivered && !bytes.is_empty() && deliver(bytes).is_err() {
*outcome = Drained::Discarded;
}
};
loop {
let n = match src.read(&mut buf) {
Ok(0) => {
send(&held, &mut outcome);
return outcome;
}
Ok(n) => n,
Err(e) if e.kind() == ErrorKind::Interrupted => continue,
Err(_) => {
send(&held, &mut outcome);
return Drained::InputFailed;
}
};
let chunk = &buf[..n];
match chunk.iter().rposition(|&b| b == b'\n') {
Some(last) => {
if held.is_empty() {
send(&chunk[..=last], &mut outcome);
} else {
held.extend_from_slice(&chunk[..=last]);
send(&held, &mut outcome);
held.clear();
}
held.extend_from_slice(&chunk[last + 1..]);
}
None => held.extend_from_slice(chunk),
}
if held.len() >= LINE_HOLD {
send(&held, &mut outcome);
held.clear();
}
}
}
fn note_drain(stream: &str, command: &str, drained: &std::thread::Result<Drained>) {
let what = match drained {
Ok(Drained::Delivered | Drained::Discarded) => return,
Ok(Drained::InputFailed) => "could not be read to end",
Err(_) => "forwarding panicked",
};
diagnostic(format_args!(
"[verify] warning: the child's {stream} {what} while running `{command}`; its output is incomplete and the verdict is unaffected"
));
}
fn run_one(repo: &Path, command: &str, child_stack: &str, json: bool) -> Result<ExitStatus, Error> {
let mut cmd = Command::new("sh");
cmd.arg("-c")
.arg(command)
.current_dir(repo)
.env(STACK_VAR, child_stack);
if !json {
return cmd
.status()
.map_err(|e| Error::Io(format!("cannot run `{command}`: {e}")));
}
let mut child = cmd
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(|e| Error::Io(format!("cannot run `{command}`: {e}")))?;
let mut child_out = child.stdout.take().expect("stdout was piped");
let pump = std::thread::spawn(move || drain_to_stderr(&mut child_out));
let mut child_err = child.stderr.take().expect("stderr was piped");
let err_drained = Ok(drain_to_stderr(&mut child_err));
let waited = child.wait();
let out_drained = pump.join();
note_drain("stdout", command, &out_drained);
note_drain("stderr", command, &err_drained);
let status = waited.map_err(|e| Error::Io(format!("cannot wait for `{command}`: {e}")))?;
Ok(status)
}
pub fn run(repo: &Path, id: &str, json: bool, plan_only: bool) -> Result<u8, Error> {
let cfg = load_repo_config(repo)?;
let plan = verify::plan(&cfg, repo, id)?;
let mut stack: Vec<String> = std::env::var(STACK_VAR)
.ok()
.map(|s| {
s.split(',')
.filter(|p| !p.is_empty())
.map(str::to_string)
.collect()
})
.unwrap_or_default();
if stack.contains(&plan.spec_id) {
stack.push(plan.spec_id.clone());
return Err(Error::Validation(vec![
Violation::new(
RE_ENTRY_CODE,
Severity::Error,
format!(
"verification re-entered itself: {}. A `## Verification` command \
that runs `verify` on its own spec recurses without bound.",
stack.join(" -> ")
),
)
.at(format!("{}/{}/spec.md", cfg.layout.specs_dir, plan.spec_id)),
]));
}
stack.push(plan.spec_id.clone());
let child_stack = stack.join(",");
if plan_only {
if json {
let value = serde_json::to_value(&plan).map_err(|e| Error::Internal(e.to_string()))?;
out::verdict(&Verdict::report(verb::VERIFY, 0, value))?;
} else {
for command in &plan.commands {
outln!("{command}");
}
}
return Ok(0);
}
let total = plan.commands.len();
let declared = plan.is_declared();
if !json {
if let Some(from) = &plan.acceptance_from {
outln!("verify: {}", plan.spec_id);
outln!(" acceptance amended by {from} (spec 037); its block is the one that runs");
}
for s in &plan.skipped {
outln!(
"verify: {}: {} {} block(s) are driven by the orchestrator; skipped here",
plan.spec_id,
s.count,
s.tag
);
}
}
let mut ran = 0usize;
let mut failure = None;
for (i, command) in plan.commands.iter().enumerate() {
transcript(json, format_args!("[verify] $ {command}"));
let status = run_one(repo, command, &child_stack, json)?;
ran += 1;
match status.code() {
Some(c) => transcript(json, format_args!("[verify] exit {c}")),
None => transcript(json, format_args!("[verify] killed by signal")),
}
if !status.success() {
failure = Some(VerifyFailure {
index: i + 1,
command: command.clone(),
exit_code: status.code(),
});
break;
}
}
let outcome = match (&failure, declared) {
(Some(_), _) => VerifyOutcome::Failed,
(None, true) => VerifyOutcome::Passed,
(None, false) => VerifyOutcome::NotDeclared,
};
let code = outcome.exit_code();
let report = VerifyReport {
spec_id: plan.spec_id.clone(),
declared,
outcome,
ran,
total,
skipped: plan.skipped.clone(),
failure,
};
if json {
let value = serde_json::to_value(&report).map_err(|e| Error::Internal(e.to_string()))?;
out::verdict(&Verdict::report(verb::VERIFY, code, value))?;
return Ok(code);
}
match report.outcome {
VerifyOutcome::NotDeclared => outln!(
"verify: {}: not-declared (no verify:cli commands under ## Verification)",
report.spec_id
),
VerifyOutcome::Passed => outln!("verify: {}: passed ({total} command(s))", report.spec_id),
VerifyOutcome::Failed => {
let f = report.failure.as_ref().expect("failed implies a failure");
eprintln!(
"verify: {}: FAILED at command {} (exit {})",
report.spec_id,
f.index,
f.exit_code
.map_or_else(|| "signal".to_string(), |c| c.to_string())
);
}
}
Ok(code)
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Cursor;
use std::sync::{Arc, Mutex};
struct Pieces(std::collections::VecDeque<Vec<u8>>);
impl Read for Pieces {
fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
match self.0.pop_front() {
None => Ok(0),
Some(p) => {
let n = p.len().min(out.len());
out[..n].copy_from_slice(&p[..n]);
if n < p.len() {
self.0.push_front(p[n..].to_vec());
}
Ok(n)
}
}
}
}
fn pieces(ps: &[&[u8]]) -> Pieces {
Pieces(ps.iter().map(|p| p.to_vec()).collect())
}
fn deliveries(mut r: impl Read) -> (Vec<Vec<u8>>, Drained) {
let mut got = Vec::new();
let d = drain_lines(&mut r, |b| {
got.push(b.to_vec());
Ok(())
});
(got, d)
}
#[test]
fn a_line_split_across_reads_is_delivered_whole() {
let (got, d) = deliveries(pieces(&[b"OUT-1\nOU", b"T-2\nOUT", b"-3\n"]));
assert_eq!(d, Drained::Delivered);
assert_eq!(
got,
vec![
b"OUT-1\n".to_vec(),
b"OUT-2\n".to_vec(),
b"OUT-3\n".to_vec()
]
);
}
#[test]
fn a_last_line_without_a_newline_is_delivered_at_eof() {
let (got, _) = deliveries(pieces(&[b"a\nb"]));
assert_eq!(got, vec![b"a\n".to_vec(), b"b".to_vec()]);
}
#[test]
fn a_line_longer_than_the_hold_is_bounded_and_complete() {
let long = vec![b'x'; 3 * LINE_HOLD + 7];
let mut input = long.clone();
input.push(b'\n');
let (got, _) = deliveries(Cursor::new(input.clone()));
assert!(
got.iter().all(|g| g.len() <= LINE_HOLD + DRAIN_BUF),
"bounded"
);
assert_eq!(got.concat(), input, "every byte, in order");
}
#[test]
fn a_failed_delivery_still_drains_to_eof() {
let mut r = Cursor::new(b"one\ntwo\nthree\n".repeat(4096));
let mut calls = 0;
let d = drain_lines(&mut r, |_| {
calls += 1;
Err(std::io::Error::other("closed"))
});
assert_eq!(d, Drained::Discarded);
assert_eq!(calls, 1, "no delivery is attempted after the first failure");
assert_eq!(r.position() as usize, r.get_ref().len(), "read to EOF");
}
#[test]
fn two_streams_into_one_destination_splice_no_line() {
let sink = Arc::new(Mutex::new(Vec::<u8>::new()));
let stream = |tag: &'static str| {
let line = format!("{tag}-0123456789012345678901234567890123456789\n");
let all = line.repeat(2048).into_bytes();
let ps: Vec<Vec<u8>> = all.chunks(7).map(<[u8]>::to_vec).collect();
Pieces(ps.into())
};
let threads: Vec<_> = ["OUT", "ERR"]
.into_iter()
.map(|tag| {
let sink = Arc::clone(&sink);
let mut r = stream(tag);
std::thread::spawn(move || {
drain_lines(&mut r, |b| {
sink.lock().unwrap().extend_from_slice(b);
std::thread::yield_now();
Ok(())
})
})
})
.collect();
for t in threads {
assert_eq!(t.join().unwrap(), Drained::Delivered);
}
let out = String::from_utf8(sink.lock().unwrap().clone()).unwrap();
for tag in ["OUT", "ERR"] {
let whole = format!("{tag}-0123456789012345678901234567890123456789");
assert_eq!(out.lines().filter(|l| *l == whole).count(), 2048, "{tag}");
}
assert_eq!(out.lines().count(), 4096, "no line was spliced");
}
}