crates/flow/src/runner.rs at 8b7034c55402 · andymccall/oxygenflow

# 8b7034c55402 oxygenflow / crates / flow / src / runner.rs

105.1 KiB · Rust · last changed Runners offer the manifests of images the hub lacks; a fetch the hub cannot seed asks peers alone (refs #94) 9 days ago source blame history raw

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

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

Keyboard