Skip to main content

max / makenotwork

ops-core: lift Sando's split event bus into the shared crate Sando's bus deliberately diverged from ops-core's in d42fe39b (ultra-fuzz Run 2): two broadcast channels, low-rate status at 256 and high-rate log chunks at 1024, because a GateLogChunk firehose on one shared 256-slot ring evicted the PromoteComplete events an operator must not miss. ops-core still carried the original single-channel scaffold, so adopting it in Sando -- as GO task 04b05963 asked -- would have re-broken Run 2. Do the inverse: generalize Sando's design into ops-core, with the payload classifying itself via the new BusEvent::is_high_rate, and have both tools adopt it. Sando's events.rs collapses to its payload plus the classifier; all five of its bus tests pass unchanged, including status_survives_a_log_firehose -- the behavior-neutrality proof. Bento was the tool actually exposed: it used ops-core's single-channel bus with StepLogChunk riding the same ring as PublishFailed/TargetFailed, so a noisy build log could evict a publish failure from the TUI. Its WS handler now merges both channels the way Sando's already did, closing the bug with a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-16 16:37 UTC
Commit: 28a7dd49a42653c5150111d238e966f222f104ce
Parent: 9340869
7 files changed, +240 insertions, -80 deletions
M bento/daemon/src/events.rs +69 -1
@@ -3,16 +3,26 @@
3 3 //! shape the TUI parses (`{"kind":"step_start", ...}`).
4 4
5 5 use crate::domain::{AppId, Status, Step, StepRunId, Target, Version};
6 - use ops_core::eventbus;
6 + use ops_core::eventbus::{self, BusEvent};
7 7 use serde::{Deserialize, Serialize};
8 8
9 9 pub type EventEnvelope = eventbus::EventEnvelope<Event>;
10 10 pub type EventTx = eventbus::EventTx<Event>;
11 11
12 + impl BusEvent for Event {
13 + /// A busy build step emits `StepLogChunk` hundreds of times a second; every
14 + /// other variant is lifecycle status the operator must not miss.
15 + fn is_high_rate(&self) -> bool {
16 + matches!(self, Event::StepLogChunk { .. })
17 + }
18 + }
19 +
12 20 pub fn channel() -> EventTx {
13 21 eventbus::channel()
14 22 }
15 23
24 + /// Send an event without caring whether anyone is listening. `StepLogChunk` goes
25 + /// to the high-rate log channel, everything else to the status channel.
16 26 pub fn emit(tx: &EventTx, event: Event) {
17 27 eventbus::emit(tx, event)
18 28 }
@@ -41,3 +51,61 @@ pub enum Event {
41 51 PublishOk { app: AppId, target: Target, channel: String },
42 52 PublishFailed { app: AppId, target: Target, channel: String, error: String },
43 53 }
54 +
55 + #[cfg(test)]
56 + mod tests {
57 + use super::*;
58 + use ops_core::eventbus::LOG_CAPACITY;
59 +
60 + fn target() -> Target {
61 + "macos/aarch64".parse().expect("valid target")
62 + }
63 +
64 + fn publish_failed() -> Event {
65 + Event::PublishFailed {
66 + app: AppId::new("goingson"),
67 + target: target(),
68 + channel: "stable".into(),
69 + error: "boom".into(),
70 + }
71 + }
72 +
73 + #[test]
74 + fn only_log_chunks_are_high_rate() {
75 + assert!(Event::StepLogChunk { run_id: StepRunId(1), seq: 0, text: "x".into() }.is_high_rate());
76 + assert!(!publish_failed().is_high_rate());
77 + }
78 +
79 + #[tokio::test]
80 + async fn status_survives_a_log_firehose() {
81 + // A build step's chunk spam must NOT evict a publish failure. Before the
82 + // split both rode one 256-slot channel, so a lagging TUI lost the
83 + // failure alongside the noise.
84 + let tx = channel();
85 + let mut status_rx = tx.subscribe_status();
86 + let mut logs_rx = tx.subscribe_logs();
87 + emit(&tx, publish_failed());
88 + for i in 0..(LOG_CAPACITY + 50) {
89 + emit(&tx, Event::StepLogChunk { run_id: StepRunId(1), seq: i as u32, text: "x".into() });
90 + }
91 + // The status event is intact despite the log overflow.
92 + let env = status_rx.recv().await.expect("status event survived");
93 + assert!(matches!(env.event, Event::PublishFailed { .. }));
94 + // The log channel is the one that lagged.
95 + assert!(matches!(
96 + logs_rx.recv().await,
97 + Err(tokio::sync::broadcast::error::RecvError::Lagged(_))
98 + ));
99 + }
100 +
101 + #[tokio::test]
102 + async fn envelope_serializes_with_flat_kind() {
103 + // Contract for the WS handler + TUI's `format_event_line`.
104 + let env = EventEnvelope { at: chrono::Utc::now(), event: publish_failed() };
105 + let v: serde_json::Value = serde_json::from_str(&serde_json::to_string(&env).unwrap()).unwrap();
106 + assert_eq!(v["kind"], "publish_failed");
107 + assert_eq!(v["channel"], "stable");
108 + assert!(v.get("event").is_none());
109 + assert!(v.get("at").is_some());
110 + }
111 + }
M bento/daemon/src/routes.rs +11 -2
@@ -278,9 +278,18 @@ async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl Into
278 278 use tokio::sync::broadcast::error::RecvError;
279 279
280 280 ws.on_upgrade(move |mut socket| async move {
281 - let mut rx = s.events.subscribe();
281 + // Subscribe to both channels and merge them: a lag on the high-rate log
282 + // stream emits its own `lagged` frame without dropping anything on the
283 + // status stream (and vice versa), so a busy build's chunk firehose can't
284 + // evict a TargetFailed/PublishOk the operator needs to see.
285 + let mut status_rx = s.events.subscribe_status();
286 + let mut logs_rx = s.events.subscribe_logs();
282 287 loop {
283 - match rx.recv().await {
288 + let recv = tokio::select! {
289 + r = status_rx.recv() => r,
290 + r = logs_rx.recv() => r,
291 + };
292 + match recv {
284 293 Ok(env) => {
285 294 let json = match serde_json::to_string(&env) {
286 295 Ok(s) => s,
M sando/Cargo.lock +15
@@ -1315,6 +1315,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
1315 1315 checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
1316 1316
1317 1317 [[package]]
1318 + name = "ops-core"
1319 + version = "0.1.0"
1320 + dependencies = [
1321 + "anyhow",
1322 + "async-trait",
1323 + "chrono",
1324 + "ops-exec",
1325 + "serde",
1326 + "sqlx",
1327 + "tokio",
1328 + "tracing",
1329 + ]
1330 +
1331 + [[package]]
1318 1332 name = "ops-exec"
1319 1333 version = "0.1.0"
1320 1334 dependencies = [
@@ -1843,6 +1857,7 @@ dependencies = [
1843 1857 "http-body-util",
1844 1858 "metrics",
1845 1859 "metrics-exporter-prometheus",
1860 + "ops-core",
1846 1861 "ops-exec",
1847 1862 "reqwest",
1848 1863 "semver",
M sando/daemon/Cargo.toml +1
@@ -11,6 +11,7 @@ path = "src/main.rs"
11 11 [dependencies]
12 12 axum = { version = "0.8.8", features = ["macros", "ws"] }
13 13 tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread", "net", "signal", "fs", "process"] }
14 + ops-core = { path = "../../shared/ops-core" }
14 15 ops-exec = { path = "../../shared/ops-exec" }
15 16 async-trait = "0.1.83"
16 17 serde = { version = "1.0.228", features = ["derive"] }
M sando/daemon/src/events.rs +25 -60
@@ -1,59 +1,24 @@
1 - //! Event bus for live operator visibility.
1 + //! Sando's concrete event payload, carried on the generic
2 + //! [`ops_core::eventbus`] bus.
2 3 //!
3 4 //! Sites that previously logged via `tracing::info!` also emit a typed event
4 - //! onto a `broadcast::Sender<EventEnvelope>`. The WS handler at `/events`
5 - //! subscribes to the bus and forwards each envelope to the connected TUI as
6 - //! a JSON text frame.
5 + //! onto the bus. The WS handler at `/events` subscribes to both of the bus's
6 + //! channels and forwards each envelope to the connected TUI as a JSON text
7 + //! frame. `GateLogChunk` is the high-rate variant and rides the log channel;
8 + //! see [`ops_core::eventbus`] for why that split exists.
7 9
8 10 use crate::domain::{GateKind, GateRunId, GitSha, NodeId, TierId, Version};
9 11 use crate::outcome::{DeployFailureKind, GateOutcome};
10 - use chrono::{DateTime, Utc};
12 + use ops_core::eventbus::{self, BusEvent};
11 13 use serde::{Deserialize, Serialize};
12 - use tokio::sync::broadcast;
13 14
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 -
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 - }
15 + pub use eventbus::{LOG_CAPACITY, STATUS_CAPACITY};
45 16
46 17 /// 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;
50 -
51 - #[derive(Clone, Debug, Serialize, Deserialize)]
52 - pub struct EventEnvelope {
53 - pub at: DateTime<Utc>,
54 - #[serde(flatten)]
55 - pub event: Event,
56 - }
18 + /// the many `events: EventTx` fields and `s.events.clone()` sites read the same.
19 + pub type EventTx = eventbus::EventTx<Event>;
20 + pub type EventBus = eventbus::EventBus<Event>;
21 + pub type EventEnvelope = eventbus::EventEnvelope<Event>;
57 22
58 23 #[derive(Clone, Debug, Serialize, Deserialize)]
59 24 #[serde(tag = "kind", rename_all = "snake_case")]
@@ -108,29 +73,29 @@ pub enum Event {
108 73 ManualConfirm { tier: TierId, version: Version },
109 74 }
110 75
111 - pub fn channel() -> EventTx {
112 - EventBus {
113 - status: broadcast::channel(STATUS_CAPACITY).0,
114 - logs: broadcast::channel(LOG_CAPACITY).0,
76 + impl BusEvent for Event {
77 + /// A busy gate emits `GateLogChunk` hundreds of times a second; every other
78 + /// variant is lifecycle status the operator must not miss.
79 + fn is_high_rate(&self) -> bool {
80 + matches!(self, Event::GateLogChunk { .. })
115 81 }
116 82 }
117 83
84 + pub fn channel() -> EventTx {
85 + eventbus::channel()
86 + }
87 +
118 88 /// 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.
89 + /// to the high-rate log channel, everything else to the status channel.
122 90 pub fn emit(bus: &EventTx, event: Event) {
123 - let envelope = EventEnvelope { at: Utc::now(), event };
124 - let tx = match &envelope.event {
125 - Event::GateLogChunk { .. } => &bus.logs,
126 - _ => &bus.status,
127 - };
128 - let _ = tx.send(envelope);
91 + eventbus::emit(bus, event)
129 92 }
130 93
131 94 #[cfg(test)]
132 95 mod tests {
133 96 use super::*;
97 + use chrono::Utc;
98 + use tokio::sync::broadcast;
134 99
135 100 #[test]
136 101 fn emit_with_zero_subscribers_does_not_panic() {
M shared/ops-core/src/eventbus.rs +116 -17
@@ -1,7 +1,7 @@
1 1 //! Generic event bus for live operator visibility.
2 2 //!
3 - //! Sando and Bento both broadcast typed events to a connected TUI over a
4 - //! `tokio::sync::broadcast` channel; the WS handler forwards each envelope as
3 + //! Sando and Bento both broadcast typed events to a connected TUI over
4 + //! `tokio::sync::broadcast` channels; the WS handler forwards each envelope as
5 5 //! a JSON text frame. The only thing that differs between the tools is the
6 6 //! concrete event payload, so the bus is generic over `E`.
7 7 //!
@@ -9,15 +9,45 @@
9 9 //! `#[serde(tag = "kind", rename_all = "snake_case")]` serializes with a
10 10 //! top-level `kind` field plus the variant's fields inline — the wire shape
11 11 //! the TUIs parse.
12 + //!
13 + //! ## Why two channels
14 + //!
15 + //! Both tools emit two very different rates of event on one bus: a low-rate
16 + //! *status* stream (a promote completed, a build failed — the things an
17 + //! operator must not miss) and a high-rate *log chunk* stream (hundreds per
18 + //! second from a busy gate or build step). On a single broadcast channel the
19 + //! log firehose evicts the status events, because `broadcast` drops the oldest
20 + //! entries once a subscriber falls `CAPACITY` behind — a subscriber that lags
21 + //! on chunk spam loses the `PromoteComplete` sitting in the same ring.
22 + //!
23 + //! Sando hit this for real (ultra-fuzz Run 2) and split its bus in two. This
24 + //! module is that design, generalized: [`BusEvent::is_high_rate`] classifies
25 + //! each event, [`emit`] routes it to the matching channel, and a WS handler
26 + //! subscribes to both and merges. A lag on one stream can no longer drop the
27 + //! other.
12 28
13 29 use chrono::{DateTime, Utc};
14 30 use serde::{Deserialize, Serialize};
15 31 use tokio::sync::broadcast;
16 32
17 - /// Capacity of the broadcast channel. Subscribers that fall behind by more
18 - /// than this many events get `RecvError::Lagged`; WS handlers treat that as a
19 - /// recoverable hiccup (emit a `lagged` notice), not a disconnect.
20 - pub const CAPACITY: usize = 256;
33 + /// Capacity of the low-rate *status* channel (build/gate/deploy/promote
34 + /// lifecycle events). Subscribers that fall behind by more than this many
35 + /// events get `RecvError::Lagged`; WS handlers treat that as a recoverable
36 + /// hiccup (emit a `lagged` notice), not a disconnect.
37 + pub const STATUS_CAPACITY: usize = 256;
38 +
39 + /// Capacity of the high-rate *log* channel. Larger than [`STATUS_CAPACITY`] so
40 + /// a momentarily-paused TUI lags the log stream rather than the status stream.
41 + pub const LOG_CAPACITY: usize = 1024;
42 +
43 + /// Lets the bus tell a high-rate log-chunk event from a low-rate status event,
44 + /// so each rides its own channel. See the module docs for why this matters.
45 + pub trait BusEvent: Clone {
46 + /// `true` for the high-rate log-chunk variants (Sando's `GateLogChunk`,
47 + /// Bento's `StepLogChunk`) — the ones a busy run emits hundreds of per
48 + /// second. Everything an operator must not miss returns `false`.
49 + fn is_high_rate(&self) -> bool;
50 + }
21 51
22 52 /// An event payload `E` stamped with the time it was emitted.
23 53 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -27,22 +57,52 @@ pub struct EventEnvelope<E> {
27 57 pub event: E,
28 58 }
29 59
30 - /// The broadcast sender every emit site clones. `subscribe()` yields a
31 - /// receiver for the WS handler.
32 - pub type EventTx<E> = broadcast::Sender<EventEnvelope<E>>;
60 + /// Two broadcast channels behind one handle: low-rate status and high-rate log
61 + /// chunks. [`emit`] routes each event to the right one; a WS handler subscribes
62 + /// to both and merges them, so a lag on one never drops the other.
63 + ///
64 + /// This is the handle every emit site clones, and it stays named `EventTx` via
65 + /// the alias below so call sites read the same as a plain sender.
66 + #[derive(Clone)]
67 + pub struct EventBus<E> {
68 + status: broadcast::Sender<EventEnvelope<E>>,
69 + logs: broadcast::Sender<EventEnvelope<E>>,
70 + }
71 +
72 + impl<E: Clone> EventBus<E> {
73 + /// Subscribe to the low-rate status stream (every event that is not
74 + /// [`BusEvent::is_high_rate`]).
75 + pub fn subscribe_status(&self) -> broadcast::Receiver<EventEnvelope<E>> {
76 + self.status.subscribe()
77 + }
78 +
79 + /// Subscribe to the high-rate log-chunk stream.
80 + pub fn subscribe_logs(&self) -> broadcast::Receiver<EventEnvelope<E>> {
81 + self.logs.subscribe()
82 + }
83 + }
84 +
85 + /// The handle every emit site clones. `subscribe_status`/`subscribe_logs` yield
86 + /// receivers for the WS handler.
87 + pub type EventTx<E> = EventBus<E>;
33 88
34 - /// Create a fresh bus. The dropped receiver from `broadcast::channel` is fine
89 + /// Create a fresh bus. The dropped receivers from `broadcast::channel` are fine
35 90 /// — emitters don't care whether anyone is listening (see [`emit`]).
36 91 pub fn channel<E: Clone>() -> EventTx<E> {
37 - broadcast::channel(CAPACITY).0
92 + EventBus {
93 + status: broadcast::channel(STATUS_CAPACITY).0,
94 + logs: broadcast::channel(LOG_CAPACITY).0,
95 + }
38 96 }
39 97
40 - /// Send an event without caring whether anyone is subscribed. `send` errors
98 + /// Send an event without caring whether anyone is subscribed. High-rate events
99 + /// go to the log channel, everything else to the status channel. `send` errors
41 100 /// only when there are zero receivers, which is the normal idle case for an
42 101 /// operator tool, so the error is intentionally dropped.
43 - pub fn emit<E: Clone>(tx: &EventTx<E>, event: E) {
102 + pub fn emit<E: BusEvent>(tx: &EventTx<E>, event: E) {
44 103 let envelope = EventEnvelope { at: Utc::now(), event };
45 - let _ = tx.send(envelope);
104 + let sender = if envelope.event.is_high_rate() { &tx.logs } else { &tx.status };
105 + let _ = sender.send(envelope);
46 106 }
47 107
48 108 #[cfg(test)]
@@ -55,6 +115,13 @@ mod tests {
55 115 enum TestEvent {
56 116 Started { name: String },
57 117 Done { code: i32 },
118 + LogChunk { seq: u32 },
119 + }
120 +
121 + impl BusEvent for TestEvent {
122 + fn is_high_rate(&self) -> bool {
123 + matches!(self, TestEvent::LogChunk { .. })
124 + }
58 125 }
59 126
60 127 #[test]
@@ -66,12 +133,44 @@ mod tests {
66 133 #[tokio::test]
67 134 async fn emit_reaches_a_subscriber() {
68 135 let tx = channel::<TestEvent>();
69 - let mut rx = tx.subscribe();
136 + let mut rx = tx.subscribe_status();
70 137 emit(&tx, TestEvent::Done { code: 7 });
71 138 let env = rx.recv().await.expect("envelope");
72 139 assert_eq!(env.event, TestEvent::Done { code: 7 });
73 140 }
74 141
142 + #[tokio::test]
143 + async fn high_rate_events_ride_the_log_channel() {
144 + let tx = channel::<TestEvent>();
145 + let mut status_rx = tx.subscribe_status();
146 + let mut logs_rx = tx.subscribe_logs();
147 + emit(&tx, TestEvent::LogChunk { seq: 1 });
148 + // It arrives on logs...
149 + assert_eq!(logs_rx.recv().await.unwrap().event, TestEvent::LogChunk { seq: 1 });
150 + // ...and never on status.
151 + assert!(status_rx.try_recv().is_err(), "log chunk must not hit the status channel");
152 + }
153 +
154 + #[tokio::test]
155 + async fn status_survives_a_log_firehose() {
156 + // The whole point of the split: flooding the log channel past its
157 + // capacity must NOT evict a status event.
158 + let tx = channel::<TestEvent>();
159 + let mut status_rx = tx.subscribe_status();
160 + let mut logs_rx = tx.subscribe_logs();
161 + emit(&tx, TestEvent::Done { code: 0 });
162 + for i in 0..(LOG_CAPACITY + 50) {
163 + emit(&tx, TestEvent::LogChunk { seq: i as u32 });
164 + }
165 + // The status event is intact despite the log overflow.
166 + assert_eq!(status_rx.recv().await.expect("status survived").event, TestEvent::Done { code: 0 });
167 + // The log channel is the one that lagged.
168 + assert!(matches!(
169 + logs_rx.recv().await,
170 + Err(broadcast::error::RecvError::Lagged(_))
171 + ));
172 + }
173 +
75 174 #[test]
76 175 fn envelope_serializes_with_flat_kind() {
77 176 let env = EventEnvelope {
@@ -91,8 +190,8 @@ mod tests {
91 190 #[tokio::test]
92 191 async fn lagged_subscriber_observes_recv_error_lagged() {
93 192 let tx = channel::<TestEvent>();
94 - let mut rx = tx.subscribe();
95 - for i in 0..(CAPACITY + 10) {
193 + let mut rx = tx.subscribe_status();
194 + for i in 0..(STATUS_CAPACITY + 10) {
96 195 emit(&tx, TestEvent::Done { code: i as i32 });
97 196 }
98 197 let err = rx.recv().await.expect_err("expected Lagged");
M shared/ops-core/src/lib.rs +3
@@ -12,6 +12,9 @@
12 12 //! its transports live in `ops-exec` proper (re-exported here as [`ops_exec`]).
13 13 //! - [`eventbus`] — a generic `EventEnvelope<E>` broadcast bus with flat-`kind`
14 14 //! serialization (each tool supplies its own concrete `Event` enum as `E`).
15 + //! Split into a low-rate status channel and a high-rate log-chunk channel, so
16 + //! a log firehose can't evict the status events an operator must not miss;
17 + //! `E` classifies itself via [`eventbus::BusEvent`].
15 18 //! - [`live_log`] — a disk-append + callback live-log sink that implements
16 19 //! `LogSink`; parameterized over a chunk callback so it is not tied to any
17 20 //! tool's `Event` enum.