crates/flow/src/runner.rs at a63d3dcd2aef · andymccall/oxygenflow

# a63d3dcd2aef oxygenflow / crates / flow / src / runner.rs

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

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

Keyboard