Skip to main content

max / makenotwork

sando: restart reconcile + real metrics + split event bus (Perf A+, ultra-fuzz Run 2 deep Phase 4) Drive the Performance/Resilience axis from A- to A+: - startup reconcile (SERIOUS-2): recover_orphaned_running settles any build_runs left 'building' by a daemon that died mid-build, run once at startup. A crash/restart/self-update SIGKILL no longer leaves a run wedged 'building' with an ever-growing /state elapsed. - real deploy metrics: /metrics was a live endpoint scraping an empty registry. Now emits sando_builds_total{result} + sando_build_seconds, sando_gate_runs _total{gate,status} + sando_gate_seconds, sando_promotes_total{tier}, sando_rollbacks_total{tier}, and a sando_tier_partial{tier} gauge. - split event bus: GateLogChunk (high-rate) and status events now ride separate broadcast channels; the WS handler merges them, so a busy gate's log firehose can no longer evict a PromoteComplete/GateDone via a shared-bus Lagged. - self_update also refuses during an in-flight promote/rollback (try_lock on deploy_lock), not just an in-flight build — the restart would SIGKILL a deploy. Tests: orphaned-running reconcile (settles building, leaves settled, idempotent); status event survives a log-channel overflow. clippy -D warnings clean; native fw13. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 03:40 UTC
Commit: d42fe39b5852e9a1ecd0ed3691e5540eaa01306e
Parent: 215112b
7 files changed, +166 insertions, -19 deletions
M sando/daemon/src/build.rs +4
@@ -139,6 +139,8 @@ pub async fn run(
139 139 crate::events::emit(&events, crate::events::Event::BuildFailed {
140 140 sha: sha.clone(), version: version.clone(), elapsed_s,
141 141 });
142 + metrics::counter!("sando_builds_total", "result" => "failed").increment(1);
143 + metrics::histogram!("sando_build_seconds").record(elapsed_s as f64);
142 144 // Settle the run with the headline compiler diagnostic (not the raw
143 145 // 4 KB tail) so `GET /runs/{id}` answers "why" without a journald dive.
144 146 let summary = crate::classify::classify_compile_error(&out.stdout, &out.stderr).summary();
@@ -151,6 +153,8 @@ pub async fn run(
151 153 crate::events::emit(&events, crate::events::Event::BuildOk {
152 154 sha: sha.clone(), version: version.clone(), elapsed_s,
153 155 });
156 + metrics::counter!("sando_builds_total", "result" => "ok").increment(1);
157 + metrics::histogram!("sando_build_seconds").record(elapsed_s as f64);
154 158
155 159 // Binaries land under `<target>/release/`; with a shared target dir that's
156 160 // not inside the worktree, so resolve it the same way cargo did above.
M sando/daemon/src/events.rs +76 -13
@@ -11,12 +11,42 @@ use chrono::{DateTime, Utc};
11 11 use serde::{Deserialize, Serialize};
12 12 use tokio::sync::broadcast;
13 13
14 - /// Capacity of the broadcast channel. Slow subscribers that fall behind by
15 - /// more than this many events get `RecvError::Lagged`; the WS handler treats
16 - /// that as a recoverable hiccup, not a disconnect.
17 - pub const CAPACITY: usize = 256;
14 + /// Capacity of the low-rate *status* channel (build/gate/deploy/promote
15 + /// lifecycle events). A subscriber falling behind by more than this gets
16 + /// `RecvError::Lagged`; the WS handler treats that as a recoverable hiccup.
17 + pub const STATUS_CAPACITY: usize = 256;
18 18
19 - pub type EventTx = broadcast::Sender<EventEnvelope>;
19 + /// Capacity of the high-rate *log* channel (`GateLogChunk` — a busy gate emits
20 + /// hundreds per second). Larger so a momentarily-paused TUI lags the log stream
21 + /// rather than the status stream. Splitting the two is what keeps a log firehose
22 + /// from evicting a `PromoteComplete` the operator needs to see (Run-2: both
23 + /// rode one 256-slot bus, so a Lagged dropped status events alongside the spam).
24 + pub const LOG_CAPACITY: usize = 1024;
25 +
26 + /// Two broadcast channels behind one handle: low-rate status and high-rate log
27 + /// chunks. `emit` routes each event to the right channel; the WS handler
28 + /// subscribes to both and merges them, so a lag on one never drops the other.
29 + #[derive(Clone)]
30 + pub struct EventBus {
31 + status: broadcast::Sender<EventEnvelope>,
32 + logs: broadcast::Sender<EventEnvelope>,
33 + }
34 +
35 + impl EventBus {
36 + /// Subscribe to the low-rate status stream (everything but `GateLogChunk`).
37 + pub fn subscribe_status(&self) -> broadcast::Receiver<EventEnvelope> {
38 + self.status.subscribe()
39 + }
40 + /// Subscribe to the high-rate `GateLogChunk` stream.
41 + pub fn subscribe_logs(&self) -> broadcast::Receiver<EventEnvelope> {
42 + self.logs.subscribe()
43 + }
44 + }
45 +
46 + /// The handle build/gate/deploy sites hold and emit on. Kept named `EventTx` so
47 + /// the many `events: EventTx` fields and `s.events.clone()` sites are unchanged
48 + /// by the channel split.
49 + pub type EventTx = EventBus;
20 50
21 51 #[derive(Clone, Debug, Serialize, Deserialize)]
22 52 pub struct EventEnvelope {
@@ -79,14 +109,22 @@ pub enum Event {
79 109 }
80 110
81 111 pub fn channel() -> EventTx {
82 - broadcast::channel(CAPACITY).0
112 + EventBus {
113 + status: broadcast::channel(STATUS_CAPACITY).0,
114 + logs: broadcast::channel(LOG_CAPACITY).0,
115 + }
83 116 }
84 117
85 - /// Send an event without caring whether anyone is listening. The `send` call
86 - /// fails only when there are zero subscribers, which is the normal case for
87 - /// most operator-tool deployments.
88 - pub fn emit(tx: &EventTx, event: Event) {
118 + /// Send an event without caring whether anyone is listening. `GateLogChunk` goes
119 + /// to the high-rate log channel, everything else to the status channel. The
120 + /// `send` fails only when there are zero subscribers, which is the normal case
121 + /// for most operator-tool deployments.
122 + pub fn emit(bus: &EventTx, event: Event) {
89 123 let envelope = EventEnvelope { at: Utc::now(), event };
124 + let tx = match &envelope.event {
125 + Event::GateLogChunk { .. } => &bus.logs,
126 + _ => &bus.status,
127 + };
90 128 let _ = tx.send(envelope);
91 129 }
92 130
@@ -108,7 +146,7 @@ mod tests {
108 146 #[tokio::test]
109 147 async fn emit_reaches_a_subscriber() {
110 148 let tx = channel();
111 - let mut rx = tx.subscribe();
149 + let mut rx = tx.subscribe_status();
112 150 emit(&tx, Event::PromoteComplete {
113 151 tier: TierId::new("a"),
114 152 version: "0.8.12".parse().unwrap(),
@@ -124,6 +162,31 @@ mod tests {
124 162 }
125 163
126 164 #[tokio::test]
165 + async fn status_survives_a_log_firehose() {
166 + // The point of the split: flooding the log channel past its capacity
167 + // must NOT evict a status event. Subscribe to both, overflow logs, then
168 + // confirm the status event still arrives.
169 + let tx = channel();
170 + let mut status_rx = tx.subscribe_status();
171 + let mut logs_rx = tx.subscribe_logs();
172 + emit(&tx, Event::PromoteComplete {
173 + tier: TierId::new("b"),
174 + version: "0.8.12".parse().unwrap(),
175 + });
176 + for i in 0..(LOG_CAPACITY + 50) {
177 + emit(&tx, Event::GateLogChunk { run_id: GateRunId(1), seq: i as u32, text: "x".into() });
178 + }
179 + // The status event is intact despite the log overflow.
180 + let env = status_rx.recv().await.expect("status event survived");
181 + assert!(matches!(env.event, Event::PromoteComplete { .. }));
182 + // The log channel is the one that lagged.
183 + assert!(matches!(
184 + logs_rx.recv().await,
185 + Err(broadcast::error::RecvError::Lagged(_))
186 + ));
187 + }
188 +
189 + #[tokio::test]
127 190 async fn envelope_serializes_with_flat_kind() {
128 191 // Contract for the WS handler + TUI's `format_event`: the JSON has a
129 192 // top-level `kind` field, not nested under `event`. Locking this in.
@@ -151,8 +214,8 @@ mod tests {
151 214 // recv() returns RecvError::Lagged(n) — not Closed, not a panic.
152 215 // The WS handler turns this into a `lagged` envelope.
153 216 let tx = channel();
154 - let mut rx = tx.subscribe();
155 - for i in 0..(CAPACITY + 10) {
217 + let mut rx = tx.subscribe_status();
218 + for i in 0..(STATUS_CAPACITY + 10) {
156 219 // 7+ hex chars satisfy GitSha::parse; pad i into that shape.
157 220 let sha = GitSha::parse(&format!("{i:0>7x}")).unwrap();
158 221 emit(&tx, Event::RebuildRequested { sha });
M sando/daemon/src/gates.rs +6
@@ -46,6 +46,7 @@ pub struct NodeProbe {
46 46 pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
47 47 let kind = gate.kind();
48 48 let started_at = Utc::now().to_rfc3339();
49 + let t0 = std::time::Instant::now();
49 50
50 51 let id: i64 = sqlx::query_scalar(
51 52 "INSERT INTO gate_runs (version, tier, gate_kind, started_at) VALUES (?, ?, ?, ?)
@@ -124,6 +125,11 @@ pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
124 125 outcome: outcome.clone(),
125 126 });
126 127
128 + metrics::counter!("sando_gate_runs_total", "gate" => kind.as_str(), "status" => outcome.status_str())
129 + .increment(1);
130 + metrics::histogram!("sando_gate_seconds", "gate" => kind.as_str())
131 + .record(t0.elapsed().as_secs_f64());
132 +
127 133 Ok(outcome)
128 134 }
129 135
M sando/daemon/src/live_log.rs +3 -3
@@ -113,7 +113,7 @@ mod tests {
113 113 let dir = tempfile::tempdir().unwrap();
114 114 let path = dir.path().join("nested/test.log");
115 115 let events = events::channel();
116 - let mut rx = events.subscribe();
116 + let mut rx = events.subscribe_logs();
117 117 let mut log = LiveLog::open(events.clone(), GateRunId(7), path.clone()).await;
118 118 log.write_chunk(b"hello ").await;
119 119 log.write_chunk(b"world\n").await;
@@ -142,7 +142,7 @@ mod tests {
142 142 tokio::fs::write(&blocker, b"i am a file, not a dir").await.unwrap();
143 143 let path = blocker.join("inside.log"); // parent is a file
144 144 let events = events::channel();
145 - let mut rx = events.subscribe();
145 + let mut rx = events.subscribe_logs();
146 146 let mut log = LiveLog::open(events.clone(), GateRunId(1), path).await;
147 147 log.write_chunk(b"streamed despite no file\n").await;
148 148 log.close().await;
@@ -159,7 +159,7 @@ mod tests {
159 159 let dir = tempfile::tempdir().unwrap();
160 160 let path = dir.path().join("empty.log");
161 161 let events = events::channel();
162 - let mut rx = events.subscribe();
162 + let mut rx = events.subscribe_logs();
163 163 let mut log = LiveLog::open(events.clone(), GateRunId(9), path).await;
164 164 log.write_chunk(b"").await;
165 165 assert_eq!(log.chunks_emitted(), 0);
M sando/daemon/src/main.rs +8 -1
@@ -1,5 +1,5 @@
1 1 use anyhow::Result;
2 - use sando_daemon::{config, db, events, git, metrics, routes, state, sync, topology};
2 + use sando_daemon::{config, db, events, git, metrics, routes, runs, state, sync, topology};
3 3 use std::net::SocketAddr;
4 4 use std::path::Path;
5 5 use std::sync::Arc;
@@ -35,6 +35,13 @@ async fn main() -> Result<()> {
35 35 git::ensure_bare_repo(Path::new(&topo.repo.bare_path)).await?;
36 36 let pool = db::connect(&cfg.db_path).await?;
37 37 db::migrate(&pool).await?;
38 + // Reconcile any build_runs left 'building' by a previous daemon that died
39 + // mid-build, so a restart can't leave a run wedged 'building' forever.
40 + match runs::recover_orphaned_running(&pool).await {
41 + Ok(0) => {}
42 + Ok(n) => tracing::warn!(reconciled = n, "settled orphaned 'building' run(s) from a prior daemon"),
43 + Err(e) => tracing::error!(error = %e, "failed to reconcile orphaned 'building' runs at startup"),
44 + }
38 45 sync::sync(&pool, &topo).await?;
39 46 tracing::info!(tiers = topo.tiers.len(), bare = %topo.repo.bare_path, "topology synced");
40 47
M sando/daemon/src/routes.rs +24 -2
@@ -496,6 +496,7 @@ async fn promote(
496 496 crate::events::emit(&s.events, crate::events::Event::PromoteComplete {
497 497 tier: target.name.clone(), version: version.clone(),
498 498 });
499 + metrics::counter!("sando_promotes_total", "tier" => target.name.to_string()).increment(1);
499 500 tracing::info!(
500 501 version = %version, tier = %target.name,
501 502 hotfix = body.hotfix, reset_burn_in = body.reset_burn_in,
@@ -580,6 +581,7 @@ async fn set_partial(s: &AppState, tier: &crate::domain::TierId, reason: &str) {
580 581 tracing::error!(tier = %tier, reason, error = %e,
581 582 "failed to record tier partial state; the fleet may be inconsistent without a /state flag");
582 583 }
584 + metrics::gauge!("sando_tier_partial", "tier" => tier.to_string()).set(1.0);
583 585 }
584 586
585 587 /// Clear a tier's partial flag after a clean full promote or rollback. Errors are
@@ -594,6 +596,7 @@ async fn clear_partial(s: &AppState, tier: &crate::domain::TierId) {
594 596 {
595 597 tracing::warn!(tier = %tier, error = %e, "failed to clear tier partial flag");
596 598 }
599 + metrics::gauge!("sando_tier_partial", "tier" => tier.to_string()).set(0.0);
597 600 }
598 601
599 602 /// Returns the kinds of `tier`'s *configured* gates that are not satisfied for
@@ -780,6 +783,7 @@ async fn rollback(
780 783 crate::events::emit(&s.events, crate::events::Event::Rollback {
781 784 tier: tier.clone(), from: current.clone(), to: previous.clone(),
782 785 });
786 + metrics::counter!("sando_rollbacks_total", "tier" => tier.to_string()).increment(1);
783 787
784 788 Ok(Json(serde_json::json!({
785 789 "tier": tier,
@@ -969,6 +973,15 @@ async fn self_update(
969 973 "a server build is in flight; retry /self-update once it settles".into(),
970 974 ));
971 975 }
976 + // Also refuse during an in-flight promote/rollback: the updater's restart
977 + // would SIGKILL the daemon mid-deploy. try_lock (not lock) so we reject
978 + // rather than queue behind a long deploy; the guard is dropped immediately
979 + // — it only needs to observe that no deploy holds it right now.
980 + if s.deploy_lock.try_lock().is_err() {
981 + return Err(crate::error::Error::GateBlocked(
982 + "a promote/rollback is in flight; retry /self-update once it settles".into(),
983 + ));
984 + }
972 985
973 986 tracing::warn!(sha = %sha, unit = %unit, "self-update requested; triggering privileged updater");
974 987 // `--no-block`: return as soon as the job is enqueued. The build+restart
@@ -1116,9 +1129,18 @@ async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl Into
1116 1129 use tokio::sync::broadcast::error::RecvError;
1117 1130
1118 1131 ws.on_upgrade(move |mut socket| async move {
1119 - let mut rx = s.events.subscribe();
1132 + // Subscribe to both channels and merge them: a lag on the high-rate log
1133 + // stream emits its own `lagged` frame without dropping anything on the
1134 + // status stream (and vice versa), so a busy gate's chunk firehose can't
1135 + // evict a PromoteComplete/GateDone the operator needs to see.
1136 + let mut status_rx = s.events.subscribe_status();
1137 + let mut logs_rx = s.events.subscribe_logs();
1120 1138 loop {
1121 - match rx.recv().await {
1139 + let recv = tokio::select! {
1140 + r = status_rx.recv() => r,
1141 + r = logs_rx.recv() => r,
1142 + };
1143 + match recv {
1122 1144 Ok(env) => {
1123 1145 let json = match serde_json::to_string(&env) {
1124 1146 Ok(s) => s,
M sando/daemon/src/runs.rs +45
@@ -155,6 +155,24 @@ pub async fn mark_aborted(pool: &SqlitePool, run_id: RunId) -> Result<()> {
155 155 Ok(())
156 156 }
157 157
158 + /// Settle any `build_runs` left `result = 'building'` by a daemon that died
159 + /// mid-build (crash, OOM, `systemctl restart`, a self-update that SIGKILLed an
160 + /// in-flight build). Without this they stay `'building'` forever and `/state` +
161 + /// `GET /runs/{id}/wait` report an ever-growing elapsed for a build that will
162 + /// never settle (the Run-2 SERIOUS-2 gap). Run once at startup before serving.
163 + /// Returns the number of orphaned runs reconciled.
164 + pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
165 + let res = sqlx::query(
166 + "UPDATE build_runs SET result = 'aborted', phase = 'done',
167 + failure_summary = 'daemon restarted mid-build', finished_at = ?
168 + WHERE result = 'building'",
169 + )
170 + .bind(Utc::now().to_rfc3339())
171 + .execute(pool)
172 + .await?;
173 + Ok(res.rows_affected())
174 + }
175 +
158 176 /// One gate's status within a run view.
159 177 #[derive(Debug, Serialize)]
160 178 pub struct RunGateView {
@@ -346,6 +364,33 @@ mod tests {
346 364 }
347 365
348 366 #[tokio::test]
367 + async fn recover_orphaned_running_settles_building_runs() {
368 + let pool = pool().await;
369 + // Two in-flight runs (as if the daemon died mid-build) + one already
370 + // settled, which must be left untouched.
371 + let a = create(&pool, "aaaaaaa").await.unwrap();
372 + let b = create(&pool, "bbbbbbb").await.unwrap();
373 + let c = create(&pool, "ccccccc").await.unwrap();
374 + mark_passed(&pool, c).await.unwrap();
375 +
376 + let n = recover_orphaned_running(&pool).await.unwrap();
377 + assert_eq!(n, 2, "both 'building' runs reconciled");
378 +
379 + for id in [a, b] {
380 + let v = get(&pool, id).await.unwrap().unwrap();
381 + assert_eq!(v.result, "aborted");
382 + assert_eq!(v.phase, "done");
383 + assert!(v.finished_at.is_some());
384 + assert_eq!(v.failure_summary.as_deref(), Some("daemon restarted mid-build"));
385 + }
386 + // The already-settled run is unchanged.
387 + assert_eq!(get(&pool, c).await.unwrap().unwrap().result, "passed");
388 +
389 + // Idempotent: a second pass finds nothing to do.
390 + assert_eq!(recover_orphaned_running(&pool).await.unwrap(), 0);
391 + }
392 +
393 + #[tokio::test]
349 394 async fn phase_and_version_advance_then_pass() {
350 395 let pool = pool().await;
351 396 let id = create(&pool, "abc1234").await.unwrap();