Skip to main content

max / makenotwork

20.2 KB · 501 lines History Blame Raw
1 //! Handing a finished artifact to Sando.
2 //!
3 //! Design + rationale: maintainer wiki.
4 //! <!-- wiki: sando-bento-boundary -->
5 //!
6 //! The boundary is "Bento builds and packages, Sando decides whether a thing
7 //! advances a stage". Sando has an intake that proves an incoming bundle against
8 //! its record three ways and publishes it content-addressed — but `POST /intake`
9 //! takes a bundle already sitting under Sando's staging directory, and until now
10 //! nothing put one there. This is the half that moves the bytes.
11 //!
12 //! Two motions, in this order, per target:
13 //!
14 //! 1. rsync the target's collect directory into
15 //! `<staging_root>/<app>-<version>-<target>/` on the Sando host.
16 //! 2. POST that path plus the artifact record to sandod, which verifies and
17 //! publishes it, or refuses it and names the file that drifted.
18 //!
19 //! **The record does not travel with the bytes.** It names the digest of the
20 //! bundle, and the digest covers every file in the bundle, so a record copied in
21 //! among the artifacts would change the digest it names — Sando would recompute,
22 //! disagree, and refuse. It goes in the request body instead. That is the same
23 //! "evidence rides alongside, never inside" rule the contract crate is built on,
24 //! showing up here as an rsync exclude.
25 //!
26 //! **The staging directory is mirrored, not merged.** A retry after a partial
27 //! transfer would otherwise leave a file from the earlier attempt behind, and an
28 //! extra file is a manifest mismatch: the honest bundle would be refused for
29 //! carrying a leftover. `--delete` makes the directory hold exactly this attempt.
30 //!
31 //! Failure fails the target run. A build whose artifact never reached the deploy
32 //! controller has not finished the job the handoff exists to do, and the same
33 //! argument the archive deposit makes applies with more force: a silently
34 //! skipped handoff means the release nobody watched is the one Sando never heard
35 //! about.
36
37 use crate::config::{Config, Handoff};
38 use crate::domain::{AppId, Target, Version};
39 use ops_exec::{CapabilitySet, Executor, LocalExec, SshExec, SyncOpts};
40 use std::path::{Path, PathBuf};
41 use std::sync::Arc;
42
43 /// Where one target's bundle is staged on the Sando host.
44 ///
45 /// Flat and fully qualified rather than nested `<app>/<version>/<target>/`,
46 /// because Sando's intake resolves this path against its own staging root and
47 /// publishes by renaming the directory out of it. One component means one
48 /// rename, and the name still says which build it is when an operator finds it
49 /// left behind after a refused intake.
50 pub fn staged_dir(staging_root: &Path, app: &AppId, version: &Version, target: Target) -> PathBuf {
51 staging_root.join(format!(
52 "{}-{}-{}",
53 app.as_str(),
54 version,
55 crate::archive::target_slug(target)
56 ))
57 }
58
59 /// The transport that writes into Sando's staging directory.
60 ///
61 /// Granted nothing, like the archive's: it only ever calls `push_dir`, so no
62 /// capability in `ops_exec` is exercisable through this handle and the Sando
63 /// host cannot become a build host by holding one. It is not in the topology
64 /// either, so no recipe can name it.
65 fn transport(handoff: &Handoff) -> Arc<dyn Executor> {
66 const NONE: [&str; 0] = [];
67 let caps = CapabilitySet::from_tokens(NONE, NONE);
68 if handoff.host == "local" || handoff.host.is_empty() {
69 return Arc::new(LocalExec::new(caps));
70 }
71 Arc::new(SshExec::new(handoff.host.clone(), caps))
72 }
73
74 /// `POST` target for this handoff: the unprefixed mount is Sando's default
75 /// product, and a named one is nested under `/apps/<id>`.
76 fn intake_url(handoff: &Handoff) -> String {
77 let base = handoff.url.trim_end_matches('/');
78 match &handoff.sando_app {
79 Some(id) => format!("{base}/apps/{id}/intake"),
80 None => format!("{base}/intake"),
81 }
82 }
83
84 /// Read the bearer token for this handoff out of the private layer.
85 ///
86 /// Resolved under `secrets_root` the same way the `secret()` host function
87 /// resolves a recipe's credentials, so there is one place secrets live rather
88 /// than two. Never logged, and the trailing newline a file written by `echo`
89 /// carries is trimmed — it would otherwise become part of the header value.
90 async fn token(cfg: &Config, handoff: &Handoff) -> anyhow::Result<Option<String>> {
91 let Some(rel) = &handoff.token_file else {
92 return Ok(None);
93 };
94 let path = cfg.secrets_root.join(rel);
95 let raw = tokio::fs::read_to_string(&path)
96 .await
97 .map_err(|e| anyhow::anyhow!("reading the sando token at {}: {e}", path.display()))?;
98 let trimmed = raw.trim().to_string();
99 anyhow::ensure!(
100 !trimmed.is_empty(),
101 "the sando token at {} is empty",
102 path.display()
103 );
104 Ok(Some(trimmed))
105 }
106
107 /// Send one target's finished bundle to Sando and ask it to take the artifact
108 /// in. A no-op for an app with no handoff configured.
109 ///
110 /// `local_dir` is the target's collect directory and `record_path` the artifact
111 /// record beside it — written by `artifact_record::emit` after the recipe, which
112 /// is why this runs there and not inside `collect`.
113 pub async fn send(
114 cfg: &Config,
115 local_dir: &Path,
116 record_path: &Path,
117 app: &AppId,
118 version: &Version,
119 target: Target,
120 ) -> anyhow::Result<()> {
121 let Some(handoff) = cfg.handoff.get(app.as_str()) else {
122 return Ok(());
123 };
124
125 // Read the record first. It is what makes the transfer worth doing, and a
126 // missing or unreadable one means shipping bytes Sando could only refuse.
127 let record = tokio::fs::read_to_string(record_path).await.map_err(|e| {
128 anyhow::anyhow!(
129 "reading the artifact record at {}: {e}",
130 record_path.display()
131 )
132 })?;
133
134 let dest = staged_dir(&handoff.staging_root, app, version, target);
135 transport(handoff)
136 .push_dir(
137 local_dir,
138 &dest,
139 &SyncOpts {
140 // Exactly this attempt's bytes: see the module header.
141 delete: true,
142 exclude: vec![crate::artifact_record::RECORD_FILE.to_string()],
143 ..SyncOpts::archive_deposit()
144 },
145 )
146 .await
147 .map_err(|e| {
148 anyhow::anyhow!(
149 "staging {} at {}:{}: {e}",
150 local_dir.display(),
151 handoff.host,
152 dest.display()
153 )
154 })?;
155
156 let url = intake_url(handoff);
157 let mut req = crate::tls::builder()
158 .build()?
159 .post(&url)
160 .json(&serde_json::json!({
161 "staged": dest.to_string_lossy(),
162 "record": record,
163 }));
164 if let Some(t) = token(cfg, handoff).await? {
165 req = req.bearer_auth(t);
166 }
167 let resp = req
168 .send()
169 .await
170 .map_err(|e| anyhow::anyhow!("posting the intake to {url}: {e}"))?;
171
172 let status = resp.status();
173 let body = resp.text().await.unwrap_or_default();
174 anyhow::ensure!(
175 status.is_success(),
176 // Sando's refusals are the useful half — it names the file that drifted
177 // — so the body is carried into the error rather than reduced to a code.
178 "sando refused the intake at {url}: {status} {body}"
179 );
180
181 tracing::info!(
182 %app, %target, %version, host = %handoff.host, dest = %dest.display(),
183 "handed the artifact to sando"
184 );
185 Ok(())
186 }
187
188 #[cfg(test)]
189 mod tests {
190 use super::*;
191
192 fn app() -> AppId {
193 AppId::new("pom")
194 }
195 fn version() -> Version {
196 "0.3.1".parse().unwrap()
197 }
198 fn handoff(root: &Path) -> Handoff {
199 Handoff {
200 host: "local".into(),
201 staging_root: root.join("staging"),
202 url: "http://127.0.0.1:1".into(),
203 sando_app: None,
204 token_file: None,
205 }
206 }
207
208 /// One component, carrying every dimension that distinguishes a build. Two
209 /// architectures of one version are two bundles, and staging them at one
210 /// path would have the second overwrite the first.
211 #[test]
212 fn a_staged_dir_names_app_version_and_target_in_one_component() {
213 let dir = staged_dir(
214 Path::new("/srv/sando/staging"),
215 &app(),
216 &version(),
217 "linux/aarch64".parse().unwrap(),
218 );
219 assert_eq!(dir, Path::new("/srv/sando/staging/pom-0.3.1-linux-aarch64"));
220 assert_eq!(dir.parent(), Some(Path::new("/srv/sando/staging")));
221
222 let other = staged_dir(
223 Path::new("/srv/sando/staging"),
224 &app(),
225 &version(),
226 "linux/x86_64".parse().unwrap(),
227 );
228 assert_ne!(dir, other);
229 }
230
231 /// The default product is the unprefixed mount, because that is what Sando's
232 /// router does with a product it was not asked to nest.
233 #[test]
234 fn the_intake_url_nests_only_a_named_product() {
235 let mut h = handoff(Path::new("/tmp"));
236 h.url = "http://100.103.89.95:7766".into();
237 assert_eq!(intake_url(&h), "http://100.103.89.95:7766/intake");
238
239 h.sando_app = Some("pom".into());
240 assert_eq!(intake_url(&h), "http://100.103.89.95:7766/apps/pom/intake");
241
242 // A trailing slash is an operator's habit, not a second path segment.
243 h.url = "http://100.103.89.95:7766/".into();
244 assert_eq!(intake_url(&h), "http://100.103.89.95:7766/apps/pom/intake");
245 }
246
247 /// An app nobody configured a handoff for is untouched — the ordinary case
248 /// for every app Bento ships that Sando does not deploy.
249 #[tokio::test]
250 async fn an_app_with_no_handoff_sends_nothing() {
251 let tmp = tempfile::tempdir().unwrap();
252 let cfg = Config::for_tests(tmp.path());
253 assert!(cfg.handoff.is_empty());
254 send(
255 &cfg,
256 tmp.path(),
257 &tmp.path().join("record.json"),
258 &app(),
259 &version(),
260 "linux/x86_64".parse().unwrap(),
261 )
262 .await
263 .expect("a no-op cannot fail");
264 }
265
266 /// The staging transfer, up to the POST (which fails against a dead port).
267 /// What it proves is the shape of what lands: the artifacts arrive, and the
268 /// record does NOT — a record inside the bundle would change the digest it
269 /// names and Sando would refuse the honest bytes.
270 #[tokio::test]
271 async fn the_bundle_is_staged_without_its_record() {
272 let tmp = tempfile::tempdir().unwrap();
273 let mut cfg = Config::for_tests(tmp.path());
274 cfg.handoff.insert("pom".into(), handoff(tmp.path()));
275
276 let collected = tmp.path().join("collected");
277 std::fs::create_dir_all(&collected).unwrap();
278 std::fs::write(collected.join("pom"), b"binary").unwrap();
279 let record_path = collected.join(crate::artifact_record::RECORD_FILE);
280 std::fs::write(&record_path, b"{\"producer\":\"bento\"}").unwrap();
281
282 let target: Target = "linux/x86_64".parse().unwrap();
283 // The POST cannot succeed here; the transfer before it still ran.
284 let _ = send(&cfg, &collected, &record_path, &app(), &version(), target).await;
285
286 let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
287 assert_eq!(std::fs::read(staged.join("pom")).unwrap(), b"binary");
288 assert!(
289 !staged.join(crate::artifact_record::RECORD_FILE).exists(),
290 "the record must not travel inside the bundle it describes"
291 );
292 }
293
294 /// A leftover from an earlier attempt is pruned. Without `--delete` it would
295 /// survive as an extra file, and an extra file is a manifest mismatch — the
296 /// retry of an honest build would be refused for carrying it.
297 #[tokio::test]
298 async fn a_retry_leaves_nothing_from_the_previous_attempt() {
299 let tmp = tempfile::tempdir().unwrap();
300 let mut cfg = Config::for_tests(tmp.path());
301 cfg.handoff.insert("pom".into(), handoff(tmp.path()));
302
303 let collected = tmp.path().join("collected");
304 std::fs::create_dir_all(&collected).unwrap();
305 std::fs::write(collected.join("pom"), b"binary").unwrap();
306 let record_path = collected.join(crate::artifact_record::RECORD_FILE);
307 std::fs::write(&record_path, b"{}").unwrap();
308
309 let target: Target = "linux/x86_64".parse().unwrap();
310 let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
311 std::fs::create_dir_all(&staged).unwrap();
312 std::fs::write(staged.join("stale-from-last-time"), b"junk").unwrap();
313
314 let _ = send(&cfg, &collected, &record_path, &app(), &version(), target).await;
315
316 assert!(staged.join("pom").exists());
317 assert!(
318 !staged.join("stale-from-last-time").exists(),
319 "the staging dir must hold exactly this attempt"
320 );
321 }
322
323 /// A missing record is refused before any bytes move. Sando could only
324 /// refuse the bundle anyway, and staging first would leave a directory
325 /// behind for an intake that was never going to be requested.
326 #[tokio::test]
327 async fn a_missing_record_fails_before_anything_is_staged() {
328 let tmp = tempfile::tempdir().unwrap();
329 let mut cfg = Config::for_tests(tmp.path());
330 cfg.handoff.insert("pom".into(), handoff(tmp.path()));
331
332 let collected = tmp.path().join("collected");
333 std::fs::create_dir_all(&collected).unwrap();
334 std::fs::write(collected.join("pom"), b"binary").unwrap();
335
336 let target: Target = "linux/x86_64".parse().unwrap();
337 let err = send(
338 &cfg,
339 &collected,
340 &collected.join("record.json"),
341 &app(),
342 &version(),
343 target,
344 )
345 .await
346 .expect_err("no record, no handoff");
347 assert!(format!("{err:#}").contains("artifact record"), "{err:#}");
348
349 let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
350 assert!(!staged.exists(), "nothing should have been staged");
351 }
352
353 /// An empty token file is a bootstrap that half-ran. Sending the header
354 /// anyway would fail at sandod as a plain 401, which reads as a wrong token
355 /// rather than a missing one.
356 #[tokio::test]
357 async fn an_empty_token_file_is_an_error_not_an_empty_header() {
358 let tmp = tempfile::tempdir().unwrap();
359 let cfg = Config::for_tests(tmp.path());
360 std::fs::create_dir_all(&cfg.secrets_root).unwrap();
361 std::fs::write(cfg.secrets_root.join("sando-token"), "\n").unwrap();
362
363 let mut h = handoff(tmp.path());
364 h.token_file = Some(PathBuf::from("sando-token"));
365 let err = token(&cfg, &h).await.expect_err("empty is not a token");
366 assert!(format!("{err:#}").contains("empty"), "{err:#}");
367 }
368
369 /// The whole motion against a sandod-shaped listener: the bytes are staged,
370 /// and the request names the path they were staged at and carries the record
371 /// as its own field. Those two together are the contract — Sando resolves
372 /// `staged` on its own disk and proves it against `record`, so a handoff that
373 /// staged one path and reported another would be refused for corruption when
374 /// nothing had corrupted.
375 #[tokio::test]
376 async fn the_post_names_the_path_the_bytes_were_staged_at() {
377 use axum::{Json, Router, routing::post};
378 use std::sync::{Arc, Mutex};
379
380 let seen: Arc<Mutex<Option<serde_json::Value>>> = Arc::new(Mutex::new(None));
381 let auth: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
382 let (s, a) = (seen.clone(), auth.clone());
383 let app_router = Router::new().route(
384 "/apps/pom/intake",
385 post(
386 move |headers: axum::http::HeaderMap, Json(b): Json<serde_json::Value>| {
387 let (s, a) = (s.clone(), a.clone());
388 async move {
389 *s.lock().unwrap() = Some(b);
390 *a.lock().unwrap() = headers
391 .get("authorization")
392 .and_then(|v| v.to_str().ok())
393 .map(str::to_owned);
394 Json(serde_json::json!({ "accepted": true, "run_id": 1 }))
395 }
396 },
397 ),
398 );
399 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
400 let addr = listener.local_addr().unwrap();
401 tokio::spawn(async move { axum::serve(listener, app_router).await.unwrap() });
402
403 let tmp = tempfile::tempdir().unwrap();
404 let mut cfg = Config::for_tests(tmp.path());
405 std::fs::create_dir_all(&cfg.secrets_root).unwrap();
406 std::fs::write(cfg.secrets_root.join("sando-token"), "s3cr3t\n").unwrap();
407 let mut h = handoff(tmp.path());
408 h.url = format!("http://{addr}");
409 h.sando_app = Some("pom".into());
410 h.token_file = Some(PathBuf::from("sando-token"));
411 cfg.handoff.insert("pom".into(), h);
412
413 let collected = tmp.path().join("collected");
414 std::fs::create_dir_all(&collected).unwrap();
415 std::fs::write(collected.join("pom"), b"binary").unwrap();
416 let record_path = collected.join(crate::artifact_record::RECORD_FILE);
417 std::fs::write(&record_path, b"{\"producer\":\"bento\"}").unwrap();
418
419 let target: Target = "linux/x86_64".parse().unwrap();
420 send(&cfg, &collected, &record_path, &app(), &version(), target)
421 .await
422 .expect("the intake was accepted");
423
424 let body = seen.lock().unwrap().clone().expect("sandod was called");
425 let staged = staged_dir(&tmp.path().join("staging"), &app(), &version(), target);
426 assert_eq!(body["staged"], staged.to_string_lossy().as_ref());
427 assert_eq!(body["record"], "{\"producer\":\"bento\"}");
428 assert!(
429 Path::new(body["staged"].as_str().unwrap())
430 .join("pom")
431 .exists()
432 );
433 assert_eq!(auth.lock().unwrap().as_deref(), Some("Bearer s3cr3t"));
434 }
435
436 /// A refusal is a failed handoff, and it carries Sando's own words. Its
437 /// errors name the file that drifted, which is the whole return on a
438 /// per-file manifest; reducing them to a status code would throw that away
439 /// at the one moment somebody needs it.
440 #[tokio::test]
441 async fn a_refusal_fails_and_carries_what_sando_said() {
442 use axum::{Router, http::StatusCode, routing::post};
443
444 let app_router = Router::new().route(
445 "/intake",
446 post(|| async {
447 (
448 StatusCode::BAD_REQUEST,
449 "the bundle is not what its record describes; `pom` differs",
450 )
451 }),
452 );
453 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
454 let addr = listener.local_addr().unwrap();
455 tokio::spawn(async move { axum::serve(listener, app_router).await.unwrap() });
456
457 let tmp = tempfile::tempdir().unwrap();
458 let mut cfg = Config::for_tests(tmp.path());
459 let mut h = handoff(tmp.path());
460 h.url = format!("http://{addr}");
461 cfg.handoff.insert("pom".into(), h);
462
463 let collected = tmp.path().join("collected");
464 std::fs::create_dir_all(&collected).unwrap();
465 std::fs::write(collected.join("pom"), b"binary").unwrap();
466 let record_path = collected.join(crate::artifact_record::RECORD_FILE);
467 std::fs::write(&record_path, b"{}").unwrap();
468
469 let err = send(
470 &cfg,
471 &collected,
472 &record_path,
473 &app(),
474 &version(),
475 "linux/x86_64".parse().unwrap(),
476 )
477 .await
478 .expect_err("a refused intake is a failed handoff");
479 let msg = format!("{err:#}");
480 assert!(msg.contains("`pom` differs"), "{msg}");
481 }
482
483 /// The token is trimmed: a file written with `echo` ends in a newline, and a
484 /// newline inside a header value is not a credential problem an operator
485 /// would ever guess at.
486 #[tokio::test]
487 async fn a_token_is_read_trimmed() {
488 let tmp = tempfile::tempdir().unwrap();
489 let cfg = Config::for_tests(tmp.path());
490 std::fs::create_dir_all(&cfg.secrets_root).unwrap();
491 std::fs::write(cfg.secrets_root.join("sando-token"), "s3cr3t\n").unwrap();
492
493 let mut h = handoff(tmp.path());
494 h.token_file = Some(PathBuf::from("sando-token"));
495 assert_eq!(token(&cfg, &h).await.unwrap().as_deref(), Some("s3cr3t"));
496 // No file named means no header, not an empty one.
497 h.token_file = None;
498 assert_eq!(token(&cfg, &h).await.unwrap(), None);
499 }
500 }
501