Skip to main content

max / makenotwork

bento: bound request bodies, stream logs, protocol-version the agent wire (Run 2 S6 + wiring) S6 / resource bounds: - DefaultBodyLimit on the daemon router (64 KiB) and the agent router (1 MiB) so oversized POSTs can't buffer freely (axum's 2 MB default applied even to the agent's unbounded /pull before this series). - GET /logs streams the file in 64 KiB chunks instead of tokio::fs::read of the whole (verbose-build) log into RAM. Resolves CHRONIC-2 (the daemon half; the agent /pull half was streamed in CR1). - Cap the AgentRpc NDJSON reader: a stream that sends >16 MiB with no newline is rejected instead of growing the buffer unbounded (OOM vector). Wiring: - Protocol versioning: Frame gains a #[serde(other)] Unknown so an older client ignores a newer agent's unknown frame instead of failing the run (the build- fatal skew the audit flagged); HealthResponse carries the protocol version (PROTOCOL_VERSION) for preflight skew detection. - error.rs redacts 500 bodies (Db/Other logged in full, generic message returned) and standardizes a JSON {"error": ...} envelope; client-error messages preserved. - Document the agent exit-code 0..=255 contract (success/non-zero preserved; a negative sentinel collapses) at exit_status. bento-daemon 46 tests, ops-exec 38, sando 146+30 green; clippy -D warnings clean across all touched crates. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 01:36 UTC
Commit: 7eae7247a8c89ad546e7d6114b7c4831fdc28f7f
Parent: 6ce0ebe
7 files changed, +155 insertions, -18 deletions
M bento/daemon/Cargo.lock +1
@@ -178,6 +178,7 @@ dependencies = [
178 178 "tempfile",
179 179 "thiserror",
180 180 "tokio",
181 + "tokio-stream",
181 182 "toml",
182 183 "tower",
183 184 "tracing",
M bento/daemon/Cargo.toml +2 -1
@@ -12,7 +12,8 @@ path = "src/main.rs"
12 12 ops-core = { path = "../../shared/ops-core" }
13 13 ops-exec = { path = "../../shared/ops-exec", features = ["rpc"] }
14 14 axum = { version = "0.8.8", features = ["macros", "ws"] }
15 - tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread", "net", "signal", "fs", "process", "sync"] }
15 + tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread", "net", "signal", "fs", "process", "sync", "io-util"] }
16 + tokio-stream = "0.1"
16 17 serde = { version = "1.0.228", features = ["derive"] }
17 18 serde_json = "1"
18 19 toml = "0.8"
M bento/daemon/src/error.rs +15 -5
@@ -15,12 +15,22 @@ pub enum Error {
15 15
16 16 impl IntoResponse for Error {
17 17 fn into_response(self) -> Response {
18 - let status = match &self {
19 - Error::NotFound => StatusCode::NOT_FOUND,
20 - Error::BadRequest(_) => StatusCode::BAD_REQUEST,
21 - _ => StatusCode::INTERNAL_SERVER_ERROR,
18 + // Client errors carry their (already user-facing) message; server errors
19 + // are logged in full but return a generic body so internal detail (SQL
20 + // text, anyhow chains, file paths) never leaks to the client.
21 + let (status, message) = match &self {
22 + Error::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()),
23 + Error::BadRequest(m) => (StatusCode::BAD_REQUEST, m.clone()),
24 + Error::Db(e) => {
25 + tracing::error!(error = %e, "internal db error");
26 + (StatusCode::INTERNAL_SERVER_ERROR, "internal server error".to_string())
27 + }
28 + Error::Other(e) => {
29 + tracing::error!(error = format!("{e:#}"), "internal error");
30 + (StatusCode::INTERNAL_SERVER_ERROR, "internal server error".to_string())
31 + }
22 32 };
23 - (status, self.to_string()).into_response()
33 + (status, axum::Json(serde_json::json!({ "error": message }))).into_response()
24 34 }
25 35 }
26 36
M bento/daemon/src/routes.rs +80 -9
@@ -41,6 +41,9 @@ pub fn router(state: AppState) -> Router {
41 41 .merge(open)
42 42 .with_state(state)
43 43 .route("/metrics", get(crate::metrics::render).with_state(prom))
44 + // Build/retry bodies are small JSON; cap request bodies well below
45 + // axum's 2 MB default so a flood of oversized POSTs can't buffer freely.
46 + .layer(axum::extract::DefaultBodyLimit::max(64 * 1024))
44 47 }
45 48
46 49 /// Bearer-token gate for the build triggers. No token configured -> pass
@@ -231,15 +234,39 @@ async fn get_step_log(
231 234 return Err(Error::NotFound);
232 235 }
233 236 let path = s.cfg.logs_root.join(&app).join(&version).join(&target).join(format!("{step}.log"));
234 - match tokio::fs::read(&path).await {
235 - Ok(bytes) => Ok((
236 - [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
237 - bytes,
238 - )
239 - .into_response()),
240 - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(Error::NotFound),
241 - Err(e) => Err(Error::Other(e.into())),
242 - }
237 + // Stream the log in chunks rather than reading the whole (potentially large,
238 + // verbose-build) file into memory.
239 + let file = match tokio::fs::File::open(&path).await {
240 + Ok(f) => f,
241 + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Err(Error::NotFound),
242 + Err(e) => return Err(Error::Other(e.into())),
243 + };
244 + let (tx, rx) = tokio::sync::mpsc::channel::<std::result::Result<axum::body::Bytes, std::io::Error>>(8);
245 + tokio::spawn(async move {
246 + use tokio::io::AsyncReadExt;
247 + let mut file = file;
248 + let mut buf = vec![0u8; 64 * 1024];
249 + loop {
250 + match file.read(&mut buf).await {
251 + Ok(0) => break,
252 + Ok(n) => {
253 + if tx.send(Ok(axum::body::Bytes::copy_from_slice(&buf[..n]))).await.is_err() {
254 + break;
255 + }
256 + }
257 + Err(e) => {
258 + let _ = tx.send(Err(e)).await;
259 + break;
260 + }
261 + }
262 + }
263 + });
264 + let body = axum::body::Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx));
265 + Ok((
266 + [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
267 + body,
268 + )
269 + .into_response())
243 270 }
244 271
245 272 async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
@@ -407,6 +434,50 @@ targets = ["linux/x86_64"]
407 434 }
408 435
409 436 #[tokio::test]
437 + async fn step_log_streams_existing_file() {
438 + let tmp = tempfile::tempdir().unwrap();
439 + let state = test_state(tmp.path()).await;
440 + // Write a log at the path the route resolves.
441 + let dir = state.cfg.logs_root.join("goingson").join("0.4.1").join("linux-x86_64");
442 + std::fs::create_dir_all(&dir).unwrap();
443 + std::fs::write(dir.join("build.log"), b"compiling\nlinked\n").unwrap();
444 + let app = router(state);
445 + let resp = app
446 + .oneshot(
447 + Request::builder()
448 + .uri("/logs/goingson/0.4.1/linux-x86_64/build")
449 + .body(Body::empty())
450 + .unwrap(),
451 + )
452 + .await
453 + .unwrap();
454 + assert_eq!(resp.status(), StatusCode::OK);
455 + assert_eq!(body_string(resp).await, "compiling\nlinked\n");
456 + }
457 +
458 + #[tokio::test]
459 + async fn bad_request_body_is_a_json_error_envelope() {
460 + let tmp = tempfile::tempdir().unwrap();
461 + let app = router(test_state(tmp.path()).await);
462 + let resp = app
463 + .oneshot(
464 + Request::builder()
465 + .method("POST")
466 + .uri("/build")
467 + .header("content-type", "application/json")
468 + .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
469 + .unwrap(),
470 + )
471 + .await
472 + .unwrap();
473 + assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
474 + // JSON envelope, message preserved.
475 + let body = body_string(resp).await;
476 + let v: serde_json::Value = serde_json::from_str(&body).unwrap();
477 + assert!(v["error"].as_str().unwrap().contains("unknown app"));
478 + }
479 +
480 + #[tokio::test]
410 481 async fn build_rejects_unshipped_target_as_bad_request() {
411 482 // windows/x86_64 is a valid target but the test app doesn't ship it.
412 483 // That's a client error (400), not a server error (500).
M shared/ops-exec/src/agent.rs +5 -2
@@ -28,7 +28,7 @@ use crate::executor::Executor;
28 28 use crate::remote::LogSink;
29 29 use crate::step::{Action, ObserveKind};
30 30 use crate::transport::LocalExec;
31 - use crate::wire::{Frame, HealthResponse, RunRequest};
31 + use crate::wire::{Frame, HealthResponse, PROTOCOL_VERSION, RunRequest};
32 32 use anyhow::{Context, Result};
33 33 use async_trait::async_trait;
34 34 use serde::Deserialize;
@@ -221,6 +221,9 @@ pub fn router(state: AgentState) -> axum::Router {
221 221 .route("/run", post(run))
222 222 .route("/pull", get(pull))
223 223 .with_state(state)
224 + // A RunRequest is a single Step (argv + optional shell script); cap the
225 + // body so a malformed/oversized POST can't buffer freely.
226 + .layer(axum::extract::DefaultBodyLimit::max(1024 * 1024))
224 227 }
225 228
226 229 /// Resolve the caller's tailnet identity and effective grant, or an early
@@ -256,7 +259,7 @@ async fn health(
256 259 ),
257 260 Err(_) => (Vec::new(), Vec::new()),
258 261 };
259 - axum::Json(HealthResponse { ok: true, actuate, observe }).into_response()
262 + axum::Json(HealthResponse { ok: true, version: PROTOCOL_VERSION, actuate, observe }).into_response()
260 263 }
261 264
262 265 async fn run(
M shared/ops-exec/src/rpc.rs +16 -1
@@ -55,7 +55,11 @@ impl AgentRpc {
55 55 }
56 56
57 57 /// Reconstruct an `ExitStatus` from a raw exit code (Unix wait-status encoding:
58 - /// the code lives in the high byte).
58 + /// the code lives in the high byte). The agent reports a process exit code,
59 + /// which is 0..=255; `.success()` (code 0) and the non-zero distinction — all a
60 + /// recipe branches on — are preserved exactly. A negative/out-of-range code
61 + /// (e.g. the agent's `-1` "terminated by signal" sentinel) collapses into the
62 + /// low byte, so it reads back as non-zero but not its original value.
59 63 #[cfg(unix)]
60 64 fn exit_status(code: i32) -> ExitStatus {
61 65 use std::os::unix::process::ExitStatusExt;
@@ -94,6 +98,10 @@ impl Executor for AgentRpc {
94 98 let mut captured: Vec<u8> = Vec::new();
95 99 let mut exit_code: Option<i32> = None;
96 100
101 + // A single NDJSON frame must fit in this much; a stream that sends this
102 + // many bytes with no newline is malformed (and would otherwise grow the
103 + // buffer unbounded — an OOM vector against the daemon).
104 + const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
97 105 while let Some(chunk) = stream.next().await {
98 106 let chunk = chunk.context("reading /run stream")?;
99 107 buf.extend_from_slice(&chunk);
@@ -112,8 +120,15 @@ impl Executor for AgentRpc {
112 120 }
113 121 Frame::Exit { code } => exit_code = Some(code),
114 122 Frame::Error { message } => anyhow::bail!("agent error: {message}"),
123 + // Forward-compatible: a newer agent's unknown frame is ignored.
124 + Frame::Unknown => {}
115 125 }
116 126 }
127 + // `buf` now holds at most one incomplete trailing line — bound it.
128 + anyhow::ensure!(
129 + buf.len() <= MAX_FRAME_BYTES,
130 + "agent /run frame exceeded {MAX_FRAME_BYTES} bytes without a newline",
131 + );
117 132 }
118 133
119 134 let code = exit_code.context("agent closed /run stream without an exit frame")?;
M shared/ops-exec/src/wire.rs +36
@@ -9,6 +9,11 @@
9 9 use crate::step::Step;
10 10 use serde::{Deserialize, Serialize};
11 11
12 + /// The wire-protocol version the daemon and agent speak. Bump on a
13 + /// breaking change to [`Frame`]/[`RunRequest`]; surfaced in [`HealthResponse`]
14 + /// so a daemon can detect a skewed agent at preflight rather than mid-stream.
15 + pub const PROTOCOL_VERSION: u32 = 1;
16 +
12 17 /// `POST /run` request body.
13 18 #[derive(Clone, Debug, Serialize, Deserialize)]
14 19 pub struct RunRequest {
@@ -25,6 +30,11 @@ pub enum Frame {
25 30 Exit { code: i32 },
26 31 /// Terminal: the agent refused (capability/identity) or failed to spawn.
27 32 Error { message: String },
33 + /// A frame kind this client does not know — a newer agent emitted it.
34 + /// Decoding to this (instead of erroring) keeps an older client
35 + /// forward-compatible: it ignores unknown frames rather than failing the run.
36 + #[serde(other)]
37 + Unknown,
28 38 }
29 39
30 40 impl Frame {
@@ -38,10 +48,36 @@ impl Frame {
38 48 }
39 49 }
40 50
51 + #[cfg(test)]
52 + mod tests {
53 + use super::*;
54 +
55 + #[test]
56 + fn unknown_frame_kind_decodes_to_unknown_not_error() {
57 + // A newer agent emits a frame this client doesn't know.
58 + let f: Frame = serde_json::from_str(r#"{"t":"progress","pct":42}"#).unwrap();
59 + assert!(matches!(f, Frame::Unknown));
60 + // Known frames still decode normally.
61 + let f: Frame = serde_json::from_str(r#"{"t":"exit","code":0}"#).unwrap();
62 + assert!(matches!(f, Frame::Exit { code: 0 }));
63 + }
64 +
65 + #[test]
66 + fn health_without_version_defaults_to_zero() {
67 + let h: HealthResponse =
68 + serde_json::from_str(r#"{"ok":true,"actuate":[],"observe":[]}"#).unwrap();
69 + assert_eq!(h.version, 0);
70 + }
71 + }
72 +
41 73 /// `GET /health` response body.
42 74 #[derive(Clone, Debug, Serialize, Deserialize)]
43 75 pub struct HealthResponse {
44 76 pub ok: bool,
77 + /// Wire-protocol version this agent speaks ([`PROTOCOL_VERSION`]). Defaults
78 + /// to 0 when absent so an older agent (no field) still deserializes.
79 + #[serde(default)]
80 + pub version: u32,
45 81 /// The agent's own actuate grant tokens (introspection / audit).
46 82 pub actuate: Vec<String>,
47 83 /// The agent's own observe grant tokens.