Runner builds fetched through the swarm (refs #84); the WASM tier as a cargo feature, off for Haiku (refs #91) 21a3ea4aab88

Author: Andy McCall @andymccall <[email protected]>
via: agent agent:claude@andymccall <[email protected]> · keyless, under your grant for changes,refs,seal · key c72b441c9f1e
When: 7 days ago · 2026-10-28 21:27 UTC
Signature: ◈ sealed by agent:claude@andymccall
Hash: 21a3ea4aab8894f35a130036a31837139627f5b0e03bf6f0180f9abc9a0be727
Depends on: 28cc603f737d 48ed8a1db369 5cf1e8aefbb6 6f66e397539c 85275e1ef3b7 c3db750ed96e cedd12342afa e0918b887850 ea56c56eaf90
Depended on by: 412808f9f769 788f57e4d266
In bookmarks: main
In tags: v0.5.2 v0.5.3
Edits: 30 atoms, 0 tree operations, 5 files

Browse the tree at this change

hide whitespace changes

modified CHANGELOG.md +5 −0

modified crates/flow/Cargo.toml +8 −2

modified crates/flow/src/runner.rs +49 −5

modified crates/flow/src/update.rs +28 −0

modified crates/flow/src/wasm.rs +36 −1

		…	1 unchanged lines
2	2
3	3		0.2.0 is OxygenFlow's first tagged release; until then the runners ran `main`. This file starts on 2026-10-01; earlier history is in `oxy log`.
4	4
	5	+	## Unreleased
	6	+
	7	+	- Runner builds through the swarm (#84): a runner in a swarm fetches an offered build as the image its BLAKE3 names, from its site's peers and the hub together, as it fetches images, and checks its size and BLAKE3 before installing it; over HTTP as before when it is in no swarm, the hub has not shared the build yet, or the swarm cannot get it. The hub shares a build the first time it offers it (OxygenHub 0.9.15), so a site's runners fetch it from each other and the hub serves it about once per site. The build stays in the runner's images for its peers.
	8	+	- The WASM tier can be left out of a build (#91): the `wasm` cargo feature (on by default) carries wasmtime; `--no-default-features` builds `flow` for systems wasmtime does not run on, such as Haiku, whose runners then offer no WASM tier and say why.
	9	+
5	10		## 0.5.1 (2026-10-28)
6	11
7	12		- The cheap checks run as WASM (#96): `layout`, `todos`, `size` and `summary` are one WASI module, `crates/flow-checks`, built by `scripts/checks-wasm.sh` and committed as `.flow/checks.wasm`, so they need no image or microVM and any runner with the WASM tier takes them (the phone, plex, windows-1); each runs in under 100 ms. The fmt job fails unless the committed module is what the source builds, and the build image has the wasm32-wasip1 target for that.
		…	120 unchanged lines

		…	30 unchanged lines
31	31		rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] }
32	32		rcgen = "0.14"
33	33		# The WASM tier (oxygenflow#75): Cranelift-compiled WASI preview 1 modules.
34		-	wasmtime = { version = "=49.0.4", default-features = false, features = ["cranelift", "runtime", "std", "parallel-compilation"] }
35		-	wasmtime-wasi = { version = "=49.0.4", default-features = false, features = ["p1"] }
	34	+	wasmtime = { version = "=49.0.4", optional = true, default-features = false, features = ["cranelift", "runtime", "std", "parallel-compilation"] }
	35	+	wasmtime-wasi = { version = "=49.0.4", optional = true, default-features = false, features = ["p1"] }
36	36
37	37		# A Windows runner's free disk, version, and renaming a running flow.exe
38	38		# (oxygenflow#83);
		…	1 unchanged lines
40	40		[target.'cfg(windows)'.dependencies]
41	41		windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_SystemInformation", "Wdk_System_SystemServices"] }
42	42
	43	+	[features]
	44	+	default = ["wasm"]
	45	+	# The WASM tier: off for systems wasmtime does not build or run on (Haiku,
	46	+	# oxygenflow#91).
	47	+	wasm = ["dep:wasmtime", "dep:wasmtime-wasi"]
	48	+
43	49		[dev-dependencies]
44	50		tempfile.workspace = true
45	51		# Test modules written as text.
		…	1 unchanged lines

		…	544 unchanged lines
545	545		Ok(()) => have.push("syscall".to_string()),
546	546		Err(e) => missing.push(format!("syscall: {e}")),
547	547		}
548		-	// wasmtime runs on every machine this binary does (oxygenflow#75).
549		-	have.push("wasm".to_string());
	548	+	// wasmtime runs on every machine a build with it does (oxygenflow#75).
	549	+	if crate::wasm::AVAILABLE {
	550	+	have.push("wasm".to_string());
	551	+	} else {
	552	+	missing.push("wasm: this build has no WASM tier".to_string());
	553	+	}
550	554		if have.is_empty() {
551	555		return Err(format!("no isolation tier can run here: {}", missing.join("; ")));
552	556		}
		…	379 unchanged lines
932	936		if let Some(note) = crate::update::take_note(&home.root) {
933	937		let _ = hub.report_update(&note);
934	938		}
935		-	consider_update(home, &hub, &mut update_said);
	939	+	consider_update(home, &hub, swarm.as_ref(), &mut update_said);
936	940		}
937	941		match answer {
938	942		Ok(Some(claim)) => {
		…	258 unchanged lines
1197	1201		/// it offers, check it against the hub's BLAKE3 and as a runner build, and
1198	1202		/// install it, or for a dry run say what would happen. What it did goes to
1199	1203		/// the hub once per outcome (`said`).
1200		-	fn consider_update(home: &Home, hub: &Hub, said: &mut String) {
	1204	+	fn consider_update(home: &Home, hub: &Hub, swarm: Option<&crate::swarm::Swarm>, said: &mut String) {
1201	1205		let answer = match hub.update_answer() {
1202	1206		Ok(Some(a)) => a,
1203	1207		Ok(None) => return,
		…	19 unchanged lines
1223	1227		let file = dir.join(format!("flow-{}", offer.version));
1224	1228		let fresh = std::fs::read(&file).is_ok_and(\|b\| blake3::hash(&b).to_hex().as_str() == offer.blake3);
1225	1229		if !fresh {
1226		-	if let Err(e) = std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(\|_\| hub.download_build(&offer, &file)) {
	1230	+	if let Err(e) =
	1231	+	std::fs::create_dir_all(&dir).map_err(anyhow::Error::from).and_then(\|_\| fetch_build(home, hub, swarm, &offer, &file))
	1232	+	{
1227	1233		return say(format!("flow {}: not fetched: {e:#}", offer.version));
1228	1234		}
1229	1235		}
		…	8 unchanged lines
1238	1244		say(format!("installed flow {} (was {}); leaving within {} s to start it", done.to, done.from, UPDATE_NOTICE.as_secs()))
1239	1245		}
1240	1246		Err(e) => say(format!("flow {}: not installed: {e:#}", offer.version)),
	1247	+	}
	1248	+	}
	1249	+
	1250	+	/// Fetch an offered build to `file`: through the swarm when this runner is
	1251	+	/// in one (oxygenflow#84), as an image named by the build's BLAKE3, which
	1252	+	/// the hub chunks when it first offers it, so runners of a site fetch it
	1253	+	/// from each other and the hub serves it about once per site; over HTTP
	1254	+	/// otherwise, or when the swarm cannot get it. Either way it must be the
	1255	+	/// size and BLAKE3 the hub attests.
	1256	+	fn fetch_build(
	1257	+	home: &Home,
	1258	+	hub: &Hub,
	1259	+	swarm: Option<&crate::swarm::Swarm>,
	1260	+	offer: &flow_core::api::UpdateOffer,
	1261	+	file: &Path,
	1262	+	) -> Result<()> {
	1263	+	if let Some(sw) = swarm.filter(\|_\| jet::images::is_hash(&offer.blake3)) {
	1264	+	match sw.fetch(home, &offer.blake3) {
	1265	+	Ok(Some(m)) => {
	1266	+	let image = home.rootfs(&offer.blake3);
	1267	+	match crate::update::copy_checked(&image, file, offer.size, &offer.blake3) {
	1268	+	Ok(()) => {
	1269	+	eprintln!(
	1270	+	"flow runner: fetched flow {} through the swarm: {} chunks, {} ({} from peers)",
	1271	+	offer.version,
	1272	+	m.chunks,
	1273	+	crate::images::human(m.bytes),
	1274	+	crate::images::human(m.peer_bytes)
	1275	+	);
	1276	+	return Ok(());
	1277	+	}
	1278	+	Err(e) => eprintln!("flow runner: flow {} from the swarm: {e:#}; fetching it over HTTP", offer.version),
	1279	+	}
	1280	+	}
	1281	+	Ok(None) => eprintln!("flow runner: the hub has not shared flow {} with the swarm yet; fetching it over HTTP", offer.version),
	1282	+	Err(e) => eprintln!("flow runner: flow {} through the swarm: {e:#}; fetching it over HTTP", offer.version),
	1283	+	}
1241	1284		}
	1285	+	hub.download_build(offer, file)
1242	1286		}
1243	1287
1244	1288		/// Held while a job takes its cache disks and images its workspace (see
		…	1082 unchanged lines

		…	154 unchanged lines
155	155		Ok(String::from_utf8_lossy(&out.stdout).into_owned())
156	156		}
157	157
	158	+	/// Copy `from` to `to` if it is `size` bytes with BLAKE3 `blake3`, as the
	159	+	/// hub attests a build: a part file renamed into place, so `to` is either
	160	+	/// the whole checked build or absent.
	161	+	pub fn copy_checked(from: &Path, to: &Path, size: u64, blake3: &str) -> Result<()> {
	162	+	let bytes = std::fs::read(from).with_context(\|\| format!("reading {}", from.display()))?;
	163	+	let got = blake3::hash(&bytes).to_hex().to_string();
	164	+	if bytes.len() as u64 != size \|\| got != blake3 {
	165	+	bail!("{} is not the build the hub attests ({} bytes, BLAKE3 {}…)", from.display(), bytes.len(), &got[..12]);
	166	+	}
	167	+	let part = to.with_extension("partial");
	168	+	std::fs::write(&part, &bytes).with_context(\|\| format!("writing {}", part.display()))?;
	169	+	std::fs::rename(&part, to)?;
	170	+	Ok(())
	171	+	}
	172	+
158	173		/// Check a build as the runner would run it: this machine's target, it
159	174		/// starts and names its version, and it finds the tiers this home offers.
160	175		/// Returns its version.
		…	285 unchanged lines
446	461
447	462		#[cfg(test)]
448	463		mod tests {
	464	+
	465	+	#[test]
	466	+	fn a_build_from_the_swarm_is_copied_only_when_it_is_the_one_attested() {
	467	+	let dir = tempfile::tempdir().unwrap();
	468	+	let (from, to) = (dir.path().join("cas.ext4"), dir.path().join("flow-0.5.2"));
	469	+	std::fs::write(&from, b"flow build").unwrap();
	470	+	let hash = blake3::hash(b"flow build").to_hex().to_string();
	471	+	assert!(copy_checked(&from, &to, 9, &hash).is_err(), "the wrong size");
	472	+	assert!(copy_checked(&from, &to, 10, &"0".repeat(64)).is_err(), "the wrong hash");
	473	+	assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing left behind");
	474	+	copy_checked(&from, &to, 10, &hash).unwrap();
	475	+	assert_eq!(std::fs::read(&to).unwrap(), b"flow build");
	476	+	}
449	477		use super::*;
450	478
451	479		#[test]
		…	114 unchanged lines

		…	22 unchanged lines
23	23		/// The runtime, named in every receipt this tier writes.
24	24		pub const WASMTIME: &str = "49.0.4";
25	25
	26	+	/// Whether this build has the tier: a build for a system wasmtime does not
	27	+	/// run on (Haiku, oxygenflow#91) leaves it out with the `wasm` feature, and
	28	+	/// its runner does not offer it.
	29	+	pub const AVAILABLE: bool = cfg!(feature = "wasm");
	30	+
	31	+	/// What a build without the tier says when asked to run a job in it.
	32	+	const WITHOUT: &str = "this flow was built without the WASM tier (the `wasm` feature): it runs on systems wasmtime supports";
	33	+
26	34		/// A line the helper writes to say something about the job itself (which
27	35		/// module starts), shown as the runner's own note rather than the job's.
28	36		const NOTE: &str = "\u{1e}N ";
29	37
30	38		/// The exit code for a module that trapped (crashed) rather than exiting.
	39	+	#[cfg(feature = "wasm")]
31	40		const TRAPPED: i32 = 134;
32	41
33	42		/// What the helper runs, written by the runner for `flow __wasm`.
		…	88 unchanged lines
122	131		/// tier lays them out); `scratch` holds the helper's spec and the job's
123	132		/// mounts.
124	133		pub fn run(sandbox: &Sandbox, job: &Job, workspace: &Path, scratch: &Path, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {
	134	+	if !AVAILABLE {
	135	+	bail!("job {}: {WITHOUT}", job.name);
	136	+	}
125	137		if !job.secrets.is_empty() \|\| !job.identities.is_empty() {
126	138		bail!("job {}: the WASM tier takes no secrets or identities yet; run it in a microvm or syscall sandbox", job.name);
127	139		}
		…	116 unchanged lines
244	256		cfg!(target_os = "android") \|\| std::env::var_os("FLOW_WASM_IN_PROCESS").is_some()
245	257		}
246	258
	259	+	#[cfg(feature = "wasm")]
247	260		/// The most of a module's output kept before it is read, per stream.
248	261		const PIPE_BYTES: usize = 64 << 20;
249	262
	263	+	#[cfg(feature = "wasm")]
250	264		/// Run the spec's modules in a thread of this process, streaming their
251	265		/// output as it comes: the pipes are read every 100 ms, and a time limit
252	266		/// or a cancelled run interrupts the module through wasmtime's epoch.
		…	80 unchanged lines
333	347		})
334	348		}
335	349
	350	+	#[cfg(feature = "wasm")]
336	351		/// `flow __wasm SPEC`: run the spec's modules in turn, each in a fresh
337	352		/// instance; the exit code is the first non-zero one, after all have run,
338	353		/// so one failing module does not hide another's results.
		…	10 unchanged lines
349	364		125
350	365		}
351	366		}
	367	+	}
	368	+
	369	+	/// `flow __wasm SPEC` in a build without the tier: says so.
	370	+	#[cfg(not(feature = "wasm"))]
	371	+	pub fn wasm_main(_spec_path: &Path) -> i32 {
	372	+	eprintln!("{FAILED}{WITHOUT}");
	373	+	125
	374	+	}
	375	+
	376	+	#[cfg(not(feature = "wasm"))]
	377	+	fn run_in_thread(_spec: &Spec, _environment: Environment, _timeout: Duration, _sink: &mut dyn Sink) -> Result<Outcome> {
	378	+	bail!(WITHOUT)
352	379		}
353	380
	381	+	#[cfg(feature = "wasm")]
354	382		/// wasmtime's own error, as this crate's: with its whole chain of causes.
355	383		fn ww<T>(r: wasmtime::Result<T>, what: impl FnOnce() -> String) -> Result<T> {
356	384		r.map_err(\|e\| anyhow::anyhow!("{}: {e:#}", what()))
357	385		}
358	386
	387	+	#[cfg(feature = "wasm")]
359	388		/// Where a module's output and the tier's notes go: inherited by the
360	389		/// helper process (notes on stderr, marked), or pipes the runner reads.
361	390		enum Io {
		…	5 unchanged lines
367	396		},
368	397		}
369	398
	399	+	#[cfg(feature = "wasm")]
370	400		impl Io {
371	401		fn note(&self, text: String) {
372	402		match self {
		…	5 unchanged lines
378	408		}
379	409		}
380	410
	411	+	#[cfg(feature = "wasm")]
381	412		/// The sandbox's memory limit, and the most a module asked for within it.
382	413		struct Limits {
383	414		memory: usize,
384	415		peak: usize,
385	416		}
386	417
	418	+	#[cfg(feature = "wasm")]
387	419		impl wasmtime::ResourceLimiter for Limits {
388	420		fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
389	421		let ok = desired <= self.memory && maximum.is_none_or(\|m\| desired <= m);
		…	8 unchanged lines
398	430		}
399	431		}
400	432
	433	+	#[cfg(feature = "wasm")]
401	434		struct State {
402	435		wasi: wasmtime_wasi::p1::WasiP1Ctx,
403	436		limits: Limits,
404	437		}
405	438
	439	+	#[cfg(feature = "wasm")]
406	440		/// Run each module in turn: the first non-zero exit code after all ran,
407	441		/// and the most memory any of them held, in bytes.
408	442		fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> {
		…	16 unchanged lines
425	459		Ok((first_failure, peak))
426	460		}
427	461
	462	+	#[cfg(feature = "wasm")]
428	463		fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> {
429	464		use wasmtime_wasi::{FsPerms, WasiCtxBuilder};
430	465		let module = ww(wasmtime::Module::from_file(engine, spec.workspace.join(path)), \|\| format!("module {path} does not load"))?;
		…	41 unchanged lines
472	507		Ok((code, store.data().limits.peak))
473	508		}
474	509
475		-	#[cfg(test)]
	510	+	#[cfg(all(test, feature = "wasm"))]
476	511		mod tests {
477	512		use super::*;
478	513
		…	85 unchanged lines

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

Runner builds fetched through the swarm (refs #84); the WASM tier as a cargo feature, off for Haiku (refs #91) 21a3ea4aab88

Keyboard