diff --git a/crates/brokerd/src/container.rs b/crates/brokerd/src/container.rs index 5f72b5f..55ed83d 100644 --- a/crates/brokerd/src/container.rs +++ b/crates/brokerd/src/container.rs @@ -3,16 +3,15 @@ //! said goes only to `brokerd`'s log. M3b spec, section 6. use std::ffi::OsString; -use std::io::{Read, Write}; use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, ExitStatus, Stdio}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -use std::thread::JoinHandle; use std::time::{Duration, Instant}; use crate::config::Runner; +use crate::pipes::Io; use crate::podman; use crate::runner::{RunError, RunOutput, RunSpec, Runtime}; @@ -349,84 +348,3 @@ impl<'a> EgressGuard<'a> { } } } - -/// Step 4: the three threads that feed and read one container, so no pipe can block another. -struct Io { - writer: Option>, - stdout: Option, bool)>>, - stderr: Option, bool)>>, -} - -impl Io { - /// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each: - /// write `input` to standard input and then drop it; `read_capped` standard output with `cap`; - /// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread. - fn start(child: &mut Child, input: String, cap: usize) -> Io { - let writer = child.stdin.take().map(|mut stdin| { - std::thread::spawn(move || { - let _ = stdin.write_all(input.as_bytes()); - }) - }); - let stdout = child - .stdout - .take() - .map(|stdout| std::thread::spawn(move || read_capped(stdout, cap))); - let stderr = child - .stderr - .take() - .map(|stderr| std::thread::spawn(move || read_capped(stderr, STDERR_KEPT))); - Io { - writer, - stdout, - stderr, - } - } - - /// Step 6: join the three threads. A thread that is missing or panicked counts as empty - /// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was - /// more, and the kept standard error decoded with `from_utf8_lossy`. - fn finish(self) -> (Vec, bool, String) { - if let Some(handle) = self.writer { - let _ = handle.join(); - } - let (out, truncated) = match self.stdout { - Some(handle) => handle.join().ok().unwrap_or_default(), - None => (Vec::new(), false), - }; - let err = match self.stderr { - Some(handle) => { - let (bytes, _) = handle.join().ok().unwrap_or_default(); - String::from_utf8_lossy(&bytes).into_owned() - } - None => String::new(), - }; - (out, truncated, err) - } -} - -/// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past -/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`. -fn read_capped(from: impl Read, cap: usize) -> (Vec, bool) { - let mut from = from; - let mut kept: Vec = Vec::new(); - let mut truncated = false; - let mut chunk = [0u8; 8192]; - loop { - let n = match from.read(&mut chunk) { - Ok(0) => break, - Ok(n) => n, - Err(_) => break, - }; - let remaining = cap - kept.len(); - if remaining == 0 { - truncated = true; - } else { - let take = remaining.min(n); - kept.extend_from_slice(&chunk[..take]); - if n > take { - truncated = true; - } - } - } - (kept, truncated) -} diff --git a/crates/brokerd/src/lib.rs b/crates/brokerd/src/lib.rs index 3f08020..f6769d9 100644 --- a/crates/brokerd/src/lib.rs +++ b/crates/brokerd/src/lib.rs @@ -9,6 +9,7 @@ pub mod config; pub mod container; pub mod grants; pub mod ledger; +pub mod pipes; pub mod podman; pub mod policy; pub mod runner; diff --git a/crates/brokerd/src/pipes.rs b/crates/brokerd/src/pipes.rs new file mode 100644 index 0000000..dff606d --- /dev/null +++ b/crates/brokerd/src/pipes.rs @@ -0,0 +1,86 @@ +//! The three pipes of one container: its arguments go in on standard input, its output and errors +//! come back, each on its own thread so no pipe can block another. + +use std::io::{Read, Write}; +use std::process::Child; +use std::thread::JoinHandle; + +/// Step 4: the three threads that feed and read one container, so no pipe can block another. +pub(crate) struct Io { + writer: Option>, + stdout: Option, bool)>>, + stderr: Option, bool)>>, +} + +impl Io { + /// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each: + /// write `input` to standard input and then drop it; `read_capped` standard output with `cap`; + /// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread. + pub(crate) fn start(child: &mut Child, input: String, cap: usize) -> Io { + let writer = child.stdin.take().map(|mut stdin| { + std::thread::spawn(move || { + let _ = stdin.write_all(input.as_bytes()); + }) + }); + let stdout = child + .stdout + .take() + .map(|stdout| std::thread::spawn(move || read_capped(stdout, cap))); + let stderr = child.stderr.take().map(|stderr| { + std::thread::spawn(move || read_capped(stderr, crate::container::STDERR_KEPT)) + }); + Io { + writer, + stdout, + stderr, + } + } + + /// Step 6: join the three threads. A thread that is missing or panicked counts as empty + /// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was + /// more, and the kept standard error decoded with `from_utf8_lossy`. + pub(crate) fn finish(self) -> (Vec, bool, String) { + if let Some(handle) = self.writer { + let _ = handle.join(); + } + let (out, truncated) = match self.stdout { + Some(handle) => handle.join().ok().unwrap_or_default(), + None => (Vec::new(), false), + }; + let err = match self.stderr { + Some(handle) => { + let (bytes, _) = handle.join().ok().unwrap_or_default(); + String::from_utf8_lossy(&bytes).into_owned() + } + None => String::new(), + }; + (out, truncated, err) + } +} + +/// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past +/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`. +pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec, bool) { + let mut from = from; + let mut kept: Vec = Vec::new(); + let mut truncated = false; + let mut chunk = [0u8; 8192]; + loop { + let n = match from.read(&mut chunk) { + Ok(0) => break, + Ok(n) => n, + Err(_) => break, + }; + let remaining = cap - kept.len(); + if remaining == 0 { + truncated = true; + } else { + let take = remaining.min(n); + kept.extend_from_slice(&chunk[..take]); + if n > take { + truncated = true; + } + } + } + (kept, truncated) +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 9fe57d3..7127823 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M3b/14-brokerd-pipes-module | 2026-09-23 | done | 1 | pass | none | Pure move: cut `struct Io`, `impl Io { start, finish }`, and `fn read_capped` from `container.rs` and pasted them into `crates/brokerd/src/pipes.rs` with bodies unchanged and `pub(crate)` visibility; `Io::start` now calls `crate::container::STDERR_KEPT` (the constant stays in `container.rs` since the tests import it). Added the module doc comment and the `use std::io::{Read, Write}`, `use std::process::Child`, `use std::thread::JoinHandle` lines. `lib.rs` gained `pub mod pipes;` between `ledger` and `podman`. `container.rs` gained `use crate::pipes::Io;` and lost the `Read, Write` and `JoinHandle` imports the compiler reported unused; nothing else changed. `cargo fmt --all` reflowed the `STDERR_KEPT` call line in `start` to wrap. `cargo check`/clippy clean; container 11, container_egress 6, serve_runner 2 pass; `grep "struct Io\|fn read_capped" container.rs` prints nothing and `container.rs` is 350 lines. `make gate` prints `gate: ok` on the first run. | ? | | M3b/13-brokerd-serve-runner | 2026-09-23 | done | 1 | pass | none | The `Podman` runtime was already written in task 12 (`crates/brokerd/src/container.rs`), so this task only wired it into `serve`. In `main.rs`: after the config loads, added `let log: Arc = Arc::new(|line| eprintln!("{line}"))`; chose the runtime and its notice with a match on `&cfg.runner` — `Some(runner)` captures `runner.image` first then `Box::new(Podman::new(runner.clone(), cfg.egress_dir(), Arc::clone(&log)))` with notice `tools run in containers from {image}`; `None` → `Box::new(Refusing)` with notice `no [runner] section: every tool call is refused`. Passed `runtime` and `log` to `serve::start` (its signature needed no change — it already took `Box` and `Arc`), and printed the notice with `eprintln!` after the `serving tools on … and approvals on …` line. Added `use brokerd::container::Podman` and `Runtime` to the imports. Copied `tests/serve_runner.rs`; the 2 new tests plus the 9 serve tests pass; `make gate` prints `gate: ok` on the first run. | ? | | M3b/12-brokerd-egress | 2026-09-23 | done | 1 | pass | none | Wrote `crates/brokerd/src/container.rs` (432 lines). Added consts `EGRESS_WAIT` (5s) and `EGRESS_POLL` (20ms); `Podman` gained private `egress_wait` (set to `EGRESS_WAIT` by `new`) and the public `with_egress_wait`. Added private `cannot_launch(podman, e)` (the task-11 step 3 log `brokerd: cannot start {path}: {e}\n{RUNBOOK}`), which `spawn` now calls, and `cannot_make(dir, e)` returning `Err(Unavailable(CANNOT_START))` with `brokerd: cannot make {dir}: {e}\n{RUNBOOK}`. `Runtime::run` now branches on `spec.egress()`: `None` runs `tool_args(spec, &runner, &name, None)` as before; `Some(hosts)` computes `dir = egress_dir.join(&name)`, calls `start_egress`, then `tool_args(spec, &runner, &name, Some(&dir))`. `start_egress` creates the `EgressGuard` first (so every return cleans up), then step 1: `DirBuilder::new().recursive(true).mode(0o700).create(egress_dir)`, `set_permissions(egress_dir, 0o700)` anyway, `remove_dir_all(dir)` if it exists (NotFound ok), `DirBuilder::new().mode(0o700).create(dir)` non-recursively, each failure via `cannot_make`; step 2: `Command::new(&runner.podman).args(egress_args).stdin(null).stdout(null).stderr(piped()).output()`, non-success logs `brokerd: podman could not start {name}-egress: {stderr}\n{RUNBOOK}` and returns Unavailable, launch error uses `cannot_launch`; step 3: wait for `dir.join("egress.sock")` every 20 ms until `egress_wait`, then log `brokerd: {name}-egress did not make its socket within {ms} ms\n{RUNBOOK}` and return Unavailable; step 4 `Ok(_guard)`. `EgressGuard<'a>` holds `&'a Podman`, the container name `-egress` and the directory; its `Drop` runs `podman rm -f -egress` (the task-11 helper) then `remove_dir_all(dir)` (NotFound ok, else log `brokerd: cannot remove {dir}: {e}`). Added `use std::os::unix::fs::{DirBuilderExt, PermissionsExt};`. `cargo fmt` put the new `impl Podman` block after `impl Runtime`. 6 egress + 11 container tests pass ten runs in a row. First gate failed on `start_egress` being inside `impl Runtime` (not a trait member) and missing `DirBuilderExt`/`PermissionsExt`; then on the guard being created at the end instead of the start (the two "nothing left" tests need the `rm -f -egress` call on a failed proxy), fixed by moving `EgressGuard::new` to the top and returning it; then on `unused_variable` for the drop guard and `mismatched_lifetime_syntaxes` on the return type (fixed to `EgressGuard<'_>`), the guard renamed `_guard` (1.98 still lints drop-only bindings). `make gate` prints `gate: ok`. | ? | | M3b/11-brokerd-container | 2026-09-23 | done | 2 | fail | none | Wrote `crates/brokerd/src/container.rs`: the fixed-sentence constants (`COULD_NOT_RUN`, `CANNOT_START`, `KILLED`, `TIMED_OUT`, `UNEXPECTED`), `RUNBOOK`, `POLL`, `STDERR_KEPT` and the `Log` type; `Podman { runner, egress_dir, log, next: AtomicU64 }` with `new` (`next` starts at 0) and the public `egress_dir`. `Runtime::run` = `next.fetch_add` for the container number, `podman::container_name`, `podman::tool_args(spec, &runner, &name, None)`, `spec.arguments().canonical_json()`, `runner.time_limit(tool)`, then `run_container` (already written). `spawn` = `Command::new(&runner.podman)` with all three streams piped, on failure log `brokerd: cannot start {path}: {e}\n{RUNBOOK}` and return None. `Io::start` takes the three pipes and starts one thread each: write `input` then drop stdin, `read_capped` stdout with `cap`, `read_capped` stderr with `STDERR_KEPT`; `finish` joins all three (a missing or panicked thread counts as empty via `join().ok()`/`unwrap_or_default()`). `read_capped` reads past the cap with an 8 KiB buffer, keeping the first `cap` bytes and setting `truncated`, taking `remaining.min(n)` so it never indexes past what it kept. `wait` loops `child.try_wait()`, and once `limit` has elapsed runs `podman kill ` then `podman rm -f ` (each via the private `podman` helper — `.status()` with the three streams null, logging a line on non-success), then `child.kill()`/`child.wait()`, returning None. `podman` helper uses `&self.runner.podman`, not a `podman` field. `answer` follows the table: None→TIMED_OUT (log a line naming the container); exit 0/1→Ok with content `from_utf8_lossy(out)`; exit 2→COULD_NOT_RUN (log stderr); 125/126/127→CANNOT_START (log stderr then `\n{RUNBOOK}`); 137→KILLED; anything else incl. signal→UNEXPECTED (log status and stderr). The tool's output never enters a `RunError`; only the six constants do. Deleted the skeleton paragraph. Added `pub mod container;` to lib.rs after `config`. Copied `tests/support/fake_podman.rs` and `tests/container.rs`. `container` 11 passed ten runs; first gate failed on clippy `manual_range_patterns` (`125|126|127` → `125..=127`), fixed; `make gate` prints `gate: ok`. | ? |