Skip to main content

max / makenotwork

8.1 KB · 222 lines History Blame Raw
1 //! Generic event bus for live operator visibility.
2 //!
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 //! a JSON text frame. The only thing that differs between the tools is the
6 //! concrete event payload, so the bus is generic over `E`.
7 //!
8 //! The envelope flattens the payload, so a payload enum tagged
9 //! `#[serde(tag = "kind", rename_all = "snake_case")]` serializes with a
10 //! top-level `kind` field plus the variant's fields inline — the wire shape
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.
28
29 use chrono::{DateTime, Utc};
30 use serde::{Deserialize, Serialize};
31 use tokio::sync::broadcast;
32
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 }
51
52 /// An event payload `E` stamped with the time it was emitted.
53 #[derive(Clone, Debug, Serialize, Deserialize)]
54 pub struct EventEnvelope<E> {
55 pub at: DateTime<Utc>,
56 #[serde(flatten)]
57 pub event: E,
58 }
59
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>;
88
89 /// Create a fresh bus. The dropped receivers from `broadcast::channel` are fine
90 /// — emitters don't care whether anyone is listening (see [`emit`]).
91 pub fn channel<E: Clone>() -> EventTx<E> {
92 EventBus {
93 status: broadcast::channel(STATUS_CAPACITY).0,
94 logs: broadcast::channel(LOG_CAPACITY).0,
95 }
96 }
97
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
100 /// only when there are zero receivers, which is the normal idle case for an
101 /// operator tool, so the error is intentionally dropped.
102 pub fn emit<E: BusEvent>(tx: &EventTx<E>, event: E) {
103 let envelope = EventEnvelope {
104 at: Utc::now(),
105 event,
106 };
107 let sender = if envelope.event.is_high_rate() {
108 &tx.logs
109 } else {
110 &tx.status
111 };
112 let _ = sender.send(envelope);
113 }
114
115 #[cfg(test)]
116 mod tests {
117 use super::*;
118 use serde::{Deserialize, Serialize};
119
120 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
121 #[serde(tag = "kind", rename_all = "snake_case")]
122 enum TestEvent {
123 Started { name: String },
124 Done { code: i32 },
125 LogChunk { seq: u32 },
126 }
127
128 impl BusEvent for TestEvent {
129 fn is_high_rate(&self) -> bool {
130 matches!(self, TestEvent::LogChunk { .. })
131 }
132 }
133
134 #[test]
135 fn emit_with_zero_subscribers_does_not_panic() {
136 let tx = channel::<TestEvent>();
137 emit(&tx, TestEvent::Started { name: "x".into() });
138 }
139
140 #[tokio::test]
141 async fn emit_reaches_a_subscriber() {
142 let tx = channel::<TestEvent>();
143 let mut rx = tx.subscribe_status();
144 emit(&tx, TestEvent::Done { code: 7 });
145 let env = rx.recv().await.expect("envelope");
146 assert_eq!(env.event, TestEvent::Done { code: 7 });
147 }
148
149 #[tokio::test]
150 async fn high_rate_events_ride_the_log_channel() {
151 let tx = channel::<TestEvent>();
152 let mut status_rx = tx.subscribe_status();
153 let mut logs_rx = tx.subscribe_logs();
154 emit(&tx, TestEvent::LogChunk { seq: 1 });
155 // It arrives on logs...
156 assert_eq!(
157 logs_rx.recv().await.unwrap().event,
158 TestEvent::LogChunk { seq: 1 }
159 );
160 // ...and never on status.
161 assert!(
162 status_rx.try_recv().is_err(),
163 "log chunk must not hit the status channel"
164 );
165 }
166
167 #[tokio::test]
168 async fn status_survives_a_log_firehose() {
169 // The whole point of the split: flooding the log channel past its
170 // capacity must NOT evict a status event.
171 let tx = channel::<TestEvent>();
172 let mut status_rx = tx.subscribe_status();
173 let mut logs_rx = tx.subscribe_logs();
174 emit(&tx, TestEvent::Done { code: 0 });
175 for i in 0..(LOG_CAPACITY + 50) {
176 emit(&tx, TestEvent::LogChunk { seq: i as u32 });
177 }
178 // The status event is intact despite the log overflow.
179 assert_eq!(
180 status_rx.recv().await.expect("status survived").event,
181 TestEvent::Done { code: 0 }
182 );
183 // The log channel is the one that lagged.
184 assert!(matches!(
185 logs_rx.recv().await,
186 Err(broadcast::error::RecvError::Lagged(_))
187 ));
188 }
189
190 #[test]
191 fn envelope_serializes_with_flat_kind() {
192 let env = EventEnvelope {
193 at: Utc::now(),
194 event: TestEvent::Started {
195 name: "build".into(),
196 },
197 };
198 let v: serde_json::Value =
199 serde_json::from_str(&serde_json::to_string(&env).unwrap()).unwrap();
200 assert_eq!(v["kind"], "started");
201 assert_eq!(v["name"], "build");
202 // Flattened, not nested under `event`.
203 assert!(v.get("event").is_none());
204 // `at` is preserved.
205 assert!(v.get("at").is_some());
206 }
207
208 #[tokio::test]
209 async fn lagged_subscriber_observes_recv_error_lagged() {
210 let tx = channel::<TestEvent>();
211 let mut rx = tx.subscribe_status();
212 for i in 0..(STATUS_CAPACITY + 10) {
213 emit(&tx, TestEvent::Done { code: i as i32 });
214 }
215 let err = rx.recv().await.expect_err("expected Lagged");
216 match err {
217 broadcast::error::RecvError::Lagged(n) => assert!(n >= 10),
218 broadcast::error::RecvError::Closed => panic!("unexpected error: Closed"),
219 }
220 }
221 }
222