Skip to main content

max / makenotwork

bento: WAL + event-driven finalize + retention (Run 2 S1/S1b/S4/CHRONIC-1) S1: ops-core/sqlite.rs opened the pool in default rollback-journal mode with busy_timeout=0, but Bento fans concurrent target writers at it — spurious SQLITE_BUSY under contention. Add WAL + busy_timeout(5s) + synchronous=NORMAL (matches Sando's own db; ops-core::sqlite is Bento-only in practice). CHRONIC-1 (busy-poll, Run 1 -> Run 2): finalize_build polled SELECT COUNT(*) every 500ms for up to 2h. Replace with a JoinSet awaited event-driven, with an overall deadline backstop. One refactor also closes: - S1b: terminal-state writes were fire-and-forget (let _ = query); a dropped SQLITE_BUSY left a row stuck running. run_target's ok/failed writes, finish_step, and the build stamp are now checked and logged on error. - active-map leak: entries were removed only on supersede, never on completion. The map now carries the owning build_id; finalize reaps only its own slots via retain. - supervision: a panic in run_target's outer task is observed via the JoinSet and any straggling running row reconciled to failed. Also wrap begin_step's step-row insert + current_step pointer update in one transaction (atomic logical state). S4: zero retention let logs_root + dist_root grow unbounded (disk fill -> silent SQLite write loss). New retention module prunes both to the newest 5 versions per app (semver order), run at startup and every 6h. Left as NOTE (below the remediation bar): migration IF NOT EXISTS — editing the applied 0001_init.sql would break sqlx checksum validation on existing DBs; not worth the risk for a re-runnability edge case. bento-daemon 44 tests, ops-core 8, sando 146+30 green; clippy -D warnings clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 01:22 UTC
Commit: acfe960282afc4ce35123377d6fb90cfa6b5b040
Parent: 6080977
7 files changed, +225 insertions, -52 deletions
M bento/daemon/src/engine.rs +12 -4
@@ -159,6 +159,10 @@ impl RecipeCtx {
159 159 let run_id = self.rt.block_on(async move {
160 160 let started = Self::now();
161 161 let log_ref = me.log_path(step).to_string_lossy().into_owned();
162 + // The step row and the target's current_step pointer are one logical
163 + // state — write them atomically so a failure can't leave a `running`
164 + // step row while current_step still names the previous step.
165 + let mut tx = me.pool.begin().await.context("begin step tx")?;
162 166 let id: i64 = sqlx::query_scalar(
163 167 "INSERT INTO step_runs (target_run_id, step, status, log_ref, started_at)
164 168 VALUES (?, ?, 'running', ?, ?) RETURNING id",
@@ -167,15 +171,16 @@ impl RecipeCtx {
167 171 .bind(step.as_str())
168 172 .bind(&log_ref)
169 173 .bind(&started)
170 - .fetch_one(&me.pool)
174 + .fetch_one(&mut *tx)
171 175 .await
172 176 .context("insert step_run")?;
173 177 sqlx::query("UPDATE target_runs SET current_step = ? WHERE id = ?")
174 178 .bind(step.as_str())
175 179 .bind(me.target_run_id)
176 - .execute(&me.pool)
180 + .execute(&mut *tx)
177 181 .await
178 182 .context("update current_step")?;
183 + tx.commit().await.context("commit step tx")?;
179 184 anyhow::Ok(StepRunId(id))
180 185 })?;
181 186
@@ -238,14 +243,17 @@ impl RecipeCtx {
238 243 if let Ok(m) = Arc::try_unwrap(st.log) {
239 244 m.into_inner().close().await;
240 245 }
241 - let _ = sqlx::query(
246 + if let Err(e) = sqlx::query(
242 247 "UPDATE step_runs SET status = ?, finished_at = ? WHERE id = ?",
243 248 )
244 249 .bind(status.as_str())
245 250 .bind(Self::now())
246 251 .bind(st.run_id.0)
247 252 .execute(&me.pool)
248 - .await;
253 + .await
254 + {
255 + tracing::error!(step = st.step.as_str(), error = %e, "could not stamp step_run status");
256 + }
249 257 });
250 258 events::emit(
251 259 &self.events,
M bento/daemon/src/lib.rs +1
@@ -13,6 +13,7 @@ pub mod error;
13 13 pub mod events;
14 14 pub mod metrics;
15 15 pub mod ota;
16 + pub mod retention;
16 17 pub mod routes;
17 18 pub mod runner;
18 19 pub mod state;
M bento/daemon/src/main.rs +14
@@ -67,6 +67,20 @@ async fn main() -> Result<()> {
67 67 active: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
68 68 api_token,
69 69 };
70 + // Disk retention: prune logs_root + dist_root to the newest N versions per
71 + // app, at startup and every 6h, so neither root grows without bound.
72 + {
73 + let cfg = app_state.cfg.clone();
74 + tokio::spawn(async move {
75 + let mut tick = tokio::time::interval(std::time::Duration::from_secs(6 * 60 * 60));
76 + loop {
77 + tick.tick().await; // fires immediately, then every 6h
78 + let cfg = cfg.clone();
79 + let _ = tokio::task::spawn_blocking(move || bento_daemon::retention::prune_once(&cfg)).await;
80 + }
81 + });
82 + }
83 +
70 84 let app = routes::router(app_state);
71 85 tracing::info!(%addr, "bento daemon listening");
72 86 let listener = tokio::net::TcpListener::bind(addr).await?;
A bento/daemon/src/retention.rs +104
@@ -0,0 +1,104 @@
1 + //! Disk retention.
2 + //!
3 + //! Build logs (`logs_root`) and collected artifacts (`dist_root`) are written
4 + //! per `(app, version)` and never removed, so both roots grow without bound —
5 + //! a build host accumulates every historical version forever until the disk
6 + //! fills (which then surfaces as silent SQLite write failures). This module
7 + //! keeps only the most recent [`KEEP_VERSIONS_PER_APP`] versions per app under
8 + //! each root; `main` runs it at startup and on a periodic timer.
9 +
10 + use crate::config::Config;
11 + use crate::domain::Version;
12 + use std::path::{Path, PathBuf};
13 +
14 + /// How many of the newest versions to keep per app, per root. Chosen for
15 + /// rollback/debug depth without unbounded growth (operator decision, Run 2).
16 + pub const KEEP_VERSIONS_PER_APP: usize = 5;
17 +
18 + /// Prune both roots to the newest [`KEEP_VERSIONS_PER_APP`] versions per app.
19 + /// Best-effort and blocking: IO errors are logged and skipped, never fatal.
20 + /// Intended to run inside `spawn_blocking`.
21 + pub fn prune_once(cfg: &Config) {
22 + prune_root(&cfg.dist_root, "dist");
23 + prune_root(&cfg.logs_root, "logs");
24 + }
25 +
26 + fn prune_root(root: &Path, label: &str) {
27 + let Ok(apps) = std::fs::read_dir(root) else { return };
28 + for app in apps.filter_map(|e| e.ok()) {
29 + let app_dir = app.path();
30 + if app_dir.is_dir() {
31 + prune_app_dir(&app_dir, label);
32 + }
33 + }
34 + }
35 +
36 + fn prune_app_dir(app_dir: &Path, label: &str) {
37 + let Ok(entries) = std::fs::read_dir(app_dir) else { return };
38 + // Only consider subdirectories whose name parses as a semver — anything
39 + // else (a stray file, a non-version dir) is left untouched.
40 + let mut versions: Vec<(Version, PathBuf)> = entries
41 + .filter_map(|e| e.ok())
42 + .filter(|e| e.path().is_dir())
43 + .filter_map(|e| {
44 + let name = e.file_name();
45 + let v = Version::parse(name.to_string_lossy().as_ref()).ok()?;
46 + Some((v, e.path()))
47 + })
48 + .collect();
49 + if versions.len() <= KEEP_VERSIONS_PER_APP {
50 + return;
51 + }
52 + versions.sort_by(|a, b| b.0.cmp(&a.0)); // newest first
53 + for (ver, path) in versions.into_iter().skip(KEEP_VERSIONS_PER_APP) {
54 + match std::fs::remove_dir_all(&path) {
55 + Ok(()) => tracing::info!(%ver, dir = %path.display(), "retention: pruned old {label} version"),
56 + Err(e) => tracing::warn!(error = %e, dir = %path.display(), "retention: could not prune {label} dir"),
57 + }
58 + }
59 + }
60 +
61 + #[cfg(test)]
62 + mod tests {
63 + use super::*;
64 +
65 + fn touch_version(root: &Path, app: &str, ver: &str) {
66 + let d = root.join(app).join(ver);
67 + std::fs::create_dir_all(&d).unwrap();
68 + std::fs::write(d.join("artifact.bin"), b"x").unwrap();
69 + }
70 +
71 + #[test]
72 + fn keeps_newest_n_and_prunes_older_by_semver() {
73 + let tmp = tempfile::tempdir().unwrap();
74 + let root = tmp.path().join("dist");
75 + // Seven versions; lexical order would mis-rank 0.4.10 vs 0.4.9.
76 + for v in ["0.4.0", "0.4.1", "0.4.2", "0.4.9", "0.4.10", "0.5.0", "0.5.1"] {
77 + touch_version(&root, "demo", v);
78 + }
79 + // A non-version dir must be left alone.
80 + std::fs::create_dir_all(root.join("demo").join("scratch")).unwrap();
81 +
82 + prune_root(&root, "dist");
83 +
84 + let mut kept: Vec<String> = std::fs::read_dir(root.join("demo"))
85 + .unwrap()
86 + .filter_map(|e| e.ok())
87 + .map(|e| e.file_name().to_string_lossy().into_owned())
88 + .collect();
89 + kept.sort();
90 + // Newest 5 by semver: 0.5.1, 0.5.0, 0.4.10, 0.4.9, 0.4.2 (+ the non-version dir).
91 + assert_eq!(kept, vec!["0.4.10", "0.4.2", "0.4.9", "0.5.0", "0.5.1", "scratch"]);
92 + }
93 +
94 + #[test]
95 + fn no_op_when_under_limit() {
96 + let tmp = tempfile::tempdir().unwrap();
97 + let root = tmp.path().join("logs");
98 + for v in ["0.1.0", "0.2.0"] {
99 + touch_version(&root, "demo", v);
100 + }
101 + prune_root(&root, "logs");
102 + assert_eq!(std::fs::read_dir(root.join("demo")).unwrap().count(), 2);
103 + }
104 + }
M bento/daemon/src/runner.rs +72 -44
@@ -84,8 +84,11 @@ pub async fn start_build(
84 84 Event::BuildRequested { app: app.clone(), version: version.clone(), targets: targets.clone() },
85 85 );
86 86
87 + // Spawn every target into a JoinSet so finalize_build can await completion
88 + // event-driven (no DB polling) and observe each task's outcome (panic vs
89 + // clean) for supervision.
90 + let mut set: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
87 91 for target in targets {
88 - let state = state.clone();
89 92 let app = app.clone();
90 93 let version = version.clone();
91 94 let key = (app.clone(), target);
@@ -93,7 +96,7 @@ pub async fn start_build(
93 96 // Latest-wins: abort any in-flight run for this (app, target).
94 97 {
95 98 let mut active = state.active.lock().await;
96 - if let Some(prev) = active.remove(&key)
99 + if let Some((_, prev)) = active.remove(&key)
97 100 && !prev.is_finished() {
98 101 events::emit(
99 102 &state.events,
@@ -103,13 +106,13 @@ pub async fn start_build(
103 106 }
104 107 }
105 108
106 - let handle = tokio::spawn(run_target(state.clone(), build_id, app, version, target));
107 - state.active.lock().await.insert(key, handle.abort_handle());
109 + let abort = set.spawn(run_target(state.clone(), build_id, app, version, target));
110 + state.active.lock().await.insert(key, (build_id, abort));
108 111 }
109 112
110 113 // Mark the build done once all target tasks settle. Spawned so /build
111 114 // returns immediately.
112 - tokio::spawn(finalize_build(state, build_id));
115 + tokio::spawn(finalize_build(state, build_id, set));
113 116 Ok(build_id)
114 117 }
115 118
@@ -214,13 +217,18 @@ async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version
214 217 match outcome {
215 218 Ok(Ok(())) => {
216 219 let artifacts = collected_artifacts(&state, &app, &version);
217 - let _ = sqlx::query(
220 + if let Err(e) = sqlx::query(
218 221 "UPDATE target_runs SET status = 'ok', current_step = NULL, finished_at = ? WHERE id = ?",
219 222 )
220 223 .bind(chrono::Utc::now().to_rfc3339())
221 224 .bind(target_run_id)
222 225 .execute(&state.pool)
223 - .await;
226 + .await
227 + {
228 + // A swallowed terminal write would leave the row `running`;
229 + // finalize_build reconciles it, but log so the cause is visible.
230 + tracing::error!(%app, %target, error = %e, "could not stamp target_run ok");
231 + }
224 232 events::emit(&state.events, Event::TargetOk { app, version, target, artifacts });
225 233 }
226 234 Ok(Err((step, msg))) => {
@@ -243,56 +251,73 @@ async fn fail_target(
243 251 step: Step,
244 252 error: &str,
245 253 ) {
246 - let _ = sqlx::query(
254 + if let Err(e) = sqlx::query(
247 255 "UPDATE target_runs SET status = 'failed', error = ?, finished_at = ? WHERE id = ?",
248 256 )
249 257 .bind(error)
250 258 .bind(chrono::Utc::now().to_rfc3339())
251 259 .bind(target_run_id)
252 260 .execute(&state.pool)
253 - .await;
261 + .await
262 + {
263 + tracing::error!(%app, %target, error = %e, "could not stamp target_run failed");
264 + }
254 265 events::emit(
255 266 &state.events,
256 267 Event::TargetFailed { app: app.clone(), version: version.clone(), target, step, error: error.to_string() },
257 268 );
258 269 }
259 270
260 - /// Wait for all target runs of a build to leave `running`, then stamp the
261 - /// build's terminal status.
262 - async fn finalize_build(state: AppState, build_id: i64) {
263 - // Generous backstop: a build (sign + notarize can be slow) shouldn't take
264 - // more than this. Without it, a target task SIGKILLed before it stamps its
265 - // terminal status leaves a `running` row forever and this loop never exits.
271 + /// Await every target task of a build, then stamp the build's terminal status.
272 + ///
273 + /// Event-driven: it joins the `JoinSet` rather than polling the DB. Each task's
274 + /// outcome is supervised — a panicked task is logged (and its still-`running`
275 + /// row is reconciled below), an aborted (superseded) task is expected. A
276 + /// generous overall deadline bounds a wedged step (e.g. a hung ssh) so the build
277 + /// can always finalize; on deadline the remaining tasks are aborted.
278 + async fn finalize_build(state: AppState, build_id: i64, mut set: tokio::task::JoinSet<()>) {
266 279 const MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(2 * 60 * 60);
267 - let start = std::time::Instant::now();
280 + let deadline = tokio::time::Instant::now() + MAX_WAIT;
268 281 loop {
269 - let running: i64 = sqlx::query_scalar(
270 - "SELECT COUNT(*) FROM target_runs WHERE build_id = ? AND status = 'running'",
271 - )
272 - .bind(build_id)
273 - .fetch_one(&state.pool)
274 - .await
275 - .unwrap_or(0);
276 - if running == 0 {
277 - break;
278 - }
279 - if start.elapsed() >= MAX_WAIT {
280 - // Force the stragglers failed so the build can finalize and the row
281 - // reflects reality (a vanished worker, not a clean success).
282 - let _ = sqlx::query(
283 - "UPDATE target_runs SET status = 'failed', \
284 - error = COALESCE(error, 'worker vanished before finishing (finalize timeout)'), \
285 - finished_at = ? WHERE build_id = ? AND status = 'running'",
286 - )
287 - .bind(chrono::Utc::now().to_rfc3339())
288 - .bind(build_id)
289 - .execute(&state.pool)
290 - .await;
291 - tracing::error!(build_id, "finalize_build timed out waiting on target_runs; marked stragglers failed");
292 - break;
282 + match tokio::time::timeout_at(deadline, set.join_next()).await {
283 + Ok(Some(Ok(()))) => continue,
284 + Ok(Some(Err(e))) => {
285 + // A panic in run_target itself (outside its inner spawn_blocking,
286 + // which is already caught). Cancellation = a superseding build.
287 + if !e.is_cancelled() {
288 + tracing::error!(build_id, error = %e, "target task panicked");
289 + }
290 + }
291 + Ok(None) => break, // all targets settled
292 + Err(_elapsed) => {
293 + tracing::error!(build_id, "finalize_build deadline hit; aborting remaining target tasks");
294 + set.abort_all();
295 + while set.join_next().await.is_some() {}
296 + break;
297 + }
293 298 }
294 - tokio::time::sleep(std::time::Duration::from_millis(500)).await;
295 299 }
300 +
301 + // Reap this build's latest-wins slots (only our own — a superseding build
302 + // owns a different build_id and is left intact).
303 + state.active.lock().await.retain(|_, (bid, _)| *bid != build_id);
304 +
305 + let now = chrono::Utc::now().to_rfc3339();
306 + // Any row still `running` is a panicked/aborted task that never stamped its
307 + // terminal status — reconcile it so the build finalizes truthfully.
308 + if let Err(e) = sqlx::query(
309 + "UPDATE target_runs SET status = 'failed', \
310 + error = COALESCE(error, 'target task ended before stamping its status'), \
311 + finished_at = ? WHERE build_id = ? AND status = 'running'",
312 + )
313 + .bind(&now)
314 + .bind(build_id)
315 + .execute(&state.pool)
316 + .await
317 + {
318 + tracing::error!(build_id, error = %e, "could not reconcile straggling target_runs");
319 + }
320 +
296 321 let failed: i64 = sqlx::query_scalar(
297 322 "SELECT COUNT(*) FROM target_runs WHERE build_id = ? AND status = 'failed'",
298 323 )
@@ -301,12 +326,15 @@ async fn finalize_build(state: AppState, build_id: i64) {
301 326 .await
302 327 .unwrap_or(0);
303 328 let status = if failed == 0 { "ok" } else { "failed" };
304 - let _ = sqlx::query("UPDATE builds SET status = ?, finished_at = ? WHERE id = ?")
329 + if let Err(e) = sqlx::query("UPDATE builds SET status = ?, finished_at = ? WHERE id = ?")
305 330 .bind(status)
306 - .bind(chrono::Utc::now().to_rfc3339())
331 + .bind(&now)
307 332 .bind(build_id)
308 333 .execute(&state.pool)
309 - .await;
334 + .await
335 + {
336 + tracing::error!(build_id, error = %e, "could not stamp build terminal status");
337 + }
310 338 }
311 339
312 340 /// Read the recipe text for `(app, target)` from the app's checkout on the
M bento/daemon/src/state.rs +9 -1
@@ -18,6 +18,12 @@ use tokio::task::AbortHandle;
18 18 /// `ExecutorMap`.
19 19 pub type ExecutorMap = HashMap<String, Arc<dyn Executor>>;
20 20
21 + /// Latest-wins guard map: per `(app, target)`, the owning `build_id` and an
22 + /// [`AbortHandle`] for the in-flight target task. A newer build supersedes the
23 + /// slot (aborting the prior handle); a finishing build reaps only the slots it
24 + /// still owns.
25 + pub type ActiveBuilds = Arc<Mutex<HashMap<(AppId, Target), (i64, AbortHandle)>>>;
26 +
21 27 #[derive(Clone)]
22 28 pub struct AppState {
23 29 pub pool: SqlitePool,
@@ -36,7 +42,9 @@ pub struct AppState {
36 42 /// Single-slot guard per `(app, target)`: a newer build for the same
37 43 /// target aborts the in-flight one (latest request wins), mirroring
38 44 /// Sando's `active_build`. Other targets keep running — that's the fan-out.
39 - pub active: Arc<Mutex<HashMap<(AppId, Target), AbortHandle>>>,
45 + /// The value carries the owning `build_id` so a finished build reaps only
46 + /// its own slots and never a superseding build's handle.
47 + pub active: ActiveBuilds,
40 48 }
41 49
42 50 /// Build one host's executor: `LocalExec` for `ssh = "local"`, `AgentRpc` for an
M shared/ops-core/src/sqlite.rs +13 -3
@@ -7,16 +7,26 @@
7 7
8 8 use anyhow::Result;
9 9 use sqlx::SqlitePool;
10 - use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
10 + use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};
11 11 use std::path::Path;
12 12 use std::str::FromStr;
13 + use std::time::Duration;
13 14
14 - /// Open (creating if missing) a pooled SQLite connection with foreign keys on.
15 + /// Open (creating if missing) a pooled SQLite connection with sane defaults for
16 + /// a concurrent multi-writer daemon. Bento fans builds out across targets, each
17 + /// on its own blocking thread writing to this pool, so the default
18 + /// rollback-journal mode with `busy_timeout = 0` would surface spurious
19 + /// `SQLITE_BUSY` errors under contention. WAL lets readers run alongside a
20 + /// writer; `busy_timeout` makes a contending writer wait rather than fail;
21 + /// `synchronous = NORMAL` is the standard durable-enough pairing for WAL.
15 22 pub async fn connect(path: &Path) -> Result<SqlitePool> {
16 23 let url = format!("sqlite://{}?mode=rwc", path.display());
17 24 let opts = SqliteConnectOptions::from_str(&url)?
18 25 .create_if_missing(true)
19 - .foreign_keys(true);
26 + .foreign_keys(true)
27 + .journal_mode(SqliteJournalMode::Wal)
28 + .synchronous(SqliteSynchronous::Normal)
29 + .busy_timeout(Duration::from_secs(5));
20 30 let pool = SqlitePoolOptions::new().max_connections(4).connect_with(opts).await?;
21 31 Ok(pool)
22 32 }