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

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

106.3 KiB · Rust · last changed flow runner drain: drain the running runner by a request file 9 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 10 days ago ↶	547	// wasmtime runs on every machine this binary does (oxygenflow#75).
28cc603f Andy McCall 10 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 10 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 9 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 10 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 10 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 15 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 15 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() {
de1d7926 Andy McCall 22 days ago ↶	1360	note(
	1361	&mut sink,
	1362	format!(
	1363	"cache: {restored} of {} files and directories unchanged since the cached build, their times kept",
	1364	sources.len()
	1365	),
	1366	);
82cf6e91 Andy McCall 22 days ago ↶	1367	}
	1368	sources
	1369	};
f201e1bc Andy McCall 10 days ago ↶	1370	// An earlier attempt's checkpoint (oxygenflow#77): back in place,
	1371	// over the cache, before the same exec runs again; the receipt
	1372	// lists it among the job's inputs.
	1373	if let Some(blob) = &claim.checkpoint {
	1374	if dir_tier {
	1375	let bytes = hub.checkpoint(claim.job_id).context("fetching the checkpoint")?;
	1376	if Repo::blob_ref(&bytes).to_hex() != *blob {
	1377	bail!("the checkpoint fetched is not blob {}, as the claim names", &blob[..12.min(blob.len())]);
	1378	}
	1379	let place = \|root: &str\| crate::checkpoint::host_dir(root, &job, &scratch.join("tree"), &scratch, wasm_tier);
	1380	let n = crate::checkpoint::unpack(&bytes, &job.checkpoints, place)?;
	1381	note(
	1382	&mut sink,
	1383	format!("restored checkpoint {}: {} files, {} bytes, from the attempt handed back", &blob[..12], n.files, n.bytes),
	1384	);
	1385	input_blobs.push(Repo::blob_ref(&bytes).0);
	1386	} else {
	1387	note(&mut sink, "a checkpoint was saved, but only the WASM and syscall tiers restore one yet: starting again".into());
	1388	}
	1389	}
28cc603f Andy McCall 10 days ago ↶	1390	if !dir_tier {
ea56c56e Andy McCall 1 month ago ↶	1391	workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
	1392	std::fs::remove_dir_all(scratch.join("tree"))?;
	1393	}
bb643c25 Andy McCall 1 month ago ↶	1394	preparing.take();
ea56c56e Andy McCall 1 month ago ↶	1395	// Cancelled while it was being prepared: no sandbox.
280ef951 Andy McCall 1 month ago ↶	1396	if microvm::Sink::cancelled(&mut sink) {
	1397	return Ok(cancelled());
	1398	}
28cc603f Andy McCall 10 days ago ↶	1399	if wasm_tier {
	1400	note(&mut sink, format!("starting WASM (wasmtime {}): {} MiB", crate::wasm::WASMTIME, claim.sandbox.memory_mib));
	1401	} else {
	1402	note(
	1403	&mut sink,
	1404	format!(
	1405	"{}: {} {}, {} MiB, rootfs {}",
	1406	if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
	1407	claim.sandbox.cpus,
	1408	if syscall_tier { "CPU" } else { "vCPU" },
	1409	claim.sandbox.memory_mib,
	1410	&claim.sandbox.rootfs[..12]
	1411	),
	1412	);
	1413	}
ea56c56e Andy McCall 1 month ago ↶	1414	let timeout = Duration::from_secs(claim.timeout_s);
28cc603f Andy McCall 10 days ago ↶	1415	let mut outcome = if wasm_tier {
	1416	crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?
	1417	} else if syscall_tier {
ea56c56e Andy McCall 1 month ago ↶	1418	let group = format!("job-{}", claim.job_id);
	1419	crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
	1420	} else {
	1421	microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
	1422	};
280ef951 Andy McCall 1 month ago ↶	1423	if outcome.cancelled {
f201e1bc Andy McCall 10 days ago ↶	1424	// Handed back as the runner leaves: its checkpoints first, for
	1425	// the next attempt (oxygenflow#77).
	1426	if leaving().is_some() && !job.checkpoints.is_empty() {
	1427	save_checkpoint(hub, claim.job_id, &job, &scratch, dir_tier, wasm_tier, \|t\| note(&mut sink, t));
	1428	}
280ef951 Andy McCall 1 month ago ↶	1429	return Ok(cancelled());
d60aac6c Andy McCall 22 days ago ↶	1430	}
	1431	// The receipt says what a cached disk started from.
	1432	let disk_mounts: Vec<&flow_core::pipeline::Mount> =
	1433	job.mounts.iter().filter(\|m\| m.source == flow_core::pipeline::MountSource::Disk).collect();
	1434	for (i, key, how) in &cached {
	1435	outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
280ef951 Andy McCall 1 month ago ↶	1436	}
a67b97e9 Andy McCall 1 month ago ↶	1437	// A passing job's artifacts, read from its disks and uploaded; a
9a6fac2b Andy McCall 1 month ago ↶	1438	// declared artifact that cannot be kept fails the job. A failing
	1439	// job keeps only its `when "always"` ones (oxygenflow#37), outside
	1440	// its receipt, and one it cannot keep is only noted.
a67b97e9 Andy McCall 1 month ago ↶	1441	let mut outputs = vec![];
a67b97e9 Andy McCall 1 month ago ↶	1442	let mut output_error = None;
9a6fac2b Andy McCall 1 month ago ↶	1443	let passed = outcome.exit_code == Some(0);
	1444	let wanted = claim.job.outputs.iter().filter(\|o\| passed \|\| o.always);
	1445	for o in wanted {
6d53fb7d Andy McCall 25 days ago ↶	1446	if o.kind == flow_core::OutputKind::Report {
28cc603f Andy McCall 10 days ago ↶	1447	let read = if dir_tier {
6d53fb7d Andy McCall 25 days ago ↶	1448	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1449	crate::outputs::read_report_dir(&dir, &inner)
	1450	} else {
	1451	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1452	crate::outputs::read_report(&image, &inner, &scratch)
	1453	};
	1454	let kept = read.and_then(\|files\| upload_report(hub, claim.job_id, &o.name, &files));
	1455	match kept {
	1456	Ok((blob, files, bytes)) => {
	1457	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1458	let hex = blob.to_hex();
	1459	note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
	1460	if passed {
	1461	outputs.push(blob.0);
	1462	}
	1463	}
	1464	Err(e) if passed => {
	1465	output_error = Some(format!("report {}: {e:#}", o.name));
	1466	break;
	1467	}
	1468	Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
	1469	}
	1470	continue;
	1471	}
28cc603f Andy McCall 10 days ago ↶	1472	let read = if dir_tier {
9a6fac2b Andy McCall 1 month ago ↶	1473	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1474	crate::outputs::read_dir_file(&dir, &inner)
	1475	} else {
	1476	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1477	crate::outputs::read(&image, &inner, &scratch)
	1478	};
	1479	let kept = read.and_then(\|bytes\| {
	1480	let blob = Repo::blob_ref(&bytes);
	1481	let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
	1482	if named != blob.to_hex() {
	1483	bail!("the hub kept it as {named}, not {}", blob.to_hex());
	1484	}
	1485	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1486	note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
	1487	Ok(blob.0)
	1488	});
	1489	match kept {
	1490	Ok(b) if passed => outputs.push(b),
	1491	Ok(_) => {}
	1492	Err(e) if passed => {
	1493	output_error = Some(format!("artifact {}: {e:#}", o.name));
	1494	break;
a67b97e9 Andy McCall 1 month ago ↶	1495	}
9a6fac2b Andy McCall 1 month ago ↶	1496	Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
a67b97e9 Andy McCall 1 month ago ↶	1497	}
	1498	}
	1499	if let Some(e) = &output_error {
	1500	note(&mut sink, e.clone());
	1501	}
	1502	let receipt = outcome.exit_code.filter(\|_\| output_error.is_none()).map(\|code\| Receipt {
e0918b88 Andy McCall 1 month ago	1503	job_name: claim.job.name.clone(),
	1504	pipeline_hash: hex32(&claim.pipeline_hash),
	1505	scope_tree_hash: scope_hash,
46fe2c81 Andy McCall 20 days ago ↶	1506	inputs: input_blobs.clone(),
e0918b88 Andy McCall 1 month ago	1507	environment: outcome.environment.hash(),
a67b97e9 Andy McCall 1 month ago ↶	1508	outputs,
e0918b88 Andy McCall 1 month ago	1509	exit_code: code as u32,
	1510	duration_ms: outcome.duration_ms,
	1511	});
47c1d64d Andy McCall 1 month ago ↶	1512	let signed_receipt = match &receipt {
35f7a849 Andy McCall 18 days ago ↶	1513	Some(r) => {
	1514	let mut signed = sign_receipt(&repo, &heads, cfg, key, claim, r)?;
	1515	if cfg.keyless {
	1516	match sign_keyless(&repo, &heads, cfg, hub, claim, r) {
	1517	Ok(k) => signed = k,
	1518	Err(e) => {
	1519	note(&mut sink, format!("receipt: signed with the runner's key; a short-lived key was not certified: {e:#}"))
	1520	}
	1521	}
	1522	}
	1523	Some(hex::encode(signed.encode()))
	1524	}
47c1d64d Andy McCall 1 month ago ↶	1525	None => None,
47c1d64d Andy McCall 1 month ago ↶	1526	};
d60aac6c Andy McCall 22 days ago ↶	1527	// Measured before a cached disk leaves the scratch directory.
d60aac6c Andy McCall 22 days ago ↶	1528	let peak_disk_mib = crate::facts::allocated_mib(&scratch);
077a6524 Andy McCall 15 days ago ↶	1529	// A passing job's cached disks are pruned and kept (#72); a
	1530	// failing job's are thrown away, so nothing it wrote reaches a
	1531	// later job.
2594278f Andy McCall 16 days ago ↶	1532	let kept = passed && output_error.is_none();
	1533	for (i, key, _) in &cached {
	1534	let disk = crate::cache::disk_path(form, &scratch, *i);
077a6524 Andy McCall 15 days ago ↶	1535	let m = disk_mounts[*i];
	1536	let prune = crate::cache::Prune {
	1537	started: cache_started,
	1538	percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
	1539	guest: outcome.pruned.iter().find(\|(t, _)\| *t == m.target).map(\|(_, p)\| p),
	1540	};
	1541	match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
	1542	Ok(Some(line)) => note(&mut sink, line),
	1543	Ok(None) => {}
	1544	Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
d60aac6c Andy McCall 22 days ago ↶	1545	}
d60aac6c Andy McCall 22 days ago ↶	1546	}
e0918b88 Andy McCall 1 month ago	1547	Ok(Finish {
	1548	exit_code: outcome.exit_code,
	1549	duration_ms: started.elapsed().as_millis() as u64,
	1550	receipt,
47c1d64d Andy McCall 1 month ago ↶	1551	signed_receipt,
a67b97e9 Andy McCall 1 month ago ↶	1552	error: output_error.or_else(\|\| outcome.timed_out.then(\|\| "stopped at the time limit".to_string())),
8f211e20 Andy McCall 24 days ago ↶	1553	// What the job took (oxygenflow#41): its disks are sparse files
8f211e20 Andy McCall 24 days ago ↶	1554	// in the scratch directory, so their blocks now are their peak.
d60aac6c Andy McCall 22 days ago ↶	1555	peak_disk_mib,
8f211e20 Andy McCall 24 days ago ↶	1556	peak_memory_mib: outcome.peak_memory_mib,
e0918b88 Andy McCall 1 month ago	1557	})
e0918b88 Andy McCall 1 month ago	1558	})();
ea56c56e Andy McCall 1 month ago ↶	1559	let _ = workspace::remove_tree(&scratch);
e0918b88 Andy McCall 1 month ago	1560	sink.flush();
e0918b88 Andy McCall 1 month ago	1561	result
47c1d64d Andy McCall 1 month ago ↶	1562	}
	1563
	1564	/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
	1565	/// runner (Oxygen spec §14.6). The state is computed here, not taken from
	1566	/// the claim, so the runner vouches only for what it fetched.
6d53fb7d Andy McCall 25 days ago ↶	1567	/// Upload a report's files, then its manifest under the report's name, and
	1568	/// check the hub kept each as the blob it should (oxygenhub#127). Answers
	1569	/// the manifest's blob, the file count and their total size.
	1570	fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
	1571	let mut listed = Vec::with_capacity(files.len());
	1572	let mut total = 0u64;
	1573	for (path, bytes) in files {
	1574	let blob = Repo::blob_ref(bytes).to_hex();
	1575	let named = hub.report_file(job, name, path, bytes).with_context(\|\| format!("uploading {path}"))?;
	1576	if named != blob {
	1577	bail!("the hub kept {path} as {named}, not {blob}");
	1578	}
	1579	total += bytes.len() as u64;
	1580	listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
	1581	}
	1582	let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
	1583	let blob = Repo::blob_ref(&manifest);
	1584	let named = hub.artifact(job, name, &manifest)?;
	1585	if named != blob.to_hex() {
	1586	bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
	1587	}
	1588	Ok((blob, files.len(), total))
	1589	}
	1590
47c1d64d Andy McCall 1 month ago ↶	1591	fn sign_receipt(
	1592	repo: &Repo,
	1593	heads: &[ChangeHash],
	1594	cfg: &RunnerConfig,
	1595	key: &RunnerKey,
	1596	claim: &Claim,
	1597	receipt: &Receipt,
	1598	) -> Result<oxy_core::receipt::SignedReceipt> {
	1599	let state = state_of(heads);
	1600	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1601	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
	1602	}
	1603
35f7a849 Andy McCall 18 days ago ↶	1604	/// Sign `receipt` with a fresh key the hub certifies for this job now,
	1605	/// just before signing, so the certificate covers the moment the hub
	1606	/// accepts it (keyless, oxygenhub#241). The key is dropped with the
	1607	/// receipt signed: nothing long-lived signs.
	1608	fn sign_keyless(
	1609	repo: &Repo,
	1610	heads: &[ChangeHash],
	1611	cfg: &RunnerConfig,
	1612	hub: &Hub,
	1613	claim: &Claim,
	1614	receipt: &Receipt,
	1615	) -> Result<oxy_core::receipt::SignedReceipt> {
	1616	let sk = oxy_core::identity::SecretKey::generate();
	1617	let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0);
	1618	let name = format!("{} job {}", cfg.name.as_deref().unwrap_or("runner"), claim.job_id);
	1619	let id = sk.identity(&name, "", now);
	1620	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge(&claim.token)).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1621	let (cert, issuer) = hub.certify(&claim.token, &id, &proof)?;
	1622	let state = state_of(heads);
	1623	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1624	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &id, &sk).with_certificate(cert, issuer))
	1625	}
	1626
47c1d64d Andy McCall 1 month ago ↶	1627	/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
	1628	pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
	1629	use oxy_core::hash::{hash_object, ObjectType};
	1630	hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
e0918b88 Andy McCall 1 month ago	1631	}
	1632
	1633	fn hex32(s: &str) -> [u8; 32] {
	1634	let mut out = [0; 32];
	1635	if let Ok(v) = hex::decode(s) {
	1636	if v.len() == 32 {
	1637	out.copy_from_slice(&v);
	1638	}
	1639	}
	1640	out
	1641	}
	1642
f53e6d47 Andy McCall 1 month ago ↶	1643	/// Whether a network error is worth another try: DNS, a refused or reset
	1644	/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
	1645	/// hub restarts (oxygenhub#111).
	1646	fn transient(msg: &str) -> bool {
	1647	let m = msg.to_lowercase();
	1648	[
	1649	"dns failed",
	1650	"name resolution",
	1651	"connection refused",
	1652	"connection reset",
	1653	"timed out",
	1654	"timeout",
	1655	"http 502",
	1656	"http 503",
	1657	"http 504",
	1658	"broken pipe",
	1659	"unexpected eof",
	1660	]
	1661	.iter()
	1662	.any(\|k\| m.contains(k))
	1663	}
	1664
	1665	/// Run `f` again after a transient network error, waiting `waits` between
	1666	/// tries (about a minute in all), and give up on any other error at once.
	1667	fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
	1668	let mut tries = waits.iter();
	1669	loop {
	1670	match f() {
	1671	Ok(v) => return Ok(v),
	1672	Err(e) => match tries.next() {
	1673	Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
	1674	_ => return Err(e),
	1675	},
	1676	}
	1677	}
	1678	}
	1679
	1680	/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
	1681	const FETCH_WAITS: [Duration; 5] =
	1682	[Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
	1683
e0918b88 Andy McCall 1 month ago	1684	/// The runner's copy of a repository, fetched up to date with the job's token.
e0918b88 Andy McCall 1 month ago	1685	fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
c23d5191 Andy McCall 1 month ago ↶	1686	cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
	1687	}
	1688
	1689	/// The runner's copy of `repo`, cloned or fetched with `token`.
	1690	fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
	1691	let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
e0918b88 Andy McCall 1 month ago	1692	let set_remote = \|repo: &mut Repo\| -> Result<()> {
c23d5191 Andy McCall 1 month ago ↶	1693	repo.config.set("remote.origin.url", clone_url).map_err(\|e\| anyhow::anyhow!("{e}"))?;
c23d5191 Andy McCall 1 month ago ↶	1694	repo.config.set("remote.origin.token", token).map_err(\|e\| anyhow::anyhow!("{e}"))?;
e0918b88 Andy McCall 1 month ago	1695	if let Some(ca) = &cfg.ca {
	1696	repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1697	}
	1698	repo.save_config().map_err(\|e\| anyhow::anyhow!("{e}"))
	1699	};
	1700	if dir.join(".oxy").is_dir() {
	1701	let mut repo = Repo::open(&dir).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1702	set_remote(&mut repo)?;
f53e6d47 Andy McCall 1 month ago ↶	1703	with_retries(&FETCH_WAITS, \|\| repo.fetch("origin").map(\|_\| ()).map_err(\|e\| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
e0918b88 Andy McCall 1 month ago	1704	return Ok(repo);
e0918b88 Andy McCall 1 month ago	1705	}
ed4aef8b Andy McCall 19 days ago ↶	1706	// First job for this repository: clone with the job's token, which goes
	1707	// through the process environment, so one clone at a time.
	1708	let _clone = CLONE.lock().unwrap_or_else(\|p\| p.into_inner());
c23d5191 Andy McCall 1 month ago ↶	1709	std::env::set_var("OXY_TOKEN", token);
e0918b88 Andy McCall 1 month ago	1710	if let Some(ca) = &cfg.ca {
	1711	std::env::set_var("OXY_CA", ca);
	1712	}
f53e6d47 Andy McCall 1 month ago ↶	1713	let cloned = with_retries(&FETCH_WAITS, \|\| {
	1714	let _ = std::fs::remove_dir_all(&dir);
	1715	Repo::clone(clone_url, &dir).map_err(\|e\| anyhow::anyhow!("{e}"))
	1716	});
e0918b88 Andy McCall 1 month ago	1717	std::env::remove_var("OXY_TOKEN");
	1718	std::env::remove_var("OXY_CA");
	1719	let mut repo = cloned.map_err(\|e\| {
	1720	let _ = std::fs::remove_dir_all(&dir);
c23d5191 Andy McCall 1 month ago ↶	1721	anyhow::anyhow!("clone {}: {e}", clone_url)
e0918b88 Andy McCall 1 month ago	1722	})?;
	1723	set_remote(&mut repo)?;
	1724	Ok(repo)
bb643c25 Andy McCall 1 month ago ↶	1725	}
	1726
	1727	#[cfg(test)]
	1728	mod tests {
46fe2c81 Andy McCall 20 days ago ↶	1729
	1730	#[test]
	1731	fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
	1732	let dir = tempfile::tempdir().unwrap();
	1733	let bytes = b"the reviewed plan";
	1734	let blob = Repo::blob_ref(bytes).to_hex();
	1735	let input = \|path: &str, blob: &str\| flow_core::api::ArtifactInput {
	1736	artifact: "tfplan".into(),
	1737	from: "plan".into(),
	1738	path: path.into(),
	1739	blob: blob.into(),
	1740	};
	1741	let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
	1742	assert_eq!(hex::encode(got), blob);
	1743	assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
	1744	let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
	1745	assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
	1746	assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
	1747	for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
	1748	assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
	1749	}
	1750	}
	1751
bb643c25 Andy McCall 1 month ago ↶	1752	use super::*;
b11e30f5 Andy McCall 1 month ago ↶	1753
	1754	#[test]
	1755	fn a_runner_names_what_stops_a_sandbox() {
	1756	let dir = tempfile::tempdir().unwrap();
	1757	let home = Home::locate(Some(dir.path())).unwrap();
	1758	let bytes = b"not really ext4";
	1759	let hash = blake3::hash(bytes).to_hex().to_string();
	1760	let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
	1761	"name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
	1762	}))
	1763	.unwrap();
	1764	let problem = \|sb: &flow_core::Sandbox, free\| sandbox_problem(&home, sb, "x86_64", free);
	1765	assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
	1766	std::fs::write(home.rootfs(&hash), bytes).unwrap();
	1767	sb.rootfs = hash.clone();
	1768	assert_eq!(problem(&sb, Some(4096)), None);
	1769	assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
	1770	std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
	1771	assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
	1772	sb.arch = Some("aarch64".into());
	1773	assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
	1774	sb.arch = None;
	1775	std::fs::remove_file(home.image_manifest(&hash)).unwrap();
	1776	std::fs::write(home.rootfs(&hash), b"changed").unwrap();
	1777	assert!(problem(&sb, None).unwrap().contains("does not verify"));
	1778	}
f53e6d47 Andy McCall 1 month ago ↶	1779
f53e6d47 Andy McCall 1 month ago ↶	1780	#[test]
0077c748 Andy McCall 22 days ago ↶	1781	fn a_runner_reads_an_unrecorded_images_architecture_and_declines_another() {
	1782	// An image fetched or imported with no recorded architecture, built
	1783	// for riscv64: the runner reads its shell, records it, and declines
	1784	// instead of running it and failing (spec §11.2, §11.5).
	1785	let dir = tempfile::tempdir().unwrap();
	1786	let home = Home::locate(Some(dir.path().join("home").as_path())).unwrap();
	1787	let tree = dir.path().join("tree");
	1788	std::fs::create_dir_all(tree.join("bin")).unwrap();
	1789	let mut elf = vec![0u8; 64];
	1790	elf[..7].copy_from_slice(b"\x7fELF\x02\x01\x01");
	1791	elf[18..20].copy_from_slice(&0xF3u16.to_le_bytes());
	1792	std::fs::write(tree.join("bin/sh"), &elf).unwrap();
	1793	let img = dir.path().join("rootfs.ext4");
	1794	let made = std::process::Command::new("mkfs.ext4").args(["-q", "-F", "-d"]).arg(&tree).arg(&img).arg("8M").status();
	1795	if !made.is_ok_and(\|s\| s.success()) {
	1796	eprintln!("mkfs.ext4 not found: skipped");
	1797	return;
	1798	}
	1799	let hash = crate::home::hash_file(&img).unwrap();
	1800	std::fs::create_dir_all(home.rootfs(&hash).parent().unwrap()).unwrap();
	1801	std::fs::rename(&img, home.rootfs(&hash)).unwrap();
	1802	assert!(!home.image_manifest(&hash).exists());
	1803	let sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
	1804	"name": "s", "runtime": "syscall", "rootfs": hash, "memory_mib": 64, "cpus": 1, "network": "isolated"
	1805	}))
	1806	.unwrap();
	1807	assert_eq!(
	1808	sandbox_problem(&home, &sb, "x86_64", None).unwrap(),
	1809	format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12])
	1810	);
	1811	assert_eq!(sandbox_problem(&home, &sb, "riscv64", None), None, "a riscv64 runner takes it");
	1812	// Learned: recorded once, and reported with the runner's claims.
	1813	let manifest: serde_json::Value = serde_json::from_slice(&std::fs::read(home.image_manifest(&hash)).unwrap()).unwrap();
	1814	assert_eq!(manifest["arch"], "riscv64");
	1815	assert_eq!(home.image_archs().get(&hash).map(String::as_str), Some("riscv64"));
	1816	}
	1817
	1818	#[test]
f53e6d47 Andy McCall 1 month ago ↶	1819	fn transient_network_errors_are_tried_again_and_others_are_not() {
	1820	assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
	1821	assert!(transient("HTTP 503: the hub is restarting"));
	1822	assert!(!transient("HTTP 404: no such repository"));
	1823	let mut n = 0;
	1824	let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], \|\| {
	1825	n += 1;
	1826	if n < 3 {
	1827	Err(anyhow::anyhow!("connection refused"))
	1828	} else {
	1829	Ok(7)
	1830	}
	1831	});
	1832	assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
	1833	let mut m = 0;
	1834	let got: Result<u32> = with_retries(&[Duration::ZERO; 4], \|\| {
	1835	m += 1;
	1836	Err(anyhow::anyhow!("HTTP 403: forbidden"))
	1837	});
	1838	assert!(got.is_err());
	1839	assert_eq!(m, 1, "a refusal is not tried again");
	1840	let mut k = 0;
	1841	let got: Result<u32> = with_retries(&[Duration::ZERO; 2], \|\| {
	1842	k += 1;
	1843	Err(anyhow::anyhow!("timed out"))
	1844	});
	1845	assert!(got.is_err());
	1846	assert_eq!(k, 3, "gives up after the last wait");
	1847	}
1b2229da Andy McCall 1 month ago ↶	1848
	1849	/// A hub that answers each request with the next of `codes`, counting them.
	1850	fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
89171fca Andy McCall 22 days ago ↶	1851	fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
	1852	}
	1853
	1854	/// A hub at `listener` answering each request with the next of `codes`.
	1855	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 ↶	1856	use std::io::{Read, Write};
	1857	let addr = listener.local_addr().unwrap();
	1858	let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
	1859	let count = seen.clone();
	1860	std::thread::spawn(move \|\| {
	1861	for code in codes {
	1862	let (mut s, _) = listener.accept().unwrap();
f3b4fdeb Andy McCall 1 month ago ↶	1863	// Read the whole request (headers, then Content-Length bytes
	1864	// of body) before answering: closing early would reset the
	1865	// connection while the client is still sending.
	1866	let mut req = Vec::new();
1b2229da Andy McCall 1 month ago ↶	1867	let mut buf = [0u8; 8192];
f3b4fdeb Andy McCall 1 month ago ↶	1868	loop {
	1869	let n = s.read(&mut buf).unwrap_or(0);
	1870	if n == 0 {
	1871	break;
	1872	}
	1873	req.extend_from_slice(&buf[..n]);
	1874	let text = String::from_utf8_lossy(&req);
	1875	if let Some(end) = text.find("\r\n\r\n") {
	1876	let len = text[..end]
	1877	.lines()
	1878	.find_map(\|l\| {
	1879	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	1880	})
	1881	.unwrap_or(0);
	1882	if req.len() >= end + 4 + len {
	1883	break;
	1884	}
	1885	}
	1886	}
1b2229da Andy McCall 1 month ago ↶	1887	count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
	1888	let reason = if code == 204 { "No Content" } else { "Error" };
	1889	let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
	1890	}
	1891	});
	1892	(format!("http://{addr}"), seen)
f98a8f78 Andy McCall 1 month ago ↶	1893	}
	1894
	1895	/// A hub that answers one request with `body` (status 200) and hands
	1896	/// back what it was sent.
	1897	fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
	1898	use std::io::{Read, Write};
	1899	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	1900	let addr = listener.local_addr().unwrap();
	1901	let (tx, rx) = std::sync::mpsc::channel();
	1902	std::thread::spawn(move \|\| {
	1903	let (mut s, _) = listener.accept().unwrap();
	1904	let mut req = Vec::new();
	1905	let mut buf = [0u8; 65536];
	1906	loop {
	1907	let n = s.read(&mut buf).unwrap_or(0);
	1908	if n == 0 {
	1909	break;
	1910	}
	1911	req.extend_from_slice(&buf[..n]);
	1912	let text = String::from_utf8_lossy(&req);
	1913	if let Some(end) = text.find("\r\n\r\n") {
	1914	let len = text[..end]
	1915	.lines()
	1916	.find_map(\|l\| {
	1917	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	1918	})
	1919	.unwrap_or(0);
	1920	if req.len() >= end + 4 + len {
	1921	break;
	1922	}
	1923	}
	1924	}
	1925	tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
	1926	let _ = write!(
	1927	s,
	1928	"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
	1929	body.len()
	1930	);
	1931	});
	1932	(format!("http://{addr}"), rx)
	1933	}
	1934
	1935	#[test]
	1936	fn registering_binds_the_key_and_keeps_the_credential_private() {
	1937	use std::os::unix::fs::PermissionsExt;
	1938	let dir = tempfile::tempdir().unwrap();
	1939	let home = Home::locate(Some(dir.path())).unwrap();
	1940	let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
	1941	let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
	1942	assert_eq!((got.id, got.name.as_str()), (7, "plex"));
	1943	let req = sent.recv().unwrap();
	1944	assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
	1945	let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
	1946	assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
	1947	let key = home.runner_key("plex").unwrap();
	1948	assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
	1949	let path = dir.path().join(crate::home::RUNNER_TOML);
	1950	assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
	1951	// The runner then starts from runner.toml.
	1952	let again = Home::locate(Some(dir.path())).unwrap();
	1953	let cfg = again.config.runner.unwrap();
	1954	assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
	1955	assert_eq!(cfg.labels, ["gpu"]);
35f7a849 Andy McCall 18 days ago ↶	1956	}
	1957
	1958	/// A hub that answers certify with a certificate `sign_as` signs for
	1959	/// `cert_key`, and publishes `published` as its key: two requests.
	1960	fn fake_certify_hub(
	1961	sign_as: (oxy_core::identity::Identity, oxy_core::identity::SecretKey),
	1962	cert_key: oxy_core::identity::Identity,
	1963	published: oxy_core::identity::Identity,
	1964	) -> String {
	1965	use std::io::{Read, Write};
	1966	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	1967	let addr = listener.local_addr().unwrap();
	1968	std::thread::spawn(move \|\| {
	1969	for _ in 0..2 {
	1970	let Ok((mut s, _)) = listener.accept() else { return };
	1971	let mut req = Vec::new();
	1972	let mut buf = [0u8; 65536];
	1973	loop {
	1974	let n = s.read(&mut buf).unwrap_or(0);
	1975	if n == 0 {
	1976	break;
	1977	}
	1978	req.extend_from_slice(&buf[..n]);
	1979	let text = String::from_utf8_lossy(&req);
	1980	if let Some(end) = text.find("\r\n\r\n") {
	1981	let len = text[..end]
	1982	.lines()
	1983	.find_map(\|l\| {
	1984	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	1985	})
	1986	.unwrap_or(0);
	1987	if req.len() >= end + 4 + len {
	1988	break;
	1989	}
	1990	}
	1991	}
	1992	let line = String::from_utf8_lossy(&req).lines().next().unwrap_or("").to_string();
	1993	let body: Vec<u8> = if line.contains("/keyless/certify") {
	1994	let cert = oxy_core::cert::SignedCertificate::sign(
	1995	oxy_core::cert::Certificate {
	1996	key: cert_key.clone(),
	1997	issuer: sign_as.0.hash(),
	1998	subject: "repo:andy/demo:bookmark:main".into(),
	1999	claims: vec![],
	2000	scope: oxy_core::identity::Scope::SEAL,
	2001	not_before: 1,
	2002	expires: 600_001,
fc2a6d18 Andy McCall 18 days ago ↶	2003	grant: None,
35f7a849 Andy McCall 18 days ago ↶	2004	},
	2005	&sign_as.0,
	2006	&sign_as.1,
	2007	);
	2008	serde_json::json!({ "certificate": oxy_core::dsse::base64(&cert.encode()) }).to_string().into_bytes()
	2009	} else {
	2010	published.encode()
	2011	};
	2012	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
	2013	let _ = s.write_all(&body);
	2014	}
	2015	});
	2016	format!("http://{addr}")
	2017	}
	2018
	2019	/// keyless (oxygenhub#241): the runner takes a certificate only when it
	2020	/// is for its fresh key and signed by the key the hub publishes.
	2021	#[test]
	2022	fn a_certificate_is_taken_only_for_this_key_from_the_hubs_key() {
	2023	use oxy_core::identity::SecretKey;
	2024	let hub_sk = SecretKey::generate();
	2025	let hub_id = hub_sk.identity("hub", "", 1);
	2026	let sk = SecretKey::generate();
	2027	let id = sk.identity("earth-1 job 7", "", 1);
	2028	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge("flowjob_x")).unwrap();
	2029
	2030	let base = fake_certify_hub((hub_id.clone(), hub_sk.clone()), id.clone(), hub_id.clone());
	2031	let (cert, issuer) = hub_at(base).certify("flowjob_x", &id, &proof).unwrap();
	2032	assert_eq!((cert.cert.key_id(), issuer.hash()), (id.hash(), hub_id.hash()));
	2033
	2034	let mallory_sk = SecretKey::generate();
	2035	let mallory = mallory_sk.identity("mallory", "", 1);
	2036	let base = fake_certify_hub((mallory.clone(), mallory_sk), id.clone(), hub_id.clone());
	2037	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	2038	assert!(e.contains("not signed by the hub's published key"), "{e}");
	2039
	2040	let other = SecretKey::generate().identity("other", "", 1);
	2041	let base = fake_certify_hub((hub_id.clone(), hub_sk), other, hub_id);
	2042	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	2043	assert!(e.contains("another key"), "{e}");
1b2229da Andy McCall 1 month ago ↶	2044	}
	2045
	2046	fn hub_at(base: String) -> Hub {
	2047	Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
cedd1234 Andy McCall 10 days ago ↶	2048	}
	2049
	2050	/// oxygenhub#322: an offered build is kept only if it is the bytes the
	2051	/// hub attests; a peer or a proxy cannot change what is installed.
	2052	#[test]
	2053	fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {
	2054	let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());
	2055	let serve = \|bytes: Vec<u8>\| {
	2056	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	2057	let addr = listener.local_addr().unwrap();
	2058	std::thread::spawn(move \|\| {
	2059	use std::io::{Read, Write};
	2060	let (mut s, _) = listener.accept().unwrap();
	2061	let mut buf = [0u8; 4096];
	2062	let _ = s.read(&mut buf);
	2063	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());
	2064	let _ = s.write_all(&bytes);
	2065	});
	2066	format!("http://{addr}")
	2067	};
	2068	let offer = flow_core::api::UpdateOffer {
	2069	version: "0.3.2".into(),
	2070	asset: "flow-linux-x86_64".into(),
	2071	url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),
	2072	blake3: blake3::hash(&body).to_hex().to_string(),
	2073	size: body.len() as u64,
	2074	..Default::default()
	2075	};
	2076	let dir = tempfile::tempdir().unwrap();
	2077	let to = dir.path().join("flow-0.3.2");
	2078	hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();
	2079	assert_eq!(std::fs::read(&to).unwrap(), body);
	2080	std::fs::remove_file(&to).unwrap();
	2081	let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();
	2082	assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");
	2083	assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");
3007d8f0 Andy McCall 1 month ago ↶	2084	}
	2085
	2086	/// oxygenhub#71: a job's result is offered until the hub takes it,
	2087	/// through server errors, and a refusal ends the trying at once.
	2088	#[test]
	2089	fn a_result_is_reported_until_the_hub_takes_it() {
8f211e20 Andy McCall 24 days ago ↶	2090	let f = Finish {
	2091	exit_code: Some(0),
	2092	duration_ms: 1,
	2093	receipt: None,
	2094	signed_receipt: None,
	2095	error: None,
	2096	peak_disk_mib: None,
	2097	peak_memory_mib: None,
	2098	};
3007d8f0 Andy McCall 1 month ago ↶	2099	let quick = Duration::from_millis(10);
	2100	let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
	2101	hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
	2102	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
	2103	let (base, seen) = fake_hub(vec![409]);
	2104	assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
	2105	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1b2229da Andy McCall 1 month ago ↶	2106	}
	2107
	2108	#[test]
	2109	fn server_errors_are_tried_again_and_client_errors_are_not() {
	2110	let quick = [Duration::from_millis(10); 3];
	2111	let (base, seen) = fake_hub(vec![500, 503, 204]);
	2112	let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
	2113	assert!(r.is_none());
	2114	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
	2115
	2116	let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
	2117	let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
	2118	assert!(e.contains("HTTP 500"), "{e}");
	2119	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
	2120
	2121	let (base, seen) = fake_hub(vec![401]);
	2122	assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
	2123	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
06a4c472 Andy McCall 10 days ago ↶	2124	}
06a4c472 Andy McCall 10 days ago ↶	2125
1b160d37 Andy McCall 10 days ago ↶	2126	static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall 10 days ago ↶	2127
	2128	/// oxygenflow#76: once the runner leaves, a job's sink says it is
	2129	/// cancelled, having sent what the job said so far, and the job is
	2130	/// handed back to the hub.
	2131	#[test]
	2132	fn a_leaving_runner_stops_its_job_and_hands_it_back() {
	2133	use microvm::Sink;
	2134	let leaving = std::sync::Mutex::new(None);
	2135	let (base, seen) = fake_hub(vec![204, 204]);
	2136	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2137	let mut sink = LogSink {
	2138	hub: &hub,
	2139	job: 7,
	2140	buf: Vec::new(),
	2141	last: Instant::now(),
	2142	gone: false,
	2143	leaving: &leaving,
	2144	started: Instant::now(),
	2145	expected: None,
	2146	};
06a4c472 Andy McCall 10 days ago ↶	2147	sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
06a4c472 Andy McCall 10 days ago ↶	2148	assert!(!sink.cancelled());
1b160d37 Andy McCall 10 days ago ↶	2149	*leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
06a4c472 Andy McCall 10 days ago ↶	2150	assert!(sink.cancelled());
	2151	assert!(sink.buf.is_empty(), "flushed before the job stops");
	2152	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
	2153	hub.release(7, "off Wi-Fi").unwrap();
	2154	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1b160d37 Andy McCall 10 days ago ↶	2155	}
	2156
	2157	/// oxygenflow#76: with notice, a job the hub expects to finish within
	2158	/// it runs on; one that will not, or whose time is unknown, is handed
	2159	/// back at once, and any job still running when the notice is up.
	2160	#[test]
	2161	fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
	2162	use microvm::Sink;
	2163	let (base, _) = fake_hub(vec![204; 8]);
	2164	let hub = hub_at(base);
	2165	let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
	2166	let sink = \|expected: Option<u64>, ran: u64\| LogSink {
	2167	hub: &hub,
	2168	job: 1,
	2169	buf: Vec::new(),
	2170	last: Instant::now(),
	2171	gone: false,
	2172	leaving: &leaving,
	2173	started: Instant::now() - Duration::from_secs(ran),
	2174	expected: expected.map(Duration::from_secs),
	2175	};
	2176	assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
	2177	assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
	2178	assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
	2179	*leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
	2180	assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
280ef951 Andy McCall 1 month ago ↶	2181	}
	2182
	2183	/// oxygenflow#32: a quiet job asks the hub every few seconds whether
	2184	/// it still runs; a 409 marks it cancelled, and nothing more is sent.
	2185	#[test]
	2186	fn a_quiet_job_learns_that_its_run_was_cancelled() {
	2187	use microvm::Sink;
	2188	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2189	let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
	2190	let (base, seen) = fake_hub(vec![204]);
	2191	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2192	let mut sink = LogSink {
	2193	hub: &hub,
	2194	job: 1,
	2195	buf: Vec::new(),
	2196	last: quiet,
	2197	gone: false,
	2198	leaving: &NOT_LEAVING,
	2199	started: Instant::now(),
	2200	expected: None,
	2201	};
280ef951 Andy McCall 1 month ago ↶	2202	assert!(!sink.cancelled());
	2203	assert!(!sink.cancelled(), "asked again only after the heartbeat");
	2204	assert_eq!(n(&seen), 1);
	2205
	2206	let (base, seen) = fake_hub(vec![409]);
	2207	let hub = hub_at(base);
1b160d37 Andy McCall 10 days ago ↶	2208	let mut sink = LogSink {
	2209	hub: &hub,
	2210	job: 1,
	2211	buf: Vec::new(),
	2212	last: quiet,
	2213	gone: false,
	2214	leaving: &NOT_LEAVING,
	2215	started: Instant::now(),
	2216	expected: None,
	2217	};
280ef951 Andy McCall 1 month ago ↶	2218	assert!(sink.cancelled());
	2219	sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
	2220	sink.flush();
	2221	assert!(sink.cancelled());
	2222	assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
89171fca Andy McCall 22 days ago ↶	2223	}
	2224
	2225	/// oxygenflow#49: a hub that is away for a while (a deploy restarting
	2226	/// it) costs a job's uploads a wait, not the upload, and costs its log
	2227	/// no lines.
	2228	#[test]
	2229	fn a_job_outlasts_a_hub_restart() {
	2230	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2231	// A port nobody listens on yet: the hub is restarting.
	2232	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2233	let base = format!("http://127.0.0.1:{port}");
	2234	let hub = hub_at(base.clone());
	2235
	2236	// Log lines sent while it is away are kept, and go when it is back.
1b160d37 Andy McCall 10 days ago ↶	2237	let mut sink = LogSink {
	2238	hub: &hub,
	2239	job: 1,
	2240	buf: Vec::new(),
	2241	last: Instant::now(),
	2242	gone: false,
	2243	leaving: &NOT_LEAVING,
	2244	started: Instant::now(),
	2245	expected: None,
	2246	};
89171fca Andy McCall 22 days ago ↶	2247	for i in 0..450 {
	2248	sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
	2249	}
	2250	sink.send();
	2251	assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
	2252	let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
	2253	sink.send();
	2254	assert!(sink.buf.is_empty() && !sink.gone);
	2255	assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");
	2256
	2257	// An upload waits for the hub to come back, through refused
	2258	// connections and server errors alike.
	2259	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2260	let url = format!("http://127.0.0.1:{port}/x");
	2261	let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
	2262	let later = std::thread::spawn(move \|\| {
	2263	std::thread::sleep(Duration::from_millis(150));
	2264	fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
	2265	});
	2266	let quick = [Duration::from_millis(50); 20];
	2267	let r = retrying(&url, &quick, true, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
	2268	assert_eq!(r.status(), 204);
	2269	assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
	2270	// Without `connection`, a hub that cannot be reached ends it at once.
	2271	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2272	let url = format!("http://127.0.0.1:{port}/x");
	2273	assert!(retrying(&url, &quick, false, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
	2274	assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
	2275	}
	2276
	2277	/// oxygenflow#50: the first stop drains (the second, not called here,
	2278	/// exits at once).
	2279	#[test]
	2280	fn a_stop_drains_the_runner() {
	2281	assert!(!stopping());
	2282	on_stop(libc::SIGTERM);
	2283	assert!(stopping(), "told to stop, it claims nothing more");
5cf1e8ae Andy McCall 10 days ago ↶	2284	// An embedded runner starts afresh, and stop() drains it the same way.
	2285	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
	2286	stop();
	2287	assert!(stopping());
	2288	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
89171fca Andy McCall 22 days ago ↶	2289	}
	2290
	2291	/// Lines kept for a hub that stays away are bounded, saying how many went.
	2292	#[test]
	2293	fn unsent_log_lines_are_bounded() {
	2294	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2295	let hub = hub_at(format!("http://127.0.0.1:{port}"));
1b160d37 Andy McCall 10 days ago ↶	2296	let mut sink = LogSink {
	2297	hub: &hub,
	2298	job: 1,
	2299	buf: Vec::new(),
	2300	last: Instant::now(),
	2301	gone: false,
	2302	leaving: &NOT_LEAVING,
	2303	started: Instant::now(),
	2304	expected: None,
	2305	};
89171fca Andy McCall 22 days ago ↶	2306	sink.buf = (0..MAX_UNSENT as u64 + 10).map(\|i\| Record { t: i, stream: "out".into(), text: String::new() }).collect();
	2307	sink.send();
	2308	assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
	2309	assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
	2310	assert_eq!(sink.buf[1].t, 10, "the oldest go first");
1b2229da Andy McCall 1 month ago ↶	2311	}
bb643c25 Andy McCall 1 month ago ↶	2312
	2313	#[test]
	2314	fn the_budget_admits_jobs_that_fit() {
	2315	let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
	2316	let used = \|jobs, memory_mib, cpus\| Used { jobs, memory_mib, cpus };
	2317	assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
	2318	assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
	2319	assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
	2320	assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
	2321	let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
	2322	assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
	2323	assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
	2324	let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
	2325	assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
	2326	}
e0918b88 Andy McCall 1 month ago	2327	}

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

Keyboard