Skip to main content

max / makenotwork

bento: test build supersession, partial preflight, and DRY test state Cover the two orchestration paths that had none: latest-wins supersession driven through start_build (a newer build of the same (app, target) cancels + aborts the in-flight one and takes the single slot), and a failing preflight (the agent host's /health probe) failing the target before the recipe dispatches, via a FakeExec test double. Also extract the AppState construction duplicated across the runner tests into a shared test_state helper. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 02:19 UTC
Commit: edd65dd000e2eb6ed27bef3162f04f0c83712ce0
Parent: 8e94559
3 files changed, +319 insertions, -56 deletions
M bento/daemon/Cargo.lock +1
@@ -179,6 +179,7 @@ name = "bento-daemon"
179 179 version = "0.1.0"
180 180 dependencies = [
181 181 "anyhow",
182 + "async-trait",
182 183 "axum",
183 184 "chrono",
184 185 "http-body-util",
M bento/daemon/Cargo.toml +1
@@ -30,6 +30,7 @@ chrono = { version = "0.4", features = ["serde"] }
30 30 semver = { version = "1.0", features = ["serde"] }
31 31
32 32 [dev-dependencies]
33 + async-trait = "0.1"
33 34 tempfile = "3.20"
34 35 tower = { version = "0.5", features = ["util"] }
35 36 http-body-util = "0.1"
M bento/daemon/src/runner.rs +317 -56
@@ -512,10 +512,110 @@ mod tests {
512 512 use crate::config::Config;
513 513 use crate::ota::OtaRegistry;
514 514 use crate::topology::Topology;
515 + use async_trait::async_trait;
516 + use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, SyncOpts};
517 + use sqlx::SqlitePool;
515 518 use std::collections::HashMap;
519 + use std::os::unix::process::ExitStatusExt;
516 520 use std::sync::Arc;
517 521 use tokio::sync::Mutex;
518 522
523 + /// A no-transport [`Executor`] for the paths that don't need a real command
524 + /// to run: its `preflight` is programmable (the agent host hits `/health`
525 + /// there, and a dead `ops-agent` must fail the target before the recipe
526 + /// dispatches), and every actual op is a success no-op.
527 + struct FakeExec {
528 + caps: CapabilitySet,
529 + preflight_err: Option<String>,
530 + }
531 +
532 + impl FakeExec {
533 + fn preflight_fails(msg: &str) -> Arc<dyn Executor> {
534 + Arc::new(Self {
535 + caps: CapabilitySet::default(),
536 + preflight_err: Some(msg.to_string()),
537 + })
538 + }
539 + }
540 +
541 + #[async_trait]
542 + impl Executor for FakeExec {
543 + async fn run_streaming(
544 + &self,
545 + _step: &ops_exec::Step,
546 + _sink: &mut dyn LogSink,
547 + ) -> anyhow::Result<RunOutput> {
548 + Ok(RunOutput {
549 + status: std::process::ExitStatus::from_raw(0),
550 + stdout: Vec::new(),
551 + stderr: Vec::new(),
552 + })
553 + }
554 + async fn pull_file(
555 + &self,
556 + _r: &std::path::Path,
557 + _l: &std::path::Path,
558 + _o: &SyncOpts,
559 + ) -> anyhow::Result<()> {
560 + Ok(())
561 + }
562 + async fn pull_dir(
563 + &self,
564 + _r: &std::path::Path,
565 + _l: &std::path::Path,
566 + _o: &SyncOpts,
567 + ) -> anyhow::Result<()> {
568 + Ok(())
569 + }
570 + async fn pull_glob(
571 + &self,
572 + _g: &str,
573 + _l: &std::path::Path,
574 + _o: &SyncOpts,
575 + ) -> anyhow::Result<()> {
576 + Ok(())
577 + }
578 + async fn push_dir(
579 + &self,
580 + _l: &std::path::Path,
581 + _r: &std::path::Path,
582 + _o: &SyncOpts,
583 + ) -> anyhow::Result<()> {
584 + Ok(())
585 + }
586 + async fn preflight(&self) -> anyhow::Result<()> {
587 + match &self.preflight_err {
588 + Some(m) => anyhow::bail!("{m}"),
589 + None => Ok(()),
590 + }
591 + }
592 + fn capabilities(&self) -> &CapabilitySet {
593 + &self.caps
594 + }
595 + }
596 +
597 + /// Assemble an [`AppState`] from the three per-test inputs, filling in the
598 + /// executors/syncs (built from `topo`) and the fixed test scaffolding
599 + /// (metrics handle, event bus, standard OTA registry, empty active map, no
600 + /// token). Every runner test builds the same struct around a different repo
601 + /// + recipe; this is that struct in one place.
602 + fn test_state(pool: SqlitePool, topo: Topology, cfg: Config) -> AppState {
603 + let executors = Arc::new(crate::state::build_executors(&topo));
604 + let syncs = Arc::new(crate::state::build_syncs(&topo));
605 + AppState {
606 + pool,
607 + topo: Arc::new(topo),
608 + cfg: Arc::new(cfg),
609 + prom: crate::metrics::test_handle(),
610 + events: crate::events::channel(),
611 + ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
612 + executors,
613 + syncs,
614 + active: Arc::new(Mutex::new(HashMap::new())),
615 + api_token: None,
616 + }
617 + }
618 +
519 619 /// Stand up a tmp app repo + topology and run a real local recipe end to
520 620 /// end: step transitions, streamed `sh_ok`, `version_of`, `log`, and a
521 621 /// `collect` that pulls a built artifact into dist_root.
@@ -565,20 +665,7 @@ repo = "{}"
565 665 ))
566 666 .unwrap();
567 667
568 - let executors = Arc::new(crate::state::build_executors(&topo));
569 - let syncs = Arc::new(crate::state::build_syncs(&topo));
570 - let state = AppState {
571 - pool: pool.clone(),
572 - topo: Arc::new(topo),
573 - cfg: Arc::new(cfg),
574 - prom: crate::metrics::test_handle(),
575 - events: crate::events::channel(),
576 - ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
577 - executors,
578 - syncs,
579 - active: Arc::new(Mutex::new(HashMap::new())),
580 - api_token: None,
581 - };
668 + let state = test_state(pool.clone(), topo, cfg);
582 669
583 670 let app = AppId::new("demo");
584 671 let version = Version::parse("0.0.1").unwrap();
@@ -704,20 +791,7 @@ repo = "{}"
704 791 ))
705 792 .unwrap();
706 793
707 - let executors = Arc::new(crate::state::build_executors(&topo));
708 - let syncs = Arc::new(crate::state::build_syncs(&topo));
709 - let state = AppState {
710 - pool: pool.clone(),
711 - topo: Arc::new(topo),
712 - cfg: Arc::new(cfg),
713 - prom: crate::metrics::test_handle(),
714 - events: crate::events::channel(),
715 - ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
716 - executors,
717 - syncs,
718 - active: Arc::new(Mutex::new(HashMap::new())),
719 - api_token: None,
720 - };
794 + let state = test_state(pool.clone(), topo, cfg);
721 795
722 796 let build_id = start_build(
723 797 state.clone(),
@@ -801,6 +875,219 @@ repo = "{}"
801 875 );
802 876 }
803 877
878 + /// Latest-wins supersession, driven through `start_build` (not the leaf
879 + /// `begin_step` bail that's already covered). Two builds of the SAME
880 + /// (app, target) race: the second must cooperatively cancel + abort the
881 + /// first and take the single slot, the first must terminate non-`ok`, and
882 + /// the second must run to completion.
883 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
884 + async fn a_newer_build_supersedes_the_in_flight_one_for_the_same_target() {
885 + let tmp = tempfile::tempdir().unwrap();
886 + let root = tmp.path();
887 +
888 + let repo = root.join("app");
889 + std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
890 + std::fs::write(
891 + repo.join("src-tauri/tauri.conf.json"),
892 + r#"{"version":"0.0.1"}"#,
893 + )
894 + .unwrap();
895 + std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
896 + // A sleep long enough that the first build is still mid-recipe when the
897 + // second arrives; the engine checks the cancel flag at the step boundary
898 + // before "done", so the superseded run bails rather than finishing.
899 + std::fs::write(
900 + repo.join("dist/recipes/linux.rhai"),
901 + r#"
902 + step("build");
903 + sh_ok("fw13", "sleep 2");
904 + "#,
905 + )
906 + .unwrap();
907 +
908 + let cfg = Config::for_tests(root);
909 + let pool = crate::db::open(&cfg.db_path).await.unwrap();
910 + std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
911 + let topo = Topology::from_str_for_tests(&format!(
912 + r#"
913 + [[host]]
914 + name = "fw13"
915 + ssh = "local"
916 + targets = ["linux/x86_64"]
917 +
918 + [app.demo]
919 + repo = "{}"
920 + "#,
921 + repo.display()
922 + ))
923 + .unwrap();
924 + let state = test_state(pool.clone(), topo, cfg);
925 +
926 + let target = "linux/x86_64".parse().unwrap();
927 + let first = start_build(
928 + state.clone(),
929 + AppId::new("demo"),
930 + Version::parse("0.0.1").unwrap(),
931 + vec![target],
932 + )
933 + .await
934 + .unwrap();
935 + // Let the first build register its slot and enter the recipe before the
936 + // second supersedes it (start_build inserts the slot synchronously, but
937 + // the recipe runs on a spawned task).
938 + tokio::time::sleep(std::time::Duration::from_millis(100)).await;
939 + let second = start_build(
940 + state.clone(),
941 + AppId::new("demo"),
942 + Version::parse("0.0.1").unwrap(),
943 + vec![target],
944 + )
945 + .await
946 + .unwrap();
947 + assert_ne!(first, second);
948 +
949 + // Latest wins: exactly one slot for the key, owned by the second build.
950 + {
951 + let active = state.active.lock().await;
952 + assert_eq!(active.len(), 1, "supersession must not leave two slots");
953 + assert_eq!(
954 + active.values().next().unwrap().build_id,
955 + second,
956 + "the surviving slot belongs to the newer build",
957 + );
958 + }
959 +
960 + // Wait for the second (surviving) build to finalize.
961 + let mut second_status = String::new();
962 + for _ in 0..200 {
963 + second_status = sqlx::query_scalar("SELECT status FROM builds WHERE id = ?")
964 + .bind(second)
965 + .fetch_one(&pool)
966 + .await
967 + .unwrap();
968 + if second_status != "running" {
969 + break;
970 + }
971 + tokio::time::sleep(std::time::Duration::from_millis(50)).await;
972 + }
973 + assert_eq!(
974 + second_status, "ok",
975 + "the superseding build runs to completion"
976 + );
977 +
978 + // The first build's target run terminated without succeeding — it was
979 + // cancelled/aborted, never stamped `ok`.
980 + let first_status: String =
981 + sqlx::query_scalar("SELECT status FROM target_runs WHERE build_id = ?")
982 + .bind(first)
983 + .fetch_one(&pool)
984 + .await
985 + .unwrap();
986 + assert_ne!(
987 + first_status, "ok",
988 + "the superseded build must not complete successfully",
989 + );
990 +
991 + // Both builds reaped their slots; nothing left in flight.
992 + assert!(
993 + state.active.lock().await.is_empty(),
994 + "every finalized build reaps its own slot",
995 + );
996 + }
997 +
998 + /// A failing `preflight` (the agent host's `/health` probe when `ops-agent`
999 + /// is down) must fail the target BEFORE the recipe dispatches — the whole
1000 + /// point of preflighting. Covered only via a fake here, since a real
1001 + /// LocalExec/SshExec preflight is a no-op that always passes.
1002 + #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1003 + async fn a_failing_preflight_fails_the_target_before_the_recipe_runs() {
1004 + let tmp = tempfile::tempdir().unwrap();
1005 + let root = tmp.path();
1006 +
1007 + let repo = root.join("app");
1008 + std::fs::create_dir_all(repo.join("src-tauri")).unwrap();
1009 + std::fs::write(
1010 + repo.join("src-tauri/tauri.conf.json"),
1011 + r#"{"version":"0.0.1"}"#,
1012 + )
1013 + .unwrap();
1014 + std::fs::create_dir_all(repo.join("dist/recipes")).unwrap();
1015 + // If the recipe ran it would drop this marker; preflight failing first
1016 + // means it never does.
1017 + let marker = root.join("recipe-ran");
1018 + std::fs::write(
1019 + repo.join("dist/recipes/linux.rhai"),
1020 + r#"
1021 + step("build");
1022 + sh_ok("fw13", "touch MARKER");
1023 + "#
1024 + .replace("MARKER", marker.to_str().unwrap()),
1025 + )
1026 + .unwrap();
1027 +
1028 + let cfg = Config::for_tests(root);
1029 + let pool = crate::db::open(&cfg.db_path).await.unwrap();
1030 + std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
1031 + let topo = Topology::from_str_for_tests(&format!(
1032 + r#"
1033 + [[host]]
1034 + name = "fw13"
1035 + ssh = "local"
1036 + targets = ["linux/x86_64"]
1037 +
1038 + [app.demo]
1039 + repo = "{}"
1040 + "#,
1041 + repo.display()
1042 + ))
1043 + .unwrap();
1044 + let mut state = test_state(pool.clone(), topo, cfg);
1045 + // Swap fw13's executor for one whose preflight fails.
1046 + let mut execs = HashMap::new();
1047 + execs.insert(
1048 + "fw13".to_string(),
1049 + FakeExec::preflight_fails("ops-agent not reachable at /health"),
1050 + );
1051 + state.executors = Arc::new(execs);
1052 +
1053 + let build_id = start_build(
1054 + state.clone(),
1055 + AppId::new("demo"),
1056 + Version::parse("0.0.1").unwrap(),
1057 + vec!["linux/x86_64".parse().unwrap()],
1058 + )
1059 + .await
1060 + .unwrap();
1061 +
1062 + let mut status = String::new();
1063 + let mut error = String::new();
1064 + for _ in 0..100 {
1065 + let row: Option<(String, Option<String>)> =
1066 + sqlx::query_as("SELECT status, error FROM target_runs WHERE build_id = ?")
1067 + .bind(build_id)
1068 + .fetch_optional(&pool)
1069 + .await
1070 + .unwrap();
1071 + if let Some((s, e)) = row {
1072 + status = s;
1073 + error = e.unwrap_or_default();
1074 + if status != "running" {
1075 + break;
1076 + }
1077 + }
1078 + tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1079 + }
1080 + assert_eq!(status, "failed", "a failed preflight must fail the target");
1081 + assert!(
1082 + error.contains("ops-agent not reachable"),
1083 + "the failure must carry the preflight error, got: {error}"
1084 + );
1085 + assert!(
1086 + !marker.exists(),
1087 + "the recipe must NOT run when preflight fails"
1088 + );
1089 + }
1090 +
804 1091 /// Run 2 S5: a publish whose version is not strictly newer than the latest
805 1092 /// already published for the same (app, target, channel) is refused — an
806 1093 /// older build cannot republish over a live newer release.
@@ -849,20 +1136,7 @@ repo = "{}"
849 1136 repo.display()
850 1137 ))
851 1138 .unwrap();
852 - let executors = Arc::new(crate::state::build_executors(&topo));
853 - let syncs = Arc::new(crate::state::build_syncs(&topo));
854 - let state = AppState {
855 - pool: pool.clone(),
856 - topo: Arc::new(topo),
857 - cfg: Arc::new(cfg),
858 - prom: crate::metrics::test_handle(),
859 - events: crate::events::channel(),
860 - ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
861 - executors,
862 - syncs,
863 - active: Arc::new(Mutex::new(HashMap::new())),
864 - api_token: None,
865 - };
1139 + let state = test_state(pool.clone(), topo, cfg);
866 1140 let build_id = start_build(
867 1141 state.clone(),
868 1142 AppId::new("demo"),
@@ -961,20 +1235,7 @@ repo = "REPO"
961 1235 )
962 1236 .unwrap();
963 1237
964 - let executors = Arc::new(crate::state::build_executors(&topo));
965 - let syncs = Arc::new(crate::state::build_syncs(&topo));
966 - let state = AppState {
967 - pool: pool.clone(),
968 - topo: Arc::new(topo),
969 - cfg: Arc::new(cfg),
970 - prom: crate::metrics::test_handle(),
971 - events: crate::events::channel(),
972 - ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
973 - executors,
974 - syncs,
975 - active: Arc::new(Mutex::new(HashMap::new())),
976 - api_token: None,
977 - };
1238 + let state = test_state(pool.clone(), topo, cfg);
978 1239
979 1240 let build_id = start_build(
980 1241 state.clone(),