Skip to main content

max / makenotwork

bento: enforce publish monotonicity + restart recovery (Run 2 S5/S2) S5: version monotonicity was claimed in the schema/publish comments but never implemented, and Version lacked Ord so it couldn't be. Derive Ord on Version (semver precedence, not the TEXT column's lexical order) and refuse a publish whose version is not strictly newer than the latest already published for the same (app, target, channel) — an older build can no longer republish over a live newer release. Fires before the artifact/authority checks. S2: a daemon restart left in-flight builds/target_runs/step_runs stuck in `running` forever (the finalizer and target tasks die with the process). Add db::recover_orphaned_running, run at startup, marking orphaned running rows failed. Idempotent; a clean DB sweeps zero. Also clarify the sh vs sh_ok contract (sh = branch-on-code, recipe owns the outcome; sh_ok = must-succeed, fails the step and bars publish) and attribute an sh_ok failure to the current step explicitly. bento-daemon 42 tests, clippy -D warnings clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 01:17 UTC
Commit: 6080977a0438f5cb22a51eccc33c8f27dc45bc41
Parent: cf7bcfe
5 files changed, +235 insertions, -3 deletions
M bento/daemon/src/db.rs +77
@@ -12,3 +12,80 @@ pub async fn open(path: &Path) -> Result<SqlitePool> {
12 12 sqlx::migrate!("./migrations").run(&pool).await?;
13 13 Ok(pool)
14 14 }
15 +
16 + /// Reconcile state left `running` by a previous process. The finalizer and the
17 + /// per-target tasks die with the daemon, so any `running` build/target/step row
18 + /// at startup is orphaned — nothing will ever stamp it terminal. Mark them all
19 + /// `failed` so `/state` reflects reality and a re-run isn't blocked by a ghost.
20 + /// Returns the number of build rows reconciled (for logging). Idempotent: a
21 + /// clean DB sweeps zero rows.
22 + pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
23 + let now = chrono::Utc::now().to_rfc3339();
24 + let mut tx = pool.begin().await?;
25 + sqlx::query("UPDATE step_runs SET status = 'failed', finished_at = ? WHERE status = 'running'")
26 + .bind(&now)
27 + .execute(&mut *tx)
28 + .await?;
29 + sqlx::query(
30 + "UPDATE target_runs SET status = 'failed', \
31 + error = COALESCE(error, 'daemon restarted while running'), finished_at = ? \
32 + WHERE status = 'running'",
33 + )
34 + .bind(&now)
35 + .execute(&mut *tx)
36 + .await?;
37 + let builds = sqlx::query("UPDATE builds SET status = 'failed', finished_at = ? WHERE status = 'running'")
38 + .bind(&now)
39 + .execute(&mut *tx)
40 + .await?
41 + .rows_affected();
42 + tx.commit().await?;
43 + Ok(builds)
44 + }
45 +
46 + #[cfg(test)]
47 + mod tests {
48 + use super::*;
49 +
50 + #[tokio::test]
51 + async fn recovery_marks_orphaned_running_failed_and_leaves_terminal_rows() {
52 + let dir = tempfile::tempdir().unwrap();
53 + let pool = open(&dir.path().join("t.db")).await.unwrap();
54 +
55 + // One running build with a running target+step, and one already-ok build.
56 + let now = chrono::Utc::now().to_rfc3339();
57 + let running: i64 = sqlx::query_scalar(
58 + "INSERT INTO builds (app, version, status, created_at) VALUES ('demo','0.1.0','running',?) RETURNING id",
59 + )
60 + .bind(&now).fetch_one(&pool).await.unwrap();
61 + let tr: i64 = sqlx::query_scalar(
62 + "INSERT INTO target_runs (build_id, app, version, target, status, started_at) \
63 + VALUES (?, 'demo','0.1.0','linux/x86_64','running',?) RETURNING id",
64 + )
65 + .bind(running).bind(&now).fetch_one(&pool).await.unwrap();
66 + sqlx::query(
67 + "INSERT INTO step_runs (target_run_id, step, status, log_ref, started_at) \
68 + VALUES (?, 'build','running','x',?)",
69 + )
70 + .bind(tr).bind(&now).execute(&pool).await.unwrap();
71 + sqlx::query("INSERT INTO builds (app, version, status, created_at) VALUES ('demo','0.0.9','ok',?)")
72 + .bind(&now).execute(&pool).await.unwrap();
73 +
74 + let reconciled = recover_orphaned_running(&pool).await.unwrap();
75 + assert_eq!(reconciled, 1, "only the running build is reconciled");
76 +
77 + // The orphaned chain is now failed; the ok build is untouched.
78 + let running_left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds WHERE status='running'")
79 + .fetch_one(&pool).await.unwrap();
80 + assert_eq!(running_left, 0);
81 + let ok_left: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM builds WHERE status='ok'")
82 + .fetch_one(&pool).await.unwrap();
83 + assert_eq!(ok_left, 1);
84 + let err: Option<String> = sqlx::query_scalar("SELECT error FROM target_runs WHERE id = ?")
85 + .bind(tr).fetch_one(&pool).await.unwrap();
86 + assert_eq!(err.as_deref(), Some("daemon restarted while running"));
87 +
88 + // Idempotent: a second sweep finds nothing.
89 + assert_eq!(recover_orphaned_running(&pool).await.unwrap(), 0);
90 + }
91 + }
M bento/daemon/src/domain.rs +15 -2
@@ -226,8 +226,9 @@ impl FromStr for Step {
226 226 // ---------------------------------------------------------------------
227 227
228 228 /// App semver (e.g. `0.4.1`), read from `tauri.conf.json` or supplied to
229 - /// `/build`. Stored as TEXT.
230 - #[derive(Debug, Clone, PartialEq, Eq, Hash)]
229 + /// `/build`. Stored as TEXT. Ordered by semver precedence (not the TEXT
230 + /// column's lexical order) so publish can enforce version monotonicity.
231 + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
231 232 pub struct Version(semver::Version);
232 233
233 234 impl Version {
@@ -340,4 +341,16 @@ mod tests {
340 341 assert_eq!(Version::parse("0.4.1").unwrap().to_string(), "0.4.1");
341 342 assert!(Version::parse("v0.4").is_err());
342 343 }
344 +
345 + #[test]
346 + fn version_orders_by_semver_not_lexically() {
347 + let v = |s: &str| Version::parse(s).unwrap();
348 + assert!(v("0.4.0") < v("0.5.0"));
349 + // Lexical order would put "0.4.10" < "0.4.9"; semver must not.
350 + assert!(v("0.4.10") > v("0.4.9"));
351 + assert!(v("1.0.0") > v("0.99.99"));
352 + let mut vs = [v("0.4.10"), v("0.4.2"), v("0.5.0")];
353 + vs.sort();
354 + assert_eq!(vs.iter().max().unwrap().to_string(), "0.5.0");
355 + }
343 356 }
M bento/daemon/src/engine.rs +43 -1
@@ -398,6 +398,12 @@ pub fn build_engine(ctx: Arc<RecipeCtx>) -> Engine {
398 398 }
399 399
400 400 // --- sh(host, cmd) -> #{ code, stdout_tail } ---
401 + //
402 + // The branch-on-exit-code primitive: the recipe OWNS the outcome. A non-zero
403 + // exit is returned, not raised, and does NOT fail the step or bar publish —
404 + // use this only when the recipe inspects `code` and decides. For a command
405 + // that must succeed (build/sign/etc.), use `sh_ok`, which fails the step (and
406 + // therefore bars publish via the failed-step ledger) on a non-zero exit.
401 407 {
402 408 let ctx = ctx.clone();
403 409 engine.register_fn("sh", move |host: &str, cmd: &str| -> Result<Map, Box<EvalAltResult>> {
@@ -409,12 +415,19 @@ pub fn build_engine(ctx: Arc<RecipeCtx>) -> Engine {
409 415 });
410 416 }
411 417
412 - // --- sh_ok(host, cmd): run + assert exit 0 ---
418 + // --- sh_ok(host, cmd): run + assert exit 0 (the must-succeed primitive) ---
419 + //
420 + // A non-zero exit fails the current step (added to the publish-barring
421 + // ledger) and aborts the recipe, so an artifact is never shipped after a
422 + // must-succeed command failed.
413 423 {
414 424 let ctx = ctx.clone();
415 425 engine.register_fn("sh_ok", move |host: &str, cmd: &str| -> Result<(), Box<EvalAltResult>> {
416 426 let (code, _) = ctx.run(host, cmd).map_err(rhai_err)?;
417 427 if code != 0 {
428 + // Attribute the failure to the current step explicitly so the
429 + // ledger bars publish even if a future caller swallowed the error.
430 + ctx.fail_current_step();
418 431 return Err(rhai_err(format!("command on `{host}` exited {code}: {cmd}")));
419 432 }
420 433 Ok(())
@@ -642,6 +655,35 @@ impl RecipeCtx {
642 655 "publish channel `{channel}` does not support target {target}",
643 656 );
644 657
658 + // Monotonicity: never publish a version that is not strictly newer than
659 + // the latest already published for this (app, target, channel). Without
660 + // this an older build could republish over a live newer release. The
661 + // `releases` column is TEXT, so compare by parsed semver precedence
662 + // (Version: Ord), not lexically.
663 + {
664 + let (app_s, target_s, chan_s) = (app.to_string(), target.to_string(), channel.to_string());
665 + let me = self.clone();
666 + let latest: Option<Version> = self.rt.block_on(async move {
667 + let rows: Vec<(String,)> = sqlx::query_as(
668 + "SELECT version FROM releases WHERE app = ? AND target = ? AND channel = ?",
669 + )
670 + .bind(app_s)
671 + .bind(target_s)
672 + .bind(chan_s)
673 + .fetch_all(&me.pool)
674 + .await
675 + .unwrap_or_default();
676 + rows.into_iter().filter_map(|(v,)| Version::parse(&v).ok()).max()
677 + });
678 + if let Some(latest) = latest {
679 + anyhow::ensure!(
680 + version > latest,
681 + "refusing to publish {app} {version} to `{channel}` ({target}): \
682 + not newer than the last published {latest}",
683 + );
684 + }
685 + }
686 +
645 687 // Step-success ledger (the Bento analogue of Sando's gate fail-closed),
646 688 // minted as an unforgeable PublishAuthority. `backend.publish` cannot be
647 689 // called without one, so the unverified/post-failure ship path is sealed
M bento/daemon/src/main.rs +7
@@ -25,6 +25,13 @@ async fn main() -> Result<()> {
25 25 tokio::fs::create_dir_all(&cfg.dist_root).await?;
26 26 tokio::fs::create_dir_all(&cfg.logs_root).await?;
27 27 let pool = db::open(&cfg.db_path).await?;
28 + // Reconcile any builds left `running` by a previous process (the finalizer
29 + // and target tasks don't survive a restart) so they don't orphan forever.
30 + match db::recover_orphaned_running(&pool).await {
31 + Ok(0) => {}
32 + Ok(n) => tracing::warn!(reconciled = n, "marked orphaned `running` builds failed after restart"),
33 + Err(e) => tracing::error!(error = %e, "failed to reconcile orphaned running builds"),
34 + }
28 35 tracing::info!(hosts = topo.hosts.len(), apps = topo.app.len(), "topology loaded");
29 36
30 37 let prom = metrics::init();
M bento/daemon/src/runner.rs +93
@@ -440,6 +440,99 @@ targets = ["linux/x86_64"]
440 440 assert!(std::fs::read_to_string(&log).unwrap().contains("compiling"));
441 441 }
442 442
443 + /// Run 2 S5: a publish whose version is not strictly newer than the latest
444 + /// already published for the same (app, target, channel) is refused — an
445 + /// older build cannot republish over a live newer release.
446 + #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
447 + async fn publish_rejects_a_non_monotonic_version() {
448 + let tmp = tempfile::tempdir().unwrap();
449 + let root = tmp.path();
450 + let repo = root.join("app");
451 + std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
452 + std::fs::write(repo.join("src-tauri/tauri.conf.json"), r#"{"version":"0.2.0"}"#).unwrap();
453 + std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
454 + let artifact = repo.join("out/app.tar.gz");
455 + // Build an artifact, publish 0.2.0 (records a release), then try to
456 + // publish the older 0.1.0 — the second publish must fail.
457 + std::fs::write(
458 + repo.join("dist/recipes/linux.rhai"),
459 + r#"
460 + step("build");
461 + sh_ok("fw13", "mkdir -p REPO/out && echo bin > ARTIFACT");
462 + step("publish");
463 + publish("tauri-mnw", "demo", "linux/x86_64", "0.2.0", "ARTIFACT", #{});
464 + publish("tauri-mnw", "demo", "linux/x86_64", "0.1.0", "ARTIFACT", #{});
465 + "#
466 + .replace("ARTIFACT", artifact.to_str().unwrap())
467 + .replace("REPO", repo.to_str().unwrap()),
468 + )
469 + .unwrap();
470 +
471 + let cfg = Config::for_tests(root);
472 + let pool = crate::db::open(&cfg.db_path).await.unwrap();
473 + let topo: Topology = toml::from_str(&format!(
474 + r#"
475 + [[host]]
476 + name = "fw13"
477 + ssh = "local"
478 + targets = ["linux/x86_64"]
479 +
480 + [app.demo]
481 + repo = "{}"
482 + targets = ["linux/x86_64"]
483 + "#,
484 + repo.display()
485 + ))
486 + .unwrap();
487 + let executors = Arc::new(crate::state::build_executors(&topo));
488 + let state = AppState {
489 + pool: pool.clone(),
490 + topo: Arc::new(topo),
491 + cfg: Arc::new(cfg),
492 + prom: crate::metrics::test_handle(),
493 + events: crate::events::channel(),
494 + ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
495 + executors,
496 + active: Arc::new(Mutex::new(HashMap::new())),
497 + api_token: None,
498 + };
499 + let build_id = start_build(
500 + state.clone(),
501 + AppId::new("demo"),
502 + Version::parse("0.2.0").unwrap(),
503 + vec!["linux/x86_64".parse().unwrap()],
504 + )
505 + .await
506 + .unwrap();
507 +
508 + let mut status = String::new();
509 + let mut error = String::new();
510 + for _ in 0..100 {
511 + let row: Option<(String, Option<String>)> =
512 + sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
513 + .bind(build_id)
514 + .fetch_optional(&pool)
515 + .await
516 + .unwrap();
517 + if let Some((s, e)) = row {
518 + status = s;
519 + error = e.unwrap_or_default();
520 + if status != "running" {
521 + break;
522 + }
523 + }
524 + tokio::time::sleep(std::time::Duration::from_millis(50)).await;
525 + }
526 + assert_eq!(status, "failed", "non-monotonic publish must fail the run");
527 + assert!(error.contains("not newer"), "expected monotonicity error, got: {error}");
528 + // Exactly one release was recorded (0.2.0); 0.1.0 never landed.
529 + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM releases")
530 + .fetch_one(&pool)
531 + .await
532 + .unwrap();
533 + assert_eq!(count, 1, "only the first (newer) publish should record a release");
534 + }
535 +
443 536 /// The audit fix: a `build` step dispatched to a host whose executor lacks the
444 537 /// `build` capability is denied at the transport BEFORE the command runs. This
445 538 /// is the structural guarantee behind "never build on prod" — a recipe naming