crates/flow/src/runner.rs at v0.5.2 · andymccall/oxygenflow

◆ v0.5.2 oxygenflow / crates / flow / src / runner.rs

113.0 KiB · Rust · last changed Runner builds fetched through the swarm (refs #84); the WASM tier as a cargo feature, off for Haiku (refs #91) 7 days ago source blame history raw

e0918b88 Andy McCall 1 month ago	1	//! `flow runner` (docs/m0.md §5): claim a job from OxygenHub, fetch the
	2	//! repository into a cache with the job's token, write the state into a
	3	//! workspace, run it in its tier, stream the log, report the receipt.
	4	//! One job at a time.
	5
47c1d64d Andy McCall 1 month ago ↶	6	use crate::home::{Home, RunnerConfig, RunnerKey};
e0918b88 Andy McCall 1 month ago	7	use crate::{microvm, workspace};
e0918b88 Andy McCall 1 month ago	8	use anyhow::{bail, Context, Result};
b11e30f5 Andy McCall 1 month ago ↶	9	use flow_core::api::{Claim, ClaimRequest, Decline, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};
e0918b88 Andy McCall 1 month ago	10	use flow_core::{scope_tree_hash, Receipt};
	11	use oxy_core::hash::ChangeHash;
	12	use oxy_core::repo::Repo;
46fe2c81 Andy McCall 19 days ago ↶	13	use std::path::{Path, PathBuf};
e0918b88 Andy McCall 1 month ago	14	use std::time::{Duration, Instant};
e0918b88 Andy McCall 1 month ago	15
280ef951 Andy McCall 1 month ago ↶	16	/// A request the hub refused (4xx). On a job's endpoints, 409 means the
	17	/// job no longer runs: it was cancelled, or given up.
	18	#[derive(Debug)]
	19	pub struct Refused {
	20	pub code: u16,
	21	pub text: String,
	22	}
	23
	24	impl std::fmt::Display for Refused {
	25	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
	26	write!(f, "HTTP {}: {}", self.code, self.text)
	27	}
	28	}
	29
	30	impl std::error::Error for Refused {}
	31
	32	/// Whether `e` is the hub saying the job no longer runs.
	33	fn job_gone(e: &anyhow::Error) -> bool {
	34	e.downcast_ref::<Refused>().is_some_and(\|r\| r.code == 409)
	35	}
	36
	37	/// What a job ends with when its run was cancelled: nothing is reported,
	38	/// as the hub has ended it already.
	39	const CANCELLED: &str = "the run was cancelled";
	40
	41	fn cancelled() -> Finish {
8f211e20 Andy McCall 24 days ago ↶	42	Finish {
	43	exit_code: None,
	44	duration_ms: 0,
	45	receipt: None,
	46	signed_receipt: None,
	47	error: Some(CANCELLED.into()),
	48	peak_disk_mib: None,
	49	peak_memory_mib: None,
1215f05f Andy McCall 8 days ago ↶	50	phases: Default::default(),
8f211e20 Andy McCall 24 days ago ↶	51	}
280ef951 Andy McCall 1 month ago ↶	52	}
280ef951 Andy McCall 1 month ago ↶	53
e0918b88 Andy McCall 1 month ago	54	pub struct Hub {
	55	base: String,
	56	token: String,
	57	agent: ureq::Agent,
	58	}
	59
	60	impl Hub {
	61	pub fn new(cfg: &RunnerConfig) -> Result<Hub> {
	62	let tls = oxy_core::protocol::tls_client_config(cfg.ca.as_deref()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	63	let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
	64	Ok(Hub { base: cfg.hub.trim_end_matches('/').to_string(), token: cfg.token.clone(), agent })
	65	}
	66
1b2229da Andy McCall 1 month ago ↶	67	/// POST to the hub. A server error (5xx) means nothing was committed,
	68	/// so it is tried again after 1, 2 and 4 seconds before giving up: a
	69	/// busy hub costs a job a few seconds, not its log lines or its claim.
e0918b88 Andy McCall 1 month ago	70	fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
1b2229da Andy McCall 1 month ago ↶	71	self.post_with(path, body, &RETRY_DELAYS)
	72	}
	73
	74	fn post_with(&self, path: &str, body: &impl serde::Serialize, delays: &[Duration]) -> Result<Option<ureq::Response>> {
e0918b88 Andy McCall 1 month ago	75	let url = format!("{}/api/v1{path}", self.base);
89171fca Andy McCall 22 days ago ↶	76	let r = retrying(&url, delays, false, \|\| {
	77	self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)
	78	})?;
	79	Ok(if r.status() == 204 { None } else { Some(r) })
	80	}
	81
	82	/// POST what a job leaves behind (oxygenflow#49): through a hub that is
	83	/// away for a few minutes, as while a deploy restarts it, so a job's
	84	/// declines and uploads outlast it.
	85	fn post_patiently(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
	86	let url = format!("{}/api/v1{path}", self.base);
	87	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
	88	self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body).map_err(Box::new)
	89	})?;
	90	Ok(if r.status() == 204 { None } else { Some(r) })
e0918b88 Andy McCall 1 month ago	91	}
	92
	93	pub fn claim(&self, req: &ClaimRequest) -> Result<Option<Claim>> {
	94	match self.post("/flow/claim", req)? {
	95	Some(r) => Ok(Some(r.into_json()?)),
	96	None => Ok(None),
	97	}
	98	}
	99
b11e30f5 Andy McCall 1 month ago ↶	100	/// Hand job `job` back, saying why this runner will not run it.
b11e30f5 Andy McCall 1 month ago ↶	101	pub fn decline(&self, job: i64, reason: &str) -> Result<()> {
89171fca Andy McCall 22 days ago ↶	102	self.post_patiently(&format!("/flow/jobs/{job}/decline"), &Decline { reason: reason.into() })?;
b11e30f5 Andy McCall 1 month ago ↶	103	Ok(())
	104	}
	105
06a4c472 Andy McCall 10 days ago ↶	106	/// Hand back a job this runner holds as it leaves (oxygenflow#76): the
	107	/// hub queues it again ahead of new work for another runner.
	108	pub fn release(&self, job: i64, reason: &str) -> Result<()> {
	109	self.post_patiently(&format!("/flow/jobs/{job}/release"), &Decline { reason: reason.into() })?;
	110	Ok(())
	111	}
	112
a4b3f8cb Andy McCall 19 days ago ↶	113	/// Say this runner is there while it claims nothing: its slots are full,
	114	/// or it is draining (oxygenhub#243). Once, no retries: the next comes
	115	/// soon. A hub without the endpoint (404) is not an error.
35f7a849 Andy McCall 18 days ago ↶	116	/// A ten-minute certificate for `identity`, asked with a job's token
	117	/// (keyless, oxygenhub#241), and the identity of the hub's key that
	118	/// issued it, from `/.well-known/oxyhub-key`.
	119	pub fn certify(
	120	&self,
	121	job_token: &str,
	122	identity: &oxy_core::identity::Identity,
	123	proof: &oxy_core::identity::Signature,
	124	) -> Result<(oxy_core::cert::SignedCertificate, oxy_core::identity::Identity)> {
	125	use oxy_core::dsse::{base64, base64_decode};
	126	let body = serde_json::json!({
	127	"identity": base64(&identity.encode()),
	128	"proof": base64(&oxy_core::cbor::encode(&proof.to_value())),
	129	"scope": "seal",
	130	});
	131	let url = format!("{}/api/v1/keyless/certify", self.base);
	132	let r = self
	133	.agent
	134	.post(&url)
	135	.set("Authorization", &format!("Bearer {job_token}"))
	136	.send_json(body)
	137	.map_err(\|e\| anyhow::anyhow!("asking the hub to certify the key: {e}"))?;
	138	let v: serde_json::Value = r.into_json()?;
	139	let cert = base64_decode(v["certificate"].as_str().unwrap_or(""))
	140	.context("the hub's answer has no certificate")
	141	.and_then(\|b\| oxy_core::cert::SignedCertificate::decode(&b).map_err(\|e\| anyhow::anyhow!("{e}")))?;
	142	let mut bytes = Vec::new();
	143	std::io::Read::read_to_end(
	144	&mut self.agent.get(&format!("{}/.well-known/oxyhub-key", self.base)).call()?.into_reader(),
	145	&mut bytes,
	146	)?;
	147	let issuer = oxy_core::identity::Identity::decode(&bytes)?;
	148	if issuer.hash() != cert.cert.issuer \|\| cert.verify(&issuer).is_err() {
	149	bail!("the certificate is not signed by the hub's published key");
	150	}
	151	if cert.cert.key_id() != identity.hash() {
	152	bail!("the hub certified another key");
	153	}
	154	Ok((cert, issuer))
	155	}
	156
a4b3f8cb Andy McCall 19 days ago ↶	157	pub fn heartbeat(&self) -> Result<()> {
	158	match self.post_with("/flow/runners/heartbeat", &serde_json::json!({}), &[]) {
	159	Ok(_) => Ok(()),
	160	Err(e) if format!("{e:#}").contains("404") => Ok(()),
	161	Err(e) => Err(e),
	162	}
	163	}
	164
cedd1234 Andy McCall 10 days ago ↶	165	/// The update policy for this runner and the build it should run
	166	/// (oxygenhub#322); `None` from a hub without runner updates.
	167	pub fn update_answer(&self) -> Result<Option<flow_core::api::UpdateAnswer>> {
	168	let url = format!("{}/api/v1/flow/runner/update", self.base);
	169	match self.agent.get(&url).set("Authorization", &format!("Bearer {}", self.token)).call() {
	170	Ok(r) => Ok(Some(r.into_json()?)),
	171	Err(ureq::Error::Status(404, _)) => Ok(None),
	172	Err(e) => Err(anyhow::anyhow!("{url}: {e}")),
	173	}
	174	}
	175
	176	/// Tell the hub what this runner did about its update.
	177	pub fn report_update(&self, note: &str) -> Result<()> {
	178	self.post_with("/flow/runner/update", &flow_core::api::UpdateReport { note: note.to_string() }, &[])?;
	179	Ok(())
	180	}
	181
	182	/// Download an offered build to `to`, refusing it unless its BLAKE3
	183	/// is the one the hub attests and it is the size it said.
	184	pub fn download_build(&self, offer: &flow_core::api::UpdateOffer, to: &Path) -> Result<()> {
	185	use std::io::{Read, Write};
	186	let url = format!("{}{}", self.base, offer.url);
	187	let r = self.agent.get(&url).timeout(Duration::from_secs(1800)).call().map_err(\|e\| anyhow::anyhow!("{url}: {e}"))?;
	188	let part = to.with_extension("partial");
	189	let mut out = std::fs::File::create(&part).with_context(\|\| format!("writing {}", part.display()))?;
	190	let (mut hasher, mut n, mut buf) = (blake3::Hasher::new(), 0u64, vec![0u8; 1 << 16]);
	191	let mut body = r.into_reader().take(offer.size + 1);
	192	loop {
	193	let k = body.read(&mut buf)?;
	194	if k == 0 {
	195	break;
	196	}
	197	hasher.update(&buf[..k]);
	198	out.write_all(&buf[..k])?;
	199	n += k as u64;
	200	}
	201	drop(out);
	202	let got = hasher.finalize().to_hex().to_string();
	203	if n != offer.size \|\| got != offer.blake3 {
	204	let _ = std::fs::remove_file(&part);
	205	bail!("{} does not match what the hub attests ({n} bytes, BLAKE3 {}…)", offer.asset, &got[..12]);
	206	}
	207	std::fs::rename(&part, to)?;
	208	Ok(())
	209	}
	210
e0918b88 Andy McCall 1 month ago	211	pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
	212	self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
	213	Ok(())
	214	}
	215
a67b97e9 Andy McCall 1 month ago ↶	216	/// Upload artifact `name` of job `job` (spec §5.2); the hub keeps it as
	217	/// a blob of the job's repository and answers with its ObjectRef.
	218	pub fn artifact(&self, job: i64, name: &str, bytes: &[u8]) -> Result<String> {
	219	let url = format!("{}/api/v1/flow/jobs/{job}/artifacts/{name}", self.base);
89171fca Andy McCall 22 days ago ↶	220	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
89171fca Andy McCall 22 days ago ↶	221	self.agent
a67b97e9 Andy McCall 1 month ago ↶	222	.post(&url)
	223	.timeout(Duration::from_secs(600))
	224	.set("Authorization", &format!("Bearer {}", self.token))
89171fca Andy McCall 22 days ago ↶	225	.set("Content-Type", "application/octet-stream")
	226	.send_bytes(bytes)
	227	.map_err(Box::new)
	228	})?;
	229	let v: serde_json::Value = r.into_json()?;
	230	v["blob"].as_str().map(String::from).context("the hub named no blob")
a67b97e9 Andy McCall 1 month ago ↶	231	}
a67b97e9 Andy McCall 1 month ago ↶	232
f201e1bc Andy McCall 10 days ago ↶	233	/// Keep a checkpoint of job `job` (oxygenflow#77) as its runner hands
	234	/// it back: the hub answers with its blob, which the next attempt's
	235	/// claim names.
	236	pub fn save_checkpoint(&self, job: i64, bytes: &[u8]) -> Result<String> {
	237	let url = format!("{}/api/v1/flow/jobs/{job}/checkpoint", self.base);
	238	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
	239	self.agent
	240	.post(&url)
	241	.timeout(Duration::from_secs(600))
	242	.set("Authorization", &format!("Bearer {}", self.token))
	243	.set("Content-Type", "application/octet-stream")
	244	.send_bytes(bytes)
	245	.map_err(Box::new)
	246	})?;
	247	let v: serde_json::Value = r.into_json()?;
	248	v["blob"].as_str().map(String::from).context("the hub named no blob")
	249	}
	250
	251	/// The checkpoint an earlier attempt of job `job` saved (oxygenflow#77),
	252	/// which the caller checks against the blob its claim names.
	253	pub fn checkpoint(&self, job: i64) -> Result<Vec<u8>> {
	254	let url = format!("{}/api/v1/flow/jobs/{job}/checkpoint", self.base);
	255	let r = retrying(&url, &PATIENT_DELAYS, false, \|\| {
	256	self.agent
	257	.get(&url)
	258	.timeout(Duration::from_secs(600))
	259	.set("Authorization", &format!("Bearer {}", self.token))
	260	.call()
	261	.map_err(Box::new)
	262	})?;
	263	let mut bytes = Vec::new();
	264	std::io::Read::read_to_end(&mut std::io::Read::take(r.into_reader(), crate::checkpoint::MAX_BYTES as u64 + 1), &mut bytes)?;
	265	Ok(bytes)
	266	}
	267
46fe2c81 Andy McCall 19 days ago ↶	268	/// Input `index` of job `job` (oxygenflow#58): the bytes of the
	269	/// producing job's artifact, which the caller checks against the blob the
	270	/// claim names.
	271	pub fn input(&self, job: i64, index: usize) -> Result<Vec<u8>> {
	272	let url = format!("{}/api/v1/flow/jobs/{job}/inputs/{index}", self.base);
	273	let r = retrying(&url, &PATIENT_DELAYS, false, \|\| {
	274	self.agent
	275	.get(&url)
	276	.timeout(Duration::from_secs(600))
	277	.set("Authorization", &format!("Bearer {}", self.token))
	278	.call()
	279	.map_err(Box::new)
	280	})?;
	281	let mut bytes = Vec::new();
	282	let mut reader = r.into_reader();
	283	std::io::Read::read_to_end(&mut std::io::Read::take(&mut reader, MAX_INPUT_BYTES as u64 + 1), &mut bytes)?;
	284	if bytes.len() > MAX_INPUT_BYTES {
	285	bail!("larger than {} MiB", MAX_INPUT_BYTES >> 20);
	286	}
	287	Ok(bytes)
	288	}
	289
6d53fb7d Andy McCall 25 days ago ↶	290	/// Upload file `path` of report `name` of job `job` (oxygenhub#127);
	291	/// the hub keeps it as a blob and answers with its ObjectRef. The
	292	/// report's manifest follows through [`Hub::artifact`].
	293	pub fn report_file(&self, job: i64, name: &str, path: &str, bytes: &[u8]) -> Result<String> {
	294	let url = format!("{}/api/v1/flow/jobs/{job}/reports/{name}/files", self.base);
89171fca Andy McCall 22 days ago ↶	295	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
89171fca Andy McCall 22 days ago ↶	296	self.agent
6d53fb7d Andy McCall 25 days ago ↶	297	.post(&url)
	298	.timeout(Duration::from_secs(600))
	299	.set("Authorization", &format!("Bearer {}", self.token))
	300	.set("Content-Type", "application/octet-stream")
89171fca Andy McCall 22 days ago ↶	301	.set("X-Flow-Report-Path", path)
	302	.send_bytes(bytes)
	303	.map_err(Box::new)
	304	})?;
	305	let v: serde_json::Value = r.into_json()?;
	306	v["blob"].as_str().map(String::from).context("the hub named no blob")
6d53fb7d Andy McCall 25 days ago ↶	307	}
6d53fb7d Andy McCall 25 days ago ↶	308
3007d8f0 Andy McCall 1 month ago ↶	309	/// Report a job's end until the hub has it (oxygenhub#71): while the hub
	310	/// cannot be reached or fails (5xx), try again after `first_wait`,
	311	/// doubling to a minute, for up to `patience`. A refusal (4xx: the job was ended
	312	/// already, say) is final. A result lost here leaves the job running
	313	/// on the hub until its time limit.
	314	pub fn finish_until_accepted(&self, job: i64, f: &Finish, patience: Duration, first_wait: Duration) -> Result<()> {
	315	let url = format!("{}/api/v1/flow/jobs/{job}/finish", self.base);
	316	let deadline = std::time::Instant::now() + patience;
	317	let mut wait = first_wait;
	318	loop {
	319	let why = match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(f) {
	320	Ok(_) => return Ok(()),
	321	Err(ureq::Error::Status(code, r)) if code < 500 => {
	322	bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default())
	323	}
	324	Err(ureq::Error::Status(code, _)) => format!("HTTP {code}"),
	325	Err(e) => e.to_string(),
	326	};
	327	if std::time::Instant::now() + wait > deadline {
	328	bail!("{url}: gave up after {} minutes: {why}", patience.as_secs() / 60);
	329	}
	330	eprintln!("job {job}: the hub did not take the result ({why}); trying again in {} s", wait.as_secs());
	331	std::thread::sleep(wait);
	332	wait = (wait * 2).min(Duration::from_secs(60));
	333	}
	334	}
	335	}
	336
	337	/// How long a runner keeps trying to report a job's end.
	338	const FINISH_PATIENCE: Duration = Duration::from_secs(2 * 3600);
	339
	340	/// Where a job's result waits until the hub has it, so a runner that
	341	/// restarts still reports it.
	342	fn pending_path(home: &Home, job: i64) -> PathBuf {
	343	home.root.join("pending").join(format!("{job}.json"))
	344	}
	345
	346	/// Report `f`, keeping it on disk until the hub has it.
	347	fn report_finish(home: &Home, hub: &Hub, job: i64, f: &Finish) {
	348	let path = pending_path(home, job);
	349	let kept = std::fs::create_dir_all(path.parent().unwrap())
	350	.and_then(\|_\| std::fs::write(&path, serde_json::to_vec(f).unwrap_or_default()))
	351	.is_ok();
	352	match hub.finish_until_accepted(job, f, FINISH_PATIENCE, Duration::from_secs(5)) {
	353	Ok(()) => {}
	354	Err(e) => eprintln!("job {job}: reporting failed: {e:#}"),
	355	}
	356	if kept {
	357	let _ = std::fs::remove_file(&path);
f98a8f78 Andy McCall 1 month ago ↶	358	}
3007d8f0 Andy McCall 1 month ago ↶	359	}
	360
	361	/// Results a previous run of the runner could not report, to send again.
	362	fn report_pending(home: &Home) -> Vec<(i64, Finish)> {
	363	let Ok(dir) = std::fs::read_dir(home.root.join("pending")) else {
	364	return Vec::new();
	365	};
	366	dir.filter_map(\|e\| {
	367	let p = e.ok()?.path();
	368	let job: i64 = p.file_stem()?.to_str()?.parse().ok()?;
	369	let f: Finish = serde_json::from_slice(&std::fs::read(&p).ok()?).ok()?;
	370	Some((job, f))
	371	})
	372	.collect()
f98a8f78 Andy McCall 1 month ago ↶	373	}
	374
	375	/// `flow runner register`: exchange a one-time registration token from
	376	/// the hub for this runner's credential, binding the runner's signing key,
	377	/// and keep hub, credential and name in runner.toml.
	378	pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {
	379	if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {
	380	bail!("flow.toml already has a [runner] section; remove it to register this runner instead");
	381	}
	382	let key = home.runner_key(name)?;
	383	let hub = hub.trim_end_matches('/').to_string();
	384	let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	385	let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
	386	let req = Register {
	387	token: token.trim().to_string(),
	388	name: name.to_string(),
	389	identity: hex::encode(key.identity.encode()),
2d025bbf Andy McCall 1 month ago ↶	390	info: RunnerInfo { version: env!("OXY_VERSION_LONG").into(), ..RunnerInfo::here(labels.clone(), 1, None, None) },
f98a8f78 Andy McCall 1 month ago ↶	391	};
	392	let url = format!("{hub}/api/v1/flow/runners/register");
	393	let got: Registered = match agent.post(&url).send_json(&req) {
	394	Ok(r) => r.into_json()?,
	395	Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
	396	Err(e) => bail!("{url}: {e}"),
	397	};
	398	let mut text = format!(
	399	"# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",
	400	got.credential, got.name
	401	);
	402	if let Some(ca) = &ca {
0d663e91 Andy McCall 8 days ago ↶	403	let kept = keep_ca(&home.root, ca)?;
0d663e91 Andy McCall 8 days ago ↶	404	text.push_str(&format!("ca = {:?}\n", kept.display().to_string()));
f98a8f78 Andy McCall 1 month ago ↶	405	}
	406	if !labels.is_empty() {
	407	text.push_str(&format!("labels = {labels:?}\n"));
	408	}
	409	let path = home.root.join(crate::home::RUNNER_TOML);
	410	{
	411	use std::io::Write;
	412	let tmp = path.with_extension("toml.tmp");
3554afef Andy McCall 10 days ago ↶	413	let mut f = crate::portable::private(std::fs::OpenOptions::new().write(true).create(true).truncate(true)).open(&tmp)?;
f98a8f78 Andy McCall 1 month ago ↶	414	f.write_all(text.as_bytes())?;
	415	f.sync_all()?;
	416	std::fs::rename(&tmp, &path)?;
e0918b88 Andy McCall 1 month ago	417	}
f98a8f78 Andy McCall 1 month ago ↶	418	Ok(got)
e0918b88 Andy McCall 1 month ago	419	}
1b2229da Andy McCall 1 month ago ↶	420
0d663e91 Andy McCall 8 days ago ↶	421	/// Copy the hub's CA certificate `ca` into the Flow home `root` (#98) and
	422	/// give the copy's absolute path, for runner.toml: a service starts in
	423	/// another directory, and the file given may move or go.
	424	fn keep_ca(root: &Path, ca: &Path) -> Result<PathBuf> {
	425	let kept = root.join(crate::home::CA_PEM);
	426	let same = matches!((std::fs::canonicalize(ca), std::fs::canonicalize(&kept)), (Ok(a), Ok(b)) if a == b);
	427	if !same {
	428	std::fs::copy(ca, &kept).with_context(\|\| format!("copy {} into {}", ca.display(), root.display()))?;
	429	}
	430	Ok(std::fs::canonicalize(&kept)?)
	431	}
	432
1b2229da Andy McCall 1 month ago ↶	433	/// Waits before trying a hub request again after a server error.
1b2229da Andy McCall 1 month ago ↶	434	const RETRY_DELAYS: [Duration; 3] = [Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)];
89171fca Andy McCall 22 days ago ↶	435
	436	/// Waits for what a job leaves behind (oxygenflow#49): three minutes in
	437	/// all, longer than a hub takes to restart.
46fe2c81 Andy McCall 19 days ago ↶	438	/// Check an input's bytes against the blob the producing job's receipt
	439	/// names, then write them at its path under `tree` (the workspace's `/src`).
	440	fn place_input(tree: &Path, i: &flow_core::api::ArtifactInput, bytes: &[u8]) -> Result<[u8; 32]> {
	441	let Some(rel) = i.path.strip_prefix("/src/").filter(\|r\| r.split('/').all(\|p\| !p.is_empty() && p != "." && p != ".." && p != ".oxy"))
	442	else {
	443	bail!("path {:?} is not a clean path under /src", i.path);
	444	};
	445	let blob = Repo::blob_ref(bytes);
	446	if blob.to_hex() != i.blob {
	447	bail!("it is not what {}'s receipt names (blob {}, not {})", i.from, &blob.to_hex()[..12], &i.blob[..i.blob.len().min(12)]);
	448	}
	449	let dest = tree.join(rel);
	450	if let Some(dir) = dest.parent() {
	451	std::fs::create_dir_all(dir)?;
	452	}
	453	std::fs::write(&dest, bytes)?;
	454	Ok(blob.0)
	455	}
	456
	457	/// The largest input a job takes: the hub's artifact limit.
	458	pub const MAX_INPUT_BYTES: usize = 512 << 20;
	459
89171fca Andy McCall 22 days ago ↶	460	const PATIENT_DELAYS: [Duration; 15] = {
	461	let mut d = [Duration::from_secs(15); 15];
	462	d[0] = Duration::from_secs(1);
	463	d[1] = Duration::from_secs(2);
	464	d[2] = Duration::from_secs(4);
	465	d[3] = Duration::from_secs(8);
	466	d
	467	};
	468
	469	/// Send a request until the hub answers, waiting `delays` in turn after a
	470	/// server error (5xx: nothing was committed) and, when `connection`, after
	471	/// the hub could not be reached at all. A refusal (4xx) is final, as a
	472	/// [`Refused`] the caller can look at.
	473	fn retrying(
	474	url: &str,
	475	delays: &[Duration],
	476	connection: bool,
	477	mut send: impl FnMut() -> Result<ureq::Response, Box<ureq::Error>>,
	478	) -> Result<ureq::Response> {
	479	let mut delays = delays.iter();
	480	loop {
	481	let why = match send().map_err(\|e\| *e) {
	482	Ok(r) => return Ok(r),
	483	Err(ureq::Error::Status(code, r)) if code < 500 => {
	484	let text = r.into_string().unwrap_or_default();
	485	return Err(anyhow::Error::new(Refused { code, text }).context(url.to_string()));
	486	}
	487	Err(ureq::Error::Status(code, r)) => format!("HTTP {code}: {}", r.into_string().unwrap_or_default()),
	488	Err(e) if connection => e.to_string(),
	489	Err(e) => bail!("{url}: {e}"),
	490	};
	491	match delays.next() {
	492	Some(d) => std::thread::sleep(*d),
	493	None => bail!("{url}: {why}"),
	494	}
	495	}
	496	}
e0918b88 Andy McCall 1 month ago	497
bb643c25 Andy McCall 1 month ago ↶	498	/// What the runner may run at once (flow.toml `[runner]`): at most
	499	/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.
	500	#[derive(Debug, Default, PartialEq, Eq)]
	501	pub struct Budget {
	502	pub max_jobs: usize,
	503	pub memory_mib: Option<u32>,
	504	pub cpus: Option<u32>,
	505	}
	506
	507	/// What is in use now.
	508	#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
	509	pub struct Used {
	510	pub jobs: usize,
	511	pub memory_mib: u32,
	512	pub cpus: u32,
	513	}
	514
	515	impl Budget {
	516	pub fn of(cfg: &RunnerConfig) -> Budget {
	517	Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }
	518	}
	519
	520	/// The room to ask the hub for, or `None` when no job more can start.
	521	/// A lone job may use the whole budget even if it asks for more, so a
	522	/// large sandbox is never starved.
	523	pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {
	524	if used.jobs >= self.max_jobs {
	525	return None;
	526	}
	527	if used.jobs == 0 {
	528	return Some((None, None));
	529	}
	530	let mem = self.memory_mib.map(\|m\| m.saturating_sub(used.memory_mib));
	531	let cpus = self.cpus.map(\|c\| c.saturating_sub(used.cpus));
	532	if mem == Some(0) \|\| cpus == Some(0) {
	533	return None;
	534	}
	535	Some((mem, cpus))
	536	}
	537	}
	538
c5b702e5 Andy McCall 1 month ago ↶	539	/// The isolation tiers this machine can honour now (spec §11.1): the
ea56c56e Andy McCall 1 month ago ↶	540	/// MicroVM tier when `/dev/kvm` opens for reading and writing and
	541	/// flow.toml pins Firecracker; the syscall tier when this binary may use
	542	/// user namespaces, Landlock and seccomp and manage a delegated cgroup.
	543	/// `Err` says why there are none; the reasons for missing tiers come back
	544	/// with the ones there are.
	545	pub fn tiers(home: &Home) -> std::result::Result<(Vec<String>, Vec<String>), String> {
	546	let mut have = Vec::new();
	547	let mut missing = Vec::new();
c5b702e5 Andy McCall 1 month ago ↶	548	let kvm = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm");
c5b702e5 Andy McCall 1 month ago ↶	549	match (kvm, home.config.microvm.is_some()) {
e6949d87 Andy McCall 10 days ago ↶	550	_ if !cfg!(target_os = "linux") => missing.push("microvm: the tier runs only on Linux (Firecracker and KVM)".to_string()),
ea56c56e Andy McCall 1 month ago ↶	551	(Ok(_), true) => have.push("microvm".to_string()),
	552	(Ok(_), false) => missing.push("microvm: flow.toml has no [microvm] section".to_string()),
	553	(Err(e), _) => missing.push(format!(
	554	"microvm: /dev/kvm cannot be opened ({e}); the tier needs hardware virtualisation and this user in the kvm group"
c5b702e5 Andy McCall 1 month ago ↶	555	)),
c5b702e5 Andy McCall 1 month ago ↶	556	}
ea56c56e Andy McCall 1 month ago ↶	557	match crate::syscall::available() {
	558	Ok(()) => have.push("syscall".to_string()),
	559	Err(e) => missing.push(format!("syscall: {e}")),
	560	}
21a3ea4a Andy McCall 7 days ago ↶	561	// wasmtime runs on every machine a build with it does (oxygenflow#75).
	562	if crate::wasm::AVAILABLE {
	563	have.push("wasm".to_string());
	564	} else {
	565	missing.push("wasm: this build has no WASM tier".to_string());
	566	}
ea56c56e Andy McCall 1 month ago ↶	567	if have.is_empty() {
	568	return Err(format!("no isolation tier can run here: {}", missing.join("; ")));
	569	}
	570	Ok((have, missing))
b11e30f5 Andy McCall 1 month ago ↶	571	}
b11e30f5 Andy McCall 1 month ago ↶	572
4cf0c99a Andy McCall 24 days ago ↶	573	/// Fetch `claim`'s image from the hub when this home lacks it
	574	/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it
	575	/// leaves it missing, and the preflight declines the job as before.
1215f05f Andy McCall 8 days ago ↶	576	fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) -> Option<u64> {
4cf0c99a Andy McCall 24 days ago ↶	577	let hash = &claim.sandbox.rootfs;
4cf0c99a Andy McCall 24 days ago ↶	578	if hash.is_empty() \|\| home.rootfs(hash).is_file() \|\| !jet::images::is_hash(hash) {
1215f05f Andy McCall 8 days ago ↶	579	return None;
4cf0c99a Andy McCall 24 days ago ↶	580	}
	581	let note = \|text: String\| {
	582	eprintln!("job {}: {text}", claim.job_id);
	583	hub.log(claim.job_id, vec![Record { t: 0, stream: "flow".into(), text }]).ok();
	584	};
	585	// One fetch of an image at a time: two jobs wanting it wait for one.
	586	static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(());
	587	let _one = FETCHING.lock().unwrap_or_else(\|p\| p.into_inner());
	588	if home.rootfs(hash).is_file() {
1215f05f Andy McCall 8 days ago ↶	589	return None;
4cf0c99a Andy McCall 24 days ago ↶	590	}
	591	let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) {
	592	Ok(c) => c,
1215f05f Andy McCall 8 days ago ↶	593	Err(e) => {
	594	note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12]));
	595	return None;
	596	}
4cf0c99a Andy McCall 24 days ago ↶	597	};
1215f05f Andy McCall 8 days ago ↶	598	let fetching = Instant::now();
b14aeaae Andy McCall 23 days ago ↶	599	let fetched = match swarm {
	600	Some(sw) => {
	601	note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12]));
	602	sw.fetch(home, hash)
	603	}
	604	None => {
	605	note(format!("image {} is not on this runner: fetching it from the hub", &hash[..12]));
	606	client.fetch(home, hash)
	607	}
	608	};
1215f05f Andy McCall 8 days ago ↶	609	let took = fetching.elapsed().as_millis() as u64;
b14aeaae Andy McCall 23 days ago ↶	610	match fetched {
4cf0c99a Andy McCall 24 days ago ↶	611	Ok(Some(m)) => note(format!(
b14aeaae Andy McCall 23 days ago ↶	612	"image {}: fetched {} chunks, {}{} in {:.1} s",
4cf0c99a Andy McCall 24 days ago ↶	613	&hash[..12],
	614	m.chunks,
	615	crate::images::human(m.bytes),
b14aeaae Andy McCall 23 days ago ↶	616	if swarm.is_some() { format!(" ({} from peers)", crate::images::human(m.peer_bytes)) } else { String::new() },
4cf0c99a Andy McCall 24 days ago ↶	617	m.elapsed.as_secs_f64()
	618	)),
	619	Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])),
	620	Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])),
	621	}
1215f05f Andy McCall 8 days ago ↶	622	Some(took)
4cf0c99a Andy McCall 24 days ago ↶	623	}
4cf0c99a Andy McCall 24 days ago ↶	624
b11e30f5 Andy McCall 1 month ago ↶	625	/// Why this runner cannot run `claim` now (spec §11.5), checked before
	626	/// anything is fetched: the tier, the architecture, the image and the
	627	/// memory. The hub chose it from what the runner last said, which may be
	628	/// out of date.
	629	fn preflight(home: &Home, claim: &Claim) -> Option<String> {
	630	let sb = &claim.sandbox;
	631	match sb.runtime {
	632	flow_core::Runtime::Microvm => {
	633	if let Err(e) = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm") {
	634	return Some(format!("the MicroVM tier needs /dev/kvm, which cannot be opened here ({e})"));
	635	}
	636	if home.config.microvm.is_none() {
	637	return Some("the MicroVM tier is not set up here (flow.toml has no [microvm] section)".into());
	638	}
	639	}
	640	flow_core::Runtime::Syscall => {
	641	if let Err(e) = crate::syscall::available() {
	642	return Some(format!("the syscall tier cannot run here: {e}"));
	643	}
	644	}
28cc603f Andy McCall 10 days ago ↶	645	// No image and no architecture: only the memory matters.
	646	flow_core::Runtime::Wasm => {
	647	return available_memory_mib()
	648	.filter(\|free\| *free < u64::from(sb.memory_mib))
	649	.map(\|free\| format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));
	650	}
b11e30f5 Andy McCall 1 month ago ↶	651	}
	652	sandbox_problem(home, sb, std::env::consts::ARCH, available_memory_mib())
	653	}
	654
	655	/// What stops `sb` running on this home's images, on a machine of
	656	/// architecture `here` with `free_mib` of memory free.
	657	fn sandbox_problem(home: &Home, sb: &flow_core::Sandbox, here: &str, free_mib: Option<u64>) -> Option<String> {
	658	if let Some(want) = sb.arch.as_deref().filter(\|a\| *a != here) {
	659	return Some(format!("the sandbox needs {want}; this runner is {here}"));
	660	}
	661	let short = &sb.rootfs[..12.min(sb.rootfs.len())];
	662	let image = home.rootfs(&sb.rootfs);
	663	if !image.is_file() {
	664	return Some(format!("rootfs {short} is not on this runner"));
	665	}
	666	if let Some(arch) = home.image_arch(&sb.rootfs).filter(\|a\| a != here) {
	667	return Some(format!("rootfs {short} is for {arch}; this runner is {here}"));
	668	}
	669	if let Err(e) = home.verified(&image, &sb.rootfs) {
	670	return Some(format!("rootfs {short} does not verify here: {e:#}"));
	671	}
	672	if let Some(free) = free_mib.filter(\|free\| *free < u64::from(sb.memory_mib)) {
	673	return Some(format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));
	674	}
	675	None
	676	}
	677
77c8c240 Andy McCall 9 days ago ↶	678	/// The memory to claim with: the budget's room, never more than the host
	679	/// has free now. An idle runner's budget is unlimited (a lone job may use
	680	/// it all), so without this it was offered sandboxes it then declined for
	681	/// want of memory, one after another, having fetched their images first
	682	/// (an 8 GiB job on a 7 GB machine).
	683	fn claim_memory(room: Option<u32>, host_free_mib: Option<u64>) -> Option<u32> {
	684	match (room, host_free_mib) {
	685	(Some(r), Some(h)) => Some(r.min(u32::try_from(h).unwrap_or(u32::MAX))),
	686	(None, Some(h)) => Some(u32::try_from(h).unwrap_or(u32::MAX)),
	687	(room, None) => room,
	688	}
	689	}
	690
b11e30f5 Andy McCall 1 month ago ↶	691	/// `MemAvailable` from /proc/meminfo, in MiB.
	692	fn available_memory_mib() -> Option<u64> {
	693	let info = std::fs::read_to_string("/proc/meminfo").ok()?;
	694	let kib: u64 = info.lines().find_map(\|l\| l.strip_prefix("MemAvailable:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;
	695	Some(kib / 1024)
c5b702e5 Andy McCall 1 month ago ↶	696	}
c5b702e5 Andy McCall 1 month ago ↶	697
e0918b88 Andy McCall 1 month ago	698	pub fn serve(home: &Home) -> Result<()> {
5cf1e8ae Andy McCall 10 days ago ↶	699	drain_on_stop();
85275e1e Andy McCall 10 days ago ↶	700	crate::update::note_runner(&home.root);
	701	// `flow runner update` swapped a new build in and asks this one to
	702	// leave with the notice it gave (#85).
3554afef Andy McCall 10 days ago ↶	703	#[cfg(unix)]
	704	{
	705	let handler = on_update as extern "C" fn(libc::c_int) as libc::sighandler_t;
	706	unsafe { libc::signal(libc::SIGUSR1, handler) };
	707	}
85275e1e Andy McCall 10 days ago ↶	708	if let Some(p) = crate::update::pending(&home.root) {
	709	// On trial: a build that does not reach the hub in time goes back.
	710	let root = home.root.clone();
	711	std::thread::spawn(move \|\| {
	712	std::thread::sleep(crate::update::PROVE_WITHIN);
	713	if crate::update::pending(&root).is_some_and(\|now\| now.to == p.to && now.at_ms == p.at_ms) {
	714	let why = format!("did not reach the hub within {} minutes", crate::update::PROVE_WITHIN.as_secs() / 60);
	715	if crate::update::roll_back(&root, &p, &why).is_ok() {
	716	std::process::exit(1);
	717	}
	718	}
	719	});
	720	}
	721	let served = serve_until_drained(home);
	722	crate::update::forget_runner(&home.root);
494a1b9a Andy McCall 9 days ago ↶	723	// Left for a new build: on Windows, end as a failure, so a Task
	724	// Scheduler task set to restart on failure starts it (systemd's
	725	// Restart=always starts it after a clean exit).
	726	if cfg!(windows) && served.is_ok() && LEFT_FOR_UPDATE.load(std::sync::atomic::Ordering::SeqCst) {
	727	eprintln!("flow runner: left for the new build; exiting 75 so it is started");
	728	std::process::exit(75);
	729	}
85275e1e Andy McCall 10 days ago ↶	730	served
5cf1e8ae Andy McCall 10 days ago ↶	731	}
494a1b9a Andy McCall 9 days ago ↶	732
	733	/// Set once the runner leaves for a new build.
	734	static LEFT_FOR_UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
5cf1e8ae Andy McCall 10 days ago ↶	735
85275e1e Andy McCall 10 days ago ↶	736	/// Set by SIGUSR1: a new build is in place; leave for it.
	737	static UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
	738
3554afef Andy McCall 10 days ago ↶	739	#[cfg(unix)]
85275e1e Andy McCall 10 days ago ↶	740	extern "C" fn on_update(_: libc::c_int) {
	741	UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);
	742	}
	743
5cf1e8ae Andy McCall 10 days ago ↶	744	/// The runner as a library, as the Android app runs it (oxygenhub-android#8):
	745	/// no signal handlers; [`stop`] drains it, and this returns once the jobs
	746	/// it holds are done and reported. It may be started again afterwards.
	747	pub fn serve_embedded(home: &Home) -> Result<()> {
	748	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
06a4c472 Andy McCall 10 days ago ↶	749	*LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()) = None;
5cf1e8ae Andy McCall 10 days ago ↶	750	serve_until_drained(home)
06a4c472 Andy McCall 10 days ago ↶	751	}
06a4c472 Andy McCall 10 days ago ↶	752
1b160d37 Andy McCall 10 days ago ↶	753	/// A departure ([`leave_within`]): why, and when the runner must be gone.
	754	#[derive(Clone, Debug)]
	755	struct Leave {
	756	reason: String,
	757	by: Instant,
	758	}
	759
	760	/// Set once the runner is leaving: a job whose expected rest fits before
	761	/// the deadline finishes; the others are stopped and handed back
	762	/// (oxygenflow#76).
	763	static LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall 10 days ago ↶	764
06a4c472 Andy McCall 10 days ago ↶	765	fn leaving() -> Option<String> {
1b160d37 Andy McCall 10 days ago ↶	766	LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()).as_ref().map(\|l\| l.reason.clone())
06a4c472 Andy McCall 10 days ago ↶	767	}
	768
	769	/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing
	770	/// more, stop the jobs it holds and hand them back to the hub, which queues
	771	/// them ahead of new work for another runner (oxygenflow#76), then return.
	772	/// For a departure with no time to finish: a phone off the allowed network.
	773	pub fn leave(reason: &str) {
1b160d37 Andy McCall 10 days ago ↶	774	leave_within(reason, Duration::ZERO);
	775	}
	776
	777	/// [`leave`], with `notice`: a job the hub expects to finish within it
	778	/// (the claim's `expected_ms`, less the time it has run) finishes and is
	779	/// reported; the others, and any still running when the notice is up, are
	780	/// handed back. For a phone taken off charge, a spot reclaim.
	781	pub fn leave_within(reason: &str, notice: Duration) {
	782	*LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()) = Some(Leave { reason: reason.to_string(), by: Instant::now() + notice });
06a4c472 Andy McCall 10 days ago ↶	783	stop();
5cf1e8ae Andy McCall 10 days ago ↶	784	}
	785
	786	/// Ask a runner started with [`serve_embedded`] to drain: claim nothing
	787	/// more, finish and report what it holds, then return (oxygenflow#76's
	788	/// drain, for a phone taken off charge or off the allowed network).
	789	pub fn stop() {
	790	STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
	791	}
	792
	793	fn serve_until_drained(home: &Home) -> Result<()> {
e0918b88 Andy McCall 1 month ago	794	let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
c5b702e5 Andy McCall 1 month ago ↶	795	// A runner offers only what it can run: a machine without KVM claims
c5b702e5 Andy McCall 1 month ago ↶	796	// nothing rather than taking MicroVM jobs and failing them (#36).
ea56c56e Andy McCall 1 month ago ↶	797	let (runtimes, missing) = tiers(home).map_err(\|why\| anyhow::anyhow!("flow runner: {why}"))?;
	798	for why in &missing {
	799	eprintln!("flow runner: not offering {why}");
	800	}
e0918b88 Andy McCall 1 month ago	801	let hub = Hub::new(&cfg)?;
e0918b88 Andy McCall 1 month ago	802	let name = cfg.name.clone().unwrap_or_else(\|\| "flow".into());
bb643c25 Andy McCall 1 month ago ↶	803	let budget = Budget::of(&cfg);
47c1d64d Andy McCall 1 month ago ↶	804	let key = home.runner_key(&name)?;
b14aeaae Andy McCall 23 days ago ↶	805	// The swarm (docs/jetstream.md §8.1): a runner that cannot serve its
	806	// peers still runs jobs, fetching images from the hub alone.
	807	let swarm = crate::swarm::Swarm::start(home, &cfg).unwrap_or_else(\|e\| {
	808	eprintln!("flow runner: not joining the swarm: {e:#}");
	809	None
	810	});
47c1d64d Andy McCall 1 month ago ↶	811	eprintln!(
c5b702e5 Andy McCall 1 month ago ↶	812	"flow runner {name}: polling {} (up to {} jobs at once, tiers {}), signing receipts as {}",
47c1d64d Andy McCall 1 month ago ↶	813	cfg.hub,
47c1d64d Andy McCall 1 month ago ↶	814	budget.max_jobs,
c5b702e5 Andy McCall 1 month ago ↶	815	runtimes.join(", "),
47c1d64d Andy McCall 1 month ago ↶	816	key.identity.hash().to_hex()
47c1d64d Andy McCall 1 month ago ↶	817	);
4be5e4cd Andy McCall 15 days ago ↶	818	if let Some(addr) = &cfg.metrics {
	819	crate::metrics::serve(addr, cfg.metrics_token.clone(), name.clone());
	820	}
32f54b39 Andy McCall 9 days ago ↶	821	// The manifests of images held here that the hub keeps none of, so
	822	// peers in this runner's site can fetch them without anyone pushing
	823	// them (oxygenflow#94): at start and every ten minutes, which also
	824	// catches an image imported while it runs. Only a runner that serves
	825	// its peers has anything to offer.
	826	if swarm.is_some() {
	827	let (root, cfg) = (home.root.clone(), cfg.clone());
	828	std::thread::spawn(move \|\| loop {
	829	let offered = Home::locate(Some(&root))
	830	.and_then(\|home\| crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref())?.offer_held(&home));
	831	match offered {
	832	Ok(0) => {}
	833	Ok(n) => eprintln!("flow runner: sent the hub the manifests of {n} image(s) only runners hold"),
	834	Err(e) => eprintln!("flow runner: offering held images: {e:#}"),
	835	}
	836	std::thread::sleep(Duration::from_secs(600));
	837	});
	838	}
bb643c25 Andy McCall 1 month ago ↶	839	let used = std::sync::Mutex::new(Used::default());
e0918b88 Andy McCall 1 month ago	840	let mut idle = Duration::from_secs(2);
3007d8f0 Andy McCall 1 month ago ↶	841	let pending = report_pending(home);
1daca0b6 Andy McCall 1 month ago ↶	842	// The jobs held now, told to the hub at each claim so it can end the
	843	// ones this runner lost: a job is held from its claim until its result
	844	// is reported.
	845	let held = std::sync::Mutex::new(std::collections::BTreeSet::<i64>::new());
	846	held.lock().unwrap().extend(pending.iter().map(\|(job, _)\| *job));
3007d8f0 Andy McCall 1 month ago ↶	847	std::thread::scope(\|scope\| {
3007d8f0 Andy McCall 1 month ago ↶	848	for (job, f) in pending {
1daca0b6 Andy McCall 1 month ago ↶	849	let (hub, held) = (&hub, &held);
3007d8f0 Andy McCall 1 month ago ↶	850	scope.spawn(move \|\| {
	851	eprintln!("job {job}: reporting the result kept from before the restart");
	852	report_finish(home, hub, job, &f);
1daca0b6 Andy McCall 1 month ago ↶	853	held.lock().unwrap().remove(&job);
3007d8f0 Andy McCall 1 month ago ↶	854	});
3007d8f0 Andy McCall 1 month ago ↶	855	}
62829d15 Andy McCall 24 days ago ↶	856	// The speed (spec §11.1): measured now, and again every 15 minutes
	857	// while no job runs, so a job never shares the CPU with it.
	858	let measure = \|\| {
	859	let (speed, ms) = crate::facts::bench();
	860	eprintln!("flow runner {name}: speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
	861	speed
	862	};
	863	let mut speed = measure();
	864	let mut benched = Instant::now();
89171fca Andy McCall 22 days ago ↶	865	let mut said_draining = false;
a4b3f8cb Andy McCall 19 days ago ↶	866	// When the hub last heard from us (oxygenhub#243): a claim, or a
	867	// heartbeat while there is no room to claim.
	868	let mut heard = Instant::now();
	869	let beat = \|heard: &mut Instant\| {
	870	if heard.elapsed() >= HUB_HEARTBEAT {
	871	if let Err(e) = hub.heartbeat() {
	872	eprintln!("flow runner {name}: heartbeat: {e:#}");
	873	}
	874	*heard = Instant::now();
	875	}
	876	};
85275e1e Andy McCall 10 days ago ↶	877	let mut on_trial = crate::update::pending(&home.root).is_some();
cedd1234 Andy McCall 10 days ago ↶	878	// Runner updates (oxygenhub#322): asked of the hub once it has
	879	// answered a claim, then every 15 minutes.
	880	let mut update_asked: Option<Instant> = None;
	881	let mut update_said = String::new();
3007d8f0 Andy McCall 1 month ago ↶	882	loop {
4dd392d1 Andy McCall 9 days ago ↶	883	if !stopping() && crate::update::drain_requested(&home.root) {
	884	eprintln!("flow runner {name}: asked to drain (flow runner drain)");
	885	stop();
	886	}
494a1b9a Andy McCall 9 days ago ↶	887	if UPDATE.swap(false, std::sync::atomic::Ordering::SeqCst) \|\| crate::update::requested(&home.root) {
85275e1e Andy McCall 10 days ago ↶	888	let p = crate::update::pending(&home.root);
	889	let notice = Duration::from_secs(p.as_ref().map_or(0, \|p\| p.notice_secs));
	890	let to = p.map_or_else(\|\| "a new build".to_string(), \|p\| format!("flow {}", p.to));
	891	eprintln!("flow runner {name}: updating to {to}: leaving within {} s", notice.as_secs());
494a1b9a Andy McCall 9 days ago ↶	892	LEFT_FOR_UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);
85275e1e Andy McCall 10 days ago ↶	893	leave_within(&format!("updating to {to}"), notice);
85275e1e Andy McCall 10 days ago ↶	894	}
89171fca Andy McCall 22 days ago ↶	895	// Draining (oxygenflow#50): told to stop, claim nothing more and
	896	// stop once every job held is done and reported.
	897	if stopping() {
	898	let (jobs, holding) = (used.lock().unwrap().jobs, held.lock().unwrap().len());
	899	if jobs == 0 && holding == 0 {
	900	eprintln!("flow runner {name}: drained, stopping");
	901	break;
	902	}
	903	if !said_draining {
	904	eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)");
	905	said_draining = true;
	906	}
a4b3f8cb Andy McCall 19 days ago ↶	907	beat(&mut heard);
89171fca Andy McCall 22 days ago ↶	908	std::thread::sleep(Duration::from_millis(500));
	909	continue;
	910	}
62829d15 Andy McCall 24 days ago ↶	911	if used.lock().unwrap().jobs == 0 && benched.elapsed() >= Duration::from_secs(15 * 60) {
	912	speed = measure();
	913	benched = Instant::now();
	914	}
3007d8f0 Andy McCall 1 month ago ↶	915	let room = budget.room(*used.lock().unwrap());
3007d8f0 Andy McCall 1 month ago ↶	916	let Some((free_memory_mib, free_cpus)) = room else {
a4b3f8cb Andy McCall 19 days ago ↶	917	beat(&mut heard);
3007d8f0 Andy McCall 1 month ago ↶	918	std::thread::sleep(Duration::from_millis(500));
	919	continue;
	920	};
a4b3f8cb Andy McCall 19 days ago ↶	921	heard = Instant::now();
1e6b9fb1 Andy McCall 22 days ago ↶	922	let cache_times = crate::cache::Pool::new(&home.root, cfg.cache_gb).slot_times();
3007d8f0 Andy McCall 1 month ago ↶	923	let req = ClaimRequest {
3007d8f0 Andy McCall 1 month ago ↶	924	runner: name.clone(),
c5b702e5 Andy McCall 1 month ago ↶	925	runtimes: runtimes.clone(),
77c8c240 Andy McCall 9 days ago ↶	926	free_memory_mib: claim_memory(free_memory_mib, available_memory_mib()),
3007d8f0 Andy McCall 1 month ago ↶	927	free_cpus,
	928	info: Some(RunnerInfo {
	929	version: env!("OXY_VERSION_LONG").into(),
ea81796b Andy McCall 20 days ago ↶	930	..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
3007d8f0 Andy McCall 1 month ago ↶	931	}),
8c0a4e3a Andy McCall 1 month ago ↶	932	// Read each time: an image imported while the runner runs counts.
8c0a4e3a Andy McCall 1 month ago ↶	933	images: Some(home.images()),
1abadd4a Andy McCall 1 month ago ↶	934	image_archs: Some(home.image_archs()),
1daca0b6 Andy McCall 1 month ago ↶	935	holding: Some(held.lock().unwrap().iter().copied().collect()),
4cf0c99a Andy McCall 24 days ago ↶	936	fetches_images: true,
62829d15 Andy McCall 24 days ago ↶	937	facts: Some(crate::facts::now(home, Some(speed))),
b14aeaae Andy McCall 23 days ago ↶	938	swarm: swarm.as_ref().map(\|s\| s.addr.clone()),
1e6b9fb1 Andy McCall 22 days ago ↶	939	caches: Some(cache_times.keys().cloned().collect()),
1e6b9fb1 Andy McCall 22 days ago ↶	940	cache_times: Some(cache_times),
3007d8f0 Andy McCall 1 month ago ↶	941	};
85275e1e Andy McCall 10 days ago ↶	942	let answer = hub.claim(&req);
	943	if on_trial && answer.is_ok() {
	944	crate::update::proven(&home.root);
	945	on_trial = false;
cedd1234 Andy McCall 10 days ago ↶	946	}
	947	if answer.is_ok() && update_asked.is_none_or(\|t\| t.elapsed() >= UPDATE_EVERY) {
	948	update_asked = Some(Instant::now());
	949	if let Some(note) = crate::update::take_note(&home.root) {
	950	let _ = hub.report_update(&note);
	951	}
21a3ea4a Andy McCall 7 days ago ↶	952	consider_update(home, &hub, swarm.as_ref(), &mut update_said);
85275e1e Andy McCall 10 days ago ↶	953	}
85275e1e Andy McCall 10 days ago ↶	954	match answer {
3007d8f0 Andy McCall 1 month ago ↶	955	Ok(Some(claim)) => {
3007d8f0 Andy McCall 1 month ago ↶	956	idle = Duration::from_secs(2);
4be5e4cd Andy McCall 15 days ago ↶	957	crate::metrics::CLAIMS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3007d8f0 Andy McCall 1 month ago ↶	958	let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);
	959	{
	960	let mut u = used.lock().unwrap();
	961	u.jobs += 1;
	962	u.memory_mib += mem;
	963	u.cpus += cpus;
bb643c25 Andy McCall 1 month ago ↶	964	}
1daca0b6 Andy McCall 1 month ago ↶	965	held.lock().unwrap().insert(claim.job_id);
b14aeaae Andy McCall 23 days ago ↶	966	let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref());
3007d8f0 Andy McCall 1 month ago ↶	967	scope.spawn(move \|\| {
3007d8f0 Andy McCall 1 month ago ↶	968	let id = claim.job_id;
1215f05f Andy McCall 8 days ago ↶	969	let image_ms = fetch_image(home, hub, cfg, swarm, &claim);
b11e30f5 Andy McCall 1 month ago ↶	970	if let Some(why) = preflight(home, &claim) {
b11e30f5 Andy McCall 1 month ago ↶	971	eprintln!("job {id}: declined: {why}");
4be5e4cd Andy McCall 15 days ago ↶	972	crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
b11e30f5 Andy McCall 1 month ago ↶	973	if let Err(e) = hub.decline(id, &why) {
	974	eprintln!("job {id}: decline: {e:#}");
	975	}
	976	held.lock().unwrap().remove(&id);
	977	let mut u = used.lock().unwrap();
	978	u.jobs -= 1;
	979	u.memory_mib -= mem;
	980	u.cpus -= cpus;
	981	return;
	982	}
3007d8f0 Andy McCall 1 month ago ↶	983	eprintln!(
	984	"job {id}: {} {} on {}",
	985	claim.repo,
	986	claim.job.name,
	987	claim.heads.first().map(\|h\| &h[..12]).unwrap_or("-")
	988	);
4be5e4cd Andy McCall 15 days ago ↶	989	crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1215f05f Andy McCall 8 days ago ↶	990	let finish = run_claim(home, hub, cfg, key, &claim, image_ms).unwrap_or_else(\|e\| Finish {
3007d8f0 Andy McCall 1 month ago ↶	991	exit_code: None,
	992	duration_ms: 0,
	993	receipt: None,
	994	signed_receipt: None,
	995	error: Some(format!("{e:#}")),
8f211e20 Andy McCall 24 days ago ↶	996	peak_disk_mib: None,
8f211e20 Andy McCall 24 days ago ↶	997	peak_memory_mib: None,
1215f05f Andy McCall 8 days ago ↶	998	phases: Default::default(),
3007d8f0 Andy McCall 1 month ago ↶	999	});
	1000	eprintln!(
	1001	"job {id}: exit {:?}{}",
	1002	finish.exit_code,
	1003	finish.error.as_deref().map(\|e\| format!(" ({e})")).unwrap_or_default()
	1004	);
4be5e4cd Andy McCall 15 days ago ↶	1005	crate::metrics::RUNNING.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
	1006	let cancelled = finish.error.as_deref() == Some(CANCELLED);
	1007	crate::metrics::finished(finish.exit_code, finish.error.as_deref(), finish.duration_ms, cancelled);
06a4c472 Andy McCall 10 days ago ↶	1008	match leaving() {
	1009	// Stopped as the runner leaves: another runner takes it.
	1010	Some(why) if cancelled => {
	1011	eprintln!("job {id}: handed back: {why}");
	1012	if let Err(e) = hub.release(id, &why) {
	1013	eprintln!("job {id}: handing back: {e:#}; the hub requeues it once this runner's lease lapses");
	1014	}
	1015	}
	1016	_ if !cancelled => report_finish(home, hub, id, &finish),
	1017	_ => {}
280ef951 Andy McCall 1 month ago ↶	1018	}
1daca0b6 Andy McCall 1 month ago ↶	1019	held.lock().unwrap().remove(&id);
3007d8f0 Andy McCall 1 month ago ↶	1020	let mut u = used.lock().unwrap();
	1021	u.jobs -= 1;
	1022	u.memory_mib -= mem;
	1023	u.cpus -= cpus;
	1024	});
	1025	}
	1026	Ok(None) => std::thread::sleep(idle),
	1027	Err(e) => {
	1028	eprintln!("claim: {e:#}");
	1029	std::thread::sleep(idle);
	1030	idle = (idle * 2).min(Duration::from_secs(60));
	1031	}
e0918b88 Andy McCall 1 month ago	1032	}
e0918b88 Andy McCall 1 month ago	1033	}
89171fca Andy McCall 22 days ago ↶	1034	Ok(())
bb643c25 Andy McCall 1 month ago ↶	1035	})
f201e1bc Andy McCall 10 days ago ↶	1036	}
	1037
	1038	/// Pack and keep a handed-back job's checkpoints (oxygenflow#77); what
	1039	/// happened goes to its log, and a failure only means the next attempt
	1040	/// starts again.
	1041	#[allow(clippy::too_many_arguments)]
	1042	fn save_checkpoint(hub: &Hub, id: i64, job: &flow_core::Job, scratch: &Path, dir_tier: bool, wasm: bool, mut say: impl FnMut(String)) {
	1043	if !dir_tier {
	1044	say("checkpoint: only the WASM and syscall tiers save one yet; the next attempt starts again".into());
	1045	return;
	1046	}
	1047	let dirs: Vec<(String, PathBuf)> = job
	1048	.checkpoints
	1049	.iter()
	1050	.filter_map(\|c\| crate::checkpoint::host_dir(c, job, &scratch.join("tree"), scratch, wasm).map(\|d\| (c.clone(), d)))
	1051	.collect();
	1052	match crate::checkpoint::pack(&dirs).and_then(\|(bytes, n)\| Ok((hub.save_checkpoint(id, &bytes)?, n))) {
	1053	Ok((blob, n)) => {
	1054	say(format!("checkpoint {}: {} files, {} bytes, kept for the next attempt", &blob[..12.min(blob.len())], n.files, n.bytes))
	1055	}
	1056	Err(e) => say(format!("checkpoint: not kept ({e:#}); the next attempt starts again")),
	1057	}
89171fca Andy McCall 22 days ago ↶	1058	}
	1059
	1060	/// How many times the runner was told to stop (SIGTERM or SIGINT).
	1061	static STOPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
	1062
	1063	extern "C" fn on_stop(_: libc::c_int) {
	1064	// A second stop means now: only async-signal-safe calls here.
	1065	if STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= 1 {
	1066	unsafe { libc::_exit(143) };
	1067	}
	1068	}
	1069
	1070	/// Drain instead of dying on the first SIGTERM or SIGINT (oxygenflow#50),
	1071	/// so restarting a runner never kills the job it runs: the unit's
	1072	/// `KillMode=mixed` sends the signal to the runner alone, not its microVMs.
	1073	fn drain_on_stop() {
	1074	let handler = on_stop as extern "C" fn(libc::c_int) as libc::sighandler_t;
	1075	unsafe {
	1076	libc::signal(libc::SIGTERM, handler);
	1077	libc::signal(libc::SIGINT, handler);
	1078	}
e0918b88 Andy McCall 1 month ago	1079	}
e0918b88 Andy McCall 1 month ago	1080
89171fca Andy McCall 22 days ago ↶	1081	fn stopping() -> bool {
	1082	STOPS.load(std::sync::atomic::Ordering::SeqCst) > 0
	1083	}
	1084
280ef951 Andy McCall 1 month ago ↶	1085	/// Batches records to the hub every second (or every 200 lines). A quiet
	1086	/// job's sink still asks the hub every few seconds whether the job runs,
	1087	/// with an empty batch; once the hub answers 409 the job is not wanted.
e0918b88 Andy McCall 1 month ago	1088	struct LogSink<'a> {
	1089	hub: &'a Hub,
	1090	job: i64,
	1091	buf: Vec<Record>,
	1092	last: Instant,
280ef951 Andy McCall 1 month ago ↶	1093	gone: bool,
06a4c472 Andy McCall 10 days ago ↶	1094	/// Set when the runner leaves ([`LEAVING`]): the job stops, to be
1b160d37 Andy McCall 10 days ago ↶	1095	/// handed back, unless its expected rest fits the notice.
	1096	leaving: &'a std::sync::Mutex<Option<Leave>>,
	1097	/// When the job started, and how long the hub expects it to take.
	1098	started: Instant,
	1099	expected: Option<Duration>,
280ef951 Andy McCall 1 month ago ↶	1100	}
89171fca Andy McCall 22 days ago ↶	1101
	1102	/// The most log lines sent at once, as a busy job's sink flushes them.
	1103	const BATCH: usize = 200;
280ef951 Andy McCall 1 month ago ↶	1104
89171fca Andy McCall 22 days ago ↶	1105	/// The most log lines kept while the hub cannot be reached.
	1106	const MAX_UNSENT: usize = 50_000;
	1107
280ef951 Andy McCall 1 month ago ↶	1108	/// How often a quiet job asks whether it was cancelled.
	1109	const HEARTBEAT: Duration = Duration::from_secs(5);
	1110
	1111	impl microvm::Sink for LogSink<'_> {
	1112	fn record(&mut self, r: Record) {
	1113	self.push(r)
	1114	}
	1115
	1116	fn cancelled(&mut self) -> bool {
1b160d37 Andy McCall 10 days ago ↶	1117	let leave = self.leaving.lock().unwrap_or_else(\|p\| p.into_inner()).clone();
	1118	if let Some(l) = leave {
	1119	// Finished if what is left of it fits the notice; else handed
	1120	// back as the runner leaves, what the job said so far first.
	1121	let now = Instant::now();
	1122	let fits = self.expected.is_some_and(\|e\| now + e.saturating_sub(self.started.elapsed()) <= l.by);
	1123	if !fits \|\| now >= l.by {
	1124	self.flush();
	1125	return true;
	1126	}
06a4c472 Andy McCall 10 days ago ↶	1127	}
280ef951 Andy McCall 1 month ago ↶	1128	if !self.gone && self.last.elapsed() >= HEARTBEAT {
	1129	self.send();
	1130	}
	1131	self.gone
	1132	}
e0918b88 Andy McCall 1 month ago	1133	}
	1134
	1135	impl LogSink<'_> {
	1136	fn push(&mut self, mut r: Record) {
280ef951 Andy McCall 1 month ago ↶	1137	if self.gone {
	1138	return;
	1139	}
e0918b88 Andy McCall 1 month ago	1140	if r.text.len() > MAX_RECORD_TEXT {
	1141	let mut cut = MAX_RECORD_TEXT;
	1142	while !r.text.is_char_boundary(cut) {
	1143	cut -= 1;
	1144	}
	1145	r.text.truncate(cut);
	1146	}
	1147	self.buf.push(r);
89171fca Andy McCall 22 days ago ↶	1148	if self.buf.len() >= BATCH \|\| self.last.elapsed() >= Duration::from_secs(1) {
e0918b88 Andy McCall 1 month ago	1149	self.flush();
	1150	}
	1151	}
	1152
	1153	fn flush(&mut self) {
	1154	if !self.buf.is_empty() {
280ef951 Andy McCall 1 month ago ↶	1155	self.send();
	1156	}
	1157	self.last = Instant::now();
	1158	}
	1159
89171fca Andy McCall 22 days ago ↶	1160	/// Send what is buffered, even nothing, in batches of at most
89171fca Andy McCall 22 days ago ↶	1161	/// `BATCH` lines.
280ef951 Andy McCall 1 month ago ↶	1162	fn send(&mut self) {
89171fca Andy McCall 22 days ago ↶	1163	loop {
	1164	if self.gone {
	1165	return;
	1166	}
	1167	let n = self.buf.len().min(BATCH);
	1168	let batch: Vec<Record> = self.buf.drain(..n).collect();
	1169	match self.hub.log(self.job, batch.clone()) {
	1170	Ok(()) => {}
	1171	Err(e) if job_gone(&e) => self.gone = true,
	1172	Err(e) => {
	1173	// The hub is away (a restart, say): keep the lines for
	1174	// the next batch, up to a bound, so the log loses none
	1175	// (oxygenflow#49).
	1176	eprintln!("job {}: log: {e:#}; kept for the next batch", self.job);
	1177	let mut kept = batch;
	1178	kept.append(&mut self.buf);
	1179	let over = kept.len().saturating_sub(MAX_UNSENT);
	1180	if over > 0 {
	1181	kept.drain(..over);
	1182	kept.insert(
	1183	0,
	1184	Record {
	1185	t: 0,
	1186	stream: "flow".into(),
	1187	text: format!("({over} log lines lost while the hub could not be reached)"),
	1188	},
	1189	);
	1190	}
	1191	self.buf = kept;
	1192	break;
	1193	}
	1194	}
	1195	if self.buf.is_empty() {
	1196	break;
	1197	}
e0918b88 Andy McCall 1 month ago	1198	}
	1199	self.last = Instant::now();
	1200	}
	1201	}
a4b3f8cb Andy McCall 19 days ago ↶	1202
	1203	/// How often a runner with no room to claim tells the hub it is there:
	1204	/// well inside the hub's 90 s before it shows a runner offline.
	1205	const HUB_HEARTBEAT: Duration = Duration::from_secs(30);
cedd1234 Andy McCall 10 days ago ↶	1206
	1207	/// How often a runner asks the hub whether it should run another build.
	1208	const UPDATE_EVERY: Duration = Duration::from_secs(15 * 60);
	1209
	1210	/// Notice a runner gives itself to leave for an update the hub offered.
	1211	const UPDATE_NOTICE: Duration = Duration::from_secs(600);
bb643c25 Andy McCall 1 month ago ↶	1212
cedd1234 Andy McCall 10 days ago ↶	1213	/// Ask the hub for this runner's update, and act on it: download the build
	1214	/// it offers, check it against the hub's BLAKE3 and as a runner build, and
	1215	/// install it, or for a dry run say what would happen. What it did goes to
	1216	/// the hub once per outcome (`said`).
21a3ea4a Andy McCall 7 days ago ↶	1217	fn consider_update(home: &Home, hub: &Hub, swarm: Option<&crate::swarm::Swarm>, said: &mut String) {
cedd1234 Andy McCall 10 days ago ↶	1218	let answer = match hub.update_answer() {
	1219	Ok(Some(a)) => a,
	1220	Ok(None) => return,
	1221	Err(e) => {
	1222	eprintln!("flow runner: asking for an update: {e:#}");
	1223	return;
	1224	}
	1225	};
	1226	let Some(offer) = answer.offer else { return };
	1227	let mut say = \|note: String\| {
	1228	if *said != note {
	1229	eprintln!("flow runner: {note}");
	1230	if let Err(e) = hub.report_update(&note) {
	1231	eprintln!("flow runner: telling the hub: {e:#}");
	1232	}
	1233	*said = note;
	1234	}
	1235	};
	1236	if home.config.runner.as_ref().and_then(\|r\| r.update.as_ref()).is_some_and(\|u\| !u.allow) {
	1237	return say(format!("refuses flow {}: flow.toml says [runner.update] allow = false", offer.version));
	1238	}
	1239	let dir = home.root.join("updates");
	1240	let file = dir.join(format!("flow-{}", offer.version));
	1241	let fresh = std::fs::read(&file).is_ok_and(\|b\| blake3::hash(&b).to_hex().as_str() == offer.blake3);
	1242	if !fresh {
21a3ea4a Andy McCall 7 days ago ↶	1243	if let Err(e) =
	1244	std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(\|_\| fetch_build(home, hub, swarm, &offer, &file))
	1245	{
cedd1234 Andy McCall 10 days ago ↶	1246	return say(format!("flow {}: not fetched: {e:#}", offer.version));
	1247	}
	1248	}
	1249	let exe = match std::env::current_exe() {
	1250	Ok(e) => e,
	1251	Err(e) => return say(format!("flow {}: cannot find its own binary: {e}", offer.version)),
	1252	};
	1253	match crate::update::install(home, &exe, &file, UPDATE_NOTICE, answer.policy.dry_run, offer.older) {
	1254	Ok(done) if answer.policy.dry_run => say(format!("dry run: flow {} checks out here and would replace flow {}", done.to, done.from)),
	1255	Ok(done) => {
	1256	let _ = std::fs::remove_file(&file);
	1257	say(format!("installed flow {} (was {}); leaving within {} s to start it", done.to, done.from, UPDATE_NOTICE.as_secs()))
	1258	}
	1259	Err(e) => say(format!("flow {}: not installed: {e:#}", offer.version)),
21a3ea4a Andy McCall 7 days ago ↶	1260	}
	1261	}
	1262
	1263	/// Fetch an offered build to `file`: through the swarm when this runner is
	1264	/// in one (oxygenflow#84), as an image named by the build's BLAKE3, which
	1265	/// the hub chunks when it first offers it, so runners of a site fetch it
	1266	/// from each other and the hub serves it about once per site; over HTTP
	1267	/// otherwise, or when the swarm cannot get it. Either way it must be the
	1268	/// size and BLAKE3 the hub attests.
	1269	fn fetch_build(
	1270	home: &Home,
	1271	hub: &Hub,
	1272	swarm: Option<&crate::swarm::Swarm>,
	1273	offer: &flow_core::api::UpdateOffer,
	1274	file: &Path,
	1275	) -> Result<()> {
	1276	if let Some(sw) = swarm.filter(\|_\| jet::images::is_hash(&offer.blake3)) {
	1277	match sw.fetch(home, &offer.blake3) {
	1278	Ok(Some(m)) => {
	1279	let image = home.rootfs(&offer.blake3);
	1280	match crate::update::copy_checked(&image, file, offer.size, &offer.blake3) {
	1281	Ok(()) => {
	1282	eprintln!(
	1283	"flow runner: fetched flow {} through the swarm: {} chunks, {} ({} from peers)",
	1284	offer.version,
	1285	m.chunks,
	1286	crate::images::human(m.bytes),
	1287	crate::images::human(m.peer_bytes)
	1288	);
	1289	return Ok(());
	1290	}
	1291	Err(e) => eprintln!("flow runner: flow {} from the swarm: {e:#}; fetching it over HTTP", offer.version),
	1292	}
	1293	}
	1294	Ok(None) => eprintln!("flow runner: the hub has not shared flow {} with the swarm yet; fetching it over HTTP", offer.version),
	1295	Err(e) => eprintln!("flow runner: flow {} through the swarm: {e:#}; fetching it over HTTP", offer.version),
	1296	}
cedd1234 Andy McCall 10 days ago ↶	1297	}
21a3ea4a Andy McCall 7 days ago ↶	1298	hub.download_build(offer, file)
cedd1234 Andy McCall 10 days ago ↶	1299	}
cedd1234 Andy McCall 10 days ago ↶	1300
ed4aef8b Andy McCall 19 days ago ↶	1301	/// Held while a job takes its cache disks and images its workspace (see
ed4aef8b Andy McCall 19 days ago ↶	1302	/// `run_claim`): one job at a time on the runner's disk and cache pool.
bb643c25 Andy McCall 1 month ago ↶	1303	static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());
e0918b88 Andy McCall 1 month ago	1304
ed4aef8b Andy McCall 19 days ago ↶	1305	/// Held while the first clone of any repository passes its token through
	1306	/// the process environment.
	1307	static CLONE: std::sync::Mutex<()> = std::sync::Mutex::new(());
	1308
	1309	/// One lock per cached repository: fetches into it, and exports from it,
	1310	/// never overlap, while jobs of other repositories go on. A slow repository
	1311	/// (one kept in a bucket, say) once held every job on the runner waiting
	1312	/// behind a single global lock.
	1313	fn repo_lock(name: &str) -> std::sync::Arc<std::sync::Mutex<()>> {
	1314	static LOCKS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<std::sync::Mutex<()>>>>> =
	1315	std::sync::OnceLock::new();
	1316	let mut locks = LOCKS.get_or_init(Default::default).lock().unwrap_or_else(\|p\| p.into_inner());
	1317	locks.entry(name.to_string()).or_default().clone()
	1318	}
	1319
	1320	/// Seconds, to one decimal place, for the job's log.
	1321	fn secs(d: std::time::Duration) -> String {
	1322	format!("{:.1} s", d.as_secs_f64())
	1323	}
	1324
1215f05f Andy McCall 8 days ago ↶	1325	/// `image_ms`: how long fetching its image took, when the runner lacked it.
1215f05f Andy McCall 8 days ago ↶	1326	fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim, image_ms: Option<u64>) -> Result<Finish> {
e0918b88 Andy McCall 1 month ago	1327	let started = Instant::now();
1b160d37 Andy McCall 10 days ago ↶	1328	let mut sink = LogSink {
	1329	hub,
	1330	job: claim.job_id,
	1331	buf: Vec::new(),
	1332	last: Instant::now(),
	1333	gone: false,
	1334	leaving: &LEAVING,
	1335	started: Instant::now(),
	1336	expected: claim.expected_ms.map(Duration::from_millis),
	1337	};
e0918b88 Andy McCall 1 month ago	1338	let note =
	1339	\|sink: &mut LogSink, text: String\| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
	1340	let heads: Vec<ChangeHash> = claim.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
	1341	note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(\|h\| &h[..12]).collect::<Vec<_>>().join("+")));
ed4aef8b Andy McCall 19 days ago ↶	1342	// Preparing a workspace: fetches into a cached repository, and exports
	1343	// from it, are one job at a time per repository (`repo_lock`); taking
	1344	// cache disks and imaging the workspace are one job at a time on the
	1345	// runner (`PREPARE`); a first clone, whose token goes through the process
	1346	// environment, one at a time (`CLONE`).
	1347	let main_lock = repo_lock(&claim.repo);
	1348	let asked = Instant::now();
	1349	let mut main_guard = Some(main_lock.lock().unwrap_or_else(\|p\| p.into_inner()));
	1350	if asked.elapsed().as_secs() >= 2 {
	1351	note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), claim.repo));
	1352	}
	1353	let fetching = Instant::now();
e0918b88 Andy McCall 1 month ago	1354	let repo = cached_repo(home, cfg, claim)?;
ed4aef8b Andy McCall 19 days ago ↶	1355	let fetched = fetching.elapsed();
ed4aef8b Andy McCall 19 days ago ↶	1356	let mut preparing = None;
e0918b88 Andy McCall 1 month ago	1357	let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
	1358	let _ = std::fs::remove_dir_all(&scratch);
	1359	std::fs::create_dir_all(scratch.join("tree"))?;
	1360	let result = (\|\| -> Result<Finish> {
ed4aef8b Andy McCall 19 days ago ↶	1361	let exporting = Instant::now();
c23d5191 Andy McCall 1 month ago ↶	1362	let mut files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
ed4aef8b Andy McCall 19 days ago ↶	1363	main_guard.take();
ed4aef8b Andy McCall 19 days ago ↶	1364	note(&mut sink, format!("fetched in {}, {} files exported in {}", secs(fetched), files.len(), secs(exporting.elapsed())));
c23d5191 Andy McCall 1 month ago ↶	1365	// Pinned subrepos (Oxygen spec §7.8), at their paths, as part of the
	1366	// workspace and of the scope hash.
	1367	for sub in &claim.subrepos {
	1368	if sub.path.split('/').any(\|p\| p.is_empty() \|\| p == "." \|\| p == ".." \|\| p == ".oxy") {
	1369	bail!("subrepo path {:?} is not a clean relative path", sub.path);
	1370	}
	1371	note(
	1372	&mut sink,
	1373	format!(
	1374	"fetching subrepo {} at {} for {}",
	1375	sub.repo,
	1376	sub.heads.iter().map(\|h\| &h[..12]).collect::<Vec<_>>().join("+"),
	1377	sub.path
	1378	),
	1379	);
ed4aef8b Andy McCall 19 days ago ↶	1380	let sub_lock = repo_lock(&sub.repo);
	1381	let asked = Instant::now();
	1382	let sub_guard = sub_lock.lock().unwrap_or_else(\|p\| p.into_inner());
	1383	if asked.elapsed().as_secs() >= 2 {
	1384	note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), sub.repo));
	1385	}
c23d5191 Andy McCall 1 month ago ↶	1386	let sub_repo = cached_repo_at(home, cfg, &sub.repo, &sub.clone_url, &claim.token)?;
	1387	let sub_heads: Vec<ChangeHash> =
	1388	sub.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad subrepo head")).collect::<Result<_>>()?;
	1389	let dir = scratch.join("tree").join(&sub.path);
	1390	std::fs::create_dir_all(&dir)?;
	1391	let sub_files = workspace::export_state(&sub_repo, &sub_heads, &dir)?;
ed4aef8b Andy McCall 19 days ago ↶	1392	drop(sub_guard);
c23d5191 Andy McCall 1 month ago ↶	1393	files.extend(sub_files.into_iter().map(\|(p, h)\| (format!("{}/{p}", sub.path), h)));
c23d5191 Andy McCall 1 month ago ↶	1394	}
e0918b88 Andy McCall 1 month ago	1395	let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(\|(p, h)\| (p.as_str(), *h)));
46fe2c81 Andy McCall 19 days ago ↶	1396	// Inputs (oxygenflow#58): each producing job's artifact, checked
	1397	// against the blob its receipt names, then put at its path. Not part
	1398	// of the scope: they are the run's, not the repository's.
	1399	let mut input_blobs = Vec::new();
	1400	for (index, i) in claim.inputs.iter().enumerate() {
	1401	let what = format!("input {} from {}", i.artifact, i.from);
	1402	let bytes = hub.input(claim.job_id, index).with_context(\|\| format!("{what}: fetching it"))?;
	1403	let blob = place_input(&scratch.join("tree"), i, &bytes).with_context(\|\| what.clone())?;
	1404	note(&mut sink, format!("{what}: {} bytes at {}, blob {} as its receipt names", bytes.len(), i.path, &i.blob[..12]));
	1405	input_blobs.push(blob);
	1406	}
ea56c56e Andy McCall 1 month ago ↶	1407	// The syscall tier runs on the tree itself; a microVM gets an image.
ea56c56e Andy McCall 1 month ago ↶	1408	let syscall_tier = claim.sandbox.runtime == flow_core::Runtime::Syscall;
28cc603f Andy McCall 10 days ago ↶	1409	let wasm_tier = claim.sandbox.runtime == flow_core::Runtime::Wasm;
	1410	// The syscall and WASM tiers see the job's files as a directory;
	1411	// the MicroVM tier as an ext4 image.
	1412	let dir_tier = syscall_tier \|\| wasm_tier;
82cf6e91 Andy McCall 22 days ago ↶	1413	let job = claim.job.in_sandbox(&claim.sandbox).with_flow_env(&claim.flow_env());
2594278f Andy McCall 16 days ago ↶	1414	// Disk mounts that name a cache start from a kept disk (#28): a
	1415	// microVM takes an image already in place as it is, the syscall tier
	1416	// a directory (#70). Taken before the workspace is imaged: the
	1417	// sources the disk was built from give unchanged files their old
	1418	// times back (#52).
ed4aef8b Andy McCall 19 days ago ↶	1419	let asked = Instant::now();
	1420	preparing = Some(PREPARE.lock().unwrap_or_else(\|p\| p.into_inner()));
	1421	if asked.elapsed().as_secs() >= 2 {
	1422	note(&mut sink, format!("waited {} for another job imaging its workspace", secs(asked.elapsed())));
	1423	}
82cf6e91 Andy McCall 22 days ago ↶	1424	let pool = crate::cache::Pool::new(&home.root, cfg.cache_gb);
28cc603f Andy McCall 10 days ago ↶	1425	let form = if dir_tier { crate::cache::Form::Dir } else { crate::cache::Form::Image };
82cf6e91 Andy McCall 22 days ago ↶	1426	let cache_ctx =
2594278f Andy McCall 16 days ago ↶	1427	crate::cache::Ctx { repo: &claim.repo, bookmark: &claim.bookmark, job: &claim.job.name, image: &claim.sandbox.rootfs, form };
82cf6e91 Andy McCall 22 days ago ↶	1428	let mut cached: Vec<(usize, String, String)> = Vec::new();
82cf6e91 Andy McCall 22 days ago ↶	1429	let mut built_from: Option<crate::cache::Sources> = None;
2594278f Andy McCall 16 days ago ↶	1430	for (i, m) in job.mounts.iter().filter(\|m\| m.source == flow_core::pipeline::MountSource::Disk).enumerate() {
	1431	let Some(name) = &m.cache else { continue };
	1432	let key = crate::cache::Pool::key(&cache_ctx, m, \|p\| files.iter().find(\|(f, _)\| f == p).map(\|(_, h)\| hex::encode(h)));
	1433	let how = match pool.take(&cache_ctx, m, &key, &crate::cache::disk_path(form, &scratch, i)) {
	1434	Some(t) => {
	1435	if built_from.is_none() {
	1436	built_from = t.sources;
82cf6e91 Andy McCall 22 days ago ↶	1437	}
2594278f Andy McCall 16 days ago ↶	1438	t.how
	1439	}
	1440	None => format!("cache {name}: none kept yet, an empty disk"),
	1441	};
	1442	note(&mut sink, format!("{} ({})", how, m.target));
077a6524 Andy McCall 15 days ago ↶	1443	// What the job reads is seen when the disk is pruned (#72).
	1444	if let Err(e) = crate::cache::forget_reads(form, &crate::cache::disk_path(form, &scratch, i)) {
	1445	note(&mut sink, format!("cache {name}: {e:#}"));
	1446	}
2594278f Andy McCall 16 days ago ↶	1447	cached.push((i, key, how));
82cf6e91 Andy McCall 22 days ago ↶	1448	}
077a6524 Andy McCall 15 days ago ↶	1449	// What the job reads or writes on its cache from now on is what it
	1450	// used; the rest may be pruned before the cache is kept (#72).
	1451	let cache_started = if cached.is_empty() { std::time::SystemTime::now() } else { flow_init::prune::start_clock() };
82cf6e91 Andy McCall 22 days ago ↶	1452	let sources = if cached.is_empty() {
	1453	crate::cache::Sources::new()
	1454	} else {
	1455	let (sources, restored) = workspace::restore_times(&scratch.join("tree"), &files, built_from.as_ref());
	1456	if built_from.is_some() {
de1d7926 Andy McCall 22 days ago ↶	1457	note(
	1458	&mut sink,
	1459	format!(
	1460	"cache: {restored} of {} files and directories unchanged since the cached build, their times kept",
	1461	sources.len()
	1462	),
	1463	);
82cf6e91 Andy McCall 22 days ago ↶	1464	}
	1465	sources
	1466	};
f201e1bc Andy McCall 10 days ago ↶	1467	// An earlier attempt's checkpoint (oxygenflow#77): back in place,
	1468	// over the cache, before the same exec runs again; the receipt
	1469	// lists it among the job's inputs.
	1470	if let Some(blob) = &claim.checkpoint {
	1471	if dir_tier {
	1472	let bytes = hub.checkpoint(claim.job_id).context("fetching the checkpoint")?;
	1473	if Repo::blob_ref(&bytes).to_hex() != *blob {
	1474	bail!("the checkpoint fetched is not blob {}, as the claim names", &blob[..12.min(blob.len())]);
	1475	}
	1476	let place = \|root: &str\| crate::checkpoint::host_dir(root, &job, &scratch.join("tree"), &scratch, wasm_tier);
	1477	let n = crate::checkpoint::unpack(&bytes, &job.checkpoints, place)?;
	1478	note(
	1479	&mut sink,
	1480	format!("restored checkpoint {}: {} files, {} bytes, from the attempt handed back", &blob[..12], n.files, n.bytes),
	1481	);
	1482	input_blobs.push(Repo::blob_ref(&bytes).0);
	1483	} else {
	1484	note(&mut sink, "a checkpoint was saved, but only the WASM and syscall tiers restore one yet: starting again".into());
	1485	}
	1486	}
28cc603f Andy McCall 10 days ago ↶	1487	if !dir_tier {
ea56c56e Andy McCall 1 month ago ↶	1488	workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
	1489	std::fs::remove_dir_all(scratch.join("tree"))?;
	1490	}
bb643c25 Andy McCall 1 month ago ↶	1491	preparing.take();
ea56c56e Andy McCall 1 month ago ↶	1492	// Cancelled while it was being prepared: no sandbox.
280ef951 Andy McCall 1 month ago ↶	1493	if microvm::Sink::cancelled(&mut sink) {
	1494	return Ok(cancelled());
	1495	}
28cc603f Andy McCall 10 days ago ↶	1496	if wasm_tier {
	1497	note(&mut sink, format!("starting WASM (wasmtime {}): {} MiB", crate::wasm::WASMTIME, claim.sandbox.memory_mib));
	1498	} else {
	1499	note(
	1500	&mut sink,
	1501	format!(
	1502	"{}: {} {}, {} MiB, rootfs {}",
	1503	if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
	1504	claim.sandbox.cpus,
	1505	if syscall_tier { "CPU" } else { "vCPU" },
	1506	claim.sandbox.memory_mib,
	1507	&claim.sandbox.rootfs[..12]
	1508	),
	1509	);
	1510	}
ea56c56e Andy McCall 1 month ago ↶	1511	let timeout = Duration::from_secs(claim.timeout_s);
1215f05f Andy McCall 8 days ago ↶	1512	// Where the time goes (oxygenflow#35): what came before the
	1513	// sandbox, the sandbox, and what comes after.
	1514	let prepare_ms = started.elapsed().as_millis() as u64;
	1515	let sandbox_started = Instant::now();
28cc603f Andy McCall 10 days ago ↶	1516	let mut outcome = if wasm_tier {
	1517	crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?
	1518	} else if syscall_tier {
ea56c56e Andy McCall 1 month ago ↶	1519	let group = format!("job-{}", claim.job_id);
	1520	crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
	1521	} else {
	1522	microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
	1523	};
1215f05f Andy McCall 8 days ago ↶	1524	let run_ms = sandbox_started.elapsed().as_millis() as u64;
280ef951 Andy McCall 1 month ago ↶	1525	if outcome.cancelled {
f201e1bc Andy McCall 10 days ago ↶	1526	// Handed back as the runner leaves: its checkpoints first, for
	1527	// the next attempt (oxygenflow#77).
	1528	if leaving().is_some() && !job.checkpoints.is_empty() {
	1529	save_checkpoint(hub, claim.job_id, &job, &scratch, dir_tier, wasm_tier, \|t\| note(&mut sink, t));
	1530	}
280ef951 Andy McCall 1 month ago ↶	1531	return Ok(cancelled());
d60aac6c Andy McCall 22 days ago ↶	1532	}
	1533	// The receipt says what a cached disk started from.
	1534	let disk_mounts: Vec<&flow_core::pipeline::Mount> =
	1535	job.mounts.iter().filter(\|m\| m.source == flow_core::pipeline::MountSource::Disk).collect();
	1536	for (i, key, how) in &cached {
	1537	outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
280ef951 Andy McCall 1 month ago ↶	1538	}
a67b97e9 Andy McCall 1 month ago ↶	1539	// A passing job's artifacts, read from its disks and uploaded; a
9a6fac2b Andy McCall 1 month ago ↶	1540	// declared artifact that cannot be kept fails the job. A failing
	1541	// job keeps only its `when "always"` ones (oxygenflow#37), outside
	1542	// its receipt, and one it cannot keep is only noted.
a67b97e9 Andy McCall 1 month ago ↶	1543	let mut outputs = vec![];
a67b97e9 Andy McCall 1 month ago ↶	1544	let mut output_error = None;
9a6fac2b Andy McCall 1 month ago ↶	1545	let passed = outcome.exit_code == Some(0);
	1546	let wanted = claim.job.outputs.iter().filter(\|o\| passed \|\| o.always);
	1547	for o in wanted {
6d53fb7d Andy McCall 25 days ago ↶	1548	if o.kind == flow_core::OutputKind::Report {
28cc603f Andy McCall 10 days ago ↶	1549	let read = if dir_tier {
6d53fb7d Andy McCall 25 days ago ↶	1550	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1551	crate::outputs::read_report_dir(&dir, &inner)
	1552	} else {
	1553	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1554	crate::outputs::read_report(&image, &inner, &scratch)
	1555	};
	1556	let kept = read.and_then(\|files\| upload_report(hub, claim.job_id, &o.name, &files));
	1557	match kept {
	1558	Ok((blob, files, bytes)) => {
	1559	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1560	let hex = blob.to_hex();
	1561	note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
	1562	if passed {
	1563	outputs.push(blob.0);
	1564	}
	1565	}
	1566	Err(e) if passed => {
	1567	output_error = Some(format!("report {}: {e:#}", o.name));
	1568	break;
	1569	}
	1570	Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
	1571	}
	1572	continue;
	1573	}
28cc603f Andy McCall 10 days ago ↶	1574	let read = if dir_tier {
9a6fac2b Andy McCall 1 month ago ↶	1575	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1576	crate::outputs::read_dir_file(&dir, &inner)
	1577	} else {
	1578	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1579	crate::outputs::read(&image, &inner, &scratch)
	1580	};
	1581	let kept = read.and_then(\|bytes\| {
	1582	let blob = Repo::blob_ref(&bytes);
	1583	let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
	1584	if named != blob.to_hex() {
	1585	bail!("the hub kept it as {named}, not {}", blob.to_hex());
	1586	}
	1587	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1588	note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
	1589	Ok(blob.0)
	1590	});
	1591	match kept {
	1592	Ok(b) if passed => outputs.push(b),
	1593	Ok(_) => {}
	1594	Err(e) if passed => {
	1595	output_error = Some(format!("artifact {}: {e:#}", o.name));
	1596	break;
a67b97e9 Andy McCall 1 month ago ↶	1597	}
9a6fac2b Andy McCall 1 month ago ↶	1598	Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
a67b97e9 Andy McCall 1 month ago ↶	1599	}
	1600	}
	1601	if let Some(e) = &output_error {
	1602	note(&mut sink, e.clone());
	1603	}
	1604	let receipt = outcome.exit_code.filter(\|_\| output_error.is_none()).map(\|code\| Receipt {
e0918b88 Andy McCall 1 month ago	1605	job_name: claim.job.name.clone(),
	1606	pipeline_hash: hex32(&claim.pipeline_hash),
	1607	scope_tree_hash: scope_hash,
46fe2c81 Andy McCall 19 days ago ↶	1608	inputs: input_blobs.clone(),
e0918b88 Andy McCall 1 month ago	1609	environment: outcome.environment.hash(),
a67b97e9 Andy McCall 1 month ago ↶	1610	outputs,
e0918b88 Andy McCall 1 month ago	1611	exit_code: code as u32,
	1612	duration_ms: outcome.duration_ms,
	1613	});
47c1d64d Andy McCall 1 month ago ↶	1614	let signed_receipt = match &receipt {
35f7a849 Andy McCall 18 days ago ↶	1615	Some(r) => {
	1616	let mut signed = sign_receipt(&repo, &heads, cfg, key, claim, r)?;
	1617	if cfg.keyless {
	1618	match sign_keyless(&repo, &heads, cfg, hub, claim, r) {
	1619	Ok(k) => signed = k,
	1620	Err(e) => {
	1621	note(&mut sink, format!("receipt: signed with the runner's key; a short-lived key was not certified: {e:#}"))
	1622	}
	1623	}
	1624	}
	1625	Some(hex::encode(signed.encode()))
	1626	}
47c1d64d Andy McCall 1 month ago ↶	1627	None => None,
47c1d64d Andy McCall 1 month ago ↶	1628	};
d60aac6c Andy McCall 22 days ago ↶	1629	// Measured before a cached disk leaves the scratch directory.
d60aac6c Andy McCall 22 days ago ↶	1630	let peak_disk_mib = crate::facts::allocated_mib(&scratch);
077a6524 Andy McCall 15 days ago ↶	1631	// A passing job's cached disks are pruned and kept (#72); a
	1632	// failing job's are thrown away, so nothing it wrote reaches a
	1633	// later job.
2594278f Andy McCall 16 days ago ↶	1634	let kept = passed && output_error.is_none();
	1635	for (i, key, _) in &cached {
	1636	let disk = crate::cache::disk_path(form, &scratch, *i);
077a6524 Andy McCall 15 days ago ↶	1637	let m = disk_mounts[*i];
	1638	let prune = crate::cache::Prune {
	1639	started: cache_started,
	1640	percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
	1641	guest: outcome.pruned.iter().find(\|(t, _)\| *t == m.target).map(\|(_, p)\| p),
	1642	};
	1643	match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
	1644	Ok(Some(line)) => note(&mut sink, line),
	1645	Ok(None) => {}
	1646	Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
d60aac6c Andy McCall 22 days ago ↶	1647	}
1215f05f Andy McCall 8 days ago ↶	1648	}
	1649	let total_ms = started.elapsed().as_millis() as u64;
	1650	let mut phases: std::collections::BTreeMap<String, u64> =
	1651	[("prepare", prepare_ms), ("run", run_ms), ("finish", total_ms.saturating_sub(prepare_ms + run_ms))]
	1652	.into_iter()
	1653	.map(\|(k, v)\| (k.to_string(), v))
	1654	.collect();
	1655	if let Some(ms) = image_ms {
	1656	phases.insert("image".into(), ms);
d60aac6c Andy McCall 22 days ago ↶	1657	}
e0918b88 Andy McCall 1 month ago	1658	Ok(Finish {
1215f05f Andy McCall 8 days ago ↶	1659	phases,
e0918b88 Andy McCall 1 month ago	1660	exit_code: outcome.exit_code,
1215f05f Andy McCall 8 days ago ↶	1661	duration_ms: total_ms,
e0918b88 Andy McCall 1 month ago	1662	receipt,
47c1d64d Andy McCall 1 month ago ↶	1663	signed_receipt,
a67b97e9 Andy McCall 1 month ago ↶	1664	error: output_error.or_else(\|\| outcome.timed_out.then(\|\| "stopped at the time limit".to_string())),
8f211e20 Andy McCall 24 days ago ↶	1665	// What the job took (oxygenflow#41): its disks are sparse files
8f211e20 Andy McCall 24 days ago ↶	1666	// in the scratch directory, so their blocks now are their peak.
d60aac6c Andy McCall 22 days ago ↶	1667	peak_disk_mib,
8f211e20 Andy McCall 24 days ago ↶	1668	peak_memory_mib: outcome.peak_memory_mib,
e0918b88 Andy McCall 1 month ago	1669	})
e0918b88 Andy McCall 1 month ago	1670	})();
ea56c56e Andy McCall 1 month ago ↶	1671	let _ = workspace::remove_tree(&scratch);
e0918b88 Andy McCall 1 month ago	1672	sink.flush();
e0918b88 Andy McCall 1 month ago	1673	result
47c1d64d Andy McCall 1 month ago ↶	1674	}
	1675
	1676	/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
	1677	/// runner (Oxygen spec §14.6). The state is computed here, not taken from
	1678	/// the claim, so the runner vouches only for what it fetched.
6d53fb7d Andy McCall 25 days ago ↶	1679	/// Upload a report's files, then its manifest under the report's name, and
	1680	/// check the hub kept each as the blob it should (oxygenhub#127). Answers
	1681	/// the manifest's blob, the file count and their total size.
	1682	fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
	1683	let mut listed = Vec::with_capacity(files.len());
	1684	let mut total = 0u64;
	1685	for (path, bytes) in files {
	1686	let blob = Repo::blob_ref(bytes).to_hex();
	1687	let named = hub.report_file(job, name, path, bytes).with_context(\|\| format!("uploading {path}"))?;
	1688	if named != blob {
	1689	bail!("the hub kept {path} as {named}, not {blob}");
	1690	}
	1691	total += bytes.len() as u64;
	1692	listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
	1693	}
	1694	let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
	1695	let blob = Repo::blob_ref(&manifest);
	1696	let named = hub.artifact(job, name, &manifest)?;
	1697	if named != blob.to_hex() {
	1698	bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
	1699	}
	1700	Ok((blob, files.len(), total))
	1701	}
	1702
47c1d64d Andy McCall 1 month ago ↶	1703	fn sign_receipt(
	1704	repo: &Repo,
	1705	heads: &[ChangeHash],
	1706	cfg: &RunnerConfig,
	1707	key: &RunnerKey,
	1708	claim: &Claim,
	1709	receipt: &Receipt,
	1710	) -> Result<oxy_core::receipt::SignedReceipt> {
	1711	let state = state_of(heads);
	1712	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1713	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
	1714	}
	1715
35f7a849 Andy McCall 18 days ago ↶	1716	/// Sign `receipt` with a fresh key the hub certifies for this job now,
	1717	/// just before signing, so the certificate covers the moment the hub
	1718	/// accepts it (keyless, oxygenhub#241). The key is dropped with the
	1719	/// receipt signed: nothing long-lived signs.
	1720	fn sign_keyless(
	1721	repo: &Repo,
	1722	heads: &[ChangeHash],
	1723	cfg: &RunnerConfig,
	1724	hub: &Hub,
	1725	claim: &Claim,
	1726	receipt: &Receipt,
	1727	) -> Result<oxy_core::receipt::SignedReceipt> {
	1728	let sk = oxy_core::identity::SecretKey::generate();
	1729	let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0);
	1730	let name = format!("{} job {}", cfg.name.as_deref().unwrap_or("runner"), claim.job_id);
	1731	let id = sk.identity(&name, "", now);
	1732	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge(&claim.token)).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1733	let (cert, issuer) = hub.certify(&claim.token, &id, &proof)?;
	1734	let state = state_of(heads);
	1735	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1736	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &id, &sk).with_certificate(cert, issuer))
	1737	}
	1738
47c1d64d Andy McCall 1 month ago ↶	1739	/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
	1740	pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
	1741	use oxy_core::hash::{hash_object, ObjectType};
	1742	hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
e0918b88 Andy McCall 1 month ago	1743	}
	1744
	1745	fn hex32(s: &str) -> [u8; 32] {
	1746	let mut out = [0; 32];
	1747	if let Ok(v) = hex::decode(s) {
	1748	if v.len() == 32 {
	1749	out.copy_from_slice(&v);
	1750	}
	1751	}
	1752	out
	1753	}
	1754
f53e6d47 Andy McCall 1 month ago ↶	1755	/// Whether a network error is worth another try: DNS, a refused or reset
	1756	/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
	1757	/// hub restarts (oxygenhub#111).
	1758	fn transient(msg: &str) -> bool {
	1759	let m = msg.to_lowercase();
	1760	[
	1761	"dns failed",
	1762	"name resolution",
	1763	"connection refused",
	1764	"connection reset",
	1765	"timed out",
	1766	"timeout",
	1767	"http 502",
	1768	"http 503",
	1769	"http 504",
	1770	"broken pipe",
	1771	"unexpected eof",
	1772	]
	1773	.iter()
	1774	.any(\|k\| m.contains(k))
	1775	}
	1776
	1777	/// Run `f` again after a transient network error, waiting `waits` between
	1778	/// tries (about a minute in all), and give up on any other error at once.
	1779	fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
	1780	let mut tries = waits.iter();
	1781	loop {
	1782	match f() {
	1783	Ok(v) => return Ok(v),
	1784	Err(e) => match tries.next() {
	1785	Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
	1786	_ => return Err(e),
	1787	},
	1788	}
	1789	}
	1790	}
	1791
	1792	/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
	1793	const FETCH_WAITS: [Duration; 5] =
	1794	[Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
	1795
e0918b88 Andy McCall 1 month ago	1796	/// The runner's copy of a repository, fetched up to date with the job's token.
e0918b88 Andy McCall 1 month ago	1797	fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
c23d5191 Andy McCall 1 month ago ↶	1798	cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
	1799	}
	1800
	1801	/// The runner's copy of `repo`, cloned or fetched with `token`.
	1802	fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
	1803	let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
e0918b88 Andy McCall 1 month ago	1804	let set_remote = \|repo: &mut Repo\| -> Result<()> {
c23d5191 Andy McCall 1 month ago ↶	1805	repo.config.set("remote.origin.url", clone_url).map_err(\|e\| anyhow::anyhow!("{e}"))?;
c23d5191 Andy McCall 1 month ago ↶	1806	repo.config.set("remote.origin.token", token).map_err(\|e\| anyhow::anyhow!("{e}"))?;
e0918b88 Andy McCall 1 month ago	1807	if let Some(ca) = &cfg.ca {
	1808	repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1809	}
	1810	repo.save_config().map_err(\|e\| anyhow::anyhow!("{e}"))
	1811	};
	1812	if dir.join(".oxy").is_dir() {
	1813	let mut repo = Repo::open(&dir).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1814	set_remote(&mut repo)?;
f53e6d47 Andy McCall 1 month ago ↶	1815	with_retries(&FETCH_WAITS, \|\| repo.fetch("origin").map(\|_\| ()).map_err(\|e\| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
e0918b88 Andy McCall 1 month ago	1816	return Ok(repo);
e0918b88 Andy McCall 1 month ago	1817	}
ed4aef8b Andy McCall 19 days ago ↶	1818	// First job for this repository: clone with the job's token, which goes
	1819	// through the process environment, so one clone at a time.
	1820	let _clone = CLONE.lock().unwrap_or_else(\|p\| p.into_inner());
c23d5191 Andy McCall 1 month ago ↶	1821	std::env::set_var("OXY_TOKEN", token);
e0918b88 Andy McCall 1 month ago	1822	if let Some(ca) = &cfg.ca {
	1823	std::env::set_var("OXY_CA", ca);
	1824	}
f53e6d47 Andy McCall 1 month ago ↶	1825	let cloned = with_retries(&FETCH_WAITS, \|\| {
	1826	let _ = std::fs::remove_dir_all(&dir);
	1827	Repo::clone(clone_url, &dir).map_err(\|e\| anyhow::anyhow!("{e}"))
	1828	});
e0918b88 Andy McCall 1 month ago	1829	std::env::remove_var("OXY_TOKEN");
	1830	std::env::remove_var("OXY_CA");
	1831	let mut repo = cloned.map_err(\|e\| {
	1832	let _ = std::fs::remove_dir_all(&dir);
c23d5191 Andy McCall 1 month ago ↶	1833	anyhow::anyhow!("clone {}: {e}", clone_url)
e0918b88 Andy McCall 1 month ago	1834	})?;
	1835	set_remote(&mut repo)?;
	1836	Ok(repo)
bb643c25 Andy McCall 1 month ago ↶	1837	}
	1838
	1839	#[cfg(test)]
	1840	mod tests {
0d663e91 Andy McCall 8 days ago ↶	1841
	1842	/// #98: register keeps a copy of the CA in the home, by absolute path,
	1843	/// and keeping the home's own copy again is a no-op.
	1844	#[test]
	1845	fn the_ca_is_kept_in_the_home_by_absolute_path() {
	1846	let home = tempfile::tempdir().unwrap();
	1847	let elsewhere = tempfile::tempdir().unwrap();
	1848	let given = elsewhere.path().join("ca.pem");
	1849	std::fs::write(&given, "PEM").unwrap();
	1850	let kept = super::keep_ca(home.path(), &given).unwrap();
	1851	assert!(kept.is_absolute() && kept.ends_with(crate::home::CA_PEM), "{kept:?}");
	1852	assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
	1853	std::fs::remove_file(&given).unwrap();
	1854	assert_eq!(super::keep_ca(home.path(), &kept).unwrap(), kept, "the home's copy again");
	1855	assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
	1856	}
46fe2c81 Andy McCall 19 days ago ↶	1857
	1858	#[test]
	1859	fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
	1860	let dir = tempfile::tempdir().unwrap();
	1861	let bytes = b"the reviewed plan";
	1862	let blob = Repo::blob_ref(bytes).to_hex();
	1863	let input = \|path: &str, blob: &str\| flow_core::api::ArtifactInput {
	1864	artifact: "tfplan".into(),
	1865	from: "plan".into(),
	1866	path: path.into(),
	1867	blob: blob.into(),
	1868	};
	1869	let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
	1870	assert_eq!(hex::encode(got), blob);
	1871	assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
	1872	let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
	1873	assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
	1874	assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
	1875	for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
	1876	assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
	1877	}
	1878	}
	1879
bb643c25 Andy McCall 1 month ago ↶	1880	use super::*;
b11e30f5 Andy McCall 1 month ago ↶	1881
	1882	#[test]
	1883	fn a_runner_names_what_stops_a_sandbox() {
	1884	let dir = tempfile::tempdir().unwrap();
	1885	let home = Home::locate(Some(dir.path())).unwrap();
	1886	let bytes = b"not really ext4";
	1887	let hash = blake3::hash(bytes).to_hex().to_string();
	1888	let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
	1889	"name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
	1890	}))
	1891	.unwrap();
	1892	let problem = \|sb: &flow_core::Sandbox, free\| sandbox_problem(&home, sb, "x86_64", free);
	1893	assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
	1894	std::fs::write(home.rootfs(&hash), bytes).unwrap();
	1895	sb.rootfs = hash.clone();
	1896	assert_eq!(problem(&sb, Some(4096)), None);
	1897	assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
	1898	std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
	1899	assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
	1900	sb.arch = Some("aarch64".into());
	1901	assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
	1902	sb.arch = None;
	1903	std::fs::remove_file(home.image_manifest(&hash)).unwrap();
	1904	std::fs::write(home.rootfs(&hash), b"changed").unwrap();
	1905	assert!(problem(&sb, None).unwrap().contains("does not verify"));
	1906	}
f53e6d47 Andy McCall 1 month ago ↶	1907
f53e6d47 Andy McCall 1 month ago ↶	1908	#[test]
0077c748 Andy McCall 22 days ago ↶	1909	fn a_runner_reads_an_unrecorded_images_architecture_and_declines_another() {
	1910	// An image fetched or imported with no recorded architecture, built
	1911	// for riscv64: the runner reads its shell, records it, and declines
	1912	// instead of running it and failing (spec §11.2, §11.5).
	1913	let dir = tempfile::tempdir().unwrap();
	1914	let home = Home::locate(Some(dir.path().join("home").as_path())).unwrap();
	1915	let tree = dir.path().join("tree");
	1916	std::fs::create_dir_all(tree.join("bin")).unwrap();
	1917	let mut elf = vec![0u8; 64];
	1918	elf[..7].copy_from_slice(b"\x7fELF\x02\x01\x01");
	1919	elf[18..20].copy_from_slice(&0xF3u16.to_le_bytes());
	1920	std::fs::write(tree.join("bin/sh"), &elf).unwrap();
	1921	let img = dir.path().join("rootfs.ext4");
	1922	let made = std::process::Command::new("mkfs.ext4").args(["-q", "-F", "-d"]).arg(&tree).arg(&img).arg("8M").status();
	1923	if !made.is_ok_and(\|s\| s.success()) {
	1924	eprintln!("mkfs.ext4 not found: skipped");
	1925	return;
	1926	}
	1927	let hash = crate::home::hash_file(&img).unwrap();
	1928	std::fs::create_dir_all(home.rootfs(&hash).parent().unwrap()).unwrap();
	1929	std::fs::rename(&img, home.rootfs(&hash)).unwrap();
	1930	assert!(!home.image_manifest(&hash).exists());
	1931	let sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
	1932	"name": "s", "runtime": "syscall", "rootfs": hash, "memory_mib": 64, "cpus": 1, "network": "isolated"
	1933	}))
	1934	.unwrap();
	1935	assert_eq!(
	1936	sandbox_problem(&home, &sb, "x86_64", None).unwrap(),
	1937	format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12])
	1938	);
	1939	assert_eq!(sandbox_problem(&home, &sb, "riscv64", None), None, "a riscv64 runner takes it");
	1940	// Learned: recorded once, and reported with the runner's claims.
	1941	let manifest: serde_json::Value = serde_json::from_slice(&std::fs::read(home.image_manifest(&hash)).unwrap()).unwrap();
	1942	assert_eq!(manifest["arch"], "riscv64");
	1943	assert_eq!(home.image_archs().get(&hash).map(String::as_str), Some("riscv64"));
	1944	}
	1945
	1946	#[test]
f53e6d47 Andy McCall 1 month ago ↶	1947	fn transient_network_errors_are_tried_again_and_others_are_not() {
	1948	assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
	1949	assert!(transient("HTTP 503: the hub is restarting"));
	1950	assert!(!transient("HTTP 404: no such repository"));
	1951	let mut n = 0;
	1952	let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], \|\| {
	1953	n += 1;
	1954	if n < 3 {
	1955	Err(anyhow::anyhow!("connection refused"))
	1956	} else {
	1957	Ok(7)
	1958	}
	1959	});
	1960	assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
	1961	let mut m = 0;
	1962	let got: Result<u32> = with_retries(&[Duration::ZERO; 4], \|\| {
	1963	m += 1;
	1964	Err(anyhow::anyhow!("HTTP 403: forbidden"))
	1965	});
	1966	assert!(got.is_err());
	1967	assert_eq!(m, 1, "a refusal is not tried again");
	1968	let mut k = 0;
	1969	let got: Result<u32> = with_retries(&[Duration::ZERO; 2], \|\| {
	1970	k += 1;
	1971	Err(anyhow::anyhow!("timed out"))
	1972	});
	1973	assert!(got.is_err());
	1974	assert_eq!(k, 3, "gives up after the last wait");
	1975	}
1b2229da Andy McCall 1 month ago ↶	1976
	1977	/// A hub that answers each request with the next of `codes`, counting them.
	1978	fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
89171fca Andy McCall 22 days ago ↶	1979	fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
	1980	}
	1981
	1982	/// A hub at `listener` answering each request with the next of `codes`.
	1983	fn fake_hub_on(listener: std::net::TcpListener, codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
1b2229da Andy McCall 1 month ago ↶	1984	use std::io::{Read, Write};
	1985	let addr = listener.local_addr().unwrap();
	1986	let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
	1987	let count = seen.clone();
	1988	std::thread::spawn(move \|\| {
	1989	for code in codes {
	1990	let (mut s, _) = listener.accept().unwrap();
f3b4fdeb Andy McCall 1 month ago ↶	1991	// Read the whole request (headers, then Content-Length bytes
	1992	// of body) before answering: closing early would reset the
	1993	// connection while the client is still sending.
	1994	let mut req = Vec::new();
1b2229da Andy McCall 1 month ago ↶	1995	let mut buf = [0u8; 8192];
f3b4fdeb Andy McCall 1 month ago ↶	1996	loop {
	1997	let n = s.read(&mut buf).unwrap_or(0);
	1998	if n == 0 {
	1999	break;
	2000	}
	2001	req.extend_from_slice(&buf[..n]);
	2002	let text = String::from_utf8_lossy(&req);
	2003	if let Some(end) = text.find("\r\n\r\n") {
	2004	let len = text[..end]
	2005	.lines()
	2006	.find_map(\|l\| {
	2007	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	2008	})
	2009	.unwrap_or(0);
	2010	if req.len() >= end + 4 + len {
	2011	break;
	2012	}
	2013	}
	2014	}
1b2229da Andy McCall 1 month ago ↶	2015	count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
	2016	let reason = if code == 204 { "No Content" } else { "Error" };
	2017	let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
	2018	}
	2019	});
	2020	(format!("http://{addr}"), seen)
f98a8f78 Andy McCall 1 month ago ↶	2021	}
	2022
	2023	/// A hub that answers one request with `body` (status 200) and hands
	2024	/// back what it was sent.
	2025	fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
	2026	use std::io::{Read, Write};
	2027	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	2028	let addr = listener.local_addr().unwrap();
	2029	let (tx, rx) = std::sync::mpsc::channel();
	2030	std::thread::spawn(move \|\| {
	2031	let (mut s, _) = listener.accept().unwrap();
	2032	let mut req = Vec::new();
	2033	let mut buf = [0u8; 65536];
	2034	loop {
	2035	let n = s.read(&mut buf).unwrap_or(0);
	2036	if n == 0 {
	2037	break;
	2038	}
	2039	req.extend_from_slice(&buf[..n]);
	2040	let text = String::from_utf8_lossy(&req);
	2041	if let Some(end) = text.find("\r\n\r\n") {
	2042	let len = text[..end]
	2043	.lines()
	2044	.find_map(\|l\| {
	2045	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	2046	})
	2047	.unwrap_or(0);
	2048	if req.len() >= end + 4 + len {
	2049	break;
	2050	}
	2051	}
	2052	}
	2053	tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
	2054	let _ = write!(
	2055	s,
	2056	"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
	2057	body.len()
	2058	);
	2059	});
	2060	(format!("http://{addr}"), rx)
	2061	}
	2062
	2063	#[test]
	2064	fn registering_binds_the_key_and_keeps_the_credential_private() {
	2065	use std::os::unix::fs::PermissionsExt;
	2066	let dir = tempfile::tempdir().unwrap();
	2067	let home = Home::locate(Some(dir.path())).unwrap();
	2068	let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
	2069	let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
	2070	assert_eq!((got.id, got.name.as_str()), (7, "plex"));
	2071	let req = sent.recv().unwrap();
	2072	assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
	2073	let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
	2074	assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
	2075	let key = home.runner_key("plex").unwrap();
	2076	assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
	2077	let path = dir.path().join(crate::home::RUNNER_TOML);
	2078	assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
	2079	// The runner then starts from runner.toml.
	2080	let again = Home::locate(Some(dir.path())).unwrap();
	2081	let cfg = again.config.runner.unwrap();
	2082	assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
	2083	assert_eq!(cfg.labels, ["gpu"]);
35f7a849 Andy McCall 18 days ago ↶	2084	}
	2085
	2086	/// A hub that answers certify with a certificate `sign_as` signs for
	2087	/// `cert_key`, and publishes `published` as its key: two requests.
	2088	fn fake_certify_hub(
	2089	sign_as: (oxy_core::identity::Identity, oxy_core::identity::SecretKey),
	2090	cert_key: oxy_core::identity::Identity,
	2091	published: oxy_core::identity::Identity,
	2092	) -> String {
	2093	use std::io::{Read, Write};
	2094	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	2095	let addr = listener.local_addr().unwrap();
	2096	std::thread::spawn(move \|\| {
	2097	for _ in 0..2 {
	2098	let Ok((mut s, _)) = listener.accept() else { return };
	2099	let mut req = Vec::new();
	2100	let mut buf = [0u8; 65536];
	2101	loop {
	2102	let n = s.read(&mut buf).unwrap_or(0);
	2103	if n == 0 {
	2104	break;
	2105	}
	2106	req.extend_from_slice(&buf[..n]);
	2107	let text = String::from_utf8_lossy(&req);
	2108	if let Some(end) = text.find("\r\n\r\n") {
	2109	let len = text[..end]
	2110	.lines()
	2111	.find_map(\|l\| {
	2112	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	2113	})
	2114	.unwrap_or(0);
	2115	if req.len() >= end + 4 + len {
	2116	break;
	2117	}
	2118	}
	2119	}
	2120	let line = String::from_utf8_lossy(&req).lines().next().unwrap_or("").to_string();
	2121	let body: Vec<u8> = if line.contains("/keyless/certify") {
	2122	let cert = oxy_core::cert::SignedCertificate::sign(
	2123	oxy_core::cert::Certificate {
	2124	key: cert_key.clone(),
	2125	issuer: sign_as.0.hash(),
	2126	subject: "repo:andy/demo:bookmark:main".into(),
	2127	claims: vec![],
	2128	scope: oxy_core::identity::Scope::SEAL,
	2129	not_before: 1,
	2130	expires: 600_001,
fc2a6d18 Andy McCall 18 days ago ↶	2131	grant: None,
35f7a849 Andy McCall 18 days ago ↶	2132	},
	2133	&sign_as.0,
	2134	&sign_as.1,
	2135	);
	2136	serde_json::json!({ "certificate": oxy_core::dsse::base64(&cert.encode()) }).to_string().into_bytes()
	2137	} else {
	2138	published.encode()
	2139	};
	2140	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
	2141	let _ = s.write_all(&body);
	2142	}
	2143	});
	2144	format!("http://{addr}")
	2145	}
	2146
	2147	/// keyless (oxygenhub#241): the runner takes a certificate only when it
	2148	/// is for its fresh key and signed by the key the hub publishes.
	2149	#[test]
	2150	fn a_certificate_is_taken_only_for_this_key_from_the_hubs_key() {
	2151	use oxy_core::identity::SecretKey;
	2152	let hub_sk = SecretKey::generate();
	2153	let hub_id = hub_sk.identity("hub", "", 1);
	2154	let sk = SecretKey::generate();
	2155	let id = sk.identity("earth-1 job 7", "", 1);
	2156	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge("flowjob_x")).unwrap();
	2157
	2158	let base = fake_certify_hub((hub_id.clone(), hub_sk.clone()), id.clone(), hub_id.clone());
	2159	let (cert, issuer) = hub_at(base).certify("flowjob_x", &id, &proof).unwrap();
	2160	assert_eq!((cert.cert.key_id(), issuer.hash()), (id.hash(), hub_id.hash()));
	2161
	2162	let mallory_sk = SecretKey::generate();
	2163	let mallory = mallory_sk.identity("mallory", "", 1);
	2164	let base = fake_certify_hub((mallory.clone(), mallory_sk), id.clone(), hub_id.clone());
	2165	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	2166	assert!(e.contains("not signed by the hub's published key"), "{e}");
	2167
	2168	let other = SecretKey::generate().identity("other", "", 1);
	2169	let base = fake_certify_hub((hub_id.clone(), hub_sk), other, hub_id);
	2170	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	2171	assert!(e.contains("another key"), "{e}");
1b2229da Andy McCall 1 month ago ↶	2172	}
	2173
	2174	fn hub_at(base: String) -> Hub {
	2175	Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
cedd1234 Andy McCall 10 days ago ↶	2176	}
	2177
	2178	/// oxygenhub#322: an offered build is kept only if it is the bytes the
	2179	/// hub attests; a peer or a proxy cannot change what is installed.
	2180	#[test]
	2181	fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {
	2182	let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());
	2183	let serve = \|bytes: Vec<u8>\| {
	2184	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	2185	let addr = listener.local_addr().unwrap();
	2186	std::thread::spawn(move \|\| {
	2187	use std::io::{Read, Write};
	2188	let (mut s, _) = listener.accept().unwrap();
	2189	let mut buf = [0u8; 4096];
	2190	let _ = s.read(&mut buf);
	2191	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());
	2192	let _ = s.write_all(&bytes);
	2193	});
	2194	format!("http://{addr}")
	2195	};
	2196	let offer = flow_core::api::UpdateOffer {
	2197	version: "0.3.2".into(),
	2198	asset: "flow-linux-x86_64".into(),
	2199	url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),
	2200	blake3: blake3::hash(&body).to_hex().to_string(),
	2201	size: body.len() as u64,
	2202	..Default::default()
	2203	};
	2204	let dir = tempfile::tempdir().unwrap();
	2205	let to = dir.path().join("flow-0.3.2");
	2206	hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();
	2207	assert_eq!(std::fs::read(&to).unwrap(), body);
	2208	std::fs::remove_file(&to).unwrap();
	2209	let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();
	2210	assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");
	2211	assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");
3007d8f0 Andy McCall 1 month ago ↶	2212	}
	2213
	2214	/// oxygenhub#71: a job's result is offered until the hub takes it,
	2215	/// through server errors, and a refusal ends the trying at once.
	2216	#[test]
	2217	fn a_result_is_reported_until_the_hub_takes_it() {
8f211e20 Andy McCall 24 days ago ↶	2218	let f = Finish {
	2219	exit_code: Some(0),
	2220	duration_ms: 1,
	2221	receipt: None,
	2222	signed_receipt: None,
	2223	error: None,
	2224	peak_disk_mib: None,
	2225	peak_memory_mib: None,
1215f05f Andy McCall 8 days ago ↶	2226	phases: Default::default(),
8f211e20 Andy McCall 24 days ago ↶	2227	};
3007d8f0 Andy McCall 1 month ago ↶	2228	let quick = Duration::from_millis(10);
	2229	let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
	2230	hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
	2231	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
	2232	let (base, seen) = fake_hub(vec![409]);
	2233	assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
	2234	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1b2229da Andy McCall 1 month ago ↶	2235	}
	2236
	2237	#[test]
	2238	fn server_errors_are_tried_again_and_client_errors_are_not() {
	2239	let quick = [Duration::from_millis(10); 3];
	2240	let (base, seen) = fake_hub(vec![500, 503, 204]);
	2241	let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
	2242	assert!(r.is_none());
	2243	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
	2244
	2245	let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
	2246	let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
	2247	assert!(e.contains("HTTP 500"), "{e}");
	2248	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
	2249
	2250	let (base, seen) = fake_hub(vec![401]);
	2251	assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
	2252	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
06a4c472 Andy McCall 10 days ago ↶	2253	}
06a4c472 Andy McCall 10 days ago ↶	2254
1b160d37 Andy McCall 10 days ago ↶	2255	static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall 10 days ago ↶	2256
	2257	/// oxygenflow#76: once the runner leaves, a job's sink says it is
	2258	/// cancelled, having sent what the job said so far, and the job is
	2259	/// handed back to the hub.
	2260	#[test]
	2261	fn a_leaving_runner_stops_its_job_and_hands_it_back() {
	2262	use microvm::Sink;
	2263	let leaving = std::sync::Mutex::new(None);
	2264	let (base, seen) = fake_hub(vec![204, 204]);
	2265	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2266	let mut sink = LogSink {
	2267	hub: &hub,
	2268	job: 7,
	2269	buf: Vec::new(),
	2270	last: Instant::now(),
	2271	gone: false,
	2272	leaving: &leaving,
	2273	started: Instant::now(),
	2274	expected: None,
	2275	};
06a4c472 Andy McCall 10 days ago ↶	2276	sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
06a4c472 Andy McCall 10 days ago ↶	2277	assert!(!sink.cancelled());
1b160d37 Andy McCall 10 days ago ↶	2278	*leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
06a4c472 Andy McCall 10 days ago ↶	2279	assert!(sink.cancelled());
	2280	assert!(sink.buf.is_empty(), "flushed before the job stops");
	2281	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
	2282	hub.release(7, "off Wi-Fi").unwrap();
	2283	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1b160d37 Andy McCall 10 days ago ↶	2284	}
	2285
	2286	/// oxygenflow#76: with notice, a job the hub expects to finish within
	2287	/// it runs on; one that will not, or whose time is unknown, is handed
	2288	/// back at once, and any job still running when the notice is up.
	2289	#[test]
	2290	fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
	2291	use microvm::Sink;
	2292	let (base, _) = fake_hub(vec![204; 8]);
	2293	let hub = hub_at(base);
	2294	let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
	2295	let sink = \|expected: Option<u64>, ran: u64\| LogSink {
	2296	hub: &hub,
	2297	job: 1,
	2298	buf: Vec::new(),
	2299	last: Instant::now(),
	2300	gone: false,
	2301	leaving: &leaving,
	2302	started: Instant::now() - Duration::from_secs(ran),
	2303	expected: expected.map(Duration::from_secs),
	2304	};
	2305	assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
	2306	assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
	2307	assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
	2308	*leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
	2309	assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
280ef951 Andy McCall 1 month ago ↶	2310	}
	2311
	2312	/// oxygenflow#32: a quiet job asks the hub every few seconds whether
	2313	/// it still runs; a 409 marks it cancelled, and nothing more is sent.
	2314	#[test]
	2315	fn a_quiet_job_learns_that_its_run_was_cancelled() {
	2316	use microvm::Sink;
	2317	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2318	let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
	2319	let (base, seen) = fake_hub(vec![204]);
	2320	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2321	let mut sink = LogSink {
	2322	hub: &hub,
	2323	job: 1,
	2324	buf: Vec::new(),
	2325	last: quiet,
	2326	gone: false,
	2327	leaving: &NOT_LEAVING,
	2328	started: Instant::now(),
	2329	expected: None,
	2330	};
280ef951 Andy McCall 1 month ago ↶	2331	assert!(!sink.cancelled());
	2332	assert!(!sink.cancelled(), "asked again only after the heartbeat");
	2333	assert_eq!(n(&seen), 1);
	2334
	2335	let (base, seen) = fake_hub(vec![409]);
	2336	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2337	let mut sink = LogSink {
	2338	hub: &hub,
	2339	job: 1,
	2340	buf: Vec::new(),
	2341	last: quiet,
	2342	gone: false,
	2343	leaving: &NOT_LEAVING,
	2344	started: Instant::now(),
	2345	expected: None,
	2346	};
280ef951 Andy McCall 1 month ago ↶	2347	assert!(sink.cancelled());
	2348	sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
	2349	sink.flush();
	2350	assert!(sink.cancelled());
	2351	assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
89171fca Andy McCall 22 days ago ↶	2352	}
	2353
	2354	/// oxygenflow#49: a hub that is away for a while (a deploy restarting
	2355	/// it) costs a job's uploads a wait, not the upload, and costs its log
	2356	/// no lines.
	2357	#[test]
	2358	fn a_job_outlasts_a_hub_restart() {
	2359	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2360	// A port nobody listens on yet: the hub is restarting.
	2361	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2362	let base = format!("http://127.0.0.1:{port}");
	2363	let hub = hub_at(base.clone());
	2364
	2365	// Log lines sent while it is away are kept, and go when it is back.
1b160d37 Andy McCall 10 days ago ↶	2366	let mut sink = LogSink {
	2367	hub: &hub,
	2368	job: 1,
	2369	buf: Vec::new(),
	2370	last: Instant::now(),
	2371	gone: false,
	2372	leaving: &NOT_LEAVING,
	2373	started: Instant::now(),
	2374	expected: None,
	2375	};
89171fca Andy McCall 22 days ago ↶	2376	for i in 0..450 {
	2377	sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
	2378	}
	2379	sink.send();
	2380	assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
	2381	let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
	2382	sink.send();
	2383	assert!(sink.buf.is_empty() && !sink.gone);
	2384	assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");
	2385
	2386	// An upload waits for the hub to come back, through refused
	2387	// connections and server errors alike.
	2388	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2389	let url = format!("http://127.0.0.1:{port}/x");
	2390	let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
	2391	let later = std::thread::spawn(move \|\| {
	2392	std::thread::sleep(Duration::from_millis(150));
	2393	fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
	2394	});
	2395	let quick = [Duration::from_millis(50); 20];
	2396	let r = retrying(&url, &quick, true, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
	2397	assert_eq!(r.status(), 204);
	2398	assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
	2399	// Without `connection`, a hub that cannot be reached ends it at once.
	2400	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2401	let url = format!("http://127.0.0.1:{port}/x");
	2402	assert!(retrying(&url, &quick, false, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
	2403	assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
	2404	}
	2405
	2406	/// oxygenflow#50: the first stop drains (the second, not called here,
	2407	/// exits at once).
	2408	#[test]
	2409	fn a_stop_drains_the_runner() {
	2410	assert!(!stopping());
	2411	on_stop(libc::SIGTERM);
	2412	assert!(stopping(), "told to stop, it claims nothing more");
5cf1e8ae Andy McCall 10 days ago ↶	2413	// An embedded runner starts afresh, and stop() drains it the same way.
	2414	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
	2415	stop();
	2416	assert!(stopping());
	2417	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
89171fca Andy McCall 22 days ago ↶	2418	}
	2419
	2420	/// Lines kept for a hub that stays away are bounded, saying how many went.
	2421	#[test]
	2422	fn unsent_log_lines_are_bounded() {
	2423	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2424	let hub = hub_at(format!("http://127.0.0.1:{port}"));
1b160d37 Andy McCall 10 days ago ↶	2425	let mut sink = LogSink {
	2426	hub: &hub,
	2427	job: 1,
	2428	buf: Vec::new(),
	2429	last: Instant::now(),
	2430	gone: false,
	2431	leaving: &NOT_LEAVING,
	2432	started: Instant::now(),
	2433	expected: None,
	2434	};
89171fca Andy McCall 22 days ago ↶	2435	sink.buf = (0..MAX_UNSENT as u64 + 10).map(\|i\| Record { t: i, stream: "out".into(), text: String::new() }).collect();
	2436	sink.send();
	2437	assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
	2438	assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
	2439	assert_eq!(sink.buf[1].t, 10, "the oldest go first");
1b2229da Andy McCall 1 month ago ↶	2440	}
bb643c25 Andy McCall 1 month ago ↶	2441
	2442	#[test]
	2443	fn the_budget_admits_jobs_that_fit() {
	2444	let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
	2445	let used = \|jobs, memory_mib, cpus\| Used { jobs, memory_mib, cpus };
	2446	assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
	2447	assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
	2448	assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
	2449	assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
	2450	let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
	2451	assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
	2452	assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
	2453	let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
	2454	assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
77c8c240 Andy McCall 9 days ago ↶	2455	}
	2456
	2457	#[test]
	2458	fn a_claim_never_asks_for_more_memory_than_the_host_has_free() {
	2459	assert_eq!(claim_memory(None, Some(6541)), Some(6541), "idle: the host's free memory, not unlimited");
	2460	assert_eq!(claim_memory(Some(16_384), Some(6541)), Some(6541));
	2461	assert_eq!(claim_memory(Some(4096), Some(6541)), Some(4096), "the budget when it is smaller");
	2462	assert_eq!(claim_memory(None, None), None, "nothing known: no limit, as before");
	2463	assert_eq!(claim_memory(Some(4096), None), Some(4096));
bb643c25 Andy McCall 1 month ago ↶	2464	}
e0918b88 Andy McCall 1 month ago	2465	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Keyboard