boxmaker/docs/plans/M3b/15-brokerd-pipes-grace.md

M3b task 15: no thread panics, and a grace period for the output

Branch: m3b (run git switch m3b; git status --short must be empty, otherwise stop) Commit subject: brokerd: start pipe threads safely and collect output within a grace period

Goal

Two defects from the M3b review, both in pipes.rs (task 14):

• Finding 3. std::thread::spawn panics when the system cannot make a thread. The panic drops the Child without stopping the container, which then runs on without its time limit. Use std::thread::Builder, which returns an error instead, and stop the container on that error.
• Finding 4. Io::finish joins the readers with no limit. Each reader stops only at its pipe's end, so if any other process still holds a pipe, brokerd waits for it: the call's time limit does not bound it. After the container has ended, give the output a grace period of 2 s, then abandon what has not ended.

Files

• Copy: crates/brokerd/tests/container_grace.rs
• Modify: crates/brokerd/src/pipes.rs, crates/brokerd/src/container.rs, docs/implementer-log.md

pipes.rs, new shape

use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
use std::thread::Builder;
use std::time::{Duration, Instant};

/// How long, after the container has ended, its output may take to reach its end.
pub const GRACE: Duration = Duration::from_secs(2);

/// What came back.
pub(crate) struct Finished {
    pub out: Vec<u8>,
    pub truncated: bool,
    pub err: String,
    /// A pipe had not ended by the grace deadline: something outside the container holds it.
    pub open: bool,
}

pub(crate) struct Io {
    stdout: Option<Receiver<(Vec<u8>, bool)>>,
    stderr: Option<Receiver<(Vec<u8>, bool)>>,
}

impl Io {
    pub(crate) fn start(child: &mut Child, input: String, cap: usize, err_cap: usize)
        -> std::io::Result<Io>;
    pub(crate) fn finish(self, grace: Duration) -> Finished;
}

pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool);  // as now

Write each function in turn, running cargo check -p brokerd after each.

Io::start

1. If child.stdin.take() is Some(stdin): start a thread with Builder::new().spawn(…) that writes input with write_all (ignore its error) and then drops stdin. Do not keep its handle and never join it: it ends when the input is written or the pipe breaks. A spawn error → return it (?).
2. For standard output with cap, and then standard error with err_cap: if the pipe is Some, make a channel (mpsc::channel()), start a thread with Builder::new().spawn(…) that runs read_capped(pipe, cap) and sends the result (ignore a send error), and keep the Receiver. A spawn error → return it. Write a small private function fn reader(pipe: impl Read + Send + 'static, cap: usize) -> std::io::Result<Receiver<(Vec<u8>, bool)>> and use it for both.
3. Ok(Io { stdout, stderr }).

Io::finish(grace)

One deadline for both readers: let until = Instant::now() + grace;. For each receiver (standard output first), a small private function collect(rx, until) -> (Vec<u8>, bool, bool):

• None (no pipe) → (empty, false, false).
• rx.recv_timeout(until.saturating_duration_since(Instant::now())):
  • Ok((bytes, truncated)) → (bytes, truncated, false);
  • Err(RecvTimeoutError::Timeout) → (empty, false, true): open;
  • Err(RecvTimeoutError::Disconnected) (the reader panicked) → (empty, false, false).

Return Finished { out, truncated, err: String::from_utf8_lossy(&err_bytes).into_owned(), open: stdout_open || stderr_open }.

read_capped

Unchanged in what it does, but no indexing and no subtraction that could wrap: keep let take = n.min(cap.saturating_sub(kept.len()));, copy with kept.extend_from_slice(chunk.get(..take).unwrap_or_default());, and set truncated when take < n. Retry ErrorKind::Interrupted; stop on any other error.

container.rs

1. A new constant after TIMED_OUT: pub const OUTPUT_OPEN: &str = "the tool left its output open";
2. use crate::pipes::{GRACE, Io};
3. In run_container, replace the lines from let io = Io::start(…) to the end of the function with exactly this:

let io = match Io::start(&mut child, input, cap, STDERR_KEPT) {
    Ok(io) => io,
    Err(e) => {
        // The container may be running: stop it before answering.
        self.podman(&["kill", name]);
        self.podman(&["rm", "-f", name]);
        let _ = child.kill();
        let _ = child.wait();
        (self.log)(&format!("brokerd: cannot start a thread for {name}: {e}\n{RUNBOOK}"));
        return Err(RunError::Unavailable(CANNOT_START.to_string()));
    }
};
let status = self.wait(&mut child, name, started, limit);
let done = io.finish(GRACE);
if done.open && status.is_some() {
    (self.log)(&format!(
        "brokerd: {name} ended but something still holds its output; it was abandoned"
    ));
    return Err(RunError::Failed(OUTPUT_OPEN.to_string()));
}
self.answer(name, status, done.out, done.truncated, &done.err)

When the time limit was passed (status is None), an open pipe changes nothing: the answer is still TIMED_OUT, from answer.

Also: grep -n "thread::spawn" crates/brokerd/src/ must print nothing when you are done.

Steps

• 1. Copy. git switch m3b, then cp docs/plans/M3b/files/crates/brokerd/tests/container_grace.rs crates/brokerd/tests/
• 2. See it fail. cargo test -p brokerd --test container_grace. Expected: it does not compile (OUTPUT_OPEN and pipes::GRACE do not exist).
• 3. Write pipes.rs, one function at a time, then container.rs. Run cargo fmt --all.
• 4. See it pass. cargo test -p brokerd --test container_grace. Expected: 4 passed, in about 4 s (two tests wait out the grace period on purpose). Then cargo test -p brokerd --test container --test container_egress: 11 and 6 passed. Run all three five times.
• 5. Run the gate. make gate. Expected last line: gate: ok.
• 6. Log and commit. git add crates/brokerd docs/implementer-log.md Cargo.lock && git commit

Done when

• The three suites pass five times running; no thread::spawn in crates/brokerd/src/; make gate prints gate: ok.

Stop and report if

• A test takes much longer than stated, or passes only sometimes.