Skip to main content

max / makenotwork

8.0 KB · 220 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, which is the wire
11 //! shape 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, so a subscriber that
21 //! lags on chunk spam loses the `PromoteComplete` sitting in the same ring.
22 //!
23 //! So the bus is split in two: [`BusEvent::is_high_rate`] classifies each
24 //! event, [`emit`] routes it to the matching channel, and a WS handler
25 //! subscribes to both and merges. A lag on one stream cannot drop the other.
26
27 use chrono::{DateTime, Utc};
28 use serde::{Deserialize, Serialize};
29 use tokio::sync::broadcast;
30
31 /// Capacity of the low-rate *status* channel (build/gate/deploy/promote
32 /// lifecycle events). Subscribers that fall behind by more than this many
33 /// events get `RecvError::Lagged`; WS handlers treat that as a recoverable
34 /// hiccup (emit a `lagged` notice), not a disconnect.
35 pub const STATUS_CAPACITY: usize = 256;
36
37 /// Capacity of the high-rate *log* channel. Larger than [`STATUS_CAPACITY`] so
38 /// a momentarily-paused TUI lags the log stream rather than the status stream.
39 pub const LOG_CAPACITY: usize = 1024;
40
41 /// Lets the bus tell a high-rate log-chunk event from a low-rate status event,
42 /// so each rides its own channel. See the module docs for why this matters.
43 pub trait BusEvent: Clone {
44 /// `true` for the high-rate log-chunk variants (Sando's `GateLogChunk`,
45 /// Bento's `StepLogChunk`) — the ones a busy run emits hundreds of per
46 /// second. Everything an operator must not miss returns `false`.
47 fn is_high_rate(&self) -> bool;
48 }
49
50 /// An event payload `E` stamped with the time it was emitted.
51 #[derive(Clone, Debug, Serialize, Deserialize)]
52 pub struct EventEnvelope<E> {
53 pub at: DateTime<Utc>,
54 #[serde(flatten)]
55 pub event: E,
56 }
57
58 /// Two broadcast channels behind one handle: low-rate status and high-rate log
59 /// chunks. [`emit`] routes each event to the right one; a WS handler subscribes
60 /// to both and merges them, so a lag on one never drops the other.
61 ///
62 /// This is the handle every emit site clones, and it stays named `EventTx` via
63 /// the alias below so call sites read the same as a plain sender.
64 #[derive(Clone)]
65 pub struct EventBus<E> {
66 status: broadcast::Sender<EventEnvelope<E>>,
67 logs: broadcast::Sender<EventEnvelope<E>>,
68 }
69
70 impl<E: Clone> EventBus<E> {
71 /// Subscribe to the low-rate status stream (every event that is not
72 /// [`BusEvent::is_high_rate`]).
73 pub fn subscribe_status(&self) -> broadcast::Receiver<EventEnvelope<E>> {
74 self.status.subscribe()
75 }
76
77 /// Subscribe to the high-rate log-chunk stream.
78 pub fn subscribe_logs(&self) -> broadcast::Receiver<EventEnvelope<E>> {
79 self.logs.subscribe()
80 }
81 }
82
83 /// The handle every emit site clones. `subscribe_status`/`subscribe_logs` yield
84 /// receivers for the WS handler.
85 pub type EventTx<E> = EventBus<E>;
86
87 /// Create a fresh bus. The dropped receivers from `broadcast::channel` are fine
88 /// — emitters don't care whether anyone is listening (see [`emit`]).
89 pub fn channel<E: Clone>() -> EventTx<E> {
90 EventBus {
91 status: broadcast::channel(STATUS_CAPACITY).0,
92 logs: broadcast::channel(LOG_CAPACITY).0,
93 }
94 }
95
96 /// Send an event without caring whether anyone is subscribed. High-rate events
97 /// go to the log channel, everything else to the status channel. `send` errors
98 /// only when there are zero receivers, which is the normal idle case for an
99 /// operator tool, so the error is intentionally dropped.
100 pub fn emit<E: BusEvent>(tx: &EventTx<E>, event: E) {
101 let envelope = EventEnvelope {
102 at: Utc::now(),
103 event,
104 };
105 let sender = if envelope.event.is_high_rate() {
106 &tx.logs
107 } else {
108 &tx.status
109 };
110 let _ = sender.send(envelope);
111 }
112
113 #[cfg(test)]
114 mod tests {
115 use super::*;
116 use serde::{Deserialize, Serialize};
117
118 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
119 #[serde(tag = "kind", rename_all = "snake_case")]
120 enum TestEvent {
121 Started { name: String },
122 Done { code: i32 },
123 LogChunk { seq: u32 },
124 }
125
126 impl BusEvent for TestEvent {
127 fn is_high_rate(&self) -> bool {
128 matches!(self, TestEvent::LogChunk { .. })
129 }
130 }
131
132 #[test]
133 fn emit_with_zero_subscribers_does_not_panic() {
134 let tx = channel::<TestEvent>();
135 emit(&tx, TestEvent::Started { name: "x".into() });
136 }
137
138 #[tokio::test]
139 async fn emit_reaches_a_subscriber() {
140 let tx = channel::<TestEvent>();
141 let mut rx = tx.subscribe_status();
142 emit(&tx, TestEvent::Done { code: 7 });
143 let env = rx.recv().await.expect("envelope");
144 assert_eq!(env.event, TestEvent::Done { code: 7 });
145 }
146
147 #[tokio::test]
148 async fn high_rate_events_ride_the_log_channel() {
149 let tx = channel::<TestEvent>();
150 let mut status_rx = tx.subscribe_status();
151 let mut logs_rx = tx.subscribe_logs();
152 emit(&tx, TestEvent::LogChunk { seq: 1 });
153 // It arrives on logs...
154 assert_eq!(
155 logs_rx.recv().await.unwrap().event,
156 TestEvent::LogChunk { seq: 1 }
157 );
158 // ...and never on status.
159 assert!(
160 status_rx.try_recv().is_err(),
161 "log chunk must not hit the status channel"
162 );
163 }
164
165 #[tokio::test]
166 async fn status_survives_a_log_firehose() {
167 // The whole point of the split: flooding the log channel past its
168 // capacity must NOT evict a status event.
169 let tx = channel::<TestEvent>();
170 let mut status_rx = tx.subscribe_status();
171 let mut logs_rx = tx.subscribe_logs();
172 emit(&tx, TestEvent::Done { code: 0 });
173 for i in 0..(LOG_CAPACITY + 50) {
174 emit(&tx, TestEvent::LogChunk { seq: i as u32 });
175 }
176 // The status event is intact despite the log overflow.
177 assert_eq!(
178 status_rx.recv().await.expect("status survived").event,
179 TestEvent::Done { code: 0 }
180 );
181 // The log channel is the one that lagged.
182 assert!(matches!(
183 logs_rx.recv().await,
184 Err(broadcast::error::RecvError::Lagged(_))
185 ));
186 }
187
188 #[test]
189 fn envelope_serializes_with_flat_kind() {
190 let env = EventEnvelope {
191 at: Utc::now(),
192 event: TestEvent::Started {
193 name: "build".into(),
194 },
195 };
196 let v: serde_json::Value =
197 serde_json::from_str(&serde_json::to_string(&env).unwrap()).unwrap();
198 assert_eq!(v["kind"], "started");
199 assert_eq!(v["name"], "build");
200 // Flattened, not nested under `event`.
201 assert!(v.get("event").is_none());
202 // `at` is preserved.
203 assert!(v.get("at").is_some());
204 }
205
206 #[tokio::test]
207 async fn lagged_subscriber_observes_recv_error_lagged() {
208 let tx = channel::<TestEvent>();
209 let mut rx = tx.subscribe_status();
210 for i in 0..(STATUS_CAPACITY + 10) {
211 emit(&tx, TestEvent::Done { code: i as i32 });
212 }
213 let err = rx.recv().await.expect_err("expected Lagged");
214 match err {
215 broadcast::error::RecvError::Lagged(n) => assert!(n >= 10),
216 broadcast::error::RecvError::Closed => panic!("unexpected error: Closed"),
217 }
218 }
219 }
220