Stop a job when its run is cancelled 280ef951d193

The microVM wait loop now asks its sink, a few times a second, whether
the job is still wanted. The runner's log sink sends an empty batch
every 5 s while a job is quiet. When the hub answers 409, the job is
cancelled: the microVM is stopped and nothing more is sent or reported.
A job cancelled while it was being prepared boots no microVM.

refs #32

Author: Andy McCall @andymccall <[email protected]>
When: 1 month ago · 2026-10-02 22:52 UTC
Signature: ✔ signed by Andy McCall
Hash: 280ef951d193257d4f21659467ee66d2f13d5759def4a1e1a219c80f25fe1c07
Depends on: 1b2229dab1a4 1daca0b6bb8c 3007d8f011fa a67b97e9a717 bb643c252515 e0918b887850
Depended on by: 1215f05ff8d0 f201e1bc5bcd 1b160d372d9e 06a4c472b385 28cc603f737d 4be5e4cd0af6 077a65242359 89171fca079c d60aac6c3f8a 8f211e20bea5 ea56c56eaf90
In bookmarks: main newer-pipeline push-uses-login
In tags: v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits: 29 atoms, 0 tree operations, 3 files

Browse the tree at this change

hide whitespace changes

modified crates/flow/src/microvm.rs +49 −12

modified crates/flow/src/runner.rs +103 −7

modified docs/m0.md +1 −1

		…	17 unchanged lines
18	18		pub duration_ms: u64,
19	19		pub environment: Environment,
20	20		pub timed_out: bool,
	21	+	/// Stopped because the run was cancelled (oxygenflow#32).
	22	+	pub cancelled: bool,
21	23		}
22	24
	25	+	/// Where a job's records go, and whether it is still wanted: asked a few
	26	+	/// times a second while the job runs, and the microVM is stopped on `true`.
	27	+	pub trait Sink {
	28	+	fn record(&mut self, r: Record);
	29	+	fn cancelled(&mut self) -> bool {
	30	+	false
	31	+	}
	32	+	}
	33	+
	34	+	impl<F: FnMut(Record)> Sink for F {
	35	+	fn record(&mut self, r: Record) {
	36	+	self(r)
	37	+	}
	38	+	}
	39	+
23	40		/// The receipt environment for a job in this tier.
24	41		pub fn environment(home: &Home, sandbox: &Sandbox, job: &Job) -> Result<Environment> {
25	42		let cfg = home.config.microvm.as_ref().context("flow.toml has no [microvm] section")?;
		…	23 unchanged lines
49	66		workspace: &Path,
50	67		scratch: &Path,
51	68		timeout: Duration,
52		-	sink: &mut dyn FnMut(Record),
	69	+	sink: &mut dyn Sink,
53	70		) -> Result<Outcome> {
54	71		let cfg = home.config.microvm.as_ref().context("flow.toml has no [microvm] section")?;
55	72		let firecracker = home.verified(&home.resolve(&cfg.firecracker.path), &cfg.firecracker.blake3)?;
		…	64 unchanged lines
120	137		let readers = [std::thread::spawn(move \|\| lines(out, true, tx)), std::thread::spawn(move \|\| lines(err, false, tx2))];
121	138		let mut exit_code = None;
122	139		let mut timed_out = false;
	140	+	let mut cancelled = false;
123	141		let deadline = started + timeout;
124	142		let t = \|s: &Instant\| s.elapsed().as_millis() as u64;
125	143		loop {
		…	2 unchanged lines
128	146		let text = String::from_utf8_lossy(&line).trim_end_matches('\r').to_string();
129	147		let rec = \|stream: &str, text: &str\| Record { t: t(&started), stream: stream.into(), text: text.into() };
130	148		if !from_console {
131		-	sink(rec("console", &format!("firecracker: {text}")));
	149	+	sink.record(rec("console", &format!("firecracker: {text}")));
132	150		} else if let Some(rest) = text.strip_prefix('\u{1e}') {
133	151		let (kind, body) = rest.split_at(rest.len().min(1));
134	152		let body = body.strip_prefix(' ').unwrap_or(body);
135	153		match kind {
136		-	"O" => sink(rec("out", body)),
137		-	"E" => sink(rec("err", body)),
	154	+	"O" => sink.record(rec("out", body)),
	155	+	"E" => sink.record(rec("err", body)),
138	156		"X" => exit_code = body.trim().parse().ok(),
139		-	"F" => sink(rec("flow", &format!("guest init failed: {body}"))),
140		-	_ => sink(rec("console", &text)),
	157	+	"F" => sink.record(rec("flow", &format!("guest init failed: {body}"))),
	158	+	_ => sink.record(rec("console", &text)),
141	159		}
142	160		} else if !text.is_empty() {
143		-	sink(rec("console", &text));
	161	+	sink.record(rec("console", &text));
144	162		}
145	163		}
146	164		Err(mpsc::RecvTimeoutError::Timeout) => {}
147	165		Err(mpsc::RecvTimeoutError::Disconnected) => break,
	166	+	}
	167	+	if !timed_out && !cancelled && sink.cancelled() {
	168	+	cancelled = true;
	169	+	let _ = child.kill();
	170	+	sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
148	171		}
149		-	if Instant::now() >= deadline && !timed_out {
	172	+	if Instant::now() >= deadline && !timed_out && !cancelled {
150	173		timed_out = true;
151	174		let _ = child.kill();
152		-	sink(Record { t: t(&started), stream: "flow".into(), text: format!("stopped at the time limit of {} s", timeout.as_secs()) });
	175	+	sink.record(Record {
	176	+	t: t(&started),
	177	+	stream: "flow".into(),
	178	+	text: format!("stopped at the time limit of {} s", timeout.as_secs()),
	179	+	});
153	180		}
154	181		}
155	182		let status = child.wait()?;
156	183		for r in readers {
157	184		let _ = r.join();
158	185		}
159		-	if exit_code.is_none() && !timed_out {
160		-	sink(Record { t: t(&started), stream: "flow".into(), text: format!("the microVM ended without the job's exit code ({status})") });
	186	+	if exit_code.is_none() && !timed_out && !cancelled {
	187	+	sink.record(Record {
	188	+	t: t(&started),
	189	+	stream: "flow".into(),
	190	+	text: format!("the microVM ended without the job's exit code ({status})"),
	191	+	});
161	192		}
162		-	Ok(Outcome { exit_code: if timed_out { None } else { exit_code }, duration_ms: t(&started), environment, timed_out })
	193	+	Ok(Outcome {
	194	+	exit_code: if timed_out \|\| cancelled { None } else { exit_code },
	195	+	duration_ms: t(&started),
	196	+	environment,
	197	+	timed_out,
	198	+	cancelled,
	199	+	})
163	200		}
164	201
165	202		/// What `flow-init` reads from `/job.json`: argv, env, and the mounts
		…	61 unchanged lines

		…	12 unchanged lines
13	13		use std::path::PathBuf;
14	14		use std::time::{Duration, Instant};
15	15
	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 {
	42	+	Finish { exit_code: None, duration_ms: 0, receipt: None, signed_receipt: None, error: Some(CANCELLED.into()) }
	43	+	}
	44	+
16	45		pub struct Hub {
17	46		base: String,
18	47		token: String,
		…	25 unchanged lines
44	73		let text = r.into_string().unwrap_or_default();
45	74		match delays.next() {
46	75		Some(d) if code >= 500 => std::thread::sleep(*d),
	76	+	_ if code < 500 => return Err(anyhow::Error::new(Refused { code, text }).context(url)),
47	77		_ => bail!("{url}: HTTP {code}: {text}"),
48	78		}
49	79		}
		…	279 unchanged lines
329	359		finish.exit_code,
330	360		finish.error.as_deref().map(\|e\| format!(" ({e})")).unwrap_or_default()
331	361		);
332		-	report_finish(home, hub, id, &finish);
	362	+	if finish.error.as_deref() != Some(CANCELLED) {
	363	+	report_finish(home, hub, id, &finish);
	364	+	}
333	365		held.lock().unwrap().remove(&id);
334	366		let mut u = used.lock().unwrap();
335	367		u.jobs -= 1;
		…	12 unchanged lines
348	380		})
349	381		}
350	382
351		-	/// Batches records to the hub every second (or every 200 lines).
	383	+	/// Batches records to the hub every second (or every 200 lines). A quiet
	384	+	/// job's sink still asks the hub every few seconds whether the job runs,
	385	+	/// with an empty batch; once the hub answers 409 the job is not wanted.
352	386		struct LogSink<'a> {
353	387		hub: &'a Hub,
354	388		job: i64,
355	389		buf: Vec<Record>,
356	390		last: Instant,
	391	+	gone: bool,
	392	+	}
	393	+
	394	+	/// How often a quiet job asks whether it was cancelled.
	395	+	const HEARTBEAT: Duration = Duration::from_secs(5);
	396	+
	397	+	impl microvm::Sink for LogSink<'_> {
	398	+	fn record(&mut self, r: Record) {
	399	+	self.push(r)
	400	+	}
	401	+
	402	+	fn cancelled(&mut self) -> bool {
	403	+	if !self.gone && self.last.elapsed() >= HEARTBEAT {
	404	+	self.send();
	405	+	}
	406	+	self.gone
	407	+	}
357	408		}
358	409
359	410		impl LogSink<'_> {
360	411		fn push(&mut self, mut r: Record) {
	412	+	if self.gone {
	413	+	return;
	414	+	}
361	415		if r.text.len() > MAX_RECORD_TEXT {
362	416		let mut cut = MAX_RECORD_TEXT;
363	417		while !r.text.is_char_boundary(cut) {
		…	9 unchanged lines
373	427
374	428		fn flush(&mut self) {
375	429		if !self.buf.is_empty() {
376		-	if let Err(e) = self.hub.log(self.job, std::mem::take(&mut self.buf)) {
377		-	eprintln!("job {}: log: {e:#}", self.job);
378		-	}
	430	+	self.send();
	431	+	}
	432	+	self.last = Instant::now();
	433	+	}
	434	+
	435	+	/// Send what is buffered, even nothing.
	436	+	fn send(&mut self) {
	437	+	if self.gone {
	438	+	return;
	439	+	}
	440	+	match self.hub.log(self.job, std::mem::take(&mut self.buf)) {
	441	+	Ok(()) => {}
	442	+	Err(e) if job_gone(&e) => self.gone = true,
	443	+	Err(e) => eprintln!("job {}: log: {e:#}", self.job),
379	444		}
380	445		self.last = Instant::now();
381	446		}
		…	4 unchanged lines
386	451
387	452		fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
388	453		let started = Instant::now();
389		-	let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now() };
	454	+	let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false };
390	455		let note =
391	456		\|sink: &mut LogSink, text: String\| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
392	457		let heads: Vec<ChangeHash> = claim.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
		…	36 unchanged lines
429	494		workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
430	495		std::fs::remove_dir_all(scratch.join("tree"))?;
431	496		preparing.take();
	497	+	// Cancelled while it was being prepared: no microVM.
	498	+	if microvm::Sink::cancelled(&mut sink) {
	499	+	return Ok(cancelled());
	500	+	}
432	501		note(
433	502		&mut sink,
434	503		format!(
		…	11 unchanged lines
446	515		&scratch.join("workspace.ext4"),
447	516		&scratch,
448	517		Duration::from_secs(claim.timeout_s),
449		-	&mut \|r\| sink.push(r),
	518	+	&mut sink,
450	519		)?;
	520	+	if outcome.cancelled {
	521	+	return Ok(cancelled());
	522	+	}
451	523		// A passing job's artifacts, read from its disks and uploaded; a
452	524		// declared artifact that cannot be kept fails the job.
453	525		let mut outputs = vec![];
		…	240 unchanged lines
694	766		let (base, seen) = fake_hub(vec![401]);
695	767		assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
696	768		assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
	769	+	}
	770	+
	771	+	/// oxygenflow#32: a quiet job asks the hub every few seconds whether
	772	+	/// it still runs; a 409 marks it cancelled, and nothing more is sent.
	773	+	#[test]
	774	+	fn a_quiet_job_learns_that_its_run_was_cancelled() {
	775	+	use microvm::Sink;
	776	+	let n = \|seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>\| seen.load(std::sync::atomic::Ordering::SeqCst);
	777	+	let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
	778	+	let (base, seen) = fake_hub(vec![204]);
	779	+	let hub = hub_at(base);
	780	+	let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
	781	+	assert!(!sink.cancelled());
	782	+	assert!(!sink.cancelled(), "asked again only after the heartbeat");
	783	+	assert_eq!(n(&seen), 1);
	784	+
	785	+	let (base, seen) = fake_hub(vec![409]);
	786	+	let hub = hub_at(base);
	787	+	let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
	788	+	assert!(sink.cancelled());
	789	+	sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
	790	+	sink.flush();
	791	+	assert!(sink.cancelled());
	792	+	assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
697	793		}
698	794
699	795		#[test]
		…	14 unchanged lines

		…	188 unchanged lines
189	189		compiled form and its sandbox, and a job token: read access to that
190	190		repository only, valid until the job finishes or 2 hours pass. `204`
191	191		when there is nothing.
192		-	- `POST /api/v1/flow/jobs/{id}/log` appends records `{t, stream, text}`.
	192	+	- `POST /api/v1/flow/jobs/{id}/log` appends records `{t, stream, text}`. An empty batch only asks whether the job still runs; the runner sends one every 5 s while a job is quiet, and stops the job's microVM when the hub answers `409` (the run was cancelled, oxygenflow#32). A cancelled job's end is not reported.
193	193		- `POST /api/v1/flow/jobs/{id}/finish` with the exit code, duration and
194	194		receipt; a job whose dependency failed becomes `skipped`.
195	195		- Each job is also a check named `flow/<job>` on the state (hub spec §13),
		…	20 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

Stop a job when its run is cancelled 280ef951d193

Keyboard