Skip to main content

max / makenotwork

8.3 KB · 257 lines History Blame Raw
1 //! Sando's concrete event payload, carried on the generic
2 //! [`ops_core::eventbus`] bus.
3 //!
4 //! Sites that previously logged via `tracing::info!` also emit a typed event
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.
9
10 use crate::domain::{GateKind, GateRunId, GitSha, NodeId, TierId, Version};
11 use crate::outcome::{DeployFailureKind, GateOutcome};
12 use ops_core::eventbus::{self, BusEvent};
13 use serde::{Deserialize, Serialize};
14
15 pub use eventbus::{LOG_CAPACITY, STATUS_CAPACITY};
16
17 /// The handle build/gate/deploy sites hold and emit on. Kept named `EventTx` so
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>;
22
23 #[derive(Clone, Debug, Serialize, Deserialize)]
24 #[serde(tag = "kind", rename_all = "snake_case")]
25 pub enum Event {
26 /// A /rebuild was accepted (post-receive hook or operator).
27 RebuildRequested {
28 sha: GitSha,
29 },
30 /// A previous in-flight build was aborted because a newer /rebuild arrived.
31 BuildAborted {
32 sha_aborted: GitSha,
33 },
34 BuildStart {
35 sha: GitSha,
36 version: Version,
37 },
38 BuildOk {
39 sha: GitSha,
40 version: Version,
41 elapsed_s: u64,
42 },
43 BuildFailed {
44 sha: GitSha,
45 version: Version,
46 elapsed_s: u64,
47 },
48 GateStart {
49 run_id: GateRunId,
50 tier: TierId,
51 version: Version,
52 gate: GateKind,
53 },
54 /// Chunk of combined stdout+stderr from a gate that's currently running.
55 /// `run_id` correlates back to the `GateStart` for the same gate run; the
56 /// TUI uses it to group chunks if it wants a per-run buffer. `seq` is a
57 /// monotonic counter scoped to one run (resets across runs). `text` is a
58 /// UTF-8-lossy slice of bytes — chunks reflect tokio read boundaries, not
59 /// line boundaries; the on-disk log at `outcome.log_ref` is the full,
60 /// byte-exact stream.
61 GateLogChunk {
62 run_id: GateRunId,
63 seq: u32,
64 text: String,
65 },
66 /// `outcome` carries classification, blocker variants, and the log_ref —
67 /// consumers should read `outcome.status` to decide pass/fail/blocked.
68 GateDone {
69 run_id: GateRunId,
70 tier: TierId,
71 version: Version,
72 gate: GateKind,
73 outcome: GateOutcome,
74 },
75 DeployStart {
76 tier: TierId,
77 node: NodeId,
78 version: Version,
79 },
80 DeployOk {
81 tier: TierId,
82 node: NodeId,
83 version: Version,
84 },
85 DeployFailed {
86 tier: TierId,
87 node: NodeId,
88 version: Version,
89 failure: DeployFailureKind,
90 },
91 PromoteComplete {
92 tier: TierId,
93 version: Version,
94 },
95 Rollback {
96 tier: TierId,
97 from: Version,
98 to: Version,
99 },
100 /// `source` is an ssh URL, kept freeform on purpose — it's a transport
101 /// detail, not a domain identifier.
102 BackupFetched {
103 source: String,
104 byte_size: i64,
105 },
106 ManualConfirm {
107 tier: TierId,
108 version: Version,
109 },
110 }
111
112 impl BusEvent for Event {
113 /// A busy gate emits `GateLogChunk` hundreds of times a second; every other
114 /// variant is lifecycle status the operator must not miss.
115 fn is_high_rate(&self) -> bool {
116 matches!(self, Event::GateLogChunk { .. })
117 }
118 }
119
120 pub fn channel() -> EventTx {
121 eventbus::channel()
122 }
123
124 /// Send an event without caring whether anyone is listening. `GateLogChunk` goes
125 /// to the high-rate log channel, everything else to the status channel.
126 pub fn emit(bus: &EventTx, event: Event) {
127 eventbus::emit(bus, event);
128 }
129
130 #[cfg(test)]
131 mod tests {
132 use super::*;
133 use chrono::Utc;
134 use tokio::sync::broadcast;
135
136 #[test]
137 fn emit_with_zero_subscribers_does_not_panic() {
138 // The whole point of `let _ = tx.send(...)` is that emitting into an
139 // unsubscribed bus is fine. Verify the contract — if this regresses
140 // to `.unwrap()` someday, every build/deploy site will start
141 // crashing.
142 let tx = channel();
143 emit(
144 &tx,
145 Event::RebuildRequested {
146 sha: GitSha::parse("abc1234").unwrap(),
147 },
148 );
149 emit(
150 &tx,
151 Event::BackupFetched {
152 source: "x".into(),
153 byte_size: 1,
154 },
155 );
156 }
157
158 #[tokio::test]
159 async fn emit_reaches_a_subscriber() {
160 let tx = channel();
161 let mut rx = tx.subscribe_status();
162 emit(
163 &tx,
164 Event::PromoteComplete {
165 tier: TierId::new("a"),
166 version: "0.8.12".parse().unwrap(),
167 },
168 );
169 let env = rx.recv().await.expect("envelope");
170 match env.event {
171 Event::PromoteComplete { tier, version } => {
172 assert_eq!(tier.as_str(), "a");
173 assert_eq!(version.to_string(), "0.8.12");
174 }
175 _ => panic!("wrong event kind"),
176 }
177 }
178
179 #[tokio::test]
180 async fn status_survives_a_log_firehose() {
181 // The point of the split: flooding the log channel past its capacity
182 // must NOT evict a status event. Subscribe to both, overflow logs, then
183 // confirm the status event still arrives.
184 let tx = channel();
185 let mut status_rx = tx.subscribe_status();
186 let mut logs_rx = tx.subscribe_logs();
187 emit(
188 &tx,
189 Event::PromoteComplete {
190 tier: TierId::new("b"),
191 version: "0.8.12".parse().unwrap(),
192 },
193 );
194 for i in 0..(LOG_CAPACITY + 50) {
195 emit(
196 &tx,
197 Event::GateLogChunk {
198 run_id: GateRunId(1),
199 seq: i as u32,
200 text: "x".into(),
201 },
202 );
203 }
204 // The status event is intact despite the log overflow.
205 let env = status_rx.recv().await.expect("status event survived");
206 assert!(matches!(env.event, Event::PromoteComplete { .. }));
207 // The log channel is the one that lagged.
208 assert!(matches!(
209 logs_rx.recv().await,
210 Err(broadcast::error::RecvError::Lagged(_))
211 ));
212 }
213
214 #[tokio::test]
215 async fn envelope_serializes_with_flat_kind() {
216 // Contract for the WS handler + TUI's `format_event`: the JSON has a
217 // top-level `kind` field, not nested under `event`. Locking this in.
218 let env = EventEnvelope {
219 at: Utc::now(),
220 event: Event::GateStart {
221 run_id: GateRunId(42),
222 tier: TierId::new("host"),
223 version: "0.8.12".parse().unwrap(),
224 gate: GateKind::CargoTest,
225 },
226 };
227 let s = serde_json::to_string(&env).unwrap();
228 let v: serde_json::Value = serde_json::from_str(&s).unwrap();
229 assert_eq!(v["kind"], "gate_start");
230 assert_eq!(v["tier"], "host");
231 assert_eq!(v["gate"], "cargo_test");
232 // No nested `event` object.
233 assert!(v.get("event").is_none());
234 }
235
236 #[tokio::test]
237 async fn lagged_subscriber_observes_recv_error_lagged() {
238 // If a subscriber falls behind by more than CAPACITY, the next
239 // recv() returns RecvError::Lagged(n) — not Closed, not a panic.
240 // The WS handler turns this into a `lagged` envelope.
241 let tx = channel();
242 let mut rx = tx.subscribe_status();
243 for i in 0..(STATUS_CAPACITY + 10) {
244 // 7+ hex chars satisfy GitSha::parse; pad i into that shape.
245 let sha = GitSha::parse(&format!("{i:0>7x}")).unwrap();
246 emit(&tx, Event::RebuildRequested { sha });
247 }
248 let err = rx.recv().await.expect_err("expected Lagged");
249 match err {
250 tokio::sync::broadcast::error::RecvError::Lagged(n) => assert!(n >= 10),
251 other @ tokio::sync::broadcast::error::RecvError::Closed => {
252 panic!("unexpected error: {other:?}")
253 }
254 }
255 }
256 }
257