max / makenotwork

5.2 KB · 184 lines History Blame Raw

1	//! Bento's concrete event payload, carried on the generic
2	//! [`ops_core::eventbus`] bus. Flat `kind`-tagged so it serializes to the wire
3	//! shape the TUI parses (`{"kind":"step_start", ...}`).
4
5	use crate::domain::{AppId, Status, Step, StepRunId, Target, Version};
6	use ops_core::eventbus::{self, BusEvent};
7	use serde::{Deserialize, Serialize};
8
9	pub type EventEnvelope = eventbus::EventEnvelope<Event>;
10	pub type EventTx = eventbus::EventTx<Event>;
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
20	pub fn channel() -> EventTx {
21	eventbus::channel()
22	}
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.
26	pub fn emit(tx: &EventTx, event: Event) {
27	eventbus::emit(tx, event);
28	}
29
30	#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
31	#[serde(tag = "kind", rename_all = "snake_case")]
32	pub enum Event {
33	/// A `/build` was accepted.
34	BuildRequested {
35	app: AppId,
36	version: Version,
37	targets: Vec<Target>,
38	},
39	/// A previous in-flight run for this `(app, target)` was aborted because a
40	/// newer request arrived.
41	TargetAborted { app: AppId, target: Target },
42	TargetStart {
43	app: AppId,
44	version: Version,
45	target: Target,
46	},
47	StepStart {
48	run_id: StepRunId,
49	app: AppId,
50	version: Version,
51	target: Target,
52	step: Step,
53	},
54	/// A chunk of merged stdout+stderr from the step currently running.
55	/// `run_id` ties back to the `StepStart`; `seq` is a per-run monotonic
56	/// counter; `text` is UTF-8-lossy and NOT line-aligned. The on-disk log is
57	/// the byte-exact source.
58	StepLogChunk {
59	run_id: StepRunId,
60	seq: u32,
61	text: String,
62	},
63	StepDone {
64	run_id: StepRunId,
65	app: AppId,
66	target: Target,
67	step: Step,
68	status: Status,
69	},
70	TargetOk {
71	app: AppId,
72	version: Version,
73	target: Target,
74	artifacts: Vec<String>,
75	},
76	TargetFailed {
77	app: AppId,
78	version: Version,
79	target: Target,
80	step: Step,
81	error: String,
82	},
83	/// Notarization is the one flaky, network-bound step; retries are surfaced.
84	NotarizeRetry {
85	app: AppId,
86	target: Target,
87	attempt: u32,
88	reason: String,
89	},
90	ArtifactCollected {
91	app: AppId,
92	target: Target,
93	path: String,
94	bytes: i64,
95	},
96	PublishOk {
97	app: AppId,
98	target: Target,
99	channel: String,
100	},
101	PublishFailed {
102	app: AppId,
103	target: Target,
104	channel: String,
105	error: String,
106	},
107	}
108
109	#[cfg(test)]
110	mod tests {
111	use super::*;
112	use ops_core::eventbus::LOG_CAPACITY;
113
114	fn target() -> Target {
115	"macos/aarch64".parse().expect("valid target")
116	}
117
118	fn publish_failed() -> Event {
119	Event::PublishFailed {
120	app: AppId::new("goingson"),
121	target: target(),
122	channel: "stable".into(),
123	error: "boom".into(),
124	}
125	}
126
127	#[test]
128	fn only_log_chunks_are_high_rate() {
129	assert!(
130	Event::StepLogChunk {
131	run_id: StepRunId(1),
132	seq: 0,
133	text: "x".into()
134	}
135	.is_high_rate()
136	);
137	assert!(!publish_failed().is_high_rate());
138	}
139
140	#[tokio::test]
141	async fn status_survives_a_log_firehose() {
142	// A build step's chunk spam must NOT evict a publish failure. Before the
143	// split both rode one 256-slot channel, so a lagging TUI lost the
144	// failure alongside the noise.
145	let tx = channel();
146	let mut status_rx = tx.subscribe_status();
147	let mut logs_rx = tx.subscribe_logs();
148	emit(&tx, publish_failed());
149	for i in 0..(LOG_CAPACITY + 50) {
150	emit(
151	&tx,
152	Event::StepLogChunk {
153	run_id: StepRunId(1),
154	seq: i as u32,
155	text: "x".into(),
156	},
157	);
158	}
159	// The status event is intact despite the log overflow.
160	let env = status_rx.recv().await.expect("status event survived");
161	assert!(matches!(env.event, Event::PublishFailed { .. }));
162	// The log channel is the one that lagged.
163	assert!(matches!(
164	logs_rx.recv().await,
165	Err(tokio::sync::broadcast::error::RecvError::Lagged(_))
166	));
167	}
168
169	#[tokio::test]
170	async fn envelope_serializes_with_flat_kind() {
171	// Contract for the WS handler + TUI's `format_event_line`.
172	let env = EventEnvelope {
173	at: chrono::Utc::now(),
174	event: publish_failed(),
175	};
176	let v: serde_json::Value =
177	serde_json::from_str(&serde_json::to_string(&env).unwrap()).unwrap();
178	assert_eq!(v["kind"], "publish_failed");
179	assert_eq!(v["channel"], "stable");
180	assert!(v.get("event").is_none());
181	assert!(v.get("at").is_some());
182	}
183	}
184