Skip to main content

max / makenotwork

18.2 KB · 438 lines History Blame Raw
1 //! The gates that ask a running thing whether it is serving: the staged
2 //! artifact on the build host, the deployed nodes, and the public pages through
3 //! the CDN.
4
5 use super::log::{append_to_log, gate_chunk_cb, stream_into_log};
6 use super::{GateCtx, NodeProbe};
7 use crate::classify;
8 use crate::domain::{GateKind, GateRunId};
9 use crate::outcome::{GateBlocker, GateFailure, GateOutcome, PassNote};
10 use anyhow::Result;
11 use ops_core::live_log::LiveLog;
12 use ops_core::remote::LogSink;
13 use ops_exec::{DiscardSink, sh_quote};
14
15 pub(super) async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
16 let bin: Option<(String,)> =
17 sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
18 .bind(&ctx.cfg.id)
19 .bind(&ctx.version)
20 .fetch_optional(&ctx.pool)
21 .await?;
22 let Some((bin,)) = bin else {
23 return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
24 version: ctx.version.clone(),
25 }));
26 };
27
28 // Readiness smoke: start the binary and confirm it serves `GET /health`
29 // within the window, not merely that the process stays up. Panics in main,
30 // missing config, and port-bind failures still surface as an early exit; a
31 // process that comes up but never serves /health is now its own failure.
32 //
33 // The server requires DATABASE_URL or it panics on config load before
34 // we can observe anything. We point it at the scratch DB (already
35 // migrated by the build step and refreshed by migration_dry_run if
36 // that gate ran first). SCAN_ENABLED=false skips loading YARA rules
37 // from /opt/makenotwork/yara-rules which doesn't exist on the build
38 // host. SANDO_BOOT_SMOKE_PORT tells the smoke server which loopback port
39 // to bind so we know where to probe. Other config has sane optional defaults.
40 let mut cmd = tokio::process::Command::new(&bin);
41 cmd.env("SANDO_BOOT_SMOKE", "1")
42 .env("SANDO_BOOT_SMOKE_PORT", ctx.cfg.boot_smoke_port.to_string())
43 .env("SCAN_ENABLED", "false")
44 .stdout(std::process::Stdio::piped())
45 .stderr(std::process::Stdio::piped())
46 .kill_on_drop(true);
47 if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
48 cmd.env("DATABASE_URL", scratch_url);
49 }
50 let log_path = ctx.log_path(GateKind::BootSmoke);
51 let log_ref = ctx.log_ref(GateKind::BootSmoke);
52 let mut child = match cmd.spawn() {
53 Ok(c) => c,
54 Err(e) => {
55 // Spawn failures get a one-off log line via LiveLog so the
56 // on-disk file still exists for `GET /logs/...`.
57 let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await;
58 log.write_chunk(format!("spawn: {e}\n").as_bytes()).await;
59 log.close().await;
60 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
61 message: e.to_string(),
62 })
63 .with_log_ref(log_ref));
64 }
65 };
66
67 // The boot smoke window is 3s. Drain stdout/stderr concurrently through
68 // a shared LiveLog sink so the operator sees panics/log lines stream in
69 // real time before the kill, AND the on-disk log gets the full byte
70 // stream for post-mortem reads. The drainers exit when their pipe
71 // closes — which happens when the child exits naturally or after kill.
72 let log = std::sync::Arc::new(tokio::sync::Mutex::new(
73 LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await,
74 ));
75 let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
76 let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));
77
78 // Poll readiness across the 3s window instead of a flat sleep: GET /health
79 // must return 2xx. A crash mid-window short-circuits to the exit-code
80 // failure path (try_wait below); a process that stays up but never serves
81 // /health is a distinct readiness failure.
82 let probe_timeout = std::time::Duration::from_millis(500);
83 let started = std::time::Instant::now();
84 let window = std::time::Duration::from_secs(3);
85 let mut probe_ok_after: Option<u32> = None;
86 let mut last_probe_err = "never responded".to_string();
87 let mut early_exit = None;
88 while started.elapsed() < window {
89 if let Some(status) = child.try_wait()? {
90 early_exit = Some(status);
91 break;
92 }
93 match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.boot_smoke_port)).await {
94 Ok(Ok(())) => {
95 probe_ok_after = Some(started.elapsed().as_millis() as u32);
96 break;
97 }
98 Ok(Err(e)) => last_probe_err = e,
99 Err(_) => last_probe_err = "probe timed out".to_string(),
100 }
101 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
102 }
103
104 // Stop the child unless it already exited, then drain the log tasks.
105 let exit = match early_exit {
106 Some(status) => Some(status),
107 None => {
108 let e = child.try_wait()?;
109 if e.is_none() {
110 let _ = child.kill().await;
111 }
112 e
113 }
114 };
115 // The streamed bytes already landed in the live log and the on-disk file for
116 // the post-mortem reader. Drain the join handles to avoid hangs.
117 let _ = stdout_task.await;
118 let _ = stderr_task.await;
119 // Unique owner of the Arc at this point (both tasks dropped their clones).
120 if let Ok(mutex) = std::sync::Arc::try_unwrap(log) {
121 mutex.into_inner().close().await;
122 }
123
124 match (exit, probe_ok_after) {
125 // Exited on its own within the window — a crash/panic/bind failure.
126 (Some(status), _) => {
127 let failure = classify::classify_boot_smoke(status.code());
128 Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
129 }
130 // Stayed up and served /health — readiness proven.
131 (None, Some(after_ms)) => {
132 Ok(GateOutcome::passed(PassNote::HealthyProbe { after_ms }).with_log_ref(log_ref))
133 }
134 // Stayed up but never served /health — started, not ready.
135 (None, None) => Ok(GateOutcome::failed(GateFailure::BootHealthProbeFailed {
136 last_error: last_probe_err,
137 })
138 .with_log_ref(log_ref)),
139 }
140 }
141
142 /// One readiness probe of the boot-smoke server: connect to `127.0.0.1:port`
143 /// and `GET /health`, returning `Ok(())` only on a `200`. A hand-rolled HTTP/1.0
144 /// request over a raw `TcpStream` keeps the outbound probe dependency-free
145 /// (reqwest is dev-only); the smoke server serves the one route over axum, which
146 /// speaks 1.0. `Err` carries a short reason for the operator's failure note. The
147 /// caller wraps each call in a timeout.
148 pub(super) async fn probe_health(port: u16) -> std::result::Result<(), String> {
149 use tokio::io::{AsyncReadExt, AsyncWriteExt};
150 let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port))
151 .await
152 .map_err(|e| format!("connect: {e}"))?;
153 stream
154 .write_all(b"GET /health HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n")
155 .await
156 .map_err(|e| format!("write: {e}"))?;
157 let mut buf = Vec::new();
158 stream
159 .read_to_end(&mut buf)
160 .await
161 .map_err(|e| format!("read: {e}"))?;
162 let text = String::from_utf8_lossy(&buf);
163 let status_line = text.lines().next().unwrap_or("");
164 if status_line.contains(" 200 ") {
165 Ok(())
166 } else {
167 Err(format!("unexpected status line: {status_line:?}"))
168 }
169 }
170
171 /// `node_health` — the post-deploy gate that proves the *deployed nodes* are
172 /// serving, recording one outcome per (tier, version) that the next promote
173 /// checks. Distinct from `boot_smoke`, which boots the staged artifact on the
174 /// build host: this probes each node over the same executor the deploy used, so
175 /// a node that took a corrupt artifact, wrong-arch binary, or failed restart is
176 /// caught here rather than waved through. Fails closed: any
177 /// unhealthy node fails the gate, and an empty node set is `Blocked`.
178 /// Load the tier's public pages in a real browser and fail if the JavaScript
179 /// did not run.
180 ///
181 /// The only gate here that crosses the CDN. `boot_smoke` runs on the build host
182 /// and `node_health` reaches a node over its executor, so between them nothing
183 /// requests the site the way a visitor does. A CDN holding a module from an
184 /// earlier deploy can fail to link against the fresh one beside it, and since
185 /// the bundle's entry point side-effect-imports every island, one bad link takes
186 /// all of them down together while every artifact is individually correct: right
187 /// markup, right stylesheets, every module answering 200 with current bytes.
188 /// Composition is observable in a browser and nowhere else.
189 ///
190 /// Runs on the daemon host rather than on a node, because it is a *client*: it
191 /// should reach the site through whatever the public reaches it through, and a
192 /// probe that ran on the origin would inherit the blind spot this exists to
193 /// close.
194 pub(super) async fn page_smoke(ctx: &GateCtx) -> Result<GateOutcome> {
195 let Some(cmd) = ctx.cfg.page_smoke_cmd.as_deref() else {
196 // A service with no pages says so by configuring no command. Blocked
197 // rather than passed: a gate that proves nothing must not read green.
198 return Ok(GateOutcome::blocked(GateBlocker::NotConfigured {
199 what: "page_smoke_cmd".into(),
200 }));
201 };
202 let Some(base) = ctx.public_url.as_deref() else {
203 return Ok(GateOutcome::blocked(GateBlocker::NotConfigured {
204 what: "public_url".into(),
205 }));
206 };
207
208 let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
209 let child = tokio::process::Command::new("sh")
210 .arg("-c")
211 .arg(cmd)
212 .env("BASE", base)
213 .stdout(std::process::Stdio::piped())
214 .stderr(std::process::Stdio::piped())
215 .kill_on_drop(true)
216 .spawn()?;
217
218 let out = match tokio::time::timeout(ceiling, child.wait_with_output()).await {
219 Ok(res) => res?,
220 Err(_elapsed) => {
221 return Ok(GateOutcome::failed(GateFailure::Timeout {
222 gate: GateKind::PageSmoke,
223 after_s: ctx.cfg.gate_timeout_secs as u32,
224 })
225 .with_log_ref(ctx.log_ref(GateKind::PageSmoke)));
226 }
227 };
228
229 let log = format!(
230 "{}{}",
231 String::from_utf8_lossy(&out.stdout),
232 String::from_utf8_lossy(&out.stderr)
233 );
234 append_to_log(&ctx.log_path(GateKind::PageSmoke), log.as_bytes()).await;
235
236 if out.status.success() {
237 return Ok(
238 GateOutcome::passed(PassNote::PagesClean { base: base.into() })
239 .with_log_ref(ctx.log_ref(GateKind::PageSmoke)),
240 );
241 }
242
243 // The script prints one `FAIL <url>` line per bad page and indents the
244 // reasons under it. Lift the first reason into the summary so a red gate
245 // says what broke without anyone opening the log.
246 let first = log
247 .lines()
248 .skip_while(|l| !l.starts_with("FAIL"))
249 .nth(1)
250 .map(str::trim)
251 .filter(|l| !l.is_empty())
252 .unwrap_or("see log");
253 Ok(GateOutcome::failed(GateFailure::PagesBroken {
254 base: base.into(),
255 detail: first.to_string(),
256 })
257 .with_log_ref(ctx.log_ref(GateKind::PageSmoke)))
258 }
259
260 pub(super) async fn node_health(ctx: &GateCtx) -> Result<GateOutcome> {
261 if ctx.nodes.is_empty() {
262 return Ok(GateOutcome::blocked(GateBlocker::NoNodesToProbe));
263 }
264 for probe in &ctx.nodes {
265 if let Err(detail) = probe_node(probe).await {
266 return Ok(GateOutcome::failed(GateFailure::NodeUnhealthy {
267 node: probe.node.to_string(),
268 detail,
269 }));
270 }
271 }
272 Ok(GateOutcome::passed(PassNote::NodesHealthy {
273 nodes: ctx.nodes.len() as u32,
274 }))
275 }
276
277 /// Probe one node over its executor: confirm the unit is active post-restart
278 /// and, when a `health_url` is configured, that it serves a 2xx. Retries across
279 /// ~10s because the service may still be restarting / warming. Runs under the
280 /// read-only `Observe(Health)` capability (every Sando node grants it), so the
281 /// probe needs no deploy authority. `Ok(())` = healthy; `Err(detail)` carries a
282 /// short reason for the gate's failure note.
283 async fn probe_node(probe: &NodeProbe) -> std::result::Result<(), String> {
284 use ops_exec::{Action, ObserveKind, Step};
285 let svc = sh_quote(&probe.service);
286 let url = probe
287 .health_url
288 .as_deref()
289 .map_or_else(|| "''".to_string(), sh_quote);
290 // One executor round-trip with the retry loop on the node: is-active, then
291 // (if a url is set) curl it for a 2xx. Exit 0 only when both hold.
292 let script = format!(
293 "svc={svc}; url={url}; \
294 for _ in $(seq 1 10); do \
295 if systemctl is-active --quiet \"$svc\"; then \
296 if [ -z \"$url\" ] || curl -fsS --max-time 5 \"$url\" >/dev/null 2>&1; then exit 0; fi; \
297 fi; \
298 sleep 1; \
299 done; \
300 echo 'service not active or health url not 2xx after retries' >&2; exit 1"
301 );
302 let step = Step::shell(Action::Observe(ObserveKind::Health), script);
303 let mut sink = DiscardSink;
304 let out = probe
305 .executor
306 .run_streaming(&step, &mut sink)
307 .await
308 .map_err(|e| format!("probe spawn: {e}"))?;
309 if out.status.success() {
310 Ok(())
311 } else {
312 let code = out
313 .status
314 .code()
315 .map_or_else(|| "signal".to_string(), |c| c.to_string());
316 let stderr: String = String::from_utf8_lossy(&out.stderr)
317 .chars()
318 .take(200)
319 .collect();
320 Err(format!("exit {code}: {stderr}"))
321 }
322 }
323
324 #[cfg(test)]
325 mod tests {
326 use super::super::run;
327 use super::*;
328 use crate::domain::TierId;
329 use crate::events;
330 use crate::topology::Gate;
331 use sqlx::sqlite::SqlitePoolOptions;
332 use std::collections::HashMap;
333
334 /// Spawn a one-shot loopback server that answers the first connection with
335 /// `status_line` + a tiny body, then closes. Returns the bound port.
336 async fn oneshot_http(status_line: &'static str) -> u16 {
337 use tokio::io::{AsyncReadExt, AsyncWriteExt};
338 let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
339 .await
340 .unwrap();
341 let port = listener.local_addr().unwrap().port();
342 tokio::spawn(async move {
343 if let Ok((mut sock, _)) = listener.accept().await {
344 let mut scratch = [0u8; 1024];
345 let _ = sock.read(&mut scratch).await; // drain the request line
346 let resp =
347 format!("{status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
348 let _ = sock.write_all(resp.as_bytes()).await;
349 }
350 });
351 port
352 }
353
354 #[tokio::test]
355 async fn probe_health_ok_on_200() {
356 let port = oneshot_http("HTTP/1.1 200 OK").await;
357 assert!(probe_health(port).await.is_ok());
358 }
359
360 #[tokio::test]
361 async fn probe_health_err_on_non_200() {
362 let port = oneshot_http("HTTP/1.1 503 Service Unavailable").await;
363 let err = probe_health(port).await.unwrap_err();
364 assert!(err.contains("status line"), "{err}");
365 }
366
367 #[tokio::test]
368 async fn probe_health_err_on_connection_refused() {
369 // Bind then drop to get an almost-certainly-free port nothing listens on.
370 let port = {
371 let l = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
372 .await
373 .unwrap();
374 l.local_addr().unwrap().port()
375 };
376 let err = probe_health(port).await.unwrap_err();
377 // Under load the kernel does not always refuse the connect: a socket
378 // left in TIME_WAIT on that port completes the handshake and then
379 // resets, so the failure surfaces on the write or the read instead.
380 // Refused-on-connect and reset-on-write/read are the same fact, that
381 // nothing is serving the port, and no other outcome counts as a pass.
382 let refused = err.starts_with("connect: ") && err.contains("refused");
383 let reset =
384 (err.starts_with("write: ") || err.starts_with("read: ")) && err.contains("reset");
385 assert!(refused || reset, "{err}");
386 }
387
388 /// node_health fails closed when there are no nodes to probe: a serving tier
389 /// should always carry nodes, so an empty set is a misconfiguration that must
390 /// block promotion, not pass it.
391 #[tokio::test]
392 async fn node_health_blocks_with_no_nodes() {
393 let pool = SqlitePoolOptions::new()
394 .max_connections(1)
395 .connect("sqlite::memory:")
396 .await
397 .unwrap();
398 crate::db::migrate(&pool).await.unwrap();
399 sqlx::query(
400 "INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('b', 2, 1, 'sequential')",
401 )
402 .execute(&pool)
403 .await
404 .unwrap();
405 sqlx::query("INSERT INTO tier_state (tier) VALUES ('b')")
406 .execute(&pool)
407 .await
408 .unwrap();
409 sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
410 .execute(&pool).await.unwrap();
411
412 let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests());
413 let ctx = GateCtx {
414 public_url: None,
415 pool: pool.clone(),
416 cfg,
417 tier: TierId::new("b"),
418 version: "0.1.0".parse().unwrap(),
419 worktree: None,
420 bundle: None,
421 events: events::channel(),
422 nodes: Vec::new(), // no nodes -> fail closed
423 build_id: None,
424 aux_dirs: HashMap::new(),
425 };
426 let out = run(&ctx, &Gate::NodeHealth).await.unwrap();
427 assert_eq!(out.status_str(), "blocked");
428 assert!(!out.is_passed());
429 let row: (Option<String>, Option<String>) =
430 sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
431 .fetch_one(&pool)
432 .await
433 .unwrap();
434 let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
435 assert_eq!(json["status"]["blocker"]["kind"], "no_nodes_to_probe");
436 }
437 }
438