Skip to main content

max / makenotwork

bento: latest-wins cooperative cancel + gatekeeper auto-grant + checked publish ledger (Run 3 S1/S2/S3) R3-S1: the latest-wins guard was unsound. (a) abort+spawn+insert spanned two lock scopes, so two concurrent /build+/retry for the same (app,target) could both run. Now one critical section. (b) AbortHandle::abort() can't stop a spawn_blocking recipe, so a superseded build ran to completion and could publish. Add a cooperative cancel flag (ActiveSlot.cancel) the engine checks at every step boundary and before publish; supersession sets it. A superseded recipe now stops at the next step instead of shipping. R3-S2: a stock signing-host config dead-ended every macOS/iOS publish because the gatekeeper observe grant was undeclared (only the throwaway driver set it). CapabilitySet::from_tokens — the one chokepoint all three grant sources flow through (topology executor, agent self-grant, agent caller allow-list) — now implies the gatekeeper observe whenever sign is granted, so it can't be misconfigured away. R3-S3: the releases INSERT was fire-and-forget; a swallowed write silently re-armed the monotonicity guard (which reads that table), letting an older version republish. The INSERT is now checked and propagates. Concurrent same-key publishers are serialized by the now-atomic latest-wins slot. Also: null current_step on fail/recovery reconciliation; recursive dir_size (a .app bundle reported ~0); local collect glob is validated-not-quoted so the wildcard expands. bento-daemon 47 tests, ops-exec 39, sando 146 green; clippy -D clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 02:44 UTC
Commit: 446609c2036503a84ab394e3648769301585bd7c
Parent: 5ad2fd9
5 files changed, +160 insertions, -54 deletions
M bento/daemon/src/db.rs +1 -1
@@ -27,7 +27,7 @@ pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
27 27 .execute(&mut *tx)
28 28 .await?;
29 29 sqlx::query(
30 - "UPDATE target_runs SET status = 'failed', \
30 + "UPDATE target_runs SET status = 'failed', current_step = NULL, \
31 31 error = COALESCE(error, 'daemon restarted while running'), finished_at = ? \
32 32 WHERE status = 'running'",
33 33 )
M bento/daemon/src/engine.rs +89 -27
@@ -25,6 +25,7 @@ use rhai::{Engine, EvalAltResult, Map};
25 25 use sqlx::SqlitePool;
26 26 use std::collections::HashMap;
27 27 use std::path::{Path, PathBuf};
28 + use std::sync::atomic::{AtomicBool, Ordering};
28 29 use std::sync::{Arc, Mutex};
29 30 use tokio::runtime::Handle;
30 31 use tokio::sync::Mutex as AsyncMutex;
@@ -97,6 +98,10 @@ pub struct RecipeCtx {
97 98 /// `publish` — an artifact is never shipped after a step failed, even if the
98 99 /// recipe ignored the failure and ran on.
99 100 failed_steps: Mutex<Vec<Step>>,
101 + /// Set when a newer build supersedes this run. Checked at step boundaries
102 + /// and before publish so a superseded recipe stops promptly rather than
103 + /// running to completion (the blocking Rhai body can't be `abort()`ed).
104 + cancel: Arc<AtomicBool>,
100 105 }
101 106
102 107 impl RecipeCtx {
@@ -115,6 +120,7 @@ impl RecipeCtx {
115 120 cfg: Arc<Config>,
116 121 ota: Arc<OtaRegistry>,
117 122 rt: Handle,
123 + cancel: Arc<AtomicBool>,
118 124 ) -> Self {
119 125 Self {
120 126 app,
@@ -133,9 +139,15 @@ impl RecipeCtx {
133 139 current: Mutex::new(None),
134 140 gatekeeper_ok: Mutex::new(None),
135 141 failed_steps: Mutex::new(Vec::new()),
142 + cancel,
136 143 }
137 144 }
138 145
146 + /// Whether a newer build has superseded this run.
147 + fn is_cancelled(&self) -> bool {
148 + self.cancel.load(Ordering::SeqCst)
149 + }
150 +
139 151 fn now() -> String {
140 152 chrono::Utc::now().to_rfc3339()
141 153 }
@@ -154,6 +166,11 @@ impl RecipeCtx {
154 166 /// Close the previous step (as `Ok`), open a new one: insert its DB row,
155 167 /// open a live log whose chunks broadcast `StepLogChunk`, emit `StepStart`.
156 168 fn begin_step(self: &Arc<Self>, step: Step) -> Result<()> {
169 + anyhow::ensure!(
170 + !self.is_cancelled(),
171 + "build superseded by a newer request; aborting before `{}`",
172 + step.as_str()
173 + );
157 174 self.finish_step(Status::Ok)?;
158 175 let me = self.clone();
159 176 let run_id = self.rt.block_on(async move {
@@ -595,22 +612,20 @@ impl RecipeCtx {
595 612 let dest_s = dest.to_string_lossy().into_owned();
596 613 let ssh = self.host_ssh(host)?;
597 614 let is_local = ssh == "local" || ssh.is_empty();
598 - // The daemon does the transfer locally: cp for a local host, scp for a
599 - // remote one. The glob is sh-quoted on the local side; on the remote side
600 - // the wildcard must stay live for the remote shell to expand, so we quote
601 - // the host prefix and leave the glob unquoted but validated.
615 + // Both branches run the transfer through `sh -c`, and both need the
616 + // wildcard (`*?[]`) to reach the shell live — so the glob is left
617 + // UNQUOTED in both and validated against shell metacharacters instead.
618 + // (Quoting it on the local side, as before, neutered the wildcard so
619 + // `cp '*.dmg'` matched nothing.) Path/wildcard chars are allowed; command
620 + // metacharacters are not.
621 + anyhow::ensure!(
622 + !glob.chars().any(|c| matches!(c, ';' | '&' | '|' | '$' | '`' | '\'' | '"' | '\\' | ' ' | '\n' | '(' | ')' | '<' | '>')),
623 + "collect glob `{glob}` contains shell metacharacters"
624 + );
602 625 let d = ops_core::remote::sh_quote(&dest_s);
603 626 let cmd = if is_local {
604 - let g = ops_core::remote::sh_quote(glob);
605 - format!("mkdir -p {d} && cp -vR {g} {d}/")
627 + format!("mkdir -p {d} && cp -vR {glob} {d}/")
606 628 } else {
607 - // Reject shell metacharacters that would break out of the scp source
608 - // (the wildcard chars `*?[]` and path chars are allowed; `;`, `$`,
609 - // backticks, quotes, whitespace are not).
610 - anyhow::ensure!(
611 - !glob.chars().any(|c| matches!(c, ';' | '&' | '|' | '$' | '`' | '\'' | '"' | '\\' | ' ' | '\n' | '(' | ')' | '<' | '>')),
612 - "collect glob `{glob}` contains shell metacharacters"
613 - );
614 629 format!(
615 630 "mkdir -p {d} && scp -r {flags} {tgt}:{glob} {d}/",
616 631 flags = ops_core::remote::SSH_FLAGS.join(" "),
@@ -647,6 +662,10 @@ impl RecipeCtx {
647 662 artifact: &str,
648 663 meta: Map,
649 664 ) -> Result<String> {
665 + // Never let a superseded build ship. This is the last and most important
666 + // cooperative-cancel checkpoint: even if a long-running step finished
667 + // after supersession, the artifact must not reach the backend.
668 + anyhow::ensure!(!self.is_cancelled(), "build superseded by a newer request; refusing to publish");
650 669 let backend = self
651 670 .ota
652 671 .get(channel)
@@ -715,23 +734,31 @@ impl RecipeCtx {
715 734 let receipt = backend
716 735 .publish(&rel, &artifact_path, &authority)
717 736 .with_context(|| format!("publish to `{channel}`"))?;
718 - // Record for idempotency / monotonicity.
737 + // Record for idempotency / monotonicity. This write is CHECKED, not
738 + // fire-and-forget: a swallowed failure here would silently re-arm the
739 + // monotonicity guard (which reads this same table), letting an older
740 + // version republish over a live release. Concurrent same-(app,target)
741 + // publishers can't race the read-then-insert because the latest-wins slot
742 + // (state::ActiveSlot) serializes them and a superseded run is cancelled
743 + // before it reaches publish.
719 744 let me = self.clone();
720 745 let (app_s, target_s, ver_s, chan_s) =
721 746 (app.to_string(), target.to_string(), version.to_string(), channel.to_string());
722 - self.rt.block_on(async move {
723 - let _ = sqlx::query(
724 - "INSERT OR IGNORE INTO releases (app, target, version, channel, published_at)
725 - VALUES (?, ?, ?, ?, ?)",
726 - )
727 - .bind(app_s)
728 - .bind(target_s)
729 - .bind(ver_s)
730 - .bind(chan_s)
731 - .bind(Self::now())
732 - .execute(&me.pool)
733 - .await;
734 - });
747 + self.rt
748 + .block_on(async move {
749 + sqlx::query(
750 + "INSERT OR IGNORE INTO releases (app, target, version, channel, published_at)
751 + VALUES (?, ?, ?, ?, ?)",
752 + )
753 + .bind(app_s)
754 + .bind(target_s)
755 + .bind(ver_s)
756 + .bind(chan_s)
757 + .bind(Self::now())
758 + .execute(&me.pool)
759 + .await
760 + })
761 + .context("recording release in the idempotency ledger (artifact published but ledger write failed)")?;
735 762 events::emit(
736 763 &self.events,
737 764 Event::PublishOk { app: self.app.clone(), target: self.target, channel: channel.to_string() },
@@ -747,6 +774,9 @@ fn dir_size(p: &Path) -> Option<i64> {
747 774 let md = entry.metadata().ok()?;
748 775 if md.is_file() {
749 776 total += md.len() as i64;
777 + } else if md.is_dir() {
778 + // Recurse so a bundle dir (a `.app`) reports its real size, not ~0.
779 + total += dir_size(&entry.path()).unwrap_or(0);
750 780 }
751 781 }
752 782 Some(total)
@@ -917,6 +947,38 @@ fn notary_accepted(output: &str) -> bool {
917 947 mod tests {
918 948 use super::*;
919 949
950 + /// Run 3 S1: once the cooperative cancel flag is set (a newer build
951 + /// superseded this run), a step boundary refuses to proceed — the blocking
952 + /// recipe stops at the next `step()` instead of running on and publishing.
953 + #[tokio::test]
954 + async fn begin_step_bails_when_cancelled() {
955 + let dir = tempfile::tempdir().unwrap();
956 + let cfg = Arc::new(Config::for_tests(dir.path()));
957 + let pool = crate::db::open(&cfg.db_path).await.unwrap();
958 + let cancel = Arc::new(AtomicBool::new(true));
959 + let ctx = Arc::new(RecipeCtx::new(
960 + AppId::new("demo"),
961 + Version::parse("0.1.0").unwrap(),
962 + "linux/x86_64".parse().unwrap(),
963 + "fw13".into(),
964 + "/tmp".into(),
965 + 1,
966 + Arc::new(HashMap::new()),
967 + HashMap::new(),
968 + pool,
969 + crate::events::channel(),
970 + cfg,
971 + Arc::new(OtaRegistry::standard("https://makenot.work")),
972 + tokio::runtime::Handle::current(),
973 + cancel.clone(),
974 + ));
975 + // Cancelled: begin_step refuses before touching the DB (the ensure! is
976 + // ahead of any block_on, so this is safe to call from the async test).
977 + let err = ctx.begin_step(Step::Build).unwrap_err();
978 + assert!(err.to_string().contains("supersede"), "got: {err}");
979 + assert!(ctx.is_cancelled());
980 + }
981 +
920 982 #[test]
921 983 fn expand_tilde_handles_home() {
922 984 unsafe { std::env::set_var("HOME", "/home/test") };
M bento/daemon/src/runner.rs +29 -18
@@ -93,23 +93,24 @@ pub async fn start_build(
93 93 let app = app.clone();
94 94 let version = version.clone();
95 95 let key = (app.clone(), target);
96 + let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));
96 97
97 - // Latest-wins: abort any in-flight run for this (app, target).
98 + // Latest-wins, done as ONE critical section: supersede the prior occupant
99 + // (cooperatively cancel its recipe + abort its task) and install ours
100 + // without ever releasing the lock, so two concurrent /build+/retry for the
101 + // same (app, target) can't both pass "nothing to abort" and run together.
102 + let mut active = state.active.lock().await;
103 + if let Some(prev) = active.remove(&key)
104 + && !prev.abort.is_finished()
98 105 {
99 - let mut active = state.active.lock().await;
100 - if let Some((_, prev)) = active.remove(&key)
101 - && !prev.is_finished() {
102 - events::emit(
103 - &state.events,
104 - Event::TargetAborted { app: app.clone(), target },
105 - );
106 - prev.abort();
107 - }
106 + // The recipe runs on a blocking thread; abort() alone can't stop it,
107 + // so set the cooperative flag the engine checks at step boundaries.
108 + prev.cancel.store(true, std::sync::atomic::Ordering::SeqCst);
109 + prev.abort.abort();
110 + events::emit(&state.events, Event::TargetAborted { app: app.clone(), target });
108 111 }
109 -
110 - let abort = set.spawn(run_target(state.clone(), build_id, app, version, target));
111 - let mut active = state.active.lock().await;
112 - active.insert(key, (build_id, abort));
112 + let abort = set.spawn(run_target(state.clone(), build_id, app, version, target, cancel.clone()));
113 + active.insert(key, crate::state::ActiveSlot { build_id, abort, cancel });
113 114 crate::metrics::set_in_flight(active.len());
114 115 }
115 116
@@ -119,8 +120,17 @@ pub async fn start_build(
119 120 Ok(build_id)
120 121 }
121 122
122 - /// Run one target's recipe end to end, updating its `target_runs` row.
123 - async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version, target: Target) {
123 + /// Run one target's recipe end to end, updating its `target_runs` row. `cancel`
124 + /// is set when a newer build supersedes this one; the engine checks it at step
125 + /// boundaries and before publish so a superseded recipe can't advance or ship.
126 + async fn run_target(
127 + state: AppState,
128 + build_id: i64,
129 + app: AppId,
130 + version: Version,
131 + target: Target,
132 + cancel: Arc<std::sync::atomic::AtomicBool>,
133 + ) {
124 134 let started = std::time::Instant::now();
125 135 let target_run_id: i64 = match sqlx::query_scalar(
126 136 "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
@@ -198,6 +208,7 @@ async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version
198 208 state.cfg.clone(),
199 209 state.ota.clone(),
200 210 tokio::runtime::Handle::current(),
211 + cancel,
201 212 ));
202 213
203 214 // Rhai is synchronous; run the recipe (and its final step finalization) on
@@ -262,7 +273,7 @@ async fn fail_target(
262 273 error: &str,
263 274 ) {
264 275 if let Err(e) = sqlx::query(
265 - "UPDATE target_runs SET status = 'failed', error = ?, finished_at = ? WHERE id = ?",
276 + "UPDATE target_runs SET status = 'failed', current_step = NULL, error = ?, finished_at = ? WHERE id = ?",
266 277 )
267 278 .bind(error)
268 279 .bind(chrono::Utc::now().to_rfc3339())
@@ -312,7 +323,7 @@ async fn finalize_build(state: AppState, build_id: i64, mut set: tokio::task::Jo
312 323 // owns a different build_id and is left intact).
313 324 {
314 325 let mut active = state.active.lock().await;
315 - active.retain(|_, (bid, _)| *bid != build_id);
326 + active.retain(|_, slot| slot.build_id != build_id);
316 327 crate::metrics::set_in_flight(active.len());
317 328 }
318 329
M bento/daemon/src/state.rs +16 -5
@@ -8,6 +8,7 @@ use ops_exec::{AgentRpc, CapabilitySet, Executor, LocalExec, SshExec};
8 8 use sqlx::SqlitePool;
9 9 use std::collections::HashMap;
10 10 use std::sync::Arc;
11 + use std::sync::atomic::AtomicBool;
11 12 use tokio::sync::Mutex;
12 13 use tokio::task::AbortHandle;
13 14
@@ -18,11 +19,21 @@ use tokio::task::AbortHandle;
18 19 /// `ExecutorMap`.
19 20 pub type ExecutorMap = HashMap<String, Arc<dyn Executor>>;
20 21
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)>>>;
22 + /// One occupant of the latest-wins guard: the owning `build_id`, an
23 + /// [`AbortHandle`] for the spawned target task (stops the async wrapper), and a
24 + /// cooperative `cancel` flag the recipe checks at step boundaries and before
25 + /// publish (stops the *blocking* Rhai body, which `abort()` alone cannot reach).
26 + #[derive(Clone)]
27 + pub struct ActiveSlot {
28 + pub build_id: i64,
29 + pub abort: AbortHandle,
30 + pub cancel: Arc<AtomicBool>,
31 + }
32 +
33 + /// Latest-wins guard map: per `(app, target)`, the in-flight occupant. A newer
34 + /// build supersedes the slot (setting the prior `cancel` and aborting its
35 + /// handle); a finishing build reaps only the slots it still owns.
36 + pub type ActiveBuilds = Arc<Mutex<HashMap<(AppId, Target), ActiveSlot>>>;
26 37
27 38 #[derive(Clone)]
28 39 pub struct AppState {
M shared/ops-exec/src/capability.rs +25 -3
@@ -36,10 +36,21 @@ impl CapabilitySet {
36 36 O: IntoIterator,
37 37 O::Item: AsRef<str>,
38 38 {
39 - Self {
40 - actuate: actuate.into_iter().map(|t| t.as_ref().to_string()).collect(),
41 - observe: observe.into_iter().map(|t| ObserveKind::from_token(t.as_ref())).collect(),
39 + let actuate: BTreeSet<String> = actuate.into_iter().map(|t| t.as_ref().to_string()).collect();
40 + let mut observe: BTreeSet<ObserveKind> =
41 + observe.into_iter().map(|t| ObserveKind::from_token(t.as_ref())).collect();
42 + // A host that can `sign` can, by definition, verify the signature (spctl /
43 + // Gatekeeper). The publish gate *requires* that verification —
44 + // `verify_gatekeeper` dispatches `Observe("gatekeeper")` and
45 + // `PublishAuthority::prove` bars every macOS/iOS publish without a
46 + // Some(true) verdict. Grant the observe implicitly here, the one chokepoint
47 + // all three grant sources (topology executor, agent self-grant, agent
48 + // caller allow-list) flow through, so a single missing token in any of them
49 + // can't silently dead-end signing.
50 + if actuate.contains("sign") {
51 + observe.insert(ObserveKind::Custom("gatekeeper".into()));
42 52 }
53 + Self { actuate, observe }
43 54 }
44 55
45 56 /// A set that permits exactly the given actuate actions and no observe.
@@ -131,6 +142,17 @@ mod tests {
131 142 }
132 143
133 144 #[test]
145 + fn sign_grant_implies_gatekeeper_observe() {
146 + // A sign host can run verify_gatekeeper without declaring the observe
147 + // token — otherwise every macOS/iOS publish dead-ends at the gate.
148 + let caps = CapabilitySet::from_tokens(["build", "sign", "notarize", "staple"], ["build-log"]);
149 + assert!(caps.permits(&Action::Observe(ObserveKind::Custom("gatekeeper".into()))));
150 + // A non-signing host gets no such implicit grant.
151 + let plain = CapabilitySet::from_tokens(["build"], ["build-log"]);
152 + assert!(!plain.permits(&Action::Observe(ObserveKind::Custom("gatekeeper".into()))));
153 + }
154 +
155 + #[test]
134 156 fn custom_action_needs_exact_grant() {
135 157 let caps = CapabilitySet::from_tokens(["smoke-test"], Vec::<&str>::new());
136 158 assert!(caps.permits(&Action::Custom("smoke-test".into())));