mind-self-run: the lifecycle is one pure decide(state, event, now); the loop only turns OS facts into events
decide owns when the flush is written, the stdin closes, a signal goes out and the run ends. The loop reads the clock once per event, ends on the child's exit (a waitpid thread) as well as on stdout EOF, and never signals a child that already exited. The flush's turn begins at the first result after the flush is written: the headroom counts only later requests and the grace runs from the flush write. The child runs in its own process group and signals go to the group. Tests: tables over decide, an enumeration of event orders against an oracle, the Yggdrasil probe shapes through a real child. Builds; every test passes. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
904ac3489c
commit
870f3838c8
5 changed files with 921 additions and 167 deletions
|
|
@ -8,10 +8,14 @@
|
|||
//! `eureka-body:ruling:self-lifecycle-flush-rehydrate`: a Self run admits what
|
||||
//! the mind lacks and ends, it never summarises), closes the child's input at
|
||||
//! the first result, and stops a child that outlives its flush: an interrupt
|
||||
//! after the grace, or once the prefix has grown by the headroom beyond the
|
||||
//! prefix the flush was asked at (ruling
|
||||
//! after the grace, or once the requests of the flush's own turn have grown the
|
||||
//! prefix by the headroom beyond the prefix the flush was asked at (ruling
|
||||
//! `eureka-body:ruling:flush-always-gets-a-turn`), then SIGTERM and SIGKILL if
|
||||
//! the child ignores that.
|
||||
//! the child ignores that. Signals go to the child's process group.
|
||||
//!
|
||||
//! Every timing decision is one pure function, `decide` (the `decide` module,
|
||||
//! ruling `eureka-body:ruling:self-run-decider`). `carry` is the loop that turns
|
||||
//! the operating system's facts into events and executes the actions.
|
||||
//!
|
||||
//! The actuator admits nothing, reads neither the brake nor the environment and
|
||||
//! stops no unit: the gate and the brake hold own those. It never echoes an
|
||||
|
|
@ -19,16 +23,14 @@
|
|||
//! foreign owner's JSON: it is parsed into `serde_json::Value` here and nothing
|
||||
//! parsed is stored.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::fmt;
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::path::Path;
|
||||
use std::process::{Child as OsChild, ChildStdin, Command, Stdio};
|
||||
use std::process::{ChildStdin, Command, Stdio};
|
||||
use std::sync::mpsc::{Receiver, RecvTimeoutError, channel};
|
||||
use std::time::Duration as StdDuration;
|
||||
|
||||
use anyhow::{Context, Result, anyhow, bail};
|
||||
use chrono::{DateTime, Duration, Utc};
|
||||
use cultnet_rs::Selection;
|
||||
use eureka_pipeline::{PipelineDocument, PipelineRef, RunTurn, Slug};
|
||||
use huginn_mind::PipelinePageItems;
|
||||
|
|
@ -38,6 +40,8 @@ use uuid::Uuid;
|
|||
use crate::launch::{Clock, holders};
|
||||
use crate::queue::MindPort;
|
||||
|
||||
use decide::{Action, Event, State, decide, next_deadline};
|
||||
|
||||
/// The measured Self ceiling: at this prefix the flush is asked for.
|
||||
pub const CEILING_TOKENS: u64 = 300_000;
|
||||
/// What the prefix may grow by beyond the prefix the flush was asked at before
|
||||
|
|
@ -47,6 +51,9 @@ pub const FLUSH_HEADROOM_TOKENS: u64 = 60_000;
|
|||
pub const FLUSH_GRACE_S: i64 = 15 * 60;
|
||||
/// How long a child may ignore a signal before the next, stronger one.
|
||||
pub const ESCALATE_AFTER_S: i64 = 120;
|
||||
/// How long a stream may stay open after the child has exited (a descendant
|
||||
/// holding its stdout) before the run is over without it.
|
||||
pub const EXIT_DRAIN_S: i64 = 2;
|
||||
/// A backstop `claude` enforces; the ceiling is reached first.
|
||||
pub const MAX_TURNS: u32 = 500;
|
||||
/// The name the start message tells her to admit as.
|
||||
|
|
@ -129,13 +136,15 @@ pub fn read_usage(line: &str) -> Usage {
|
|||
}
|
||||
}
|
||||
|
||||
/// What the child's stdout gave in one wait.
|
||||
/// What the child gave in one wait: the operating system's facts about it.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum Read {
|
||||
Line(String),
|
||||
Timeout,
|
||||
/// The stream ended: the child is gone or going.
|
||||
Closed,
|
||||
/// Its stdout ended.
|
||||
Eof,
|
||||
/// It was reaped: its exit code, none when a signal ended it.
|
||||
Exited(Option<i32>),
|
||||
}
|
||||
|
||||
/// The running `claude`. A test hands in a scripted one.
|
||||
|
|
@ -145,11 +154,9 @@ pub trait Child {
|
|||
fn write_line(&mut self, line: &str);
|
||||
/// Closes its input; later writes go nowhere.
|
||||
fn close_input(&mut self);
|
||||
fn next_line(&mut self, timeout: StdDuration) -> Read;
|
||||
/// Sends it the signal; whether the signal was delivered.
|
||||
fn next(&mut self, timeout: StdDuration) -> Read;
|
||||
/// Sends the signal to its whole process group; whether it was delivered.
|
||||
fn stop(&mut self, how: Stop) -> bool;
|
||||
/// Its exit code, none when a signal ended it.
|
||||
fn wait(&mut self) -> Option<i32>;
|
||||
}
|
||||
|
||||
/// The signals the actuator sends, weakest first.
|
||||
|
|
@ -245,76 +252,45 @@ fn her_run(mind: &dyn MindPort) -> Result<Option<(PipelineRef, String)>> {
|
|||
/// Carries her one open Self run: one `claude` child from the start message to
|
||||
/// its end. The request lines go to `out`. A mind that cannot be reached or a
|
||||
/// child that cannot be started is an error; every other end is an outcome.
|
||||
/// This is only the loop: facts in, one clock read per fact, actions out.
|
||||
pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, spawner: &dyn Spawner, out: &mut dyn Write) -> Result<SelfRunOutcome> {
|
||||
let Some((run, budget)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) };
|
||||
let mut child = spawner.spawn(&claude_args(session_id(&run), &budget))?;
|
||||
let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string();
|
||||
child.write_line(&user_line(&start_message(mind.instance(), &run)));
|
||||
|
||||
let (mut seen, mut requests, mut prefix) = (HashSet::new(), 0u64, 0u64);
|
||||
let mut result_seen = false;
|
||||
// The prefix and the grace deadline of the flush, once asked for.
|
||||
let mut flush: Option<(u64, DateTime<Utc>)> = None;
|
||||
// The last signal attempted and when, and the strongest one delivered.
|
||||
let mut stopping: Option<(Stop, DateTime<Utc>)> = None;
|
||||
let mut delivered: Option<Stop> = None;
|
||||
let mut state = State::default();
|
||||
let mut now = clock.now();
|
||||
loop {
|
||||
// The one deadline the wait must not pass: the next escalation once a signal went out, else the grace.
|
||||
let deadline = match (stopping, flush) {
|
||||
(Some((_, sent)), _) => Some(sent + Duration::seconds(ESCALATE_AFTER_S)),
|
||||
(None, Some((_, grace_ends))) => Some(grace_ends),
|
||||
(None, None) => None,
|
||||
};
|
||||
let wait = deadline.map_or(POLL, |ends| (ends - clock.now()).to_std().unwrap_or_default().min(POLL));
|
||||
match child.next_line(wait) {
|
||||
let wait = next_deadline(&state).map_or(POLL, |at| (at - now).to_std().unwrap_or_default().min(POLL));
|
||||
let fact = match child.next(wait) {
|
||||
Read::Line(line) => match read_usage(&line) {
|
||||
// A message id seen before is the same request again (F24).
|
||||
Usage::Request { id, prefix: newest } if seen.insert(id.clone()) => {
|
||||
requests += 1;
|
||||
prefix = newest;
|
||||
writeln!(out, "request {requests} prefix {prefix}")?;
|
||||
}
|
||||
Usage::Result if !result_seen => {
|
||||
result_seen = true;
|
||||
child.close_input();
|
||||
}
|
||||
_ => {}
|
||||
Usage::Request { id, prefix } => Event::Request { id, prefix },
|
||||
Usage::Result => Event::Result,
|
||||
Usage::Other | Usage::Unreadable => Event::Tick,
|
||||
},
|
||||
Read::Timeout => {}
|
||||
Read::Closed => break,
|
||||
}
|
||||
if flush.is_none() && !result_seen && prefix >= CEILING_TOKENS {
|
||||
child.write_line(&user_line(FLUSH_MESSAGE));
|
||||
flush = Some((prefix, clock.now() + Duration::seconds(FLUSH_GRACE_S)));
|
||||
}
|
||||
// The backstop is relative to the flush point, so it cannot fire in the pass that wrote the flush.
|
||||
let due = match (stopping, flush) {
|
||||
(Some((_, sent)), _) => clock.now() >= sent + Duration::seconds(ESCALATE_AFTER_S),
|
||||
(None, Some((at, grace_ends))) => clock.now() >= grace_ends || at.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit),
|
||||
(None, None) => false,
|
||||
Read::Timeout => Event::Tick,
|
||||
Read::Eof => Event::StdoutEof,
|
||||
Read::Exited(code) => Event::ChildExited(code),
|
||||
};
|
||||
if due {
|
||||
let how = stopping.map_or(Stop::Interrupt, |(last, _)| last.next());
|
||||
if child.stop(how) {
|
||||
delivered = Some(how);
|
||||
now = clock.now();
|
||||
let mut event = Some(fact);
|
||||
while let Some(heard) = event.take() {
|
||||
let counted = state.requests;
|
||||
let (next, action) = decide(state, heard, now);
|
||||
state = next;
|
||||
if state.requests != counted {
|
||||
writeln!(out, "request {} prefix {}", state.requests, state.prefix)?;
|
||||
}
|
||||
stopping = Some((how, clock.now()));
|
||||
if how == Stop::Kill {
|
||||
// A grandchild can hold the stream open past a dead child: do not wait for its end.
|
||||
break;
|
||||
match action {
|
||||
Action::None => {}
|
||||
Action::WriteFlush => child.write_line(&user_line(FLUSH_MESSAGE)),
|
||||
Action::CloseStdin => child.close_input(),
|
||||
Action::Signal(how) => event = Some(Event::Signalled { how, delivered: child.stop(how) }),
|
||||
Action::Finish(outcome) => return Ok(outcome),
|
||||
}
|
||||
}
|
||||
}
|
||||
if matches!(stopping, Some((Stop::Kill, _))) && delivered != Some(Stop::Kill) {
|
||||
return Ok(SelfRunOutcome::Ended { cause: Cause::Unstoppable, code: None, requests, prefix });
|
||||
}
|
||||
let code = child.wait();
|
||||
let cause = match (delivered, flush) {
|
||||
(Some(how), _) => Cause::Signalled(how),
|
||||
(None, Some(_)) => Cause::FlushRequested,
|
||||
(None, None) => Cause::Own,
|
||||
};
|
||||
Ok(SelfRunOutcome::Ended { cause, code, requests, prefix })
|
||||
}
|
||||
|
||||
/// The real spawner: the `claude` on the unit's `PATH`, with the unit's
|
||||
|
|
@ -328,25 +304,38 @@ impl Spawner for ClaudeSpawner {
|
|||
}
|
||||
|
||||
fn spawn_program(program: &Path, args: &[String]) -> Result<Box<dyn Child>> {
|
||||
let mut process = Command::new(program).args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit()).spawn().context("claude could not be started")?;
|
||||
let mut command = Command::new(program);
|
||||
command.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::inherit());
|
||||
// Its own process group, so that a signal reaches what the child started too.
|
||||
#[cfg(unix)]
|
||||
std::os::unix::process::CommandExt::process_group(&mut command, 0);
|
||||
let mut process = command.spawn().context("claude could not be started")?;
|
||||
let group = process.id();
|
||||
let stdin = process.stdin.take();
|
||||
let stdout = process.stdout.take().ok_or_else(|| anyhow!("claude has no stdout"))?;
|
||||
let (sender, lines) = channel();
|
||||
let (sender, facts) = channel();
|
||||
let lines = sender.clone();
|
||||
std::thread::spawn(move || {
|
||||
for line in BufReader::new(stdout).split(b'\n') {
|
||||
let Ok(bytes) = line else { return };
|
||||
if sender.send(String::from_utf8_lossy(&bytes).into_owned()).is_err() {
|
||||
let Ok(bytes) = line else { break };
|
||||
if lines.send(Read::Line(String::from_utf8_lossy(&bytes).into_owned())).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
let _ = lines.send(Read::Eof);
|
||||
});
|
||||
Ok(Box::new(ClaudeProcess { process, stdin, lines }))
|
||||
// The one waitpid: the child's end is a fact of its own, not inferred from its stream.
|
||||
std::thread::spawn(move || {
|
||||
let _ = sender.send(Read::Exited(process.wait().ok().and_then(|status| status.code())));
|
||||
});
|
||||
Ok(Box::new(ClaudeProcess { stdin, facts, group }))
|
||||
}
|
||||
|
||||
struct ClaudeProcess {
|
||||
process: OsChild,
|
||||
stdin: Option<ChildStdin>,
|
||||
lines: Receiver<String>,
|
||||
facts: Receiver<Read>,
|
||||
/// The child's pid, which is its process group's id.
|
||||
group: u32,
|
||||
}
|
||||
|
||||
impl Child for ClaudeProcess {
|
||||
|
|
@ -360,28 +349,26 @@ impl Child for ClaudeProcess {
|
|||
self.stdin = None;
|
||||
}
|
||||
|
||||
fn next_line(&mut self, timeout: StdDuration) -> Read {
|
||||
match self.lines.recv_timeout(timeout) {
|
||||
Ok(line) => Read::Line(line),
|
||||
fn next(&mut self, timeout: StdDuration) -> Read {
|
||||
match self.facts.recv_timeout(timeout) {
|
||||
Ok(fact) => fact,
|
||||
Err(RecvTimeoutError::Timeout) => Read::Timeout,
|
||||
Err(RecvTimeoutError::Disconnected) => Read::Closed,
|
||||
Err(RecvTimeoutError::Disconnected) => Read::Eof,
|
||||
}
|
||||
}
|
||||
|
||||
fn stop(&mut self, how: Stop) -> bool {
|
||||
// The `kill` program, not a libc call: the library names no FFI.
|
||||
let signal = match how {
|
||||
Stop::Interrupt => "-INT",
|
||||
Stop::Terminate => "-TERM",
|
||||
Stop::Kill => return self.process.kill().is_ok(),
|
||||
Stop::Interrupt => "INT",
|
||||
Stop::Terminate => "TERM",
|
||||
Stop::Kill => "KILL",
|
||||
};
|
||||
Command::new("kill").args([signal, &self.process.id().to_string()]).status().is_ok_and(|status| status.success())
|
||||
}
|
||||
|
||||
fn wait(&mut self) -> Option<i32> {
|
||||
self.process.wait().ok().and_then(|status| status.code())
|
||||
Command::new("kill").args(["-s", signal, "--", &format!("-{}", self.group)]).stderr(Stdio::null()).status().is_ok_and(|status| status.success())
|
||||
}
|
||||
}
|
||||
|
||||
mod decide;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
|
|
|||
185
crates/mind-body/src/self_run/decide.rs
Normal file
185
crates/mind-body/src/self_run/decide.rs
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
//! The Self-run actuator's lifecycle decision, and nothing else.
|
||||
//!
|
||||
//! `decide(state, event, now) -> (state, action)` is the one owner of when the
|
||||
//! flush is asked for, when the stdin closes, when a signal goes out and when
|
||||
//! the run is over (ruling `eureka-body:ruling:self-run-decider`). It performs
|
||||
//! no I/O and reads no clock: the loop in `carry` turns operating-system facts
|
||||
//! into events, reads the clock once per event, and executes the action. A
|
||||
//! timing rule is pinned here, by tables and by an enumeration of event orders.
|
||||
//!
|
||||
//! The flush's turn begins at the first result after the flush was written (the
|
||||
//! turn already running ends there; the queued flush is the next one). The
|
||||
//! headroom counts only requests after that boundary, measured from the prefix
|
||||
//! the flush was written at, and the grace from the flush write (ruling
|
||||
//! `eureka-body:ruling:flush-always-gets-a-turn`, refined by the decider ruling).
|
||||
//! At most one action comes out of an event; a deadline that is already due
|
||||
//! yields its signal at the next event, and `next_deadline` says it is now.
|
||||
|
||||
use std::collections::HashSet;
|
||||
|
||||
use chrono::{DateTime, Duration, Utc};
|
||||
|
||||
use super::{CEILING_TOKENS, Cause, ESCALATE_AFTER_S, EXIT_DRAIN_S, FLUSH_GRACE_S, FLUSH_HEADROOM_TOKENS, SelfRunOutcome, Stop};
|
||||
|
||||
/// What the loop saw, already turned into a fact the decision can use.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub(super) enum Event {
|
||||
/// A request of the main thread: its message id and the context it read.
|
||||
Request { id: String, prefix: u64 },
|
||||
Result,
|
||||
/// `waitpid` returned: the child's exit code, none when a signal ended it.
|
||||
ChildExited(Option<i32>),
|
||||
StdoutEof,
|
||||
/// Time passed, or a line that no rule reads.
|
||||
Tick,
|
||||
/// What the operating system said of the signal the last action sent.
|
||||
Signalled { how: Stop, delivered: bool },
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub(super) enum Action {
|
||||
None,
|
||||
WriteFlush,
|
||||
CloseStdin,
|
||||
/// To the child's whole process group.
|
||||
Signal(Stop),
|
||||
Finish(SelfRunOutcome),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct Flush {
|
||||
/// The prefix at which the flush was written: the headroom's base.
|
||||
prefix: u64,
|
||||
written: DateTime<Utc>,
|
||||
/// A result has come since the write: the flush is the next turn.
|
||||
turn_begun: bool,
|
||||
/// A request of the flush's turn reached the headroom.
|
||||
exhausted: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub(super) struct State {
|
||||
seen: HashSet<String>,
|
||||
pub(super) requests: u64,
|
||||
pub(super) prefix: u64,
|
||||
result_seen: bool,
|
||||
flush: Option<Flush>,
|
||||
/// The last signal attempted and when.
|
||||
stopping: Option<(Stop, DateTime<Utc>)>,
|
||||
/// The strongest signal delivered.
|
||||
delivered: Option<Stop>,
|
||||
exited: Option<(Option<i32>, DateTime<Utc>)>,
|
||||
eof: bool,
|
||||
finished: bool,
|
||||
}
|
||||
|
||||
fn outcome(state: &State, cause: Cause, code: Option<i32>) -> SelfRunOutcome {
|
||||
SelfRunOutcome::Ended { cause, code, requests: state.requests, prefix: state.prefix }
|
||||
}
|
||||
|
||||
/// How the run ended, by what was done to the child: the strongest signal delivered, else the flush, else nothing.
|
||||
fn cause(state: &State) -> Cause {
|
||||
match (state.delivered, &state.flush) {
|
||||
(Some(how), _) => Cause::Signalled(how),
|
||||
(None, Some(_)) => Cause::FlushRequested,
|
||||
(None, None) => Cause::Own,
|
||||
}
|
||||
}
|
||||
|
||||
/// The moment the loop must next be heard at, if any. A moment already past
|
||||
/// means an action is due now.
|
||||
pub(super) fn next_deadline(state: &State) -> Option<DateTime<Utc>> {
|
||||
if state.finished {
|
||||
return None;
|
||||
}
|
||||
if let Some((_, exited)) = state.exited {
|
||||
return (!state.eof).then(|| exited + Duration::seconds(EXIT_DRAIN_S));
|
||||
}
|
||||
match (&state.stopping, &state.flush) {
|
||||
(Some((_, sent)), _) => Some(*sent + Duration::seconds(ESCALATE_AFTER_S)),
|
||||
(None, Some(flush)) => Some(flush.written + Duration::seconds(FLUSH_GRACE_S)),
|
||||
(None, None) => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn decide(mut state: State, event: Event, now: DateTime<Utc>) -> (State, Action) {
|
||||
if state.finished {
|
||||
return (state, Action::None);
|
||||
}
|
||||
let mut action = Action::None;
|
||||
match event {
|
||||
Event::Request { id, prefix } => {
|
||||
if state.seen.insert(id) {
|
||||
state.requests += 1;
|
||||
state.prefix = prefix;
|
||||
if let Some(flush) = state.flush.as_mut()
|
||||
&& flush.turn_begun
|
||||
&& flush.prefix.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| prefix >= limit)
|
||||
{
|
||||
flush.exhausted = true;
|
||||
}
|
||||
if state.flush.is_none() && !state.result_seen && state.exited.is_none() && prefix >= CEILING_TOKENS {
|
||||
state.flush = Some(Flush { prefix, written: now, turn_begun: false, exhausted: false });
|
||||
action = Action::WriteFlush;
|
||||
}
|
||||
}
|
||||
}
|
||||
Event::Result => {
|
||||
if !state.result_seen {
|
||||
state.result_seen = true;
|
||||
if let Some(flush) = state.flush.as_mut() {
|
||||
flush.turn_begun = true;
|
||||
}
|
||||
action = Action::CloseStdin;
|
||||
}
|
||||
}
|
||||
Event::ChildExited(code) => {
|
||||
state.exited.get_or_insert((code, now));
|
||||
}
|
||||
Event::StdoutEof => state.eof = true,
|
||||
Event::Tick => {}
|
||||
Event::Signalled { how, delivered } => {
|
||||
if delivered {
|
||||
state.delivered = Some(how);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A child that has exited is never written to or signalled: the run is over
|
||||
// once its stream has ended, or once the stream has had its drain.
|
||||
if let Some((code, exited)) = state.exited {
|
||||
if state.eof || now >= exited + Duration::seconds(EXIT_DRAIN_S) {
|
||||
state.finished = true;
|
||||
let outcome = outcome(&state, cause(&state), code);
|
||||
return (state, Action::Finish(outcome));
|
||||
}
|
||||
return (state, Action::None);
|
||||
}
|
||||
if action != Action::None {
|
||||
return (state, action);
|
||||
}
|
||||
if let Some((Stop::Kill, sent)) = state.stopping {
|
||||
// SIGKILL went out and the child still has not been reaped: stop waiting for it.
|
||||
if now >= sent + Duration::seconds(ESCALATE_AFTER_S) {
|
||||
state.finished = true;
|
||||
let cause = if state.delivered == Some(Stop::Kill) { Cause::Signalled(Stop::Kill) } else { Cause::Unstoppable };
|
||||
let outcome = outcome(&state, cause, None);
|
||||
return (state, Action::Finish(outcome));
|
||||
}
|
||||
return (state, Action::None);
|
||||
}
|
||||
let how = match (&state.stopping, &state.flush) {
|
||||
(Some((last, sent)), _) => (now >= *sent + Duration::seconds(ESCALATE_AFTER_S)).then(|| last.next()),
|
||||
(None, Some(flush)) => (flush.exhausted || now >= flush.written + Duration::seconds(FLUSH_GRACE_S)).then_some(Stop::Interrupt),
|
||||
(None, None) => None,
|
||||
};
|
||||
match how {
|
||||
Some(how) => {
|
||||
state.stopping = Some((how, now));
|
||||
(state, Action::Signal(how))
|
||||
}
|
||||
None => (state, Action::None),
|
||||
}
|
||||
}
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
443
crates/mind-body/src/self_run/decide/tests.rs
Normal file
443
crates/mind-body/src/self_run/decide/tests.rs
Normal file
|
|
@ -0,0 +1,443 @@
|
|||
//! `decide` is pure, so its rules are pinned here with no child, no mind and no
|
||||
//! clock: tables, one row per boundary, and an enumeration of event orders that
|
||||
//! asserts the invariants against an oracle kept from the event history alone.
|
||||
|
||||
use super::*;
|
||||
use crate::testkit::at;
|
||||
|
||||
fn t(seconds: i64) -> DateTime<Utc> {
|
||||
at(seconds * 1000)
|
||||
}
|
||||
|
||||
fn req(id: &str, prefix: u64) -> Event {
|
||||
Event::Request { id: id.to_string(), prefix }
|
||||
}
|
||||
|
||||
fn feed(state: State, event: Event, seconds: i64) -> (State, Action) {
|
||||
decide(state, event, t(seconds))
|
||||
}
|
||||
|
||||
/// Runs the events in order, each at its own second, and keeps every action.
|
||||
fn run(events: Vec<(Event, i64)>) -> (State, Vec<Action>) {
|
||||
let mut state = State::default();
|
||||
let mut actions = Vec::new();
|
||||
for (event, seconds) in events {
|
||||
let (next, action) = feed(state, event, seconds);
|
||||
state = next;
|
||||
actions.push(action);
|
||||
}
|
||||
(state, actions)
|
||||
}
|
||||
|
||||
fn last(events: Vec<(Event, i64)>) -> Action {
|
||||
run(events).1.pop().unwrap()
|
||||
}
|
||||
|
||||
fn delivered(how: Stop) -> Event {
|
||||
Event::Signalled { how, delivered: true }
|
||||
}
|
||||
|
||||
fn ended(cause: Cause, code: Option<i32>, requests: u64, prefix: u64) -> Action {
|
||||
Action::Finish(SelfRunOutcome::Ended { cause, code, requests, prefix })
|
||||
}
|
||||
|
||||
const GRACE: i64 = FLUSH_GRACE_S;
|
||||
const ESC: i64 = ESCALATE_AFTER_S;
|
||||
const LIMIT: u64 = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS;
|
||||
|
||||
#[test]
|
||||
fn the_flush_is_asked_once_at_the_ceiling_and_never_after_a_result_or_an_exit_or_for_a_repeated_request() {
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS - 1), 0)]), Action::None);
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0)]), Action::WriteFlush);
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS + 1), 0)]), Action::WriteFlush);
|
||||
// Once.
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", CEILING_TOKENS + 5), 1)]), Action::None);
|
||||
// A result first: the child is finishing, there is nothing to flush.
|
||||
assert_eq!(last(vec![(Event::Result, 0), (req("a", CEILING_TOKENS), 1)]), Action::None);
|
||||
// A child that is gone is not written to.
|
||||
assert_eq!(last(vec![(Event::ChildExited(Some(0)), 0), (req("a", CEILING_TOKENS), 1)]), Action::None);
|
||||
// The same message id is the same request again: it neither counts nor decides.
|
||||
let (state, actions) = run(vec![(req("a", 10), 0), (req("a", CEILING_TOKENS), 1)]);
|
||||
assert_eq!((actions[1].clone(), state.requests, state.prefix), (Action::None, 1, 10));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_input_closes_once_at_the_first_result() {
|
||||
assert_eq!(last(vec![(Event::Result, 0)]), Action::CloseStdin);
|
||||
assert_eq!(last(vec![(Event::Result, 0), (Event::Result, 1)]), Action::None);
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1)]), Action::CloseStdin);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_grace_runs_from_the_flush_write_to_the_second_and_no_later_request_moves_it() {
|
||||
let flushed = |rest: Vec<(Event, i64)>| last([vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100)], rest].concat());
|
||||
assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE - 1)]), Action::None);
|
||||
assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt));
|
||||
assert_eq!(flushed(vec![(Event::Tick, 100 + GRACE + 5000)]), Action::Signal(Stop::Interrupt));
|
||||
// Requests after the flush, however late, do not refresh it.
|
||||
assert_eq!(flushed(vec![(req("c", CEILING_TOKENS + 1), 100 + GRACE - 10), (Event::Tick, 100 + GRACE)]), Action::Signal(Stop::Interrupt));
|
||||
// Not from the first request, not from the start.
|
||||
assert_eq!(last(vec![(req("a", 10), 0), (req("b", CEILING_TOKENS), 100), (Event::Tick, GRACE)]), Action::None);
|
||||
// The deadline the loop waits for is the same moment.
|
||||
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 100)]);
|
||||
assert_eq!(next_deadline(&state), Some(t(100 + GRACE)));
|
||||
assert_eq!(next_deadline(&State::default()), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_headroom_counts_the_requests_after_the_first_result_after_the_flush_from_the_prefix_of_the_flush() {
|
||||
let after_result = |request: u64| last(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1), (req("b", request), 2)]);
|
||||
assert_eq!(after_result(LIMIT - 1), Action::None);
|
||||
assert_eq!(after_result(LIMIT), Action::Signal(Stop::Interrupt));
|
||||
assert_eq!(after_result(LIMIT + 1), Action::Signal(Stop::Interrupt));
|
||||
// Measured from the flush's own prefix, not from the ceiling.
|
||||
let high = |request: u64| last(vec![(req("a", 400_000), 0), (Event::Result, 1), (req("b", request), 2)]);
|
||||
assert_eq!((high(459_999), high(460_000)), (Action::None, Action::Signal(Stop::Interrupt)));
|
||||
// The turn already running is not the flush's turn: before a result no request counts.
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 400_000), 1)]), Action::None);
|
||||
// A big request before the result is forgotten by the boundary; only those after it count.
|
||||
assert_eq!(last(vec![(req("a", CEILING_TOKENS), 0), (req("b", LIMIT + 1), 1), (Event::Result, 2), (req("c", LIMIT - 1), 3)]), Action::None);
|
||||
// The result before the flush is not the boundary: the flush cannot be written then.
|
||||
assert_eq!(last(vec![(Event::Result, 0), (req("a", CEILING_TOKENS), 1), (req("b", 900_000), 2)]), Action::None);
|
||||
// A prefix that saturates leaves no headroom to count: only the grace is left.
|
||||
let saturated = |rest: Vec<(Event, i64)>| last([vec![(req("a", u64::MAX), 0), (Event::Result, 1), (req("b", u64::MAX), 2)], rest].concat());
|
||||
assert_eq!(saturated(vec![]), Action::None);
|
||||
assert_eq!(saturated(vec![(Event::Tick, GRACE)]), Action::Signal(Stop::Interrupt));
|
||||
// A request counted past the headroom stays counted: a later smaller one does not take the interrupt back.
|
||||
let (state, actions) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Result, 1), (req("b", LIMIT), 2)]);
|
||||
assert_eq!(actions[2], Action::Signal(Stop::Interrupt));
|
||||
assert_eq!(next_deadline(&state), Some(t(2 + ESC)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_flush_pass_never_signals_whatever_the_clock_does() {
|
||||
// Two events apart by more than the grace: the write is its own pass, the signal comes in the next.
|
||||
let (state, action) = feed(State::default(), req("a", CEILING_TOKENS), 0);
|
||||
assert_eq!(action, Action::WriteFlush);
|
||||
assert_eq!(feed(state, Event::Tick, 10 * GRACE).1, Action::Signal(Stop::Interrupt));
|
||||
// A request that both crosses the ceiling and is far past the headroom writes the flush and nothing else.
|
||||
assert_eq!(last(vec![(req("a", u64::MAX), 0)]), Action::WriteFlush);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn each_signal_follows_the_last_by_the_bound_and_the_run_ends_when_the_kill_has_had_its_time() {
|
||||
let interrupted = vec![(req("a", CEILING_TOKENS), 0), (Event::Tick, GRACE), (delivered(Stop::Interrupt), GRACE)];
|
||||
let then = |rest: Vec<(Event, i64)>| last([interrupted.clone(), rest].concat());
|
||||
assert_eq!(then(vec![(Event::Tick, GRACE + ESC - 1)]), Action::None);
|
||||
assert_eq!(then(vec![(Event::Tick, GRACE + ESC)]), Action::Signal(Stop::Terminate));
|
||||
// The bound runs from the last signal, not from the first or from the flush.
|
||||
let late_term = [interrupted.clone(), vec![(Event::Tick, GRACE + ESC + 30), (delivered(Stop::Terminate), GRACE + ESC + 30)]].concat();
|
||||
let kill_at = |seconds: i64| last([late_term.clone(), vec![(Event::Tick, seconds)]].concat());
|
||||
assert_eq!(kill_at(GRACE + 2 * ESC + 29), Action::None);
|
||||
assert_eq!(kill_at(GRACE + 2 * ESC + 30), Action::Signal(Stop::Kill));
|
||||
// After the kill: the child may still be reaped until the bound, then the run ends without it.
|
||||
let killed = [late_term, vec![(Event::Tick, GRACE + 2 * ESC + 30), (delivered(Stop::Kill), GRACE + 2 * ESC + 30)]].concat();
|
||||
let after = |seconds: i64| last([killed.clone(), vec![(Event::Tick, seconds)]].concat());
|
||||
let end = GRACE + 3 * ESC + 30;
|
||||
assert_eq!(after(end - 1), Action::None);
|
||||
assert_eq!(after(end), ended(Cause::Signalled(Stop::Kill), None, 1, CEILING_TOKENS));
|
||||
// A kill that was not delivered, with the child still there at the bound: Unstoppable, whatever else got through.
|
||||
let refused_kill = [killed[..killed.len() - 1].to_vec(), vec![(Event::Signalled { how: Stop::Kill, delivered: false }, GRACE + 2 * ESC + 30)]].concat();
|
||||
assert_eq!(last([refused_kill, vec![(Event::Tick, end)]].concat()), ended(Cause::Unstoppable, None, 1, CEILING_TOKENS));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_kill_that_was_never_delivered_ends_the_run_as_unstoppable_when_nothing_was_delivered_at_all() {
|
||||
let all_refused = |how: Stop| Event::Signalled { how, delivered: false };
|
||||
let (state, _) = run(vec![
|
||||
(req("a", CEILING_TOKENS), 0),
|
||||
(Event::Tick, GRACE),
|
||||
(all_refused(Stop::Interrupt), GRACE),
|
||||
(Event::Tick, GRACE + ESC),
|
||||
(all_refused(Stop::Terminate), GRACE + ESC),
|
||||
(Event::Tick, GRACE + 2 * ESC),
|
||||
(all_refused(Stop::Kill), GRACE + 2 * ESC),
|
||||
]);
|
||||
assert_eq!(next_deadline(&state), Some(t(GRACE + 3 * ESC)));
|
||||
assert_eq!(feed(state.clone(), Event::Tick, GRACE + 3 * ESC - 1).1, Action::None);
|
||||
assert_eq!(feed(state, Event::Tick, GRACE + 3 * ESC).1, ended(Cause::Unstoppable, None, 1, CEILING_TOKENS));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_child_that_ends_while_the_kill_was_in_flight_ends_the_run_by_its_own_cause_not_unstoppable() {
|
||||
// The kill was refused because the child was already gone; the reap is on its way.
|
||||
let (state, _) = run(vec![
|
||||
(req("a", CEILING_TOKENS), 0),
|
||||
(Event::Tick, GRACE),
|
||||
(delivered(Stop::Interrupt), GRACE),
|
||||
(Event::Tick, GRACE + ESC),
|
||||
(Event::Tick, GRACE + 2 * ESC),
|
||||
(Event::Signalled { how: Stop::Kill, delivered: false }, GRACE + 2 * ESC),
|
||||
(Event::ChildExited(Some(0)), GRACE + 2 * ESC + 1),
|
||||
]);
|
||||
assert_eq!(feed(state, Event::StdoutEof, GRACE + 2 * ESC + 1).1, ended(Cause::Signalled(Stop::Interrupt), Some(0), 1, CEILING_TOKENS));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_child_that_has_exited_is_never_signalled_and_the_run_ends_with_its_stream_or_after_the_drain() {
|
||||
// A flush asked for, the grace long past, the child already gone: no signal, only the end.
|
||||
let gone = vec![(req("a", CEILING_TOKENS), 0), (Event::ChildExited(Some(0)), 10)];
|
||||
let tick = |seconds: i64| last([gone.clone(), vec![(Event::Tick, seconds)]].concat());
|
||||
assert_eq!(tick(11), Action::None);
|
||||
assert_eq!(tick(10 + EXIT_DRAIN_S - 1), Action::None);
|
||||
assert_eq!(tick(10 + EXIT_DRAIN_S), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
|
||||
assert_eq!(tick(10 + GRACE * 4), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
|
||||
// With the stream ended too, it ends at once, in either order.
|
||||
assert_eq!(last([gone.clone(), vec![(Event::StdoutEof, 11)]].concat()), ended(Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
|
||||
assert_eq!(last(vec![(Event::StdoutEof, 0), (Event::ChildExited(None), 1)]), ended(Cause::Own, None, 0, 0));
|
||||
// The end of the stream alone is not the end of the child.
|
||||
assert_eq!(last(vec![(Event::StdoutEof, 0), (Event::Tick, 5)]), Action::None);
|
||||
// No flush, a signal's cause named by what was delivered, a code from the exit.
|
||||
let (state, _) = run(vec![(req("a", CEILING_TOKENS), 0), (Event::Tick, GRACE), (delivered(Stop::Interrupt), GRACE), (Event::ChildExited(Some(130)), GRACE + 1)]);
|
||||
assert_eq!(feed(state, Event::StdoutEof, GRACE + 1).1, ended(Cause::Signalled(Stop::Interrupt), Some(130), 1, CEILING_TOKENS));
|
||||
// The deadline of a child that has gone is the end of its drain.
|
||||
let (state, _) = run(vec![(Event::ChildExited(Some(0)), 7)]);
|
||||
assert_eq!(next_deadline(&state), Some(t(7 + EXIT_DRAIN_S)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn without_a_flush_nothing_is_ever_signalled_and_a_finished_run_decides_nothing_more() {
|
||||
let quiet = vec![(req("a", 10), 0), (Event::Result, 1), (req("b", 900_000), 2), (Event::Tick, 100_000)];
|
||||
assert!(run(quiet).1.iter().all(|action| !matches!(action, Action::Signal(_))));
|
||||
let (state, actions) = run(vec![(Event::ChildExited(Some(3)), 0), (Event::StdoutEof, 0)]);
|
||||
assert_eq!(actions[1], ended(Cause::Own, Some(3), 0, 0));
|
||||
assert_eq!(next_deadline(&state), None);
|
||||
assert_eq!(feed(state, Event::Tick, 10_000).1, Action::None);
|
||||
}
|
||||
|
||||
// ---- every order of events ------------------------------------------------------
|
||||
|
||||
/// What can happen next, each at a time after the last.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum Sym {
|
||||
Low,
|
||||
AtCeiling,
|
||||
Huge,
|
||||
Repeat,
|
||||
Result,
|
||||
Exit(Option<i32>),
|
||||
Eof,
|
||||
Soon,
|
||||
Bound,
|
||||
Grace,
|
||||
}
|
||||
|
||||
const ALPHABET: [Sym; 11] = [
|
||||
Sym::Low,
|
||||
Sym::AtCeiling,
|
||||
Sym::Huge,
|
||||
Sym::Repeat,
|
||||
Sym::Result,
|
||||
Sym::Exit(Some(0)),
|
||||
Sym::Exit(None),
|
||||
Sym::Eof,
|
||||
Sym::Soon,
|
||||
Sym::Bound,
|
||||
Sym::Grace,
|
||||
];
|
||||
|
||||
/// Which signals the operating system lets through.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum Policy {
|
||||
All,
|
||||
Nothing,
|
||||
NoKill,
|
||||
}
|
||||
|
||||
/// The facts of the history, kept apart from `State`: what the invariants are asked of.
|
||||
#[derive(Default)]
|
||||
struct Oracle {
|
||||
ids: Vec<String>,
|
||||
flush: Option<(i64, u64)>,
|
||||
flushes: u32,
|
||||
closes: u32,
|
||||
boundary: bool,
|
||||
exceeded: bool,
|
||||
result_seen: bool,
|
||||
exited: Option<i64>,
|
||||
eof: bool,
|
||||
signals: Vec<(Stop, i64)>,
|
||||
finished: bool,
|
||||
}
|
||||
|
||||
fn check_step(oracle: &mut Oracle, event: &Event, now: i64, action: &Action) {
|
||||
assert!(!oracle.finished || *action == Action::None, "an action after the run ended: {action:?} on {event:?}");
|
||||
let flush_before = oracle.flush;
|
||||
let (was_result, was_exited) = (oracle.result_seen, oracle.exited.is_some());
|
||||
match event {
|
||||
Event::Request { id, prefix } => {
|
||||
if !oracle.ids.contains(id) {
|
||||
oracle.ids.push(id.clone());
|
||||
if oracle.boundary
|
||||
&& let Some((_, base)) = oracle.flush
|
||||
&& base.checked_add(FLUSH_HEADROOM_TOKENS).is_some_and(|limit| *prefix >= limit)
|
||||
{
|
||||
oracle.exceeded = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
Event::Result => {
|
||||
if oracle.flush.is_some() {
|
||||
oracle.boundary = true;
|
||||
}
|
||||
oracle.result_seen = true;
|
||||
}
|
||||
Event::ChildExited(_) => {
|
||||
oracle.exited.get_or_insert(now);
|
||||
}
|
||||
Event::StdoutEof => oracle.eof = true,
|
||||
Event::Tick | Event::Signalled { .. } => {}
|
||||
}
|
||||
match action {
|
||||
Action::None => {}
|
||||
Action::WriteFlush => {
|
||||
oracle.flushes += 1;
|
||||
let Event::Request { prefix, .. } = event else { panic!("a flush on {event:?}") };
|
||||
assert!(*prefix >= CEILING_TOKENS && !was_result && !was_exited && flush_before.is_none(), "flush at {prefix} after {was_result} {was_exited}");
|
||||
oracle.flush = Some((now, *prefix));
|
||||
}
|
||||
Action::CloseStdin => {
|
||||
oracle.closes += 1;
|
||||
assert!(matches!(event, Event::Result) && !was_result && !was_exited);
|
||||
}
|
||||
Action::Signal(how) => {
|
||||
let (flushed_at, _) = flush_before.expect("a signal before the flush was written");
|
||||
assert!(oracle.exited.is_none(), "a signal to a child that already exited");
|
||||
match oracle.signals.last() {
|
||||
None => {
|
||||
assert_eq!(*how, Stop::Interrupt);
|
||||
assert!(now >= flushed_at + GRACE || oracle.exceeded, "interrupt before the flush had its turn: now {now}, flush {flushed_at}, exceeded {}", oracle.exceeded);
|
||||
assert!(now > flushed_at || oracle.exceeded, "interrupt in the pass that wrote the flush");
|
||||
}
|
||||
Some((last, sent)) => {
|
||||
assert_eq!(*how, last.next(), "escalation out of order");
|
||||
assert_ne!(*last, Stop::Kill, "a signal after the kill");
|
||||
assert!(now >= sent + ESC, "escalated after {} s, bound {ESC}", now - sent);
|
||||
}
|
||||
}
|
||||
oracle.signals.push((*how, now));
|
||||
}
|
||||
Action::Finish(outcome) => {
|
||||
oracle.finished = true;
|
||||
let SelfRunOutcome::Ended { cause, code, .. } = outcome else { panic!("{outcome:?}") };
|
||||
match oracle.exited {
|
||||
Some(at) => assert!(oracle.eof || now >= at + EXIT_DRAIN_S, "finished before the stream or the drain"),
|
||||
None => {
|
||||
let (last, sent) = *oracle.signals.last().expect("finished with the child alive and no signal sent");
|
||||
assert!(last == Stop::Kill && now >= sent + ESC, "finished with the child alive");
|
||||
assert_eq!(*code, None);
|
||||
assert!(matches!(cause, Cause::Signalled(Stop::Kill) | Cause::Unstoppable), "{cause:?}");
|
||||
}
|
||||
}
|
||||
if matches!(cause, Cause::Own) {
|
||||
assert!(oracle.flush.is_none() && oracle.signals.is_empty());
|
||||
}
|
||||
if matches!(cause, Cause::FlushRequested) {
|
||||
assert!(oracle.flush.is_some());
|
||||
}
|
||||
}
|
||||
}
|
||||
assert!(oracle.flushes <= 1 && oracle.closes <= 1);
|
||||
assert!(oracle.signals.len() <= 3);
|
||||
}
|
||||
|
||||
fn play(symbols: &[Sym], policy: Policy) {
|
||||
let mut state = State::default();
|
||||
let mut oracle = Oracle::default();
|
||||
let mut now = 0i64;
|
||||
let mut unique = 0u32;
|
||||
let mut events: Vec<Event> = Vec::new();
|
||||
for symbol in symbols {
|
||||
let (event, dt) = match symbol {
|
||||
Sym::Low => (req_new(&mut unique, 1_000), 1),
|
||||
Sym::AtCeiling => (req_new(&mut unique, CEILING_TOKENS), 1),
|
||||
Sym::Huge => (req_new(&mut unique, LIMIT + 400_000), 1),
|
||||
Sym::Repeat => (req("u1", CEILING_TOKENS + 7), 1),
|
||||
Sym::Result => (Event::Result, 1),
|
||||
Sym::Exit(code) => (Event::ChildExited(*code), 1),
|
||||
Sym::Eof => (Event::StdoutEof, 1),
|
||||
Sym::Soon => (Event::Tick, 1),
|
||||
Sym::Bound => (Event::Tick, ESC + 1),
|
||||
Sym::Grace => (Event::Tick, GRACE + 1),
|
||||
};
|
||||
now += dt;
|
||||
events.push(event.clone());
|
||||
let mut next = Some(event);
|
||||
while let Some(event) = next.take() {
|
||||
let (after, action) = decide(state, event.clone(), t(now));
|
||||
state = after;
|
||||
check_step(&mut oracle, &event, now, &action);
|
||||
if let Action::Signal(how) = action {
|
||||
let delivered = match policy {
|
||||
Policy::All => true,
|
||||
Policy::Nothing => false,
|
||||
Policy::NoKill => how != Stop::Kill,
|
||||
};
|
||||
next = Some(Event::Signalled { how, delivered });
|
||||
}
|
||||
}
|
||||
}
|
||||
// No spinning: a deadline that has come is never answered with nothing.
|
||||
if let Some(deadline) = next_deadline(&state) {
|
||||
let at = deadline.max(t(now));
|
||||
let (_, action) = decide(state.clone(), Event::Tick, at);
|
||||
assert_ne!(action, Action::None, "a due deadline gave no action: {symbols:?} {policy:?}");
|
||||
}
|
||||
// Once the child has exited and its stream has ended, the run is over, whatever came before.
|
||||
for event in [Event::ChildExited(Some(0)), Event::StdoutEof] {
|
||||
now += 1;
|
||||
let mut next = Some(event);
|
||||
while let Some(event) = next.take() {
|
||||
let (after, action) = decide(state, event.clone(), t(now));
|
||||
state = after;
|
||||
check_step(&mut oracle, &event, now, &action);
|
||||
if let Action::Signal(how) = action {
|
||||
next = Some(Event::Signalled { how, delivered: true });
|
||||
}
|
||||
}
|
||||
}
|
||||
assert!(oracle.finished, "no end after the exit and the end of the stream: {symbols:?} {policy:?} {events:?}");
|
||||
}
|
||||
|
||||
fn req_new(unique: &mut u32, prefix: u64) -> Event {
|
||||
*unique += 1;
|
||||
req(&format!("u{unique}"), prefix)
|
||||
}
|
||||
|
||||
fn enumerate(length: usize, symbols: &mut Vec<Sym>, policy: Policy) {
|
||||
if symbols.len() == length {
|
||||
play(symbols, policy);
|
||||
return;
|
||||
}
|
||||
for symbol in ALPHABET {
|
||||
symbols.push(symbol);
|
||||
enumerate(length, symbols, policy);
|
||||
symbols.pop();
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_order_of_up_to_five_events_keeps_the_invariants_under_every_policy_of_delivery() {
|
||||
for policy in [Policy::All, Policy::Nothing, Policy::NoKill] {
|
||||
for length in 0..=5 {
|
||||
enumerate(length, &mut Vec::new(), policy);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_long_random_order_of_events_keeps_the_invariants_and_always_ends_once_the_child_is_gone() {
|
||||
// xorshift: the same orders every run.
|
||||
let mut seed = 0x2545_F491_4F6C_DD1Du64;
|
||||
let mut next = || {
|
||||
seed ^= seed << 13;
|
||||
seed ^= seed >> 7;
|
||||
seed ^= seed << 17;
|
||||
seed
|
||||
};
|
||||
for round in 0..40_000 {
|
||||
let length = 6 + (next() % 30) as usize;
|
||||
let symbols: Vec<Sym> = (0..length).map(|_| ALPHABET[(next() % ALPHABET.len() as u64) as usize]).collect();
|
||||
play(&symbols, [Policy::All, Policy::Nothing, Policy::NoKill][round % 3]);
|
||||
}
|
||||
}
|
||||
|
|
@ -69,7 +69,6 @@ enum Event {
|
|||
/// A signal was attempted, delivered or not.
|
||||
Stopped(Stop),
|
||||
Asked(StdDuration),
|
||||
Waited,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
|
|
@ -77,9 +76,15 @@ enum Step {
|
|||
Line(String),
|
||||
/// Time passes with no output: the clock moves and the read times out.
|
||||
Idle(i64),
|
||||
/// The clock moves, then this line arrives.
|
||||
After(i64, String),
|
||||
/// The child is reaped with this code; its stream stays open.
|
||||
Exit(Option<i32>),
|
||||
/// Its stdout ends; the child is not reaped yet.
|
||||
Eof,
|
||||
}
|
||||
|
||||
use Step::{Idle, Line as Out};
|
||||
use Step::{Eof, Exit, Idle, Line as Out};
|
||||
|
||||
struct Scripted {
|
||||
steps: VecDeque<Step>,
|
||||
|
|
@ -90,6 +95,10 @@ struct Scripted {
|
|||
dies_on: Option<Stop>,
|
||||
/// Signals that cannot be delivered (no `kill` program, say).
|
||||
refuses: Vec<Stop>,
|
||||
/// When the steps run out it neither exits nor closes its stream: time just passes.
|
||||
hangs: bool,
|
||||
exited: bool,
|
||||
eof: bool,
|
||||
}
|
||||
|
||||
impl Child for Scripted {
|
||||
|
|
@ -101,7 +110,7 @@ impl Child for Scripted {
|
|||
self.log.borrow_mut().push(Event::InputClosed);
|
||||
}
|
||||
|
||||
fn next_line(&mut self, timeout: StdDuration) -> Read {
|
||||
fn next(&mut self, timeout: StdDuration) -> Read {
|
||||
self.log.borrow_mut().push(Event::Asked(timeout));
|
||||
match self.steps.pop_front() {
|
||||
Some(Step::Line(line)) => Read::Line(line),
|
||||
|
|
@ -109,7 +118,32 @@ impl Child for Scripted {
|
|||
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
|
||||
Read::Timeout
|
||||
}
|
||||
None => Read::Closed,
|
||||
Some(Step::After(seconds, line)) => {
|
||||
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(seconds));
|
||||
Read::Line(line)
|
||||
}
|
||||
Some(Step::Exit(code)) => {
|
||||
self.exited = true;
|
||||
Read::Exited(code)
|
||||
}
|
||||
Some(Step::Eof) => {
|
||||
self.eof = true;
|
||||
Read::Eof
|
||||
}
|
||||
// The stream is over: the child is reaped, then its stdout ends.
|
||||
None if self.hangs => {
|
||||
self.clock.0.set(self.clock.0.get() + ChronoDuration::seconds(timeout.as_secs() as i64));
|
||||
Read::Timeout
|
||||
}
|
||||
None if !self.exited => {
|
||||
self.exited = true;
|
||||
Read::Exited(self.code)
|
||||
}
|
||||
None if !self.eof => {
|
||||
self.eof = true;
|
||||
Read::Eof
|
||||
}
|
||||
None => panic!("the loop kept waiting on a child that ended"),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -120,14 +154,10 @@ impl Child for Scripted {
|
|||
}
|
||||
if self.dies_on.is_some_and(|weakest| how >= weakest) {
|
||||
self.steps.clear();
|
||||
self.hangs = false;
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn wait(&mut self) -> Option<i32> {
|
||||
self.log.borrow_mut().push(Event::Waited);
|
||||
self.code
|
||||
}
|
||||
}
|
||||
|
||||
struct Fake {
|
||||
|
|
@ -235,10 +265,14 @@ fn carry_with(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on_interrup
|
|||
}
|
||||
|
||||
fn carry_signalled(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Option<Stop>, refuses: &[Stop]) -> Ran {
|
||||
carry_scripted(seen, steps, code, dies_on, refuses, false)
|
||||
}
|
||||
|
||||
fn carry_scripted(seen: &Seen, steps: Vec<Step>, code: Option<i32>, dies_on: Option<Stop>, refuses: &[Stop], hangs: bool) -> Ran {
|
||||
let log = Rc::new(RefCell::new(Vec::new()));
|
||||
let clock = Rc::new(Fixed(Cell::new(at_start())));
|
||||
let fake = Fake {
|
||||
child: RefCell::new(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec() })),
|
||||
child: RefCell::new(Some(Scripted { steps: steps.into(), log: log.clone(), clock: clock.clone(), code, dies_on, refuses: refuses.to_vec(), hangs, exited: false, eof: false })),
|
||||
spawns: RefCell::new(Vec::new()),
|
||||
refuse: false,
|
||||
};
|
||||
|
|
@ -529,50 +563,59 @@ fn after_the_flush_the_grace_interrupts_exactly_once_and_not_a_second_early() {
|
|||
// A child that ignores it is interrupted once and still waited for.
|
||||
let stubborn = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(grace), Idle(grace), Idle(grace)], Some(0), false);
|
||||
assert_eq!(stubborn.count(&Event::Stopped(Stop::Interrupt)), 1);
|
||||
assert_eq!(stubborn.count(&Event::Waited), 1);
|
||||
assert_eq!(stubborn.ended().1, Some(0), "its own end is waited for");
|
||||
assert_eq!(FLUSH_GRACE_S, 900);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_wait_for_the_child_never_passes_the_grace_deadline_and_never_spins_after_the_interrupt() {
|
||||
let ran = go(vec![Out(at("m1", CEILING_TOKENS)), Idle(840), Idle(59), Idle(1)]);
|
||||
assert_eq!(ran.asked(), [60, 60, 60, 1, 60], "{:?}", ran.asked());
|
||||
// 60 until the grace is near, the one second left, then the escalation's own wait, then the drain after the child's exit.
|
||||
assert_eq!(ran.asked(), [60, 60, 60, 1, 60, 2], "{:?}", ran.asked());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_prefix_at_the_ceiling_plus_the_allowance_interrupts_at_once_and_one_short_of_it_does_not() {
|
||||
fn the_headroom_counts_the_requests_after_the_first_result_that_follows_the_flush_and_one_short_of_it_does_not_interrupt() {
|
||||
let hard = CEILING_TOKENS + FLUSH_HEADROOM_TOKENS;
|
||||
let short = go(vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard - 1))]);
|
||||
let short = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard - 1))]);
|
||||
assert_eq!(short.count(&Event::Stopped(Stop::Interrupt)), 0);
|
||||
assert_eq!(short.ended().0, Cause::FlushRequested);
|
||||
|
||||
let at_hard = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true);
|
||||
let at_hard = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Out(result_line()), Out(at("m2", hard)), Out(at("m3", hard + 1))], Some(130), true);
|
||||
assert_eq!(at_hard.count(&Event::Stopped(Stop::Interrupt)), 1);
|
||||
assert_eq!(at_hard.ended().0, Cause::Signalled(Stop::Interrupt));
|
||||
assert_eq!(FLUSH_HEADROOM_TOKENS, 60_000);
|
||||
|
||||
// Before that result the running turn's own requests are not the flush's turn: however far they grow, no interrupt.
|
||||
let same_turn = go(vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", hard)), Out(at("m3", hard + 500_000))]);
|
||||
assert_eq!(same_turn.count(&Event::Stopped(Stop::Interrupt)), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_gives_it_a_turn_before_any_interrupt() {
|
||||
let jump = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 430_000)), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))], Some(130), true);
|
||||
fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_the_running_turn_cannot_use_up_the_flushs_turn() {
|
||||
let jump = carry_with(
|
||||
&Seen::of(&with_run()),
|
||||
vec![Out(at("m1", 290_000)), Out(at("m2", 400_000)), Out(at("m3", 470_000)), Out(result_line()), Out(at("m4", 459_999)), Out(at("m5", 460_000)), Out(at("m6", 470_000))],
|
||||
Some(130),
|
||||
true,
|
||||
);
|
||||
let kinds: Vec<&str> = jump
|
||||
.events
|
||||
.iter()
|
||||
.filter_map(|event| match event {
|
||||
Event::Wrote(line) if line.contains("context ceiling") => Some("flush"),
|
||||
Event::Stopped(_) => Some("interrupt"),
|
||||
Event::InputClosed => Some("closed"),
|
||||
Event::Asked(_) => Some("read"),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
// Reads: m1, m2 (the jump, which writes the flush), then a read after it before anything else.
|
||||
assert_eq!(kinds[..4], ["read", "read", "flush", "read"], "{kinds:?}");
|
||||
// The headroom is measured from the prefix the flush was asked at (400k), not from the ceiling:
|
||||
// 430k is past 300k + 60k and does not interrupt; 459_999 does not; 460_000 does, once.
|
||||
// Reads: m1, m2 (the jump, which writes the flush), m3 (the running turn's 470k, past the headroom: no interrupt).
|
||||
assert_eq!(kinds[..5], ["read", "read", "flush", "read", "read"], "{kinds:?}");
|
||||
// The headroom is measured from the prefix the flush was asked at (400k) and counts the requests after the
|
||||
// first result only: 459_999 does not interrupt; 460_000 does, once, and m6 is never read.
|
||||
assert_eq!(jump.count(&Event::Stopped(Stop::Interrupt)), 1, "{kinds:?}");
|
||||
assert_eq!(kinds.iter().filter(|kind| **kind == "read").count(), 6, "interrupted on m5, the sixth read ends the stream: {kinds:?}");
|
||||
assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 430000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\n");
|
||||
assert_eq!(jump.out, "request 1 prefix 290000\nrequest 2 prefix 400000\nrequest 3 prefix 470000\nrequest 4 prefix 459999\nrequest 5 prefix 460000\n");
|
||||
assert_eq!(jump.ended().0, Cause::Signalled(Stop::Interrupt));
|
||||
assert_eq!(flushes(&jump), 1);
|
||||
}
|
||||
|
|
@ -581,7 +624,7 @@ fn a_jump_past_the_ceiling_plus_the_headroom_writes_the_flush_and_gives_it_a_tur
|
|||
fn a_child_that_ends_before_the_grace_is_never_interrupted() {
|
||||
let ran = go(vec![Out(at("m1", CEILING_TOKENS)), Out(result_line())]);
|
||||
assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 0);
|
||||
assert_eq!(ran.count(&Event::Waited), 1);
|
||||
assert_eq!(ran.ended().0, Cause::FlushRequested);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -649,7 +692,6 @@ fn a_child_that_ignores_a_signal_gets_the_next_one_after_the_bound_and_the_outco
|
|||
let killed = hung(Some(Stop::Kill), &[], vec![Idle(after - 1), Idle(1), Idle(after - 1), Idle(1), Idle(after)]);
|
||||
assert_eq!(stops(&killed), [Event::Stopped(Stop::Interrupt), Event::Stopped(Stop::Terminate), Event::Stopped(Stop::Kill)]);
|
||||
assert_eq!(killed.ended().0, Cause::Signalled(Stop::Kill));
|
||||
assert_eq!(killed.count(&Event::Waited), 1);
|
||||
|
||||
// One second short of the bound there is no second signal.
|
||||
let waiting = hung(None, &[], vec![Idle(after - 1)]);
|
||||
|
|
@ -675,13 +717,74 @@ fn a_signal_that_could_not_be_delivered_is_never_reported_as_sent() {
|
|||
let partly = hung(&[Stop::Terminate], vec![Idle(ESCALATE_AFTER_S)]);
|
||||
assert_eq!(partly.ended().0, Cause::Signalled(Stop::Interrupt));
|
||||
|
||||
// Not even SIGKILL is delivered: the run ends at once, the child is not waited for, and the label says so.
|
||||
let stuck = hung(&[Stop::Interrupt, Stop::Terminate, Stop::Kill], vec![Idle(ESCALATE_AFTER_S), Idle(ESCALATE_AFTER_S), Idle(ESCALATE_AFTER_S)]);
|
||||
// Not even SIGKILL is delivered and the child never ends: after the bound the run is over without it, and the label says so.
|
||||
let stuck = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Idle(FLUSH_GRACE_S)], Some(0), None, &[Stop::Interrupt, Stop::Terminate, Stop::Kill], true);
|
||||
assert_eq!(stuck.ended(), (Cause::Unstoppable, None, 1, CEILING_TOKENS));
|
||||
assert_eq!(stuck.count(&Event::Waited), 0);
|
||||
assert_eq!(stuck.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 3);
|
||||
assert_eq!(stuck.outcome.unwrap().exit_status(), 1);
|
||||
}
|
||||
|
||||
/// A clock that counts its reads.
|
||||
struct Counting(Fixed, Cell<u32>);
|
||||
|
||||
impl Clock for Counting {
|
||||
fn now(&self) -> chrono::DateTime<chrono::Utc> {
|
||||
self.1.set(self.1.get() + 1);
|
||||
self.0.now()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_clock_is_read_once_per_event_so_a_step_of_the_wall_clock_cannot_split_a_pass() {
|
||||
let mind = with_run();
|
||||
let log = Rc::new(RefCell::new(Vec::new()));
|
||||
let clock = Rc::new(Counting(Fixed(Cell::new(at_start())), Cell::new(0)));
|
||||
// The child's reads move the clock by far more than the grace, as a stepped wall clock would.
|
||||
struct Stepped(Rc<Counting>);
|
||||
impl Clock for Stepped {
|
||||
fn now(&self) -> chrono::DateTime<chrono::Utc> {
|
||||
let now = self.0.now();
|
||||
self.0.0.0.set(now + ChronoDuration::seconds(FLUSH_GRACE_S + 1));
|
||||
now
|
||||
}
|
||||
}
|
||||
let steps: VecDeque<Step> = vec![Out(at("m1", CEILING_TOKENS)), Out(at("m2", CEILING_TOKENS + 1)), Idle(0)].into();
|
||||
let child = Scripted { steps, log: log.clone(), clock: Rc::new(Fixed(Cell::new(at_start()))), code: Some(0), dies_on: None, refuses: vec![], hangs: false, exited: false, eof: false };
|
||||
let fake = Fake { child: RefCell::new(Some(child)), spawns: RefCell::new(Vec::new()), refuse: false };
|
||||
let outcome = carry(&Seen::of(&mind), &Stepped(clock.clone()), &fake, &mut Vec::new()).unwrap();
|
||||
let events = log.borrow().clone();
|
||||
let reads = events.iter().filter(|event| matches!(event, Event::Asked(_))).count() as u32;
|
||||
// One read before the first wait, then one per fact heard: never a second read inside a pass.
|
||||
assert_eq!(clock.1.get(), 1 + reads, "{events:?}");
|
||||
// The flush was written on the first request at time 0; the very next pass sees the stepped clock.
|
||||
// The flush pass itself has no interrupt, whatever the step: the interrupt is on a later event.
|
||||
let flush_at = events.iter().position(|event| matches!(event, Event::Wrote(line) if line.contains("context ceiling"))).unwrap();
|
||||
let stop_at = events.iter().position(|event| matches!(event, Event::Stopped(_))).unwrap();
|
||||
assert!(events[flush_at..stop_at].iter().any(|event| matches!(event, Event::Asked(_))), "a read between the flush and the interrupt: {events:?}");
|
||||
assert_eq!(outcome.exit_status(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_deadline_that_came_during_a_line_is_answered_without_waiting_for_the_next_poll() {
|
||||
// The result arrives after the grace has run out: the input closes, and the interrupt is due at once.
|
||||
let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Step::After(FLUSH_GRACE_S + 5, result_line())], Some(130), Some(Stop::Interrupt), &[], true);
|
||||
assert_eq!(ran.count(&Event::Stopped(Stop::Interrupt)), 1);
|
||||
assert_eq!(ran.count(&Event::InputClosed), 1);
|
||||
let asks = ran.asked();
|
||||
assert_eq!(asks[2], 0, "the wait after the result: the deadline is past, not a poll away: {asks:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_child_that_exited_is_never_signalled_however_long_its_stream_stays_open() {
|
||||
// Flush asked for, the child exits, then the grace and more pass with its stream still open.
|
||||
let ran = carry_scripted(&Seen::of(&with_run()), vec![Out(at("m1", CEILING_TOKENS)), Exit(Some(0)), Idle(1)], Some(0), None, &[], true);
|
||||
assert_eq!(ran.events.iter().filter(|event| matches!(event, Event::Stopped(_))).count(), 0);
|
||||
assert_eq!(ran.ended(), (Cause::FlushRequested, Some(0), 1, CEILING_TOKENS));
|
||||
// The stream ending first and the exit after: the same end.
|
||||
let eof_first = carry_with(&Seen::of(&with_run()), vec![Out(at("m1", 10)), Eof], Some(0), false);
|
||||
assert_eq!(eof_first.ended(), (Cause::Own, Some(0), 1, 10));
|
||||
}
|
||||
|
||||
// ---- how it ends --------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
|
|
@ -768,17 +871,27 @@ mod process {
|
|||
panic!("the script stayed busy");
|
||||
}
|
||||
|
||||
fn drain(child: &mut dyn Child) -> Vec<String> {
|
||||
let mut lines = Vec::new();
|
||||
/// Everything the child gives until it has been reaped and its stdout has ended.
|
||||
fn drain(child: &mut dyn Child) -> (Vec<String>, Option<i32>) {
|
||||
let (mut lines, mut code, mut eof) = (Vec::new(), None, false);
|
||||
loop {
|
||||
match child.next_line(StdDuration::from_secs(20)) {
|
||||
match child.next(StdDuration::from_secs(20)) {
|
||||
Read::Line(line) => lines.push(line),
|
||||
Read::Closed => return lines,
|
||||
Read::Eof => eof = true,
|
||||
Read::Exited(status) => code = Some(status),
|
||||
Read::Timeout => panic!("the fake claude hung"),
|
||||
}
|
||||
if let (true, Some(status)) = (eof, code) {
|
||||
return (lines, status);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a process of this id still exists.
|
||||
fn alive(pid: &str) -> bool {
|
||||
Command::new("kill").args(["-0", pid]).stderr(Stdio::null()).status().unwrap().success()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_child_gets_its_arguments_and_lines_in_order_and_ends_when_its_input_closes() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
|
@ -787,8 +900,7 @@ mod process {
|
|||
child.write_line("first");
|
||||
child.write_line("second");
|
||||
child.close_input();
|
||||
assert_eq!(drain(&mut *child), ["args: -p --max-turns 500", "got: first", "got: second", "bye"]);
|
||||
assert_eq!(child.wait(), Some(7));
|
||||
assert_eq!(drain(&mut *child), (["args: -p --max-turns 500", "got: first", "got: second", "bye"].map(String::from).to_vec(), Some(7)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -796,15 +908,13 @@ mod process {
|
|||
let dir = tempfile::tempdir().unwrap();
|
||||
let program = script(&dir, "kill -9 $$");
|
||||
let mut child = start(&program, &[]);
|
||||
assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed);
|
||||
assert_eq!(child.wait(), None);
|
||||
assert_eq!(drain(&mut *child), (Vec::new(), None));
|
||||
|
||||
let quiet = script(&dir, "exec sleep 5");
|
||||
let mut child = start(&quiet, &[]);
|
||||
assert_eq!(child.next_line(StdDuration::from_millis(100)), Read::Timeout);
|
||||
assert_eq!(child.next(StdDuration::from_millis(100)), Read::Timeout);
|
||||
child.stop(Stop::Interrupt);
|
||||
assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed);
|
||||
assert_eq!(child.wait(), None, "SIGINT ended the unprepared child");
|
||||
assert_eq!(drain(&mut *child), (Vec::new(), None), "SIGINT ended the unprepared child");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -812,10 +922,9 @@ mod process {
|
|||
let dir = tempfile::tempdir().unwrap();
|
||||
let program = script(&dir, "trap 'echo interrupted; exit 130' INT\necho ready\nwhile :; do sleep 0.05; done");
|
||||
let mut child = start(&program, &[]);
|
||||
assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Line("ready".into()));
|
||||
assert_eq!(child.next(StdDuration::from_secs(20)), Read::Line("ready".into()));
|
||||
child.stop(Stop::Interrupt);
|
||||
assert_eq!(drain(&mut *child), ["interrupted"]);
|
||||
assert_eq!(child.wait(), Some(130));
|
||||
assert_eq!(drain(&mut *child), (vec!["interrupted".to_string()], Some(130)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -835,8 +944,8 @@ mod process {
|
|||
fn close_input(&mut self) {
|
||||
self.0.close_input()
|
||||
}
|
||||
fn next_line(&mut self, timeout: StdDuration) -> Read {
|
||||
match self.0.next_line(StdDuration::from_secs(20).min(timeout)) {
|
||||
fn next(&mut self, timeout: StdDuration) -> Read {
|
||||
match self.0.next(StdDuration::from_secs(20).min(timeout)) {
|
||||
Read::Timeout => panic!("the fake claude went quiet"),
|
||||
read => read,
|
||||
}
|
||||
|
|
@ -844,9 +953,6 @@ mod process {
|
|||
fn stop(&mut self, how: Stop) -> bool {
|
||||
self.0.stop(how)
|
||||
}
|
||||
fn wait(&mut self) -> Option<i32> {
|
||||
self.0.wait()
|
||||
}
|
||||
}
|
||||
|
||||
struct Spawn(PathBuf);
|
||||
|
|
@ -856,31 +962,27 @@ mod process {
|
|||
}
|
||||
}
|
||||
|
||||
/// The real child as it is: a read may time out, which is how the drain after an exit is waited.
|
||||
struct Patient(PathBuf);
|
||||
impl Spawner for Patient {
|
||||
fn spawn(&self, args: &[String]) -> Result<Box<dyn Child>> {
|
||||
Ok(start(&self.0, args))
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_whole_run_through_the_real_child_flushes_at_the_ceiling_and_ends_on_the_result() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let lines = [at("m1", 10), at("m2", CEILING_TOKENS)].join("
|
||||
");
|
||||
let lines = [at("m1", 10), at("m2", CEILING_TOKENS)].join("\n");
|
||||
let result = result_line();
|
||||
// Prints two requests, waits for the flush on stdin, then the result, as a stand-in for a Self that obeys.
|
||||
let body = format!("cat <<'EOF'
|
||||
{lines}
|
||||
EOF
|
||||
read -t 5 start
|
||||
read -t 5 flush
|
||||
cat <<'EOF'
|
||||
{result}
|
||||
EOF
|
||||
while read more; do :; done
|
||||
exit 0");
|
||||
let body = format!("cat <<'EOF'\n{lines}\nEOF\nread -t 5 start\nread -t 5 flush\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 0");
|
||||
let program = script(&dir, &body);
|
||||
let mind = with_run();
|
||||
let mut out = Vec::new();
|
||||
let outcome = carry(&Seen::of(&mind), &SystemClock, &Spawn(program), &mut out).unwrap();
|
||||
assert_eq!(outcome, SelfRunOutcome::Ended { cause: Cause::FlushRequested, code: Some(0), requests: 2, prefix: 300_000 });
|
||||
assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10
|
||||
request 2 prefix 300000
|
||||
");
|
||||
assert_eq!(String::from_utf8(out).unwrap(), "request 1 prefix 10\nrequest 2 prefix 300000\n");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -889,12 +991,7 @@ request 2 prefix 300000
|
|||
let result = result_line();
|
||||
// An authentication or API failure: the error result is the first thing it prints, then it
|
||||
// waits for its input to end, as `claude -p` over stream-json does, and exits 1.
|
||||
let program = script(&dir, &format!("read -t 5 start
|
||||
cat <<'EOF'
|
||||
{result}
|
||||
EOF
|
||||
while read more; do :; done
|
||||
exit 1"));
|
||||
let program = script(&dir, &format!("read -t 5 start\ncat <<'EOF'\n{result}\nEOF\nwhile read more; do :; done\nexit 1"));
|
||||
let started = std::time::Instant::now();
|
||||
let mut out = Vec::new();
|
||||
let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Spawn(program), &mut out).unwrap();
|
||||
|
|
@ -904,24 +1001,65 @@ exit 1"));
|
|||
assert!(started.elapsed() < StdDuration::from_secs(20), "it ended promptly");
|
||||
}
|
||||
|
||||
/// The Yggdrasil probe shape: the child prints its result and exits, and a background
|
||||
/// process it started still holds its stdout, so the stream never ends. The run is over
|
||||
/// when the child is, and the actuator signals nothing: the child is gone.
|
||||
#[test]
|
||||
fn a_child_that_exits_while_a_descendant_holds_its_stdout_ends_the_run_without_a_signal() {
|
||||
for flushed in [false, true] {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let pids = dir.path().join("holder.pid");
|
||||
let (result, ceiling) = (result_line(), at("m1", CEILING_TOKENS));
|
||||
let before = if flushed { format!("cat <<'EOF'\n{ceiling}\nEOF\nread -t 5 start\nread -t 5 flush\n") } else { "read -t 5 start\n".to_string() };
|
||||
let program = script(&dir, &format!("{before}cat <<'EOF'\n{result}\nEOF\nsleep 30 &\necho $! > {}\nexit 0", pids.display()));
|
||||
let started = std::time::Instant::now();
|
||||
let mut out = Vec::new();
|
||||
let outcome = carry(&Seen::of(&with_run()), &SystemClock, &Patient(program), &mut out).unwrap();
|
||||
let holder = std::fs::read_to_string(&pids).unwrap().trim().to_string();
|
||||
let cause = if flushed { Cause::FlushRequested } else { Cause::Own };
|
||||
let (requests, prefix) = if flushed { (1, CEILING_TOKENS) } else { (0, 0) };
|
||||
assert_eq!(outcome, SelfRunOutcome::Ended { cause, code: Some(0), requests, prefix }, "flushed: {flushed}");
|
||||
assert!(alive(&holder), "the actuator signals no one after the child is gone");
|
||||
assert!(started.elapsed() < StdDuration::from_secs(20), "it ended on the child's exit: {:?}", started.elapsed());
|
||||
Command::new("kill").arg(&holder).status().unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn each_signal_reaches_a_child_that_ignores_the_weaker_ones_and_one_for_a_gone_child_is_not_delivered() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let program = script(&dir, "trap '' INT TERM
|
||||
echo ready
|
||||
while :; do sleep 0.05; done");
|
||||
let program = script(&dir, "trap '' INT TERM\necho ready\nwhile :; do sleep 0.05; done");
|
||||
let mut child = start(&program, &[]);
|
||||
assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Line("ready".into()));
|
||||
assert_eq!(child.next(StdDuration::from_secs(20)), Read::Line("ready".into()));
|
||||
assert!(child.stop(Stop::Interrupt) && child.stop(Stop::Terminate), "delivered, and ignored");
|
||||
assert_eq!(child.next_line(StdDuration::from_millis(300)), Read::Timeout, "still running");
|
||||
assert_eq!(child.next(StdDuration::from_millis(300)), Read::Timeout, "still running");
|
||||
assert!(child.stop(Stop::Kill));
|
||||
assert_eq!(child.next_line(StdDuration::from_secs(20)), Read::Closed);
|
||||
assert_eq!(child.wait(), None, "SIGKILL leaves no code");
|
||||
assert_eq!(drain(&mut *child), (Vec::new(), None), "SIGKILL leaves no code");
|
||||
|
||||
let quick = script(&dir, "exit 0");
|
||||
let mut gone = start(&quick, &[]);
|
||||
assert_eq!(gone.next_line(StdDuration::from_secs(20)), Read::Closed);
|
||||
assert_eq!(gone.wait(), Some(0));
|
||||
assert_eq!(drain(&mut *gone), (Vec::new(), Some(0)));
|
||||
assert!(!gone.stop(Stop::Interrupt), "no such process");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_signal_reaches_the_whole_process_group_the_child_started() {
|
||||
// Not SIGINT: a background job of a non-interactive shell ignores it, which is the shell's rule, not ours.
|
||||
for how in [Stop::Terminate, Stop::Kill] {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
// The child starts a descendant of its own and waits for it.
|
||||
let program = script(&dir, "sleep 300 &\necho $!\nwait");
|
||||
let mut child = start(&program, &[]);
|
||||
let Read::Line(holder) = child.next(StdDuration::from_secs(20)) else { panic!("no pid") };
|
||||
assert!(alive(&holder));
|
||||
assert!(child.stop(how), "{how:?} delivered to the group");
|
||||
drain(&mut *child);
|
||||
let mut waited = 0;
|
||||
while alive(&holder) && waited < 100 {
|
||||
std::thread::sleep(StdDuration::from_millis(50));
|
||||
waited += 1;
|
||||
}
|
||||
assert!(!alive(&holder), "{how:?} reached the descendant too");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -75,12 +75,13 @@ mind_body::queue: pub trait MindPort { fn instance (& self) -> & Slug ; fn query
|
|||
mind_body::queue: struct HuginnMind< 'a > pub-fields[]
|
||||
mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq , PartialOrd , Ord)] pub enum Stop { Interrupt , Terminate , Kill , }
|
||||
mind_body::self_run: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Cause { Own , FlushRequested , Signalled (Stop) , Unstoppable , }
|
||||
mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Read { Line (String) , Timeout , Closed , }
|
||||
mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Read { Line (String) , Timeout , Eof , Exited (Option < i32 >) , }
|
||||
mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum SelfRunOutcome { NoRun , Ended { cause : Cause , code : Option < i32 > , requests : u64 , prefix : u64 } , }
|
||||
mind_body::self_run: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Usage { Request { id : String , prefix : u64 } , Result , Other , Unreadable , }
|
||||
mind_body::self_run: const AGENT : & str
|
||||
mind_body::self_run: const CEILING_TOKENS : u64
|
||||
mind_body::self_run: const ESCALATE_AFTER_S : i64
|
||||
mind_body::self_run: const EXIT_DRAIN_S : i64
|
||||
mind_body::self_run: const FLUSH_GRACE_S : i64
|
||||
mind_body::self_run: const FLUSH_HEADROOM_TOKENS : u64
|
||||
mind_body::self_run: const FLUSH_MESSAGE : & str
|
||||
|
|
@ -97,7 +98,7 @@ mind_body::self_run: impl SelfRunOutcome: fn exit_status (& self) -> i32
|
|||
mind_body::self_run: impl Spawner for ClaudeSpawner
|
||||
mind_body::self_run: impl Stop
|
||||
mind_body::self_run: impl fmt :: Display for SelfRunOutcome
|
||||
mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next_line (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; fn wait (& mut self) -> Option < i32 > ; }
|
||||
mind_body::self_run: pub trait Child { fn write_line (& mut self , line : & str) ; fn close_input (& mut self) ; fn next (& mut self , timeout : StdDuration) -> Read ; fn stop (& mut self , how : Stop) -> bool ; }
|
||||
mind_body::self_run: pub trait Spawner { fn spawn (& self , args : & [String]) -> Result < Box < dyn Child > > ; }
|
||||
mind_body::self_run: struct ClaudeSpawner pub-fields[]
|
||||
mind_body::wake: # [derive (Clone , Debug , PartialEq , Eq)] pub enum WakeOutcome { Held { reason : HeldReason } , NotYet { until : DateTime < Utc > } , Busy { run : PipelineRef } , Launched { run : PipelineRef } , Refused (Declined) , }
|
||||
|
|
|
|||
Loading…
Reference in a new issue