Skip to main content

max / makenotwork

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