crates/flow/src/runner.rs at 6f66e397539c · andymccall/oxygenflow

# 6f66e397539c oxygenflow / crates / flow / src / runner.rs

102.1 KiB · Rust · last changed Windows runner: helpers start from the running binary; microVM says Linux only 11 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 21 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 25 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 24 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 24 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 11 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 20 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 20 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 20 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 11 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 24 days ago ↶	219	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
89171fca Andy McCall 24 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 24 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 11 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 21 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 26 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 24 days ago ↶	294	let r = retrying(&url, &PATIENT_DELAYS, true, \|\| {
89171fca Andy McCall 24 days ago ↶	295	self.agent
6d53fb7d Andy McCall 26 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 24 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 26 days ago ↶	306	}
6d53fb7d Andy McCall 26 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 11 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 24 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 21 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 24 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 11 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 12 days ago ↶	547	// wasmtime runs on every machine this binary does (oxygenflow#75).
28cc603f Andy McCall 12 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 25 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 25 days ago ↶	558	fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) {
4cf0c99a Andy McCall 25 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 25 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 25 days ago ↶	588	Ok(Some(m)) => note(format!(
b14aeaae Andy McCall 25 days ago ↶	589	"image {}: fetched {} chunks, {}{} in {:.1} s",
4cf0c99a Andy McCall 25 days ago ↶	590	&hash[..12],
	591	m.chunks,
	592	crate::images::human(m.bytes),
b14aeaae Andy McCall 25 days ago ↶	593	if swarm.is_some() { format!(" ({} from peers)", crate::images::human(m.peer_bytes)) } else { String::new() },
4cf0c99a Andy McCall 25 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 12 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 12 days ago ↶	662	drain_on_stop();
85275e1e Andy McCall 11 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 11 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 11 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);
	686	served
5cf1e8ae Andy McCall 12 days ago ↶	687	}
5cf1e8ae Andy McCall 12 days ago ↶	688
85275e1e Andy McCall 11 days ago ↶	689	/// Set by SIGUSR1: a new build is in place; leave for it.
	690	static UPDATE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
	691
3554afef Andy McCall 11 days ago ↶	692	#[cfg(unix)]
85275e1e Andy McCall 11 days ago ↶	693	extern "C" fn on_update(_: libc::c_int) {
	694	UPDATE.store(true, std::sync::atomic::Ordering::SeqCst);
	695	}
	696
5cf1e8ae Andy McCall 12 days ago ↶	697	/// The runner as a library, as the Android app runs it (oxygenhub-android#8):
	698	/// no signal handlers; [`stop`] drains it, and this returns once the jobs
	699	/// it holds are done and reported. It may be started again afterwards.
	700	pub fn serve_embedded(home: &Home) -> Result<()> {
	701	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
06a4c472 Andy McCall 11 days ago ↶	702	*LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()) = None;
5cf1e8ae Andy McCall 12 days ago ↶	703	serve_until_drained(home)
06a4c472 Andy McCall 11 days ago ↶	704	}
06a4c472 Andy McCall 11 days ago ↶	705
1b160d37 Andy McCall 11 days ago ↶	706	/// A departure ([`leave_within`]): why, and when the runner must be gone.
	707	#[derive(Clone, Debug)]
	708	struct Leave {
	709	reason: String,
	710	by: Instant,
	711	}
	712
	713	/// Set once the runner is leaving: a job whose expected rest fits before
	714	/// the deadline finishes; the others are stopped and handed back
	715	/// (oxygenflow#76).
	716	static LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall 11 days ago ↶	717
06a4c472 Andy McCall 11 days ago ↶	718	fn leaving() -> Option<String> {
1b160d37 Andy McCall 11 days ago ↶	719	LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()).as_ref().map(\|l\| l.reason.clone())
06a4c472 Andy McCall 11 days ago ↶	720	}
	721
	722	/// Ask a runner started with [`serve_embedded`] to leave now: claim nothing
	723	/// more, stop the jobs it holds and hand them back to the hub, which queues
	724	/// them ahead of new work for another runner (oxygenflow#76), then return.
	725	/// For a departure with no time to finish: a phone off the allowed network.
	726	pub fn leave(reason: &str) {
1b160d37 Andy McCall 11 days ago ↶	727	leave_within(reason, Duration::ZERO);
	728	}
	729
	730	/// [`leave`], with `notice`: a job the hub expects to finish within it
	731	/// (the claim's `expected_ms`, less the time it has run) finishes and is
	732	/// reported; the others, and any still running when the notice is up, are
	733	/// handed back. For a phone taken off charge, a spot reclaim.
	734	pub fn leave_within(reason: &str, notice: Duration) {
	735	*LEAVING.lock().unwrap_or_else(\|p\| p.into_inner()) = Some(Leave { reason: reason.to_string(), by: Instant::now() + notice });
06a4c472 Andy McCall 11 days ago ↶	736	stop();
5cf1e8ae Andy McCall 12 days ago ↶	737	}
	738
	739	/// Ask a runner started with [`serve_embedded`] to drain: claim nothing
	740	/// more, finish and report what it holds, then return (oxygenflow#76's
	741	/// drain, for a phone taken off charge or off the allowed network).
	742	pub fn stop() {
	743	STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
	744	}
	745
	746	fn serve_until_drained(home: &Home) -> Result<()> {
e0918b88 Andy McCall 1 month ago	747	let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
c5b702e5 Andy McCall 1 month ago ↶	748	// A runner offers only what it can run: a machine without KVM claims
c5b702e5 Andy McCall 1 month ago ↶	749	// nothing rather than taking MicroVM jobs and failing them (#36).
ea56c56e Andy McCall 1 month ago ↶	750	let (runtimes, missing) = tiers(home).map_err(\|why\| anyhow::anyhow!("flow runner: {why}"))?;
	751	for why in &missing {
	752	eprintln!("flow runner: not offering {why}");
	753	}
e0918b88 Andy McCall 1 month ago	754	let hub = Hub::new(&cfg)?;
e0918b88 Andy McCall 1 month ago	755	let name = cfg.name.clone().unwrap_or_else(\|\| "flow".into());
bb643c25 Andy McCall 1 month ago ↶	756	let budget = Budget::of(&cfg);
47c1d64d Andy McCall 1 month ago ↶	757	let key = home.runner_key(&name)?;
b14aeaae Andy McCall 25 days ago ↶	758	// The swarm (docs/jetstream.md §8.1): a runner that cannot serve its
	759	// peers still runs jobs, fetching images from the hub alone.
	760	let swarm = crate::swarm::Swarm::start(home, &cfg).unwrap_or_else(\|e\| {
	761	eprintln!("flow runner: not joining the swarm: {e:#}");
	762	None
	763	});
47c1d64d Andy McCall 1 month ago ↶	764	eprintln!(
c5b702e5 Andy McCall 1 month ago ↶	765	"flow runner {name}: polling {} (up to {} jobs at once, tiers {}), signing receipts as {}",
47c1d64d Andy McCall 1 month ago ↶	766	cfg.hub,
47c1d64d Andy McCall 1 month ago ↶	767	budget.max_jobs,
c5b702e5 Andy McCall 1 month ago ↶	768	runtimes.join(", "),
47c1d64d Andy McCall 1 month ago ↶	769	key.identity.hash().to_hex()
47c1d64d Andy McCall 1 month ago ↶	770	);
4be5e4cd Andy McCall 16 days ago ↶	771	if let Some(addr) = &cfg.metrics {
	772	crate::metrics::serve(addr, cfg.metrics_token.clone(), name.clone());
	773	}
bb643c25 Andy McCall 1 month ago ↶	774	let used = std::sync::Mutex::new(Used::default());
e0918b88 Andy McCall 1 month ago	775	let mut idle = Duration::from_secs(2);
3007d8f0 Andy McCall 1 month ago ↶	776	let pending = report_pending(home);
1daca0b6 Andy McCall 1 month ago ↶	777	// The jobs held now, told to the hub at each claim so it can end the
	778	// ones this runner lost: a job is held from its claim until its result
	779	// is reported.
	780	let held = std::sync::Mutex::new(std::collections::BTreeSet::<i64>::new());
	781	held.lock().unwrap().extend(pending.iter().map(\|(job, _)\| *job));
3007d8f0 Andy McCall 1 month ago ↶	782	std::thread::scope(\|scope\| {
3007d8f0 Andy McCall 1 month ago ↶	783	for (job, f) in pending {
1daca0b6 Andy McCall 1 month ago ↶	784	let (hub, held) = (&hub, &held);
3007d8f0 Andy McCall 1 month ago ↶	785	scope.spawn(move \|\| {
	786	eprintln!("job {job}: reporting the result kept from before the restart");
	787	report_finish(home, hub, job, &f);
1daca0b6 Andy McCall 1 month ago ↶	788	held.lock().unwrap().remove(&job);
3007d8f0 Andy McCall 1 month ago ↶	789	});
3007d8f0 Andy McCall 1 month ago ↶	790	}
62829d15 Andy McCall 25 days ago ↶	791	// The speed (spec §11.1): measured now, and again every 15 minutes
	792	// while no job runs, so a job never shares the CPU with it.
	793	let measure = \|\| {
	794	let (speed, ms) = crate::facts::bench();
	795	eprintln!("flow runner {name}: speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
	796	speed
	797	};
	798	let mut speed = measure();
	799	let mut benched = Instant::now();
89171fca Andy McCall 24 days ago ↶	800	let mut said_draining = false;
a4b3f8cb Andy McCall 20 days ago ↶	801	// When the hub last heard from us (oxygenhub#243): a claim, or a
	802	// heartbeat while there is no room to claim.
	803	let mut heard = Instant::now();
	804	let beat = \|heard: &mut Instant\| {
	805	if heard.elapsed() >= HUB_HEARTBEAT {
	806	if let Err(e) = hub.heartbeat() {
	807	eprintln!("flow runner {name}: heartbeat: {e:#}");
	808	}
	809	*heard = Instant::now();
	810	}
	811	};
85275e1e Andy McCall 11 days ago ↶	812	let mut on_trial = crate::update::pending(&home.root).is_some();
cedd1234 Andy McCall 11 days ago ↶	813	// Runner updates (oxygenhub#322): asked of the hub once it has
	814	// answered a claim, then every 15 minutes.
	815	let mut update_asked: Option<Instant> = None;
	816	let mut update_said = String::new();
3007d8f0 Andy McCall 1 month ago ↶	817	loop {
85275e1e Andy McCall 11 days ago ↶	818	if UPDATE.swap(false, std::sync::atomic::Ordering::SeqCst) {
	819	let p = crate::update::pending(&home.root);
	820	let notice = Duration::from_secs(p.as_ref().map_or(0, \|p\| p.notice_secs));
	821	let to = p.map_or_else(\|\| "a new build".to_string(), \|p\| format!("flow {}", p.to));
	822	eprintln!("flow runner {name}: updating to {to}: leaving within {} s", notice.as_secs());
	823	leave_within(&format!("updating to {to}"), notice);
	824	}
89171fca Andy McCall 24 days ago ↶	825	// Draining (oxygenflow#50): told to stop, claim nothing more and
	826	// stop once every job held is done and reported.
	827	if stopping() {
	828	let (jobs, holding) = (used.lock().unwrap().jobs, held.lock().unwrap().len());
	829	if jobs == 0 && holding == 0 {
	830	eprintln!("flow runner {name}: drained, stopping");
	831	break;
	832	}
	833	if !said_draining {
	834	eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)");
	835	said_draining = true;
	836	}
a4b3f8cb Andy McCall 20 days ago ↶	837	beat(&mut heard);
89171fca Andy McCall 24 days ago ↶	838	std::thread::sleep(Duration::from_millis(500));
	839	continue;
	840	}
62829d15 Andy McCall 25 days ago ↶	841	if used.lock().unwrap().jobs == 0 && benched.elapsed() >= Duration::from_secs(15 * 60) {
	842	speed = measure();
	843	benched = Instant::now();
	844	}
3007d8f0 Andy McCall 1 month ago ↶	845	let room = budget.room(*used.lock().unwrap());
3007d8f0 Andy McCall 1 month ago ↶	846	let Some((free_memory_mib, free_cpus)) = room else {
a4b3f8cb Andy McCall 20 days ago ↶	847	beat(&mut heard);
3007d8f0 Andy McCall 1 month ago ↶	848	std::thread::sleep(Duration::from_millis(500));
	849	continue;
	850	};
a4b3f8cb Andy McCall 20 days ago ↶	851	heard = Instant::now();
1e6b9fb1 Andy McCall 23 days ago ↶	852	let cache_times = crate::cache::Pool::new(&home.root, cfg.cache_gb).slot_times();
3007d8f0 Andy McCall 1 month ago ↶	853	let req = ClaimRequest {
3007d8f0 Andy McCall 1 month ago ↶	854	runner: name.clone(),
c5b702e5 Andy McCall 1 month ago ↶	855	runtimes: runtimes.clone(),
3007d8f0 Andy McCall 1 month ago ↶	856	free_memory_mib,
	857	free_cpus,
	858	info: Some(RunnerInfo {
	859	version: env!("OXY_VERSION_LONG").into(),
ea81796b Andy McCall 21 days ago ↶	860	..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
3007d8f0 Andy McCall 1 month ago ↶	861	}),
8c0a4e3a Andy McCall 1 month ago ↶	862	// Read each time: an image imported while the runner runs counts.
8c0a4e3a Andy McCall 1 month ago ↶	863	images: Some(home.images()),
1abadd4a Andy McCall 1 month ago ↶	864	image_archs: Some(home.image_archs()),
1daca0b6 Andy McCall 1 month ago ↶	865	holding: Some(held.lock().unwrap().iter().copied().collect()),
4cf0c99a Andy McCall 25 days ago ↶	866	fetches_images: true,
62829d15 Andy McCall 25 days ago ↶	867	facts: Some(crate::facts::now(home, Some(speed))),
b14aeaae Andy McCall 25 days ago ↶	868	swarm: swarm.as_ref().map(\|s\| s.addr.clone()),
1e6b9fb1 Andy McCall 23 days ago ↶	869	caches: Some(cache_times.keys().cloned().collect()),
1e6b9fb1 Andy McCall 23 days ago ↶	870	cache_times: Some(cache_times),
3007d8f0 Andy McCall 1 month ago ↶	871	};
85275e1e Andy McCall 11 days ago ↶	872	let answer = hub.claim(&req);
	873	if on_trial && answer.is_ok() {
	874	crate::update::proven(&home.root);
	875	on_trial = false;
cedd1234 Andy McCall 11 days ago ↶	876	}
	877	if answer.is_ok() && update_asked.is_none_or(\|t\| t.elapsed() >= UPDATE_EVERY) {
	878	update_asked = Some(Instant::now());
	879	if let Some(note) = crate::update::take_note(&home.root) {
	880	let _ = hub.report_update(&note);
	881	}
	882	consider_update(home, &hub, &mut update_said);
85275e1e Andy McCall 11 days ago ↶	883	}
85275e1e Andy McCall 11 days ago ↶	884	match answer {
3007d8f0 Andy McCall 1 month ago ↶	885	Ok(Some(claim)) => {
3007d8f0 Andy McCall 1 month ago ↶	886	idle = Duration::from_secs(2);
4be5e4cd Andy McCall 16 days ago ↶	887	crate::metrics::CLAIMS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3007d8f0 Andy McCall 1 month ago ↶	888	let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);
	889	{
	890	let mut u = used.lock().unwrap();
	891	u.jobs += 1;
	892	u.memory_mib += mem;
	893	u.cpus += cpus;
bb643c25 Andy McCall 1 month ago ↶	894	}
1daca0b6 Andy McCall 1 month ago ↶	895	held.lock().unwrap().insert(claim.job_id);
b14aeaae Andy McCall 25 days ago ↶	896	let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref());
3007d8f0 Andy McCall 1 month ago ↶	897	scope.spawn(move \|\| {
3007d8f0 Andy McCall 1 month ago ↶	898	let id = claim.job_id;
b14aeaae Andy McCall 25 days ago ↶	899	fetch_image(home, hub, cfg, swarm, &claim);
b11e30f5 Andy McCall 1 month ago ↶	900	if let Some(why) = preflight(home, &claim) {
b11e30f5 Andy McCall 1 month ago ↶	901	eprintln!("job {id}: declined: {why}");
4be5e4cd Andy McCall 16 days ago ↶	902	crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
b11e30f5 Andy McCall 1 month ago ↶	903	if let Err(e) = hub.decline(id, &why) {
	904	eprintln!("job {id}: decline: {e:#}");
	905	}
	906	held.lock().unwrap().remove(&id);
	907	let mut u = used.lock().unwrap();
	908	u.jobs -= 1;
	909	u.memory_mib -= mem;
	910	u.cpus -= cpus;
	911	return;
	912	}
3007d8f0 Andy McCall 1 month ago ↶	913	eprintln!(
	914	"job {id}: {} {} on {}",
	915	claim.repo,
	916	claim.job.name,
	917	claim.heads.first().map(\|h\| &h[..12]).unwrap_or("-")
	918	);
4be5e4cd Andy McCall 16 days ago ↶	919	crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3007d8f0 Andy McCall 1 month ago ↶	920	let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(\|e\| Finish {
	921	exit_code: None,
	922	duration_ms: 0,
	923	receipt: None,
	924	signed_receipt: None,
	925	error: Some(format!("{e:#}")),
8f211e20 Andy McCall 25 days ago ↶	926	peak_disk_mib: None,
8f211e20 Andy McCall 25 days ago ↶	927	peak_memory_mib: None,
3007d8f0 Andy McCall 1 month ago ↶	928	});
	929	eprintln!(
	930	"job {id}: exit {:?}{}",
	931	finish.exit_code,
	932	finish.error.as_deref().map(\|e\| format!(" ({e})")).unwrap_or_default()
	933	);
4be5e4cd Andy McCall 16 days ago ↶	934	crate::metrics::RUNNING.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
	935	let cancelled = finish.error.as_deref() == Some(CANCELLED);
	936	crate::metrics::finished(finish.exit_code, finish.error.as_deref(), finish.duration_ms, cancelled);
06a4c472 Andy McCall 11 days ago ↶	937	match leaving() {
	938	// Stopped as the runner leaves: another runner takes it.
	939	Some(why) if cancelled => {
	940	eprintln!("job {id}: handed back: {why}");
	941	if let Err(e) = hub.release(id, &why) {
	942	eprintln!("job {id}: handing back: {e:#}; the hub requeues it once this runner's lease lapses");
	943	}
	944	}
	945	_ if !cancelled => report_finish(home, hub, id, &finish),
	946	_ => {}
280ef951 Andy McCall 1 month ago ↶	947	}
1daca0b6 Andy McCall 1 month ago ↶	948	held.lock().unwrap().remove(&id);
3007d8f0 Andy McCall 1 month ago ↶	949	let mut u = used.lock().unwrap();
	950	u.jobs -= 1;
	951	u.memory_mib -= mem;
	952	u.cpus -= cpus;
	953	});
	954	}
	955	Ok(None) => std::thread::sleep(idle),
	956	Err(e) => {
	957	eprintln!("claim: {e:#}");
	958	std::thread::sleep(idle);
	959	idle = (idle * 2).min(Duration::from_secs(60));
	960	}
e0918b88 Andy McCall 1 month ago	961	}
e0918b88 Andy McCall 1 month ago	962	}
89171fca Andy McCall 24 days ago ↶	963	Ok(())
bb643c25 Andy McCall 1 month ago ↶	964	})
f201e1bc Andy McCall 11 days ago ↶	965	}
	966
	967	/// Pack and keep a handed-back job's checkpoints (oxygenflow#77); what
	968	/// happened goes to its log, and a failure only means the next attempt
	969	/// starts again.
	970	#[allow(clippy::too_many_arguments)]
	971	fn save_checkpoint(hub: &Hub, id: i64, job: &flow_core::Job, scratch: &Path, dir_tier: bool, wasm: bool, mut say: impl FnMut(String)) {
	972	if !dir_tier {
	973	say("checkpoint: only the WASM and syscall tiers save one yet; the next attempt starts again".into());
	974	return;
	975	}
	976	let dirs: Vec<(String, PathBuf)> = job
	977	.checkpoints
	978	.iter()
	979	.filter_map(\|c\| crate::checkpoint::host_dir(c, job, &scratch.join("tree"), scratch, wasm).map(\|d\| (c.clone(), d)))
	980	.collect();
	981	match crate::checkpoint::pack(&dirs).and_then(\|(bytes, n)\| Ok((hub.save_checkpoint(id, &bytes)?, n))) {
	982	Ok((blob, n)) => {
	983	say(format!("checkpoint {}: {} files, {} bytes, kept for the next attempt", &blob[..12.min(blob.len())], n.files, n.bytes))
	984	}
	985	Err(e) => say(format!("checkpoint: not kept ({e:#}); the next attempt starts again")),
	986	}
89171fca Andy McCall 24 days ago ↶	987	}
	988
	989	/// How many times the runner was told to stop (SIGTERM or SIGINT).
	990	static STOPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
	991
	992	extern "C" fn on_stop(_: libc::c_int) {
	993	// A second stop means now: only async-signal-safe calls here.
	994	if STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= 1 {
	995	unsafe { libc::_exit(143) };
	996	}
	997	}
	998
	999	/// Drain instead of dying on the first SIGTERM or SIGINT (oxygenflow#50),
	1000	/// so restarting a runner never kills the job it runs: the unit's
	1001	/// `KillMode=mixed` sends the signal to the runner alone, not its microVMs.
	1002	fn drain_on_stop() {
	1003	let handler = on_stop as extern "C" fn(libc::c_int) as libc::sighandler_t;
	1004	unsafe {
	1005	libc::signal(libc::SIGTERM, handler);
	1006	libc::signal(libc::SIGINT, handler);
	1007	}
e0918b88 Andy McCall 1 month ago	1008	}
e0918b88 Andy McCall 1 month ago	1009
89171fca Andy McCall 24 days ago ↶	1010	fn stopping() -> bool {
	1011	STOPS.load(std::sync::atomic::Ordering::SeqCst) > 0
	1012	}
	1013
280ef951 Andy McCall 1 month ago ↶	1014	/// Batches records to the hub every second (or every 200 lines). A quiet
	1015	/// job's sink still asks the hub every few seconds whether the job runs,
	1016	/// with an empty batch; once the hub answers 409 the job is not wanted.
e0918b88 Andy McCall 1 month ago	1017	struct LogSink<'a> {
	1018	hub: &'a Hub,
	1019	job: i64,
	1020	buf: Vec<Record>,
	1021	last: Instant,
280ef951 Andy McCall 1 month ago ↶	1022	gone: bool,
06a4c472 Andy McCall 11 days ago ↶	1023	/// Set when the runner leaves ([`LEAVING`]): the job stops, to be
1b160d37 Andy McCall 11 days ago ↶	1024	/// handed back, unless its expected rest fits the notice.
	1025	leaving: &'a std::sync::Mutex<Option<Leave>>,
	1026	/// When the job started, and how long the hub expects it to take.
	1027	started: Instant,
	1028	expected: Option<Duration>,
280ef951 Andy McCall 1 month ago ↶	1029	}
89171fca Andy McCall 24 days ago ↶	1030
	1031	/// The most log lines sent at once, as a busy job's sink flushes them.
	1032	const BATCH: usize = 200;
280ef951 Andy McCall 1 month ago ↶	1033
89171fca Andy McCall 24 days ago ↶	1034	/// The most log lines kept while the hub cannot be reached.
	1035	const MAX_UNSENT: usize = 50_000;
	1036
280ef951 Andy McCall 1 month ago ↶	1037	/// How often a quiet job asks whether it was cancelled.
	1038	const HEARTBEAT: Duration = Duration::from_secs(5);
	1039
	1040	impl microvm::Sink for LogSink<'_> {
	1041	fn record(&mut self, r: Record) {
	1042	self.push(r)
	1043	}
	1044
	1045	fn cancelled(&mut self) -> bool {
1b160d37 Andy McCall 11 days ago ↶	1046	let leave = self.leaving.lock().unwrap_or_else(\|p\| p.into_inner()).clone();
	1047	if let Some(l) = leave {
	1048	// Finished if what is left of it fits the notice; else handed
	1049	// back as the runner leaves, what the job said so far first.
	1050	let now = Instant::now();
	1051	let fits = self.expected.is_some_and(\|e\| now + e.saturating_sub(self.started.elapsed()) <= l.by);
	1052	if !fits \|\| now >= l.by {
	1053	self.flush();
	1054	return true;
	1055	}
06a4c472 Andy McCall 11 days ago ↶	1056	}
280ef951 Andy McCall 1 month ago ↶	1057	if !self.gone && self.last.elapsed() >= HEARTBEAT {
	1058	self.send();
	1059	}
	1060	self.gone
	1061	}
e0918b88 Andy McCall 1 month ago	1062	}
	1063
	1064	impl LogSink<'_> {
	1065	fn push(&mut self, mut r: Record) {
280ef951 Andy McCall 1 month ago ↶	1066	if self.gone {
	1067	return;
	1068	}
e0918b88 Andy McCall 1 month ago	1069	if r.text.len() > MAX_RECORD_TEXT {
	1070	let mut cut = MAX_RECORD_TEXT;
	1071	while !r.text.is_char_boundary(cut) {
	1072	cut -= 1;
	1073	}
	1074	r.text.truncate(cut);
	1075	}
	1076	self.buf.push(r);
89171fca Andy McCall 24 days ago ↶	1077	if self.buf.len() >= BATCH \|\| self.last.elapsed() >= Duration::from_secs(1) {
e0918b88 Andy McCall 1 month ago	1078	self.flush();
	1079	}
	1080	}
	1081
	1082	fn flush(&mut self) {
	1083	if !self.buf.is_empty() {
280ef951 Andy McCall 1 month ago ↶	1084	self.send();
	1085	}
	1086	self.last = Instant::now();
	1087	}
	1088
89171fca Andy McCall 24 days ago ↶	1089	/// Send what is buffered, even nothing, in batches of at most
89171fca Andy McCall 24 days ago ↶	1090	/// `BATCH` lines.
280ef951 Andy McCall 1 month ago ↶	1091	fn send(&mut self) {
89171fca Andy McCall 24 days ago ↶	1092	loop {
	1093	if self.gone {
	1094	return;
	1095	}
	1096	let n = self.buf.len().min(BATCH);
	1097	let batch: Vec<Record> = self.buf.drain(..n).collect();
	1098	match self.hub.log(self.job, batch.clone()) {
	1099	Ok(()) => {}
	1100	Err(e) if job_gone(&e) => self.gone = true,
	1101	Err(e) => {
	1102	// The hub is away (a restart, say): keep the lines for
	1103	// the next batch, up to a bound, so the log loses none
	1104	// (oxygenflow#49).
	1105	eprintln!("job {}: log: {e:#}; kept for the next batch", self.job);
	1106	let mut kept = batch;
	1107	kept.append(&mut self.buf);
	1108	let over = kept.len().saturating_sub(MAX_UNSENT);
	1109	if over > 0 {
	1110	kept.drain(..over);
	1111	kept.insert(
	1112	0,
	1113	Record {
	1114	t: 0,
	1115	stream: "flow".into(),
	1116	text: format!("({over} log lines lost while the hub could not be reached)"),
	1117	},
	1118	);
	1119	}
	1120	self.buf = kept;
	1121	break;
	1122	}
	1123	}
	1124	if self.buf.is_empty() {
	1125	break;
	1126	}
e0918b88 Andy McCall 1 month ago	1127	}
	1128	self.last = Instant::now();
	1129	}
	1130	}
a4b3f8cb Andy McCall 20 days ago ↶	1131
	1132	/// How often a runner with no room to claim tells the hub it is there:
	1133	/// well inside the hub's 90 s before it shows a runner offline.
	1134	const HUB_HEARTBEAT: Duration = Duration::from_secs(30);
cedd1234 Andy McCall 11 days ago ↶	1135
	1136	/// How often a runner asks the hub whether it should run another build.
	1137	const UPDATE_EVERY: Duration = Duration::from_secs(15 * 60);
	1138
	1139	/// Notice a runner gives itself to leave for an update the hub offered.
	1140	const UPDATE_NOTICE: Duration = Duration::from_secs(600);
bb643c25 Andy McCall 1 month ago ↶	1141
cedd1234 Andy McCall 11 days ago ↶	1142	/// Ask the hub for this runner's update, and act on it: download the build
	1143	/// it offers, check it against the hub's BLAKE3 and as a runner build, and
	1144	/// install it, or for a dry run say what would happen. What it did goes to
	1145	/// the hub once per outcome (`said`).
	1146	fn consider_update(home: &Home, hub: &Hub, said: &mut String) {
	1147	let answer = match hub.update_answer() {
	1148	Ok(Some(a)) => a,
	1149	Ok(None) => return,
	1150	Err(e) => {
	1151	eprintln!("flow runner: asking for an update: {e:#}");
	1152	return;
	1153	}
	1154	};
	1155	let Some(offer) = answer.offer else { return };
	1156	let mut say = \|note: String\| {
	1157	if *said != note {
	1158	eprintln!("flow runner: {note}");
	1159	if let Err(e) = hub.report_update(&note) {
	1160	eprintln!("flow runner: telling the hub: {e:#}");
	1161	}
	1162	*said = note;
	1163	}
	1164	};
	1165	if home.config.runner.as_ref().and_then(\|r\| r.update.as_ref()).is_some_and(\|u\| !u.allow) {
	1166	return say(format!("refuses flow {}: flow.toml says [runner.update] allow = false", offer.version));
	1167	}
	1168	let dir = home.root.join("updates");
	1169	let file = dir.join(format!("flow-{}", offer.version));
	1170	let fresh = std::fs::read(&file).is_ok_and(\|b\| blake3::hash(&b).to_hex().as_str() == offer.blake3);
	1171	if !fresh {
	1172	if let Err(e) = std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(\|_\| hub.download_build(&offer, &file)) {
	1173	return say(format!("flow {}: not fetched: {e:#}", offer.version));
	1174	}
	1175	}
	1176	let exe = match std::env::current_exe() {
	1177	Ok(e) => e,
	1178	Err(e) => return say(format!("flow {}: cannot find its own binary: {e}", offer.version)),
	1179	};
	1180	match crate::update::install(home, &exe, &file, UPDATE_NOTICE, answer.policy.dry_run, offer.older) {
	1181	Ok(done) if answer.policy.dry_run => say(format!("dry run: flow {} checks out here and would replace flow {}", done.to, done.from)),
	1182	Ok(done) => {
	1183	let _ = std::fs::remove_file(&file);
	1184	say(format!("installed flow {} (was {}); leaving within {} s to start it", done.to, done.from, UPDATE_NOTICE.as_secs()))
	1185	}
	1186	Err(e) => say(format!("flow {}: not installed: {e:#}", offer.version)),
	1187	}
	1188	}
	1189
ed4aef8b Andy McCall 20 days ago ↶	1190	/// Held while a job takes its cache disks and images its workspace (see
ed4aef8b Andy McCall 20 days ago ↶	1191	/// `run_claim`): one job at a time on the runner's disk and cache pool.
bb643c25 Andy McCall 1 month ago ↶	1192	static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());
e0918b88 Andy McCall 1 month ago	1193
ed4aef8b Andy McCall 20 days ago ↶	1194	/// Held while the first clone of any repository passes its token through
	1195	/// the process environment.
	1196	static CLONE: std::sync::Mutex<()> = std::sync::Mutex::new(());
	1197
	1198	/// One lock per cached repository: fetches into it, and exports from it,
	1199	/// never overlap, while jobs of other repositories go on. A slow repository
	1200	/// (one kept in a bucket, say) once held every job on the runner waiting
	1201	/// behind a single global lock.
	1202	fn repo_lock(name: &str) -> std::sync::Arc<std::sync::Mutex<()>> {
	1203	static LOCKS: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<std::sync::Mutex<()>>>>> =
	1204	std::sync::OnceLock::new();
	1205	let mut locks = LOCKS.get_or_init(Default::default).lock().unwrap_or_else(\|p\| p.into_inner());
	1206	locks.entry(name.to_string()).or_default().clone()
	1207	}
	1208
	1209	/// Seconds, to one decimal place, for the job's log.
	1210	fn secs(d: std::time::Duration) -> String {
	1211	format!("{:.1} s", d.as_secs_f64())
	1212	}
	1213
47c1d64d Andy McCall 1 month ago ↶	1214	fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
e0918b88 Andy McCall 1 month ago	1215	let started = Instant::now();
1b160d37 Andy McCall 11 days ago ↶	1216	let mut sink = LogSink {
	1217	hub,
	1218	job: claim.job_id,
	1219	buf: Vec::new(),
	1220	last: Instant::now(),
	1221	gone: false,
	1222	leaving: &LEAVING,
	1223	started: Instant::now(),
	1224	expected: claim.expected_ms.map(Duration::from_millis),
	1225	};
e0918b88 Andy McCall 1 month ago	1226	let note =
	1227	\|sink: &mut LogSink, text: String\| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
	1228	let heads: Vec<ChangeHash> = claim.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
	1229	note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(\|h\| &h[..12]).collect::<Vec<_>>().join("+")));
ed4aef8b Andy McCall 20 days ago ↶	1230	// Preparing a workspace: fetches into a cached repository, and exports
	1231	// from it, are one job at a time per repository (`repo_lock`); taking
	1232	// cache disks and imaging the workspace are one job at a time on the
	1233	// runner (`PREPARE`); a first clone, whose token goes through the process
	1234	// environment, one at a time (`CLONE`).
	1235	let main_lock = repo_lock(&claim.repo);
	1236	let asked = Instant::now();
	1237	let mut main_guard = Some(main_lock.lock().unwrap_or_else(\|p\| p.into_inner()));
	1238	if asked.elapsed().as_secs() >= 2 {
	1239	note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), claim.repo));
	1240	}
	1241	let fetching = Instant::now();
e0918b88 Andy McCall 1 month ago	1242	let repo = cached_repo(home, cfg, claim)?;
ed4aef8b Andy McCall 20 days ago ↶	1243	let fetched = fetching.elapsed();
ed4aef8b Andy McCall 20 days ago ↶	1244	let mut preparing = None;
e0918b88 Andy McCall 1 month ago	1245	let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
	1246	let _ = std::fs::remove_dir_all(&scratch);
	1247	std::fs::create_dir_all(scratch.join("tree"))?;
	1248	let result = (\|\| -> Result<Finish> {
ed4aef8b Andy McCall 20 days ago ↶	1249	let exporting = Instant::now();
c23d5191 Andy McCall 1 month ago ↶	1250	let mut files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
ed4aef8b Andy McCall 20 days ago ↶	1251	main_guard.take();
ed4aef8b Andy McCall 20 days ago ↶	1252	note(&mut sink, format!("fetched in {}, {} files exported in {}", secs(fetched), files.len(), secs(exporting.elapsed())));
c23d5191 Andy McCall 1 month ago ↶	1253	// Pinned subrepos (Oxygen spec §7.8), at their paths, as part of the
	1254	// workspace and of the scope hash.
	1255	for sub in &claim.subrepos {
	1256	if sub.path.split('/').any(\|p\| p.is_empty() \|\| p == "." \|\| p == ".." \|\| p == ".oxy") {
	1257	bail!("subrepo path {:?} is not a clean relative path", sub.path);
	1258	}
	1259	note(
	1260	&mut sink,
	1261	format!(
	1262	"fetching subrepo {} at {} for {}",
	1263	sub.repo,
	1264	sub.heads.iter().map(\|h\| &h[..12]).collect::<Vec<_>>().join("+"),
	1265	sub.path
	1266	),
	1267	);
ed4aef8b Andy McCall 20 days ago ↶	1268	let sub_lock = repo_lock(&sub.repo);
	1269	let asked = Instant::now();
	1270	let sub_guard = sub_lock.lock().unwrap_or_else(\|p\| p.into_inner());
	1271	if asked.elapsed().as_secs() >= 2 {
	1272	note(&mut sink, format!("waited {} for another job fetching {}", secs(asked.elapsed()), sub.repo));
	1273	}
c23d5191 Andy McCall 1 month ago ↶	1274	let sub_repo = cached_repo_at(home, cfg, &sub.repo, &sub.clone_url, &claim.token)?;
	1275	let sub_heads: Vec<ChangeHash> =
	1276	sub.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad subrepo head")).collect::<Result<_>>()?;
	1277	let dir = scratch.join("tree").join(&sub.path);
	1278	std::fs::create_dir_all(&dir)?;
	1279	let sub_files = workspace::export_state(&sub_repo, &sub_heads, &dir)?;
ed4aef8b Andy McCall 20 days ago ↶	1280	drop(sub_guard);
c23d5191 Andy McCall 1 month ago ↶	1281	files.extend(sub_files.into_iter().map(\|(p, h)\| (format!("{}/{p}", sub.path), h)));
c23d5191 Andy McCall 1 month ago ↶	1282	}
e0918b88 Andy McCall 1 month ago	1283	let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(\|(p, h)\| (p.as_str(), *h)));
46fe2c81 Andy McCall 21 days ago ↶	1284	// Inputs (oxygenflow#58): each producing job's artifact, checked
	1285	// against the blob its receipt names, then put at its path. Not part
	1286	// of the scope: they are the run's, not the repository's.
	1287	let mut input_blobs = Vec::new();
	1288	for (index, i) in claim.inputs.iter().enumerate() {
	1289	let what = format!("input {} from {}", i.artifact, i.from);
	1290	let bytes = hub.input(claim.job_id, index).with_context(\|\| format!("{what}: fetching it"))?;
	1291	let blob = place_input(&scratch.join("tree"), i, &bytes).with_context(\|\| what.clone())?;
	1292	note(&mut sink, format!("{what}: {} bytes at {}, blob {} as its receipt names", bytes.len(), i.path, &i.blob[..12]));
	1293	input_blobs.push(blob);
	1294	}
ea56c56e Andy McCall 1 month ago ↶	1295	// The syscall tier runs on the tree itself; a microVM gets an image.
ea56c56e Andy McCall 1 month ago ↶	1296	let syscall_tier = claim.sandbox.runtime == flow_core::Runtime::Syscall;
28cc603f Andy McCall 12 days ago ↶	1297	let wasm_tier = claim.sandbox.runtime == flow_core::Runtime::Wasm;
	1298	// The syscall and WASM tiers see the job's files as a directory;
	1299	// the MicroVM tier as an ext4 image.
	1300	let dir_tier = syscall_tier \|\| wasm_tier;
82cf6e91 Andy McCall 24 days ago ↶	1301	let job = claim.job.in_sandbox(&claim.sandbox).with_flow_env(&claim.flow_env());
2594278f Andy McCall 18 days ago ↶	1302	// Disk mounts that name a cache start from a kept disk (#28): a
	1303	// microVM takes an image already in place as it is, the syscall tier
	1304	// a directory (#70). Taken before the workspace is imaged: the
	1305	// sources the disk was built from give unchanged files their old
	1306	// times back (#52).
ed4aef8b Andy McCall 20 days ago ↶	1307	let asked = Instant::now();
	1308	preparing = Some(PREPARE.lock().unwrap_or_else(\|p\| p.into_inner()));
	1309	if asked.elapsed().as_secs() >= 2 {
	1310	note(&mut sink, format!("waited {} for another job imaging its workspace", secs(asked.elapsed())));
	1311	}
82cf6e91 Andy McCall 24 days ago ↶	1312	let pool = crate::cache::Pool::new(&home.root, cfg.cache_gb);
28cc603f Andy McCall 12 days ago ↶	1313	let form = if dir_tier { crate::cache::Form::Dir } else { crate::cache::Form::Image };
82cf6e91 Andy McCall 24 days ago ↶	1314	let cache_ctx =
2594278f Andy McCall 18 days ago ↶	1315	crate::cache::Ctx { repo: &claim.repo, bookmark: &claim.bookmark, job: &claim.job.name, image: &claim.sandbox.rootfs, form };
82cf6e91 Andy McCall 24 days ago ↶	1316	let mut cached: Vec<(usize, String, String)> = Vec::new();
82cf6e91 Andy McCall 24 days ago ↶	1317	let mut built_from: Option<crate::cache::Sources> = None;
2594278f Andy McCall 18 days ago ↶	1318	for (i, m) in job.mounts.iter().filter(\|m\| m.source == flow_core::pipeline::MountSource::Disk).enumerate() {
	1319	let Some(name) = &m.cache else { continue };
	1320	let key = crate::cache::Pool::key(&cache_ctx, m, \|p\| files.iter().find(\|(f, _)\| f == p).map(\|(_, h)\| hex::encode(h)));
	1321	let how = match pool.take(&cache_ctx, m, &key, &crate::cache::disk_path(form, &scratch, i)) {
	1322	Some(t) => {
	1323	if built_from.is_none() {
	1324	built_from = t.sources;
82cf6e91 Andy McCall 24 days ago ↶	1325	}
2594278f Andy McCall 18 days ago ↶	1326	t.how
	1327	}
	1328	None => format!("cache {name}: none kept yet, an empty disk"),
	1329	};
	1330	note(&mut sink, format!("{} ({})", how, m.target));
077a6524 Andy McCall 17 days ago ↶	1331	// What the job reads is seen when the disk is pruned (#72).
	1332	if let Err(e) = crate::cache::forget_reads(form, &crate::cache::disk_path(form, &scratch, i)) {
	1333	note(&mut sink, format!("cache {name}: {e:#}"));
	1334	}
2594278f Andy McCall 18 days ago ↶	1335	cached.push((i, key, how));
82cf6e91 Andy McCall 24 days ago ↶	1336	}
077a6524 Andy McCall 17 days ago ↶	1337	// What the job reads or writes on its cache from now on is what it
	1338	// used; the rest may be pruned before the cache is kept (#72).
	1339	let cache_started = if cached.is_empty() { std::time::SystemTime::now() } else { flow_init::prune::start_clock() };
82cf6e91 Andy McCall 24 days ago ↶	1340	let sources = if cached.is_empty() {
	1341	crate::cache::Sources::new()
	1342	} else {
	1343	let (sources, restored) = workspace::restore_times(&scratch.join("tree"), &files, built_from.as_ref());
	1344	if built_from.is_some() {
	1345	note(&mut sink, format!("cache: {restored} of {} files unchanged since the cached build, their times kept", sources.len()));
	1346	}
	1347	sources
	1348	};
f201e1bc Andy McCall 11 days ago ↶	1349	// An earlier attempt's checkpoint (oxygenflow#77): back in place,
	1350	// over the cache, before the same exec runs again; the receipt
	1351	// lists it among the job's inputs.
	1352	if let Some(blob) = &claim.checkpoint {
	1353	if dir_tier {
	1354	let bytes = hub.checkpoint(claim.job_id).context("fetching the checkpoint")?;
	1355	if Repo::blob_ref(&bytes).to_hex() != *blob {
	1356	bail!("the checkpoint fetched is not blob {}, as the claim names", &blob[..12.min(blob.len())]);
	1357	}
	1358	let place = \|root: &str\| crate::checkpoint::host_dir(root, &job, &scratch.join("tree"), &scratch, wasm_tier);
	1359	let n = crate::checkpoint::unpack(&bytes, &job.checkpoints, place)?;
	1360	note(
	1361	&mut sink,
	1362	format!("restored checkpoint {}: {} files, {} bytes, from the attempt handed back", &blob[..12], n.files, n.bytes),
	1363	);
	1364	input_blobs.push(Repo::blob_ref(&bytes).0);
	1365	} else {
	1366	note(&mut sink, "a checkpoint was saved, but only the WASM and syscall tiers restore one yet: starting again".into());
	1367	}
	1368	}
28cc603f Andy McCall 12 days ago ↶	1369	if !dir_tier {
ea56c56e Andy McCall 1 month ago ↶	1370	workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
	1371	std::fs::remove_dir_all(scratch.join("tree"))?;
	1372	}
bb643c25 Andy McCall 1 month ago ↶	1373	preparing.take();
ea56c56e Andy McCall 1 month ago ↶	1374	// Cancelled while it was being prepared: no sandbox.
280ef951 Andy McCall 1 month ago ↶	1375	if microvm::Sink::cancelled(&mut sink) {
	1376	return Ok(cancelled());
	1377	}
28cc603f Andy McCall 12 days ago ↶	1378	if wasm_tier {
	1379	note(&mut sink, format!("starting WASM (wasmtime {}): {} MiB", crate::wasm::WASMTIME, claim.sandbox.memory_mib));
	1380	} else {
	1381	note(
	1382	&mut sink,
	1383	format!(
	1384	"{}: {} {}, {} MiB, rootfs {}",
	1385	if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
	1386	claim.sandbox.cpus,
	1387	if syscall_tier { "CPU" } else { "vCPU" },
	1388	claim.sandbox.memory_mib,
	1389	&claim.sandbox.rootfs[..12]
	1390	),
	1391	);
	1392	}
ea56c56e Andy McCall 1 month ago ↶	1393	let timeout = Duration::from_secs(claim.timeout_s);
28cc603f Andy McCall 12 days ago ↶	1394	let mut outcome = if wasm_tier {
	1395	crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?
	1396	} else if syscall_tier {
ea56c56e Andy McCall 1 month ago ↶	1397	let group = format!("job-{}", claim.job_id);
	1398	crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
	1399	} else {
	1400	microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
	1401	};
280ef951 Andy McCall 1 month ago ↶	1402	if outcome.cancelled {
f201e1bc Andy McCall 11 days ago ↶	1403	// Handed back as the runner leaves: its checkpoints first, for
	1404	// the next attempt (oxygenflow#77).
	1405	if leaving().is_some() && !job.checkpoints.is_empty() {
	1406	save_checkpoint(hub, claim.job_id, &job, &scratch, dir_tier, wasm_tier, \|t\| note(&mut sink, t));
	1407	}
280ef951 Andy McCall 1 month ago ↶	1408	return Ok(cancelled());
d60aac6c Andy McCall 24 days ago ↶	1409	}
	1410	// The receipt says what a cached disk started from.
	1411	let disk_mounts: Vec<&flow_core::pipeline::Mount> =
	1412	job.mounts.iter().filter(\|m\| m.source == flow_core::pipeline::MountSource::Disk).collect();
	1413	for (i, key, how) in &cached {
	1414	outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
280ef951 Andy McCall 1 month ago ↶	1415	}
a67b97e9 Andy McCall 1 month ago ↶	1416	// A passing job's artifacts, read from its disks and uploaded; a
9a6fac2b Andy McCall 1 month ago ↶	1417	// declared artifact that cannot be kept fails the job. A failing
	1418	// job keeps only its `when "always"` ones (oxygenflow#37), outside
	1419	// its receipt, and one it cannot keep is only noted.
a67b97e9 Andy McCall 1 month ago ↶	1420	let mut outputs = vec![];
a67b97e9 Andy McCall 1 month ago ↶	1421	let mut output_error = None;
9a6fac2b Andy McCall 1 month ago ↶	1422	let passed = outcome.exit_code == Some(0);
	1423	let wanted = claim.job.outputs.iter().filter(\|o\| passed \|\| o.always);
	1424	for o in wanted {
6d53fb7d Andy McCall 26 days ago ↶	1425	if o.kind == flow_core::OutputKind::Report {
28cc603f Andy McCall 12 days ago ↶	1426	let read = if dir_tier {
6d53fb7d Andy McCall 26 days ago ↶	1427	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1428	crate::outputs::read_report_dir(&dir, &inner)
	1429	} else {
	1430	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1431	crate::outputs::read_report(&image, &inner, &scratch)
	1432	};
	1433	let kept = read.and_then(\|files\| upload_report(hub, claim.job_id, &o.name, &files));
	1434	match kept {
	1435	Ok((blob, files, bytes)) => {
	1436	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1437	let hex = blob.to_hex();
	1438	note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
	1439	if passed {
	1440	outputs.push(blob.0);
	1441	}
	1442	}
	1443	Err(e) if passed => {
	1444	output_error = Some(format!("report {}: {e:#}", o.name));
	1445	break;
	1446	}
	1447	Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
	1448	}
	1449	continue;
	1450	}
28cc603f Andy McCall 12 days ago ↶	1451	let read = if dir_tier {
9a6fac2b Andy McCall 1 month ago ↶	1452	let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
	1453	crate::outputs::read_dir_file(&dir, &inner)
	1454	} else {
	1455	let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
	1456	crate::outputs::read(&image, &inner, &scratch)
	1457	};
	1458	let kept = read.and_then(\|bytes\| {
	1459	let blob = Repo::blob_ref(&bytes);
	1460	let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
	1461	if named != blob.to_hex() {
	1462	bail!("the hub kept it as {named}, not {}", blob.to_hex());
	1463	}
	1464	let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
	1465	note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
	1466	Ok(blob.0)
	1467	});
	1468	match kept {
	1469	Ok(b) if passed => outputs.push(b),
	1470	Ok(_) => {}
	1471	Err(e) if passed => {
	1472	output_error = Some(format!("artifact {}: {e:#}", o.name));
	1473	break;
a67b97e9 Andy McCall 1 month ago ↶	1474	}
9a6fac2b Andy McCall 1 month ago ↶	1475	Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
a67b97e9 Andy McCall 1 month ago ↶	1476	}
	1477	}
	1478	if let Some(e) = &output_error {
	1479	note(&mut sink, e.clone());
	1480	}
	1481	let receipt = outcome.exit_code.filter(\|_\| output_error.is_none()).map(\|code\| Receipt {
e0918b88 Andy McCall 1 month ago	1482	job_name: claim.job.name.clone(),
	1483	pipeline_hash: hex32(&claim.pipeline_hash),
	1484	scope_tree_hash: scope_hash,
46fe2c81 Andy McCall 21 days ago ↶	1485	inputs: input_blobs.clone(),
e0918b88 Andy McCall 1 month ago	1486	environment: outcome.environment.hash(),
a67b97e9 Andy McCall 1 month ago ↶	1487	outputs,
e0918b88 Andy McCall 1 month ago	1488	exit_code: code as u32,
	1489	duration_ms: outcome.duration_ms,
	1490	});
47c1d64d Andy McCall 1 month ago ↶	1491	let signed_receipt = match &receipt {
35f7a849 Andy McCall 20 days ago ↶	1492	Some(r) => {
	1493	let mut signed = sign_receipt(&repo, &heads, cfg, key, claim, r)?;
	1494	if cfg.keyless {
	1495	match sign_keyless(&repo, &heads, cfg, hub, claim, r) {
	1496	Ok(k) => signed = k,
	1497	Err(e) => {
	1498	note(&mut sink, format!("receipt: signed with the runner's key; a short-lived key was not certified: {e:#}"))
	1499	}
	1500	}
	1501	}
	1502	Some(hex::encode(signed.encode()))
	1503	}
47c1d64d Andy McCall 1 month ago ↶	1504	None => None,
47c1d64d Andy McCall 1 month ago ↶	1505	};
d60aac6c Andy McCall 24 days ago ↶	1506	// Measured before a cached disk leaves the scratch directory.
d60aac6c Andy McCall 24 days ago ↶	1507	let peak_disk_mib = crate::facts::allocated_mib(&scratch);
077a6524 Andy McCall 17 days ago ↶	1508	// A passing job's cached disks are pruned and kept (#72); a
	1509	// failing job's are thrown away, so nothing it wrote reaches a
	1510	// later job.
2594278f Andy McCall 18 days ago ↶	1511	let kept = passed && output_error.is_none();
	1512	for (i, key, _) in &cached {
	1513	let disk = crate::cache::disk_path(form, &scratch, *i);
077a6524 Andy McCall 17 days ago ↶	1514	let m = disk_mounts[*i];
	1515	let prune = crate::cache::Prune {
	1516	started: cache_started,
	1517	percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
	1518	guest: outcome.pruned.iter().find(\|(t, _)\| *t == m.target).map(\|(_, p)\| p),
	1519	};
	1520	match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
	1521	Ok(Some(line)) => note(&mut sink, line),
	1522	Ok(None) => {}
	1523	Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
d60aac6c Andy McCall 24 days ago ↶	1524	}
d60aac6c Andy McCall 24 days ago ↶	1525	}
e0918b88 Andy McCall 1 month ago	1526	Ok(Finish {
	1527	exit_code: outcome.exit_code,
	1528	duration_ms: started.elapsed().as_millis() as u64,
	1529	receipt,
47c1d64d Andy McCall 1 month ago ↶	1530	signed_receipt,
a67b97e9 Andy McCall 1 month ago ↶	1531	error: output_error.or_else(\|\| outcome.timed_out.then(\|\| "stopped at the time limit".to_string())),
8f211e20 Andy McCall 25 days ago ↶	1532	// What the job took (oxygenflow#41): its disks are sparse files
8f211e20 Andy McCall 25 days ago ↶	1533	// in the scratch directory, so their blocks now are their peak.
d60aac6c Andy McCall 24 days ago ↶	1534	peak_disk_mib,
8f211e20 Andy McCall 25 days ago ↶	1535	peak_memory_mib: outcome.peak_memory_mib,
e0918b88 Andy McCall 1 month ago	1536	})
e0918b88 Andy McCall 1 month ago	1537	})();
ea56c56e Andy McCall 1 month ago ↶	1538	let _ = workspace::remove_tree(&scratch);
e0918b88 Andy McCall 1 month ago	1539	sink.flush();
e0918b88 Andy McCall 1 month ago	1540	result
47c1d64d Andy McCall 1 month ago ↶	1541	}
	1542
	1543	/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
	1544	/// runner (Oxygen spec §14.6). The state is computed here, not taken from
	1545	/// the claim, so the runner vouches only for what it fetched.
6d53fb7d Andy McCall 26 days ago ↶	1546	/// Upload a report's files, then its manifest under the report's name, and
	1547	/// check the hub kept each as the blob it should (oxygenhub#127). Answers
	1548	/// the manifest's blob, the file count and their total size.
	1549	fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
	1550	let mut listed = Vec::with_capacity(files.len());
	1551	let mut total = 0u64;
	1552	for (path, bytes) in files {
	1553	let blob = Repo::blob_ref(bytes).to_hex();
	1554	let named = hub.report_file(job, name, path, bytes).with_context(\|\| format!("uploading {path}"))?;
	1555	if named != blob {
	1556	bail!("the hub kept {path} as {named}, not {blob}");
	1557	}
	1558	total += bytes.len() as u64;
	1559	listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
	1560	}
	1561	let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
	1562	let blob = Repo::blob_ref(&manifest);
	1563	let named = hub.artifact(job, name, &manifest)?;
	1564	if named != blob.to_hex() {
	1565	bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
	1566	}
	1567	Ok((blob, files.len(), total))
	1568	}
	1569
47c1d64d Andy McCall 1 month ago ↶	1570	fn sign_receipt(
	1571	repo: &Repo,
	1572	heads: &[ChangeHash],
	1573	cfg: &RunnerConfig,
	1574	key: &RunnerKey,
	1575	claim: &Claim,
	1576	receipt: &Receipt,
	1577	) -> Result<oxy_core::receipt::SignedReceipt> {
	1578	let state = state_of(heads);
	1579	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1580	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
	1581	}
	1582
35f7a849 Andy McCall 20 days ago ↶	1583	/// Sign `receipt` with a fresh key the hub certifies for this job now,
	1584	/// just before signing, so the certificate covers the moment the hub
	1585	/// accepts it (keyless, oxygenhub#241). The key is dropped with the
	1586	/// receipt signed: nothing long-lived signs.
	1587	fn sign_keyless(
	1588	repo: &Repo,
	1589	heads: &[ChangeHash],
	1590	cfg: &RunnerConfig,
	1591	hub: &Hub,
	1592	claim: &Claim,
	1593	receipt: &Receipt,
	1594	) -> Result<oxy_core::receipt::SignedReceipt> {
	1595	let sk = oxy_core::identity::SecretKey::generate();
	1596	let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(\|d\| d.as_millis() as u64).unwrap_or(0);
	1597	let name = format!("{} job {}", cfg.name.as_deref().unwrap_or("runner"), claim.job_id);
	1598	let id = sk.identity(&name, "", now);
	1599	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge(&claim.token)).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1600	let (cert, issuer) = hub.certify(&claim.token, &id, &proof)?;
	1601	let state = state_of(heads);
	1602	let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
	1603	Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &id, &sk).with_certificate(cert, issuer))
	1604	}
	1605
47c1d64d Andy McCall 1 month ago ↶	1606	/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
	1607	pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
	1608	use oxy_core::hash::{hash_object, ObjectType};
	1609	hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
e0918b88 Andy McCall 1 month ago	1610	}
	1611
	1612	fn hex32(s: &str) -> [u8; 32] {
	1613	let mut out = [0; 32];
	1614	if let Ok(v) = hex::decode(s) {
	1615	if v.len() == 32 {
	1616	out.copy_from_slice(&v);
	1617	}
	1618	}
	1619	out
	1620	}
	1621
f53e6d47 Andy McCall 1 month ago ↶	1622	/// Whether a network error is worth another try: DNS, a refused or reset
	1623	/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
	1624	/// hub restarts (oxygenhub#111).
	1625	fn transient(msg: &str) -> bool {
	1626	let m = msg.to_lowercase();
	1627	[
	1628	"dns failed",
	1629	"name resolution",
	1630	"connection refused",
	1631	"connection reset",
	1632	"timed out",
	1633	"timeout",
	1634	"http 502",
	1635	"http 503",
	1636	"http 504",
	1637	"broken pipe",
	1638	"unexpected eof",
	1639	]
	1640	.iter()
	1641	.any(\|k\| m.contains(k))
	1642	}
	1643
	1644	/// Run `f` again after a transient network error, waiting `waits` between
	1645	/// tries (about a minute in all), and give up on any other error at once.
	1646	fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
	1647	let mut tries = waits.iter();
	1648	loop {
	1649	match f() {
	1650	Ok(v) => return Ok(v),
	1651	Err(e) => match tries.next() {
	1652	Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
	1653	_ => return Err(e),
	1654	},
	1655	}
	1656	}
	1657	}
	1658
	1659	/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
	1660	const FETCH_WAITS: [Duration; 5] =
	1661	[Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
	1662
e0918b88 Andy McCall 1 month ago	1663	/// The runner's copy of a repository, fetched up to date with the job's token.
e0918b88 Andy McCall 1 month ago	1664	fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
c23d5191 Andy McCall 1 month ago ↶	1665	cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
	1666	}
	1667
	1668	/// The runner's copy of `repo`, cloned or fetched with `token`.
	1669	fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
	1670	let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
e0918b88 Andy McCall 1 month ago	1671	let set_remote = \|repo: &mut Repo\| -> Result<()> {
c23d5191 Andy McCall 1 month ago ↶	1672	repo.config.set("remote.origin.url", clone_url).map_err(\|e\| anyhow::anyhow!("{e}"))?;
c23d5191 Andy McCall 1 month ago ↶	1673	repo.config.set("remote.origin.token", token).map_err(\|e\| anyhow::anyhow!("{e}"))?;
e0918b88 Andy McCall 1 month ago	1674	if let Some(ca) = &cfg.ca {
	1675	repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1676	}
	1677	repo.save_config().map_err(\|e\| anyhow::anyhow!("{e}"))
	1678	};
	1679	if dir.join(".oxy").is_dir() {
	1680	let mut repo = Repo::open(&dir).map_err(\|e\| anyhow::anyhow!("{e}"))?;
	1681	set_remote(&mut repo)?;
f53e6d47 Andy McCall 1 month ago ↶	1682	with_retries(&FETCH_WAITS, \|\| repo.fetch("origin").map(\|_\| ()).map_err(\|e\| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
e0918b88 Andy McCall 1 month ago	1683	return Ok(repo);
e0918b88 Andy McCall 1 month ago	1684	}
ed4aef8b Andy McCall 20 days ago ↶	1685	// First job for this repository: clone with the job's token, which goes
	1686	// through the process environment, so one clone at a time.
	1687	let _clone = CLONE.lock().unwrap_or_else(\|p\| p.into_inner());
c23d5191 Andy McCall 1 month ago ↶	1688	std::env::set_var("OXY_TOKEN", token);
e0918b88 Andy McCall 1 month ago	1689	if let Some(ca) = &cfg.ca {
	1690	std::env::set_var("OXY_CA", ca);
	1691	}
f53e6d47 Andy McCall 1 month ago ↶	1692	let cloned = with_retries(&FETCH_WAITS, \|\| {
	1693	let _ = std::fs::remove_dir_all(&dir);
	1694	Repo::clone(clone_url, &dir).map_err(\|e\| anyhow::anyhow!("{e}"))
	1695	});
e0918b88 Andy McCall 1 month ago	1696	std::env::remove_var("OXY_TOKEN");
	1697	std::env::remove_var("OXY_CA");
	1698	let mut repo = cloned.map_err(\|e\| {
	1699	let _ = std::fs::remove_dir_all(&dir);
c23d5191 Andy McCall 1 month ago ↶	1700	anyhow::anyhow!("clone {}: {e}", clone_url)
e0918b88 Andy McCall 1 month ago	1701	})?;
	1702	set_remote(&mut repo)?;
	1703	Ok(repo)
bb643c25 Andy McCall 1 month ago ↶	1704	}
	1705
	1706	#[cfg(test)]
	1707	mod tests {
46fe2c81 Andy McCall 21 days ago ↶	1708
	1709	#[test]
	1710	fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
	1711	let dir = tempfile::tempdir().unwrap();
	1712	let bytes = b"the reviewed plan";
	1713	let blob = Repo::blob_ref(bytes).to_hex();
	1714	let input = \|path: &str, blob: &str\| flow_core::api::ArtifactInput {
	1715	artifact: "tfplan".into(),
	1716	from: "plan".into(),
	1717	path: path.into(),
	1718	blob: blob.into(),
	1719	};
	1720	let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
	1721	assert_eq!(hex::encode(got), blob);
	1722	assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
	1723	let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
	1724	assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
	1725	assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
	1726	for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
	1727	assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
	1728	}
	1729	}
	1730
bb643c25 Andy McCall 1 month ago ↶	1731	use super::*;
b11e30f5 Andy McCall 1 month ago ↶	1732
	1733	#[test]
	1734	fn a_runner_names_what_stops_a_sandbox() {
	1735	let dir = tempfile::tempdir().unwrap();
	1736	let home = Home::locate(Some(dir.path())).unwrap();
	1737	let bytes = b"not really ext4";
	1738	let hash = blake3::hash(bytes).to_hex().to_string();
	1739	let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
	1740	"name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
	1741	}))
	1742	.unwrap();
	1743	let problem = \|sb: &flow_core::Sandbox, free\| sandbox_problem(&home, sb, "x86_64", free);
	1744	assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
	1745	std::fs::write(home.rootfs(&hash), bytes).unwrap();
	1746	sb.rootfs = hash.clone();
	1747	assert_eq!(problem(&sb, Some(4096)), None);
	1748	assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
	1749	std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
	1750	assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
	1751	sb.arch = Some("aarch64".into());
	1752	assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
	1753	sb.arch = None;
	1754	std::fs::remove_file(home.image_manifest(&hash)).unwrap();
	1755	std::fs::write(home.rootfs(&hash), b"changed").unwrap();
	1756	assert!(problem(&sb, None).unwrap().contains("does not verify"));
	1757	}
f53e6d47 Andy McCall 1 month ago ↶	1758
	1759	#[test]
	1760	fn transient_network_errors_are_tried_again_and_others_are_not() {
	1761	assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
	1762	assert!(transient("HTTP 503: the hub is restarting"));
	1763	assert!(!transient("HTTP 404: no such repository"));
	1764	let mut n = 0;
	1765	let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], \|\| {
	1766	n += 1;
	1767	if n < 3 {
	1768	Err(anyhow::anyhow!("connection refused"))
	1769	} else {
	1770	Ok(7)
	1771	}
	1772	});
	1773	assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
	1774	let mut m = 0;
	1775	let got: Result<u32> = with_retries(&[Duration::ZERO; 4], \|\| {
	1776	m += 1;
	1777	Err(anyhow::anyhow!("HTTP 403: forbidden"))
	1778	});
	1779	assert!(got.is_err());
	1780	assert_eq!(m, 1, "a refusal is not tried again");
	1781	let mut k = 0;
	1782	let got: Result<u32> = with_retries(&[Duration::ZERO; 2], \|\| {
	1783	k += 1;
	1784	Err(anyhow::anyhow!("timed out"))
	1785	});
	1786	assert!(got.is_err());
	1787	assert_eq!(k, 3, "gives up after the last wait");
	1788	}
1b2229da Andy McCall 1 month ago ↶	1789
	1790	/// A hub that answers each request with the next of `codes`, counting them.
	1791	fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
89171fca Andy McCall 24 days ago ↶	1792	fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
	1793	}
	1794
	1795	/// A hub at `listener` answering each request with the next of `codes`.
	1796	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 ↶	1797	use std::io::{Read, Write};
	1798	let addr = listener.local_addr().unwrap();
	1799	let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
	1800	let count = seen.clone();
	1801	std::thread::spawn(move \|\| {
	1802	for code in codes {
	1803	let (mut s, _) = listener.accept().unwrap();
	1804	let mut buf = [0u8; 8192];
	1805	let _ = s.read(&mut buf);
	1806	count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
	1807	let reason = if code == 204 { "No Content" } else { "Error" };
	1808	let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
	1809	}
	1810	});
	1811	(format!("http://{addr}"), seen)
f98a8f78 Andy McCall 1 month ago ↶	1812	}
	1813
	1814	/// A hub that answers one request with `body` (status 200) and hands
	1815	/// back what it was sent.
	1816	fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
	1817	use std::io::{Read, Write};
	1818	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	1819	let addr = listener.local_addr().unwrap();
	1820	let (tx, rx) = std::sync::mpsc::channel();
	1821	std::thread::spawn(move \|\| {
	1822	let (mut s, _) = listener.accept().unwrap();
	1823	let mut req = Vec::new();
	1824	let mut buf = [0u8; 65536];
	1825	loop {
	1826	let n = s.read(&mut buf).unwrap_or(0);
	1827	if n == 0 {
	1828	break;
	1829	}
	1830	req.extend_from_slice(&buf[..n]);
	1831	let text = String::from_utf8_lossy(&req);
	1832	if let Some(end) = text.find("\r\n\r\n") {
	1833	let len = text[..end]
	1834	.lines()
	1835	.find_map(\|l\| {
	1836	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	1837	})
	1838	.unwrap_or(0);
	1839	if req.len() >= end + 4 + len {
	1840	break;
	1841	}
	1842	}
	1843	}
	1844	tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
	1845	let _ = write!(
	1846	s,
	1847	"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
	1848	body.len()
	1849	);
	1850	});
	1851	(format!("http://{addr}"), rx)
	1852	}
	1853
	1854	#[test]
	1855	fn registering_binds_the_key_and_keeps_the_credential_private() {
	1856	use std::os::unix::fs::PermissionsExt;
	1857	let dir = tempfile::tempdir().unwrap();
	1858	let home = Home::locate(Some(dir.path())).unwrap();
	1859	let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
	1860	let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
	1861	assert_eq!((got.id, got.name.as_str()), (7, "plex"));
	1862	let req = sent.recv().unwrap();
	1863	assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
	1864	let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
	1865	assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
	1866	let key = home.runner_key("plex").unwrap();
	1867	assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
	1868	let path = dir.path().join(crate::home::RUNNER_TOML);
	1869	assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
	1870	// The runner then starts from runner.toml.
	1871	let again = Home::locate(Some(dir.path())).unwrap();
	1872	let cfg = again.config.runner.unwrap();
	1873	assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
	1874	assert_eq!(cfg.labels, ["gpu"]);
35f7a849 Andy McCall 20 days ago ↶	1875	}
	1876
	1877	/// A hub that answers certify with a certificate `sign_as` signs for
	1878	/// `cert_key`, and publishes `published` as its key: two requests.
	1879	fn fake_certify_hub(
	1880	sign_as: (oxy_core::identity::Identity, oxy_core::identity::SecretKey),
	1881	cert_key: oxy_core::identity::Identity,
	1882	published: oxy_core::identity::Identity,
	1883	) -> String {
	1884	use std::io::{Read, Write};
	1885	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	1886	let addr = listener.local_addr().unwrap();
	1887	std::thread::spawn(move \|\| {
	1888	for _ in 0..2 {
	1889	let Ok((mut s, _)) = listener.accept() else { return };
	1890	let mut req = Vec::new();
	1891	let mut buf = [0u8; 65536];
	1892	loop {
	1893	let n = s.read(&mut buf).unwrap_or(0);
	1894	if n == 0 {
	1895	break;
	1896	}
	1897	req.extend_from_slice(&buf[..n]);
	1898	let text = String::from_utf8_lossy(&req);
	1899	if let Some(end) = text.find("\r\n\r\n") {
	1900	let len = text[..end]
	1901	.lines()
	1902	.find_map(\|l\| {
	1903	l.to_ascii_lowercase().strip_prefix("content-length:").map(\|v\| v.trim().parse::<usize>().unwrap_or(0))
	1904	})
	1905	.unwrap_or(0);
	1906	if req.len() >= end + 4 + len {
	1907	break;
	1908	}
	1909	}
	1910	}
	1911	let line = String::from_utf8_lossy(&req).lines().next().unwrap_or("").to_string();
	1912	let body: Vec<u8> = if line.contains("/keyless/certify") {
	1913	let cert = oxy_core::cert::SignedCertificate::sign(
	1914	oxy_core::cert::Certificate {
	1915	key: cert_key.clone(),
	1916	issuer: sign_as.0.hash(),
	1917	subject: "repo:andy/demo:bookmark:main".into(),
	1918	claims: vec![],
	1919	scope: oxy_core::identity::Scope::SEAL,
	1920	not_before: 1,
	1921	expires: 600_001,
fc2a6d18 Andy McCall 19 days ago ↶	1922	grant: None,
35f7a849 Andy McCall 20 days ago ↶	1923	},
	1924	&sign_as.0,
	1925	&sign_as.1,
	1926	);
	1927	serde_json::json!({ "certificate": oxy_core::dsse::base64(&cert.encode()) }).to_string().into_bytes()
	1928	} else {
	1929	published.encode()
	1930	};
	1931	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
	1932	let _ = s.write_all(&body);
	1933	}
	1934	});
	1935	format!("http://{addr}")
	1936	}
	1937
	1938	/// keyless (oxygenhub#241): the runner takes a certificate only when it
	1939	/// is for its fresh key and signed by the key the hub publishes.
	1940	#[test]
	1941	fn a_certificate_is_taken_only_for_this_key_from_the_hubs_key() {
	1942	use oxy_core::identity::SecretKey;
	1943	let hub_sk = SecretKey::generate();
	1944	let hub_id = hub_sk.identity("hub", "", 1);
	1945	let sk = SecretKey::generate();
	1946	let id = sk.identity("earth-1 job 7", "", 1);
	1947	let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge("flowjob_x")).unwrap();
	1948
	1949	let base = fake_certify_hub((hub_id.clone(), hub_sk.clone()), id.clone(), hub_id.clone());
	1950	let (cert, issuer) = hub_at(base).certify("flowjob_x", &id, &proof).unwrap();
	1951	assert_eq!((cert.cert.key_id(), issuer.hash()), (id.hash(), hub_id.hash()));
	1952
	1953	let mallory_sk = SecretKey::generate();
	1954	let mallory = mallory_sk.identity("mallory", "", 1);
	1955	let base = fake_certify_hub((mallory.clone(), mallory_sk), id.clone(), hub_id.clone());
	1956	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	1957	assert!(e.contains("not signed by the hub's published key"), "{e}");
	1958
	1959	let other = SecretKey::generate().identity("other", "", 1);
	1960	let base = fake_certify_hub((hub_id.clone(), hub_sk), other, hub_id);
	1961	let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
	1962	assert!(e.contains("another key"), "{e}");
1b2229da Andy McCall 1 month ago ↶	1963	}
	1964
	1965	fn hub_at(base: String) -> Hub {
	1966	Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
cedd1234 Andy McCall 11 days ago ↶	1967	}
	1968
	1969	/// oxygenhub#322: an offered build is kept only if it is the bytes the
	1970	/// hub attests; a peer or a proxy cannot change what is installed.
	1971	#[test]
	1972	fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {
	1973	let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());
	1974	let serve = \|bytes: Vec<u8>\| {
	1975	let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
	1976	let addr = listener.local_addr().unwrap();
	1977	std::thread::spawn(move \|\| {
	1978	use std::io::{Read, Write};
	1979	let (mut s, _) = listener.accept().unwrap();
	1980	let mut buf = [0u8; 4096];
	1981	let _ = s.read(&mut buf);
	1982	let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());
	1983	let _ = s.write_all(&bytes);
	1984	});
	1985	format!("http://{addr}")
	1986	};
	1987	let offer = flow_core::api::UpdateOffer {
	1988	version: "0.3.2".into(),
	1989	asset: "flow-linux-x86_64".into(),
	1990	url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),
	1991	blake3: blake3::hash(&body).to_hex().to_string(),
	1992	size: body.len() as u64,
	1993	..Default::default()
	1994	};
	1995	let dir = tempfile::tempdir().unwrap();
	1996	let to = dir.path().join("flow-0.3.2");
	1997	hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();
	1998	assert_eq!(std::fs::read(&to).unwrap(), body);
	1999	std::fs::remove_file(&to).unwrap();
	2000	let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();
	2001	assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");
	2002	assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");
3007d8f0 Andy McCall 1 month ago ↶	2003	}
	2004
	2005	/// oxygenhub#71: a job's result is offered until the hub takes it,
	2006	/// through server errors, and a refusal ends the trying at once.
	2007	#[test]
	2008	fn a_result_is_reported_until_the_hub_takes_it() {
8f211e20 Andy McCall 25 days ago ↶	2009	let f = Finish {
	2010	exit_code: Some(0),
	2011	duration_ms: 1,
	2012	receipt: None,
	2013	signed_receipt: None,
	2014	error: None,
	2015	peak_disk_mib: None,
	2016	peak_memory_mib: None,
	2017	};
3007d8f0 Andy McCall 1 month ago ↶	2018	let quick = Duration::from_millis(10);
	2019	let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
	2020	hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
	2021	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
	2022	let (base, seen) = fake_hub(vec![409]);
	2023	assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
	2024	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1b2229da Andy McCall 1 month ago ↶	2025	}
	2026
	2027	#[test]
	2028	fn server_errors_are_tried_again_and_client_errors_are_not() {
	2029	let quick = [Duration::from_millis(10); 3];
	2030	let (base, seen) = fake_hub(vec![500, 503, 204]);
	2031	let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
	2032	assert!(r.is_none());
	2033	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
	2034
	2035	let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
	2036	let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
	2037	assert!(e.contains("HTTP 500"), "{e}");
	2038	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
	2039
	2040	let (base, seen) = fake_hub(vec![401]);
	2041	assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
	2042	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
06a4c472 Andy McCall 11 days ago ↶	2043	}
06a4c472 Andy McCall 11 days ago ↶	2044
1b160d37 Andy McCall 11 days ago ↶	2045	static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall 11 days ago ↶	2046
	2047	/// oxygenflow#76: once the runner leaves, a job's sink says it is
	2048	/// cancelled, having sent what the job said so far, and the job is
	2049	/// handed back to the hub.
	2050	#[test]
	2051	fn a_leaving_runner_stops_its_job_and_hands_it_back() {
	2052	use microvm::Sink;
	2053	let leaving = std::sync::Mutex::new(None);
	2054	let (base, seen) = fake_hub(vec![204, 204]);
	2055	let hub = hub_at(base);
1b160d37 Andy McCall 11 days ago ↶	2056	let mut sink = LogSink {
	2057	hub: &hub,
	2058	job: 7,
	2059	buf: Vec::new(),
	2060	last: Instant::now(),
	2061	gone: false,
	2062	leaving: &leaving,
	2063	started: Instant::now(),
	2064	expected: None,
	2065	};
06a4c472 Andy McCall 11 days ago ↶	2066	sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
06a4c472 Andy McCall 11 days ago ↶	2067	assert!(!sink.cancelled());
1b160d37 Andy McCall 11 days ago ↶	2068	*leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
06a4c472 Andy McCall 11 days ago ↶	2069	assert!(sink.cancelled());
	2070	assert!(sink.buf.is_empty(), "flushed before the job stops");
	2071	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
	2072	hub.release(7, "off Wi-Fi").unwrap();
	2073	assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1b160d37 Andy McCall 11 days ago ↶	2074	}
	2075
	2076	/// oxygenflow#76: with notice, a job the hub expects to finish within
	2077	/// it runs on; one that will not, or whose time is unknown, is handed
	2078	/// back at once, and any job still running when the notice is up.
	2079	#[test]
	2080	fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
	2081	use microvm::Sink;
	2082	let (base, _) = fake_hub(vec![204; 8]);
	2083	let hub = hub_at(base);
	2084	let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
	2085	let sink = \|expected: Option<u64>, ran: u64\| LogSink {
	2086	hub: &hub,
	2087	job: 1,
	2088	buf: Vec::new(),
	2089	last: Instant::now(),
	2090	gone: false,
	2091	leaving: &leaving,
	2092	started: Instant::now() - Duration::from_secs(ran),
	2093	expected: expected.map(Duration::from_secs),
	2094	};
	2095	assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
	2096	assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
	2097	assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
	2098	*leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
	2099	assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
280ef951 Andy McCall 1 month ago ↶	2100	}
	2101
	2102	/// oxygenflow#32: a quiet job asks the hub every few seconds whether
	2103	/// it still runs; a 409 marks it cancelled, and nothing more is sent.
	2104	#[test]
	2105	fn a_quiet_job_learns_that_its_run_was_cancelled() {
	2106	use microvm::Sink;
	2107	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2108	let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
	2109	let (base, seen) = fake_hub(vec![204]);
	2110	let hub = hub_at(base);
1b160d37 Andy McCall 11 days ago ↶	2111	let mut sink = LogSink {
	2112	hub: &hub,
	2113	job: 1,
	2114	buf: Vec::new(),
	2115	last: quiet,
	2116	gone: false,
	2117	leaving: &NOT_LEAVING,
	2118	started: Instant::now(),
	2119	expected: None,
	2120	};
280ef951 Andy McCall 1 month ago ↶	2121	assert!(!sink.cancelled());
	2122	assert!(!sink.cancelled(), "asked again only after the heartbeat");
	2123	assert_eq!(n(&seen), 1);
	2124
	2125	let (base, seen) = fake_hub(vec![409]);
	2126	let hub = hub_at(base);
1b160d37 Andy McCall 11 days ago ↶	2127	let mut sink = LogSink {
	2128	hub: &hub,
	2129	job: 1,
	2130	buf: Vec::new(),
	2131	last: quiet,
	2132	gone: false,
	2133	leaving: &NOT_LEAVING,
	2134	started: Instant::now(),
	2135	expected: None,
	2136	};
280ef951 Andy McCall 1 month ago ↶	2137	assert!(sink.cancelled());
	2138	sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
	2139	sink.flush();
	2140	assert!(sink.cancelled());
	2141	assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
89171fca Andy McCall 24 days ago ↶	2142	}
	2143
	2144	/// oxygenflow#49: a hub that is away for a while (a deploy restarting
	2145	/// it) costs a job's uploads a wait, not the upload, and costs its log
	2146	/// no lines.
	2147	#[test]
	2148	fn a_job_outlasts_a_hub_restart() {
	2149	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	2150	// A port nobody listens on yet: the hub is restarting.
	2151	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2152	let base = format!("http://127.0.0.1:{port}");
	2153	let hub = hub_at(base.clone());
	2154
	2155	// Log lines sent while it is away are kept, and go when it is back.
1b160d37 Andy McCall 11 days ago ↶	2156	let mut sink = LogSink {
	2157	hub: &hub,
	2158	job: 1,
	2159	buf: Vec::new(),
	2160	last: Instant::now(),
	2161	gone: false,
	2162	leaving: &NOT_LEAVING,
	2163	started: Instant::now(),
	2164	expected: None,
	2165	};
89171fca Andy McCall 24 days ago ↶	2166	for i in 0..450 {
	2167	sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
	2168	}
	2169	sink.send();
	2170	assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
	2171	let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
	2172	sink.send();
	2173	assert!(sink.buf.is_empty() && !sink.gone);
	2174	assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");
	2175
	2176	// An upload waits for the hub to come back, through refused
	2177	// connections and server errors alike.
	2178	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2179	let url = format!("http://127.0.0.1:{port}/x");
	2180	let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
	2181	let later = std::thread::spawn(move \|\| {
	2182	std::thread::sleep(Duration::from_millis(150));
	2183	fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
	2184	});
	2185	let quick = [Duration::from_millis(50); 20];
	2186	let r = retrying(&url, &quick, true, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
	2187	assert_eq!(r.status(), 204);
	2188	assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
	2189	// Without `connection`, a hub that cannot be reached ends it at once.
	2190	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2191	let url = format!("http://127.0.0.1:{port}/x");
	2192	assert!(retrying(&url, &quick, false, \|\| agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
	2193	assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
	2194	}
	2195
	2196	/// oxygenflow#50: the first stop drains (the second, not called here,
	2197	/// exits at once).
	2198	#[test]
	2199	fn a_stop_drains_the_runner() {
	2200	assert!(!stopping());
	2201	on_stop(libc::SIGTERM);
	2202	assert!(stopping(), "told to stop, it claims nothing more");
5cf1e8ae Andy McCall 12 days ago ↶	2203	// An embedded runner starts afresh, and stop() drains it the same way.
	2204	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
	2205	stop();
	2206	assert!(stopping());
	2207	STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
89171fca Andy McCall 24 days ago ↶	2208	}
	2209
	2210	/// Lines kept for a hub that stays away are bounded, saying how many went.
	2211	#[test]
	2212	fn unsent_log_lines_are_bounded() {
	2213	let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
	2214	let hub = hub_at(format!("http://127.0.0.1:{port}"));
1b160d37 Andy McCall 11 days ago ↶	2215	let mut sink = LogSink {
	2216	hub: &hub,
	2217	job: 1,
	2218	buf: Vec::new(),
	2219	last: Instant::now(),
	2220	gone: false,
	2221	leaving: &NOT_LEAVING,
	2222	started: Instant::now(),
	2223	expected: None,
	2224	};
89171fca Andy McCall 24 days ago ↶	2225	sink.buf = (0..MAX_UNSENT as u64 + 10).map(\|i\| Record { t: i, stream: "out".into(), text: String::new() }).collect();
	2226	sink.send();
	2227	assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
	2228	assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
	2229	assert_eq!(sink.buf[1].t, 10, "the oldest go first");
1b2229da Andy McCall 1 month ago ↶	2230	}
bb643c25 Andy McCall 1 month ago ↶	2231
	2232	#[test]
	2233	fn the_budget_admits_jobs_that_fit() {
	2234	let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
	2235	let used = \|jobs, memory_mib, cpus\| Used { jobs, memory_mib, cpus };
	2236	assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
	2237	assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
	2238	assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
	2239	assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
	2240	let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
	2241	assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
	2242	assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
	2243	let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
	2244	assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
	2245	}
e0918b88 Andy McCall 1 month ago	2246	}

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

Keyboard