OXYGENHUB
# 2be1fe4ad345 oxygenflow / crates / flow / src / runner.rs
66.2 KiB · Rust · last changed Artifacts between jobs: input { artifact "x" from "job"; path "/src/x" } 20 days ago source blame history raw
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
6 
use crate::home::{Home, RunnerConfig, RunnerKey};
7 
use crate::{microvm, workspace};
8 
use anyhow::{bail, Context, Result};
9 
use flow_core::api::{Claim, ClaimRequest, Decline, Finish, LogBatch, Record, Register, Registered, RunnerInfo, MAX_RECORD_TEXT};
10 
use flow_core::{scope_tree_hash, Receipt};
11 
use oxy_core::hash::ChangeHash;
12 
use oxy_core::repo::Repo;
13 
use std::path::{Path, PathBuf};
14 
use std::time::{Duration, Instant};
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 {
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 
    }
51 
}
52
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
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.
69 
    fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
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>> {
74 
        let url = format!("{}/api/v1{path}", self.base);
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) })
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
99 
    /// Hand job `job` back, saying why this runner will not run it.
100 
    pub fn decline(&self, job: i64, reason: &str) -> Result<()> {
101 
        self.post_patiently(&format!("/flow/jobs/{job}/decline"), &Decline { reason: reason.into() })?;
102 
        Ok(())
103 
    }
104
105 
    pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
106 
        self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
107 
        Ok(())
108 
    }
109
110 
    /// Upload artifact `name` of job `job` (spec §5.2); the hub keeps it as
111 
    /// a blob of the job's repository and answers with its ObjectRef.
112 
    pub fn artifact(&self, job: i64, name: &str, bytes: &[u8]) -> Result<String> {
113 
        let url = format!("{}/api/v1/flow/jobs/{job}/artifacts/{name}", self.base);
114 
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
115 
            self.agent
116 
                .post(&url)
117 
                .timeout(Duration::from_secs(600))
118 
                .set("Authorization", &format!("Bearer {}", self.token))
119 
                .set("Content-Type", "application/octet-stream")
120 
                .send_bytes(bytes)
121 
                .map_err(Box::new)
122 
        })?;
123 
        let v: serde_json::Value = r.into_json()?;
124 
        v["blob"].as_str().map(String::from).context("the hub named no blob")
125 
    }
126
127 
    /// Input `index` of job `job` (oxygenflow#58): the bytes of the
128 
    /// producing job's artifact, which the caller checks against the blob the
129 
    /// claim names.
130 
    pub fn input(&self, job: i64, index: usize) -> Result<Vec<u8>> {
131 
        let url = format!("{}/api/v1/flow/jobs/{job}/inputs/{index}", self.base);
132 
        let r = retrying(&url, &PATIENT_DELAYS, false, || {
133 
            self.agent
134 
                .get(&url)
135 
                .timeout(Duration::from_secs(600))
136 
                .set("Authorization", &format!("Bearer {}", self.token))
137 
                .call()
138 
                .map_err(Box::new)
139 
        })?;
140 
        let mut bytes = Vec::new();
141 
        let mut reader = r.into_reader();
142 
        std::io::Read::read_to_end(&mut std::io::Read::take(&mut reader, MAX_INPUT_BYTES as u64 + 1), &mut bytes)?;
143 
        if bytes.len() > MAX_INPUT_BYTES {
144 
            bail!("larger than {} MiB", MAX_INPUT_BYTES >> 20);
145 
        }
146 
        Ok(bytes)
147 
    }
148
149 
    /// Upload file `path` of report `name` of job `job` (oxygenhub#127);
150 
    /// the hub keeps it as a blob and answers with its ObjectRef. The
151 
    /// report's manifest follows through [`Hub::artifact`].
152 
    pub fn report_file(&self, job: i64, name: &str, path: &str, bytes: &[u8]) -> Result<String> {
153 
        let url = format!("{}/api/v1/flow/jobs/{job}/reports/{name}/files", self.base);
154 
        let r = retrying(&url, &PATIENT_DELAYS, true, || {
155 
            self.agent
156 
                .post(&url)
157 
                .timeout(Duration::from_secs(600))
158 
                .set("Authorization", &format!("Bearer {}", self.token))
159 
                .set("Content-Type", "application/octet-stream")
160 
                .set("X-Flow-Report-Path", path)
161 
                .send_bytes(bytes)
162 
                .map_err(Box::new)
163 
        })?;
164 
        let v: serde_json::Value = r.into_json()?;
165 
        v["blob"].as_str().map(String::from).context("the hub named no blob")
166 
    }
167
168 
    /// Report a job's end until the hub has it (oxygenhub#71): while the hub
169 
    /// cannot be reached or fails (5xx), try again after `first_wait`,
170 
    /// doubling to a minute, for up to `patience`. A refusal (4xx: the job was ended
171 
    /// already, say) is final. A result lost here leaves the job running
172 
    /// on the hub until its time limit.
173 
    pub fn finish_until_accepted(&self, job: i64, f: &Finish, patience: Duration, first_wait: Duration) -> Result<()> {
174 
        let url = format!("{}/api/v1/flow/jobs/{job}/finish", self.base);
175 
        let deadline = std::time::Instant::now() + patience;
176 
        let mut wait = first_wait;
177 
        loop {
178 
            let why = match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(f) {
179 
                Ok(_) => return Ok(()),
180 
                Err(ureq::Error::Status(code, r)) if code < 500 => {
181 
                    bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default())
182 
                }
183 
                Err(ureq::Error::Status(code, _)) => format!("HTTP {code}"),
184 
                Err(e) => e.to_string(),
185 
            };
186 
            if std::time::Instant::now() + wait > deadline {
187 
                bail!("{url}: gave up after {} minutes: {why}", patience.as_secs() / 60);
188 
            }
189 
            eprintln!("job {job}: the hub did not take the result ({why}); trying again in {} s", wait.as_secs());
190 
            std::thread::sleep(wait);
191 
            wait = (wait * 2).min(Duration::from_secs(60));
192 
        }
193 
    }
194 
}
195
196 
/// How long a runner keeps trying to report a job's end.
197 
const FINISH_PATIENCE: Duration = Duration::from_secs(2 * 3600);
198
199 
/// Where a job's result waits until the hub has it, so a runner that
200 
/// restarts still reports it.
201 
fn pending_path(home: &Home, job: i64) -> PathBuf {
202 
    home.root.join("pending").join(format!("{job}.json"))
203 
}
204
205 
/// Report `f`, keeping it on disk until the hub has it.
206 
fn report_finish(home: &Home, hub: &Hub, job: i64, f: &Finish) {
207 
    let path = pending_path(home, job);
208 
    let kept = std::fs::create_dir_all(path.parent().unwrap())
209 
        .and_then(|_| std::fs::write(&path, serde_json::to_vec(f).unwrap_or_default()))
210 
        .is_ok();
211 
    match hub.finish_until_accepted(job, f, FINISH_PATIENCE, Duration::from_secs(5)) {
212 
        Ok(()) => {}
213 
        Err(e) => eprintln!("job {job}: reporting failed: {e:#}"),
214 
    }
215 
    if kept {
216 
        let _ = std::fs::remove_file(&path);
217 
    }
218 
}
219
220 
/// Results a previous run of the runner could not report, to send again.
221 
fn report_pending(home: &Home) -> Vec<(i64, Finish)> {
222 
    let Ok(dir) = std::fs::read_dir(home.root.join("pending")) else {
223 
        return Vec::new();
224 
    };
225 
    dir.filter_map(|e| {
226 
        let p = e.ok()?.path();
227 
        let job: i64 = p.file_stem()?.to_str()?.parse().ok()?;
228 
        let f: Finish = serde_json::from_slice(&std::fs::read(&p).ok()?).ok()?;
229 
        Some((job, f))
230 
    })
231 
    .collect()
232 
}
233
234 
/// `flow runner register`: exchange a one-time registration token from
235 
/// the hub for this runner's credential, binding the runner's signing key,
236 
/// and keep hub, credential and name in runner.toml.
237 
pub fn register(home: &Home, hub: &str, token: &str, name: &str, ca: Option<PathBuf>, labels: Vec<String>) -> Result<Registered> {
238 
    if home.config.runner.is_some() && !home.root.join(crate::home::RUNNER_TOML).is_file() {
239 
        bail!("flow.toml already has a [runner] section; remove it to register this runner instead");
240 
    }
241 
    let key = home.runner_key(name)?;
242 
    let hub = hub.trim_end_matches('/').to_string();
243 
    let tls = oxy_core::protocol::tls_client_config(ca.as_deref()).map_err(|e| anyhow::anyhow!("{e}"))?;
244 
    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
245 
    let req = Register {
246 
        token: token.trim().to_string(),
247 
        name: name.to_string(),
248 
        identity: hex::encode(key.identity.encode()),
249 
        info: RunnerInfo { version: env!("OXY_VERSION_LONG").into(), ..RunnerInfo::here(labels.clone(), 1, None, None) },
250 
    };
251 
    let url = format!("{hub}/api/v1/flow/runners/register");
252 
    let got: Registered = match agent.post(&url).send_json(&req) {
253 
        Ok(r) => r.into_json()?,
254 
        Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
255 
        Err(e) => bail!("{url}: {e}"),
256 
    };
257 
    let mut text = format!(
258 
        "# Written by `flow runner register`; the credential is secret.\nhub = {hub:?}\ntoken = {:?}\nname = {:?}\n",
259 
        got.credential, got.name
260 
    );
261 
    if let Some(ca) = &ca {
262 
        text.push_str(&format!("ca = {:?}\n", ca.display().to_string()));
263 
    }
264 
    if !labels.is_empty() {
265 
        text.push_str(&format!("labels = {labels:?}\n"));
266 
    }
267 
    let path = home.root.join(crate::home::RUNNER_TOML);
268 
    {
269 
        use std::io::Write;
270 
        use std::os::unix::fs::OpenOptionsExt;
271 
        let tmp = path.with_extension("toml.tmp");
272 
        let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&tmp)?;
273 
        f.write_all(text.as_bytes())?;
274 
        f.sync_all()?;
275 
        std::fs::rename(&tmp, &path)?;
276 
    }
277 
    Ok(got)
278 
}
279
280 
/// Waits before trying a hub request again after a server error.
281 
const RETRY_DELAYS: [Duration; 3] = [Duration::from_secs(1), Duration::from_secs(2), Duration::from_secs(4)];
282
283 
/// Waits for what a job leaves behind (oxygenflow#49): three minutes in
284 
/// all, longer than a hub takes to restart.
285 
/// Check an input's bytes against the blob the producing job's receipt
286 
/// names, then write them at its path under `tree` (the workspace's `/src`).
287 
fn place_input(tree: &Path, i: &flow_core::api::ArtifactInput, bytes: &[u8]) -> Result<[u8; 32]> {
288 
    let Some(rel) = i.path.strip_prefix("/src/").filter(|r| r.split('/').all(|p| !p.is_empty() && p != "." && p != ".." && p != ".oxy"))
289 
    else {
290 
        bail!("path {:?} is not a clean path under /src", i.path);
291 
    };
292 
    let blob = Repo::blob_ref(bytes);
293 
    if blob.to_hex() != i.blob {
294 
        bail!("it is not what {}'s receipt names (blob {}, not {})", i.from, &blob.to_hex()[..12], &i.blob[..i.blob.len().min(12)]);
295 
    }
296 
    let dest = tree.join(rel);
297 
    if let Some(dir) = dest.parent() {
298 
        std::fs::create_dir_all(dir)?;
299 
    }
300 
    std::fs::write(&dest, bytes)?;
301 
    Ok(blob.0)
302 
}
303
304 
/// The largest input a job takes: the hub's artifact limit.
305 
pub const MAX_INPUT_BYTES: usize = 512 << 20;
306
307 
const PATIENT_DELAYS: [Duration; 15] = {
308 
    let mut d = [Duration::from_secs(15); 15];
309 
    d[0] = Duration::from_secs(1);
310 
    d[1] = Duration::from_secs(2);
311 
    d[2] = Duration::from_secs(4);
312 
    d[3] = Duration::from_secs(8);
313 
    d
314 
};
315
316 
/// Send a request until the hub answers, waiting `delays` in turn after a
317 
/// server error (5xx: nothing was committed) and, when `connection`, after
318 
/// the hub could not be reached at all. A refusal (4xx) is final, as a
319 
/// [`Refused`] the caller can look at.
320 
fn retrying(
321 
    url: &str,
322 
    delays: &[Duration],
323 
    connection: bool,
324 
    mut send: impl FnMut() -> Result<ureq::Response, Box<ureq::Error>>,
325 
) -> Result<ureq::Response> {
326 
    let mut delays = delays.iter();
327 
    loop {
328 
        let why = match send().map_err(|e| *e) {
329 
            Ok(r) => return Ok(r),
330 
            Err(ureq::Error::Status(code, r)) if code < 500 => {
331 
                let text = r.into_string().unwrap_or_default();
332 
                return Err(anyhow::Error::new(Refused { code, text }).context(url.to_string()));
333 
            }
334 
            Err(ureq::Error::Status(code, r)) => format!("HTTP {code}: {}", r.into_string().unwrap_or_default()),
335 
            Err(e) if connection => e.to_string(),
336 
            Err(e) => bail!("{url}: {e}"),
337 
        };
338 
        match delays.next() {
339 
            Some(d) => std::thread::sleep(*d),
340 
            None => bail!("{url}: {why}"),
341 
        }
342 
    }
343 
}
344
345 
/// What the runner may run at once (flow.toml `[runner]`): at most
346 
/// `max_jobs` jobs, whose sandboxes together fit `memory_mib` and `cpus`.
347 
#[derive(Debug, Default, PartialEq, Eq)]
348 
pub struct Budget {
349 
    pub max_jobs: usize,
350 
    pub memory_mib: Option<u32>,
351 
    pub cpus: Option<u32>,
352 
}
353
354 
/// What is in use now.
355 
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
356 
pub struct Used {
357 
    pub jobs: usize,
358 
    pub memory_mib: u32,
359 
    pub cpus: u32,
360 
}
361
362 
impl Budget {
363 
    pub fn of(cfg: &RunnerConfig) -> Budget {
364 
        Budget { max_jobs: cfg.max_jobs.unwrap_or(1).max(1), memory_mib: cfg.memory_mib, cpus: cfg.cpus }
365 
    }
366
367 
    /// The room to ask the hub for, or `None` when no job more can start.
368 
    /// A lone job may use the whole budget even if it asks for more, so a
369 
    /// large sandbox is never starved.
370 
    pub fn room(&self, used: Used) -> Option<(Option<u32>, Option<u32>)> {
371 
        if used.jobs >= self.max_jobs {
372 
            return None;
373 
        }
374 
        if used.jobs == 0 {
375 
            return Some((None, None));
376 
        }
377 
        let mem = self.memory_mib.map(|m| m.saturating_sub(used.memory_mib));
378 
        let cpus = self.cpus.map(|c| c.saturating_sub(used.cpus));
379 
        if mem == Some(0) || cpus == Some(0) {
380 
            return None;
381 
        }
382 
        Some((mem, cpus))
383 
    }
384 
}
385
386 
/// The isolation tiers this machine can honour now (spec §11.1): the
387 
/// MicroVM tier when `/dev/kvm` opens for reading and writing and
388 
/// flow.toml pins Firecracker; the syscall tier when this binary may use
389 
/// user namespaces, Landlock and seccomp and manage a delegated cgroup.
390 
/// `Err` says why there are none; the reasons for missing tiers come back
391 
/// with the ones there are.
392 
pub fn tiers(home: &Home) -> std::result::Result<(Vec<String>, Vec<String>), String> {
393 
    let mut have = Vec::new();
394 
    let mut missing = Vec::new();
395 
    let kvm = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm");
396 
    match (kvm, home.config.microvm.is_some()) {
397 
        (Ok(_), true) => have.push("microvm".to_string()),
398 
        (Ok(_), false) => missing.push("microvm: flow.toml has no [microvm] section".to_string()),
399 
        (Err(e), _) => missing.push(format!(
400 
            "microvm: /dev/kvm cannot be opened ({e}); the tier needs hardware virtualisation and this user in the kvm group"
401 
        )),
402 
    }
403 
    match crate::syscall::available() {
404 
        Ok(()) => have.push("syscall".to_string()),
405 
        Err(e) => missing.push(format!("syscall: {e}")),
406 
    }
407 
    if have.is_empty() {
408 
        return Err(format!("no isolation tier can run here: {}", missing.join("; ")));
409 
    }
410 
    Ok((have, missing))
411 
}
412
413 
/// Fetch `claim`'s image from the hub when this home lacks it
414 
/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it
415 
/// leaves it missing, and the preflight declines the job as before.
416 
fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) {
417 
    let hash = &claim.sandbox.rootfs;
418 
    if hash.is_empty() || home.rootfs(hash).is_file() || !jet::images::is_hash(hash) {
419 
        return;
420 
    }
421 
    let note = |text: String| {
422 
        eprintln!("job {}: {text}", claim.job_id);
423 
        hub.log(claim.job_id, vec![Record { t: 0, stream: "flow".into(), text }]).ok();
424 
    };
425 
    // One fetch of an image at a time: two jobs wanting it wait for one.
426 
    static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(());
427 
    let _one = FETCHING.lock().unwrap_or_else(|p| p.into_inner());
428 
    if home.rootfs(hash).is_file() {
429 
        return;
430 
    }
431 
    let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) {
432 
        Ok(c) => c,
433 
        Err(e) => return note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12])),
434 
    };
435 
    let fetched = match swarm {
436 
        Some(sw) => {
437 
            note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12]));
438 
            sw.fetch(home, hash)
439 
        }
440 
        None => {
441 
            note(format!("image {} is not on this runner: fetching it from the hub", &hash[..12]));
442 
            client.fetch(home, hash)
443 
        }
444 
    };
445 
    match fetched {
446 
        Ok(Some(m)) => note(format!(
447 
            "image {}: fetched {} chunks, {}{} in {:.1} s",
448 
            &hash[..12],
449 
            m.chunks,
450 
            crate::images::human(m.bytes),
451 
            if swarm.is_some() { format!(" ({} from peers)", crate::images::human(m.peer_bytes)) } else { String::new() },
452 
            m.elapsed.as_secs_f64()
453 
        )),
454 
        Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])),
455 
        Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])),
456 
    }
457 
}
458
459 
/// Why this runner cannot run `claim` now (spec §11.5), checked before
460 
/// anything is fetched: the tier, the architecture, the image and the
461 
/// memory. The hub chose it from what the runner last said, which may be
462 
/// out of date.
463 
fn preflight(home: &Home, claim: &Claim) -> Option<String> {
464 
    let sb = &claim.sandbox;
465 
    match sb.runtime {
466 
        flow_core::Runtime::Microvm => {
467 
            if let Err(e) = std::fs::OpenOptions::new().read(true).write(true).open("/dev/kvm") {
468 
                return Some(format!("the MicroVM tier needs /dev/kvm, which cannot be opened here ({e})"));
469 
            }
470 
            if home.config.microvm.is_none() {
471 
                return Some("the MicroVM tier is not set up here (flow.toml has no [microvm] section)".into());
472 
            }
473 
        }
474 
        flow_core::Runtime::Syscall => {
475 
            if let Err(e) = crate::syscall::available() {
476 
                return Some(format!("the syscall tier cannot run here: {e}"));
477 
            }
478 
        }
479 
        flow_core::Runtime::Wasm => return Some("this runner has no wasm tier".into()),
480 
    }
481 
    sandbox_problem(home, sb, std::env::consts::ARCH, available_memory_mib())
482 
}
483
484 
/// What stops `sb` running on this home's images, on a machine of
485 
/// architecture `here` with `free_mib` of memory free.
486 
fn sandbox_problem(home: &Home, sb: &flow_core::Sandbox, here: &str, free_mib: Option<u64>) -> Option<String> {
487 
    if let Some(want) = sb.arch.as_deref().filter(|a| *a != here) {
488 
        return Some(format!("the sandbox needs {want}; this runner is {here}"));
489 
    }
490 
    let short = &sb.rootfs[..12.min(sb.rootfs.len())];
491 
    let image = home.rootfs(&sb.rootfs);
492 
    if !image.is_file() {
493 
        return Some(format!("rootfs {short} is not on this runner"));
494 
    }
495 
    if let Some(arch) = home.image_arch(&sb.rootfs).filter(|a| a != here) {
496 
        return Some(format!("rootfs {short} is for {arch}; this runner is {here}"));
497 
    }
498 
    if let Err(e) = home.verified(&image, &sb.rootfs) {
499 
        return Some(format!("rootfs {short} does not verify here: {e:#}"));
500 
    }
501 
    if let Some(free) = free_mib.filter(|free| *free < u64::from(sb.memory_mib)) {
502 
        return Some(format!("only {free} MiB of memory is free here; the sandbox needs {} MiB", sb.memory_mib));
503 
    }
504 
    None
505 
}
506
507 
/// `MemAvailable` from /proc/meminfo, in MiB.
508 
fn available_memory_mib() -> Option<u64> {
509 
    let info = std::fs::read_to_string("/proc/meminfo").ok()?;
510 
    let kib: u64 = info.lines().find_map(|l| l.strip_prefix("MemAvailable:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;
511 
    Some(kib / 1024)
512 
}
513
514 
pub fn serve(home: &Home) -> Result<()> {
515 
    let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
516 
    // A runner offers only what it can run: a machine without KVM claims
517 
    // nothing rather than taking MicroVM jobs and failing them (#36).
518 
    let (runtimes, missing) = tiers(home).map_err(|why| anyhow::anyhow!("flow runner: {why}"))?;
519 
    for why in &missing {
520 
        eprintln!("flow runner: not offering {why}");
521 
    }
522 
    let hub = Hub::new(&cfg)?;
523 
    let name = cfg.name.clone().unwrap_or_else(|| "flow".into());
524 
    let budget = Budget::of(&cfg);
525 
    drain_on_stop();
526 
    let key = home.runner_key(&name)?;
527 
    // The swarm (docs/jetstream.md §8.1): a runner that cannot serve its
528 
    // peers still runs jobs, fetching images from the hub alone.
529 
    let swarm = crate::swarm::Swarm::start(home, &cfg).unwrap_or_else(|e| {
530 
        eprintln!("flow runner: not joining the swarm: {e:#}");
531 
        None
532 
    });
533 
    eprintln!(
534 
        "flow runner {name}: polling {} (up to {} jobs at once, tiers {}), signing receipts as {}",
535 
        cfg.hub,
536 
        budget.max_jobs,
537 
        runtimes.join(", "),
538 
        key.identity.hash().to_hex()
539 
    );
540 
    let used = std::sync::Mutex::new(Used::default());
541 
    let mut idle = Duration::from_secs(2);
542 
    let pending = report_pending(home);
543 
    // The jobs held now, told to the hub at each claim so it can end the
544 
    // ones this runner lost: a job is held from its claim until its result
545 
    // is reported.
546 
    let held = std::sync::Mutex::new(std::collections::BTreeSet::<i64>::new());
547 
    held.lock().unwrap().extend(pending.iter().map(|(job, _)| *job));
548 
    std::thread::scope(|scope| {
549 
        for (job, f) in pending {
550 
            let (hub, held) = (&hub, &held);
551 
            scope.spawn(move || {
552 
                eprintln!("job {job}: reporting the result kept from before the restart");
553 
                report_finish(home, hub, job, &f);
554 
                held.lock().unwrap().remove(&job);
555 
            });
556 
        }
557 
        // The speed (spec §11.1): measured now, and again every 15 minutes
558 
        // while no job runs, so a job never shares the CPU with it.
559 
        let measure = || {
560 
            let (speed, ms) = crate::facts::bench();
561 
            eprintln!("flow runner {name}: speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
562 
            speed
563 
        };
564 
        let mut speed = measure();
565 
        let mut benched = Instant::now();
566 
        let mut said_draining = false;
567 
        loop {
568 
            // Draining (oxygenflow#50): told to stop, claim nothing more and
569 
            // stop once every job held is done and reported.
570 
            if stopping() {
571 
                let (jobs, holding) = (used.lock().unwrap().jobs, held.lock().unwrap().len());
572 
                if jobs == 0 && holding == 0 {
573 
                    eprintln!("flow runner {name}: drained, stopping");
574 
                    break;
575 
                }
576 
                if !said_draining {
577 
                    eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)");
578 
                    said_draining = true;
579 
                }
580 
                std::thread::sleep(Duration::from_millis(500));
581 
                continue;
582 
            }
583 
            if used.lock().unwrap().jobs == 0 && benched.elapsed() >= Duration::from_secs(15 * 60) {
584 
                speed = measure();
585 
                benched = Instant::now();
586 
            }
587 
            let room = budget.room(*used.lock().unwrap());
588 
            let Some((free_memory_mib, free_cpus)) = room else {
589 
                std::thread::sleep(Duration::from_millis(500));
590 
                continue;
591 
            };
592 
            let req = ClaimRequest {
593 
                runner: name.clone(),
594 
                runtimes: runtimes.clone(),
595 
                free_memory_mib,
596 
                free_cpus,
597 
                info: Some(RunnerInfo {
598 
                    version: env!("OXY_VERSION_LONG").into(),
599 
                    ..RunnerInfo::here(cfg.offered_labels(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)
600 
                }),
601 
                // Read each time: an image imported while the runner runs counts.
602 
                images: Some(home.images()),
603 
                image_archs: Some(home.image_archs()),
604 
                holding: Some(held.lock().unwrap().iter().copied().collect()),
605 
                fetches_images: true,
606 
                facts: Some(crate::facts::now(home, Some(speed))),
607 
                swarm: swarm.as_ref().map(|s| s.addr.clone()),
608 
            };
609 
            match hub.claim(&req) {
610 
                Ok(Some(claim)) => {
611 
                    idle = Duration::from_secs(2);
612 
                    let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);
613 
                    {
614 
                        let mut u = used.lock().unwrap();
615 
                        u.jobs += 1;
616 
                        u.memory_mib += mem;
617 
                        u.cpus += cpus;
618 
                    }
619 
                    held.lock().unwrap().insert(claim.job_id);
620 
                    let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref());
621 
                    scope.spawn(move || {
622 
                        let id = claim.job_id;
623 
                        fetch_image(home, hub, cfg, swarm, &claim);
624 
                        if let Some(why) = preflight(home, &claim) {
625 
                            eprintln!("job {id}: declined: {why}");
626 
                            if let Err(e) = hub.decline(id, &why) {
627 
                                eprintln!("job {id}: decline: {e:#}");
628 
                            }
629 
                            held.lock().unwrap().remove(&id);
630 
                            let mut u = used.lock().unwrap();
631 
                            u.jobs -= 1;
632 
                            u.memory_mib -= mem;
633 
                            u.cpus -= cpus;
634 
                            return;
635 
                        }
636 
                        eprintln!(
637 
                            "job {id}: {} {} on {}",
638 
                            claim.repo,
639 
                            claim.job.name,
640 
                            claim.heads.first().map(|h| &h[..12]).unwrap_or("-")
641 
                        );
642 
                        let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(|e| Finish {
643 
                            exit_code: None,
644 
                            duration_ms: 0,
645 
                            receipt: None,
646 
                            signed_receipt: None,
647 
                            error: Some(format!("{e:#}")),
648 
                            peak_disk_mib: None,
649 
                            peak_memory_mib: None,
650 
                        });
651 
                        eprintln!(
652 
                            "job {id}: exit {:?}{}",
653 
                            finish.exit_code,
654 
                            finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()
655 
                        );
656 
                        if finish.error.as_deref() != Some(CANCELLED) {
657 
                            report_finish(home, hub, id, &finish);
658 
                        }
659 
                        held.lock().unwrap().remove(&id);
660 
                        let mut u = used.lock().unwrap();
661 
                        u.jobs -= 1;
662 
                        u.memory_mib -= mem;
663 
                        u.cpus -= cpus;
664 
                    });
665 
                }
666 
                Ok(None) => std::thread::sleep(idle),
667 
                Err(e) => {
668 
                    eprintln!("claim: {e:#}");
669 
                    std::thread::sleep(idle);
670 
                    idle = (idle * 2).min(Duration::from_secs(60));
671 
                }
672 
            }
673 
        }
674 
        Ok(())
675 
    })
676 
}
677
678 
/// How many times the runner was told to stop (SIGTERM or SIGINT).
679 
static STOPS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
680
681 
extern "C" fn on_stop(_: libc::c_int) {
682 
    // A second stop means now: only async-signal-safe calls here.
683 
    if STOPS.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= 1 {
684 
        unsafe { libc::_exit(143) };
685 
    }
686 
}
687
688 
/// Drain instead of dying on the first SIGTERM or SIGINT (oxygenflow#50),
689 
/// so restarting a runner never kills the job it runs: the unit's
690 
/// `KillMode=mixed` sends the signal to the runner alone, not its microVMs.
691 
fn drain_on_stop() {
692 
    let handler = on_stop as extern "C" fn(libc::c_int) as libc::sighandler_t;
693 
    unsafe {
694 
        libc::signal(libc::SIGTERM, handler);
695 
        libc::signal(libc::SIGINT, handler);
696 
    }
697 
}
698
699 
fn stopping() -> bool {
700 
    STOPS.load(std::sync::atomic::Ordering::SeqCst) > 0
701 
}
702
703 
/// Batches records to the hub every second (or every 200 lines). A quiet
704 
/// job's sink still asks the hub every few seconds whether the job runs,
705 
/// with an empty batch; once the hub answers 409 the job is not wanted.
706 
struct LogSink<'a> {
707 
    hub: &'a Hub,
708 
    job: i64,
709 
    buf: Vec<Record>,
710 
    last: Instant,
711 
    gone: bool,
712 
}
713
714 
/// The most log lines sent at once, as a busy job's sink flushes them.
715 
const BATCH: usize = 200;
716
717 
/// The most log lines kept while the hub cannot be reached.
718 
const MAX_UNSENT: usize = 50_000;
719
720 
/// How often a quiet job asks whether it was cancelled.
721 
const HEARTBEAT: Duration = Duration::from_secs(5);
722
723 
impl microvm::Sink for LogSink<'_> {
724 
    fn record(&mut self, r: Record) {
725 
        self.push(r)
726 
    }
727
728 
    fn cancelled(&mut self) -> bool {
729 
        if !self.gone && self.last.elapsed() >= HEARTBEAT {
730 
            self.send();
731 
        }
732 
        self.gone
733 
    }
734 
}
735
736 
impl LogSink<'_> {
737 
    fn push(&mut self, mut r: Record) {
738 
        if self.gone {
739 
            return;
740 
        }
741 
        if r.text.len() > MAX_RECORD_TEXT {
742 
            let mut cut = MAX_RECORD_TEXT;
743 
            while !r.text.is_char_boundary(cut) {
744 
                cut -= 1;
745 
            }
746 
            r.text.truncate(cut);
747 
        }
748 
        self.buf.push(r);
749 
        if self.buf.len() >= BATCH || self.last.elapsed() >= Duration::from_secs(1) {
750 
            self.flush();
751 
        }
752 
    }
753
754 
    fn flush(&mut self) {
755 
        if !self.buf.is_empty() {
756 
            self.send();
757 
        }
758 
        self.last = Instant::now();
759 
    }
760
761 
    /// Send what is buffered, even nothing, in batches of at most
762 
    /// `BATCH` lines.
763 
    fn send(&mut self) {
764 
        loop {
765 
            if self.gone {
766 
                return;
767 
            }
768 
            let n = self.buf.len().min(BATCH);
769 
            let batch: Vec<Record> = self.buf.drain(..n).collect();
770 
            match self.hub.log(self.job, batch.clone()) {
771 
                Ok(()) => {}
772 
                Err(e) if job_gone(&e) => self.gone = true,
773 
                Err(e) => {
774 
                    // The hub is away (a restart, say): keep the lines for
775 
                    // the next batch, up to a bound, so the log loses none
776 
                    // (oxygenflow#49).
777 
                    eprintln!("job {}: log: {e:#}; kept for the next batch", self.job);
778 
                    let mut kept = batch;
779 
                    kept.append(&mut self.buf);
780 
                    let over = kept.len().saturating_sub(MAX_UNSENT);
781 
                    if over > 0 {
782 
                        kept.drain(..over);
783 
                        kept.insert(
784 
                            0,
785 
                            Record {
786 
                                t: 0,
787 
                                stream: "flow".into(),
788 
                                text: format!("({over} log lines lost while the hub could not be reached)"),
789 
                            },
790 
                        );
791 
                    }
792 
                    self.buf = kept;
793 
                    break;
794 
                }
795 
            }
796 
            if self.buf.is_empty() {
797 
                break;
798 
            }
799 
        }
800 
        self.last = Instant::now();
801 
    }
802 
}
803
804 
/// Held while a job's workspace is prepared (see `run_claim`).
805 
static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(());
806
807 
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> {
808 
    let started = Instant::now();
809 
    let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now(), gone: false };
810 
    let note =
811 
        |sink: &mut LogSink, text: String| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
812 
    let heads: Vec<ChangeHash> = claim.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
813 
    note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+")));
814 
    // Preparing a workspace (the cached repositories, the export, the
815 
    // image) is one job at a time: fetches into a cached repository must not
816 
    // overlap, and the first clone passes its token through the process
817 
    // environment. Only the microVMs run side by side.
818 
    let mut preparing = Some(PREPARE.lock().unwrap_or_else(|p| p.into_inner()));
819 
    let repo = cached_repo(home, cfg, claim)?;
820 
    let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
821 
    let _ = std::fs::remove_dir_all(&scratch);
822 
    std::fs::create_dir_all(scratch.join("tree"))?;
823 
    let result = (|| -> Result<Finish> {
824 
        let mut files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
825 
        // Pinned subrepos (Oxygen spec §7.8), at their paths, as part of the
826 
        // workspace and of the scope hash.
827 
        for sub in &claim.subrepos {
828 
            if sub.path.split('/').any(|p| p.is_empty() || p == "." || p == ".." || p == ".oxy") {
829 
                bail!("subrepo path {:?} is not a clean relative path", sub.path);
830 
            }
831 
            note(
832 
                &mut sink,
833 
                format!(
834 
                    "fetching subrepo {} at {} for {}",
835 
                    sub.repo,
836 
                    sub.heads.iter().map(|h| &h[..12]).collect::<Vec<_>>().join("+"),
837 
                    sub.path
838 
                ),
839 
            );
840 
            let sub_repo = cached_repo_at(home, cfg, &sub.repo, &sub.clone_url, &claim.token)?;
841 
            let sub_heads: Vec<ChangeHash> =
842 
                sub.heads.iter().map(|h| ChangeHash::from_hex(h).context("bad subrepo head")).collect::<Result<_>>()?;
843 
            let dir = scratch.join("tree").join(&sub.path);
844 
            std::fs::create_dir_all(&dir)?;
845 
            let sub_files = workspace::export_state(&sub_repo, &sub_heads, &dir)?;
846 
            files.extend(sub_files.into_iter().map(|(p, h)| (format!("{}/{p}", sub.path), h)));
847 
        }
848 
        let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(|(p, h)| (p.as_str(), *h)));
849 
        // Inputs (oxygenflow#58): each producing job's artifact, checked
850 
        // against the blob its receipt names, then put at its path. Not part
851 
        // of the scope: they are the run's, not the repository's.
852 
        let mut input_blobs = Vec::new();
853 
        for (index, i) in claim.inputs.iter().enumerate() {
854 
            let what = format!("input {} from {}", i.artifact, i.from);
855 
            let bytes = hub.input(claim.job_id, index).with_context(|| format!("{what}: fetching it"))?;
856 
            let blob = place_input(&scratch.join("tree"), i, &bytes).with_context(|| what.clone())?;
857 
            note(&mut sink, format!("{what}: {} bytes at {}, blob {} as its receipt names", bytes.len(), i.path, &i.blob[..12]));
858 
            input_blobs.push(blob);
859 
        }
860 
        // The syscall tier runs on the tree itself; a microVM gets an image.
861 
        let syscall_tier = claim.sandbox.runtime == flow_core::Runtime::Syscall;
862 
        if !syscall_tier {
863 
            workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
864 
            std::fs::remove_dir_all(scratch.join("tree"))?;
865 
        }
866 
        preparing.take();
867 
        // Cancelled while it was being prepared: no sandbox.
868 
        if microvm::Sink::cancelled(&mut sink) {
869 
            return Ok(cancelled());
870 
        }
871 
        note(
872 
            &mut sink,
873 
            format!(
874 
                "{}: {} {}, {} MiB, rootfs {}",
875 
                if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
876 
                claim.sandbox.cpus,
877 
                if syscall_tier { "CPU" } else { "vCPU" },
878 
                claim.sandbox.memory_mib,
879 
                &claim.sandbox.rootfs[..12]
880 
            ),
881 
        );
882 
        let job = claim.job.in_sandbox(&claim.sandbox).with_flow_env(&claim.flow_env());
883 
        let timeout = Duration::from_secs(claim.timeout_s);
884 
        // Disk mounts that name a cache start from a kept disk (#28); the
885 
        // microVM takes a disk already in place as it is.
886 
        let pool = crate::cache::Pool::new(&home.root, cfg.cache_gb);
887 
        let cache_ctx =
888 
            crate::cache::Ctx { repo: &claim.repo, bookmark: &claim.bookmark, job: &claim.job.name, image: &claim.sandbox.rootfs };
889 
        let mut cached: Vec<(usize, String, String)> = Vec::new();
890 
        if !syscall_tier {
891 
            for (i, m) in job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).enumerate() {
892 
                let Some(name) = &m.cache else { continue };
893 
                let key = crate::cache::Pool::key(&cache_ctx, m, |p| files.iter().find(|(f, _)| f == p).map(|(_, h)| hex::encode(h)));
894 
                let how = pool.take(&cache_ctx, m, &key, &scratch.join(format!("mount-{i}.ext4")));
895 
                let how = how.unwrap_or_else(|| format!("cache {name}: none kept yet, an empty disk"));
896 
                note(&mut sink, format!("{} ({})", how, m.target));
897 
                cached.push((i, key, how));
898 
            }
899 
        }
900 
        let mut outcome = if syscall_tier {
901 
            let group = format!("job-{}", claim.job_id);
902 
            crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
903 
        } else {
904 
            microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
905 
        };
906 
        if outcome.cancelled {
907 
            return Ok(cancelled());
908 
        }
909 
        // The receipt says what a cached disk started from.
910 
        let disk_mounts: Vec<&flow_core::pipeline::Mount> =
911 
            job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).collect();
912 
        for (i, key, how) in &cached {
913 
            outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
914 
        }
915 
        // A passing job's artifacts, read from its disks and uploaded; a
916 
        // declared artifact that cannot be kept fails the job. A failing
917 
        // job keeps only its `when "always"` ones (oxygenflow#37), outside
918 
        // its receipt, and one it cannot keep is only noted.
919 
        let mut outputs = vec![];
920 
        let mut output_error = None;
921 
        let passed = outcome.exit_code == Some(0);
922 
        let wanted = claim.job.outputs.iter().filter(|o| passed || o.always);
923 
        for o in wanted {
924 
            if o.kind == flow_core::OutputKind::Report {
925 
                let read = if syscall_tier {
926 
                    let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
927 
                    crate::outputs::read_report_dir(&dir, &inner)
928 
                } else {
929 
                    let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
930 
                    crate::outputs::read_report(&image, &inner, &scratch)
931 
                };
932 
                let kept = read.and_then(|files| upload_report(hub, claim.job_id, &o.name, &files));
933 
                match kept {
934 
                    Ok((blob, files, bytes)) => {
935 
                        let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
936 
                        let hex = blob.to_hex();
937 
                        note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
938 
                        if passed {
939 
                            outputs.push(blob.0);
940 
                        }
941 
                    }
942 
                    Err(e) if passed => {
943 
                        output_error = Some(format!("report {}: {e:#}", o.name));
944 
                        break;
945 
                    }
946 
                    Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
947 
                }
948 
                continue;
949 
            }
950 
            let read = if syscall_tier {
951 
                let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
952 
                crate::outputs::read_dir_file(&dir, &inner)
953 
            } else {
954 
                let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
955 
                crate::outputs::read(&image, &inner, &scratch)
956 
            };
957 
            let kept = read.and_then(|bytes| {
958 
                let blob = Repo::blob_ref(&bytes);
959 
                let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
960 
                if named != blob.to_hex() {
961 
                    bail!("the hub kept it as {named}, not {}", blob.to_hex());
962 
                }
963 
                let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
964 
                note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
965 
                Ok(blob.0)
966 
            });
967 
            match kept {
968 
                Ok(b) if passed => outputs.push(b),
969 
                Ok(_) => {}
970 
                Err(e) if passed => {
971 
                    output_error = Some(format!("artifact {}: {e:#}", o.name));
972 
                    break;
973 
                }
974 
                Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
975 
            }
976 
        }
977 
        if let Some(e) = &output_error {
978 
            note(&mut sink, e.clone());
979 
        }
980 
        let receipt = outcome.exit_code.filter(|_| output_error.is_none()).map(|code| Receipt {
981 
            job_name: claim.job.name.clone(),
982 
            pipeline_hash: hex32(&claim.pipeline_hash),
983 
            scope_tree_hash: scope_hash,
984 
            inputs: input_blobs.clone(),
985 
            environment: outcome.environment.hash(),
986 
            outputs,
987 
            exit_code: code as u32,
988 
            duration_ms: outcome.duration_ms,
989 
        });
990 
        let signed_receipt = match &receipt {
991 
            Some(r) => Some(hex::encode(sign_receipt(&repo, &heads, cfg, key, claim, r)?.encode())),
992 
            None => None,
993 
        };
994 
        // Measured before a cached disk leaves the scratch directory.
995 
        let peak_disk_mib = crate::facts::allocated_mib(&scratch);
996 
        // A passing job's cached disks are kept; a failing job's are not,
997 
        // so nothing it wrote reaches a later job.
998 
        if passed && output_error.is_none() {
999 
            for (i, key, _) in &cached {
1000 
                if let Err(e) = pool.keep(&cache_ctx, disk_mounts[*i], key, &scratch.join(format!("mount-{i}.ext4"))) {
1001 
                    note(&mut sink, format!("cache: not kept: {e:#}"));
1002 
                }
1003 
            }
1004 
        }
1005 
        Ok(Finish {
1006 
            exit_code: outcome.exit_code,
1007 
            duration_ms: started.elapsed().as_millis() as u64,
1008 
            receipt,
1009 
            signed_receipt,
1010 
            error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())),
1011 
            // What the job took (oxygenflow#41): its disks are sparse files
1012 
            // in the scratch directory, so their blocks now are their peak.
1013 
            peak_disk_mib,
1014 
            peak_memory_mib: outcome.peak_memory_mib,
1015 
        })
1016 
    })();
1017 
    let _ = workspace::remove_tree(&scratch);
1018 
    sink.flush();
1019 
    result
1020 
}
1021
1022 
/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
1023 
/// runner (Oxygen spec §14.6). The state is computed here, not taken from
1024 
/// the claim, so the runner vouches only for what it fetched.
1025 
/// Upload a report's files, then its manifest under the report's name, and
1026 
/// check the hub kept each as the blob it should (oxygenhub#127). Answers
1027 
/// the manifest's blob, the file count and their total size.
1028 
fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
1029 
    let mut listed = Vec::with_capacity(files.len());
1030 
    let mut total = 0u64;
1031 
    for (path, bytes) in files {
1032 
        let blob = Repo::blob_ref(bytes).to_hex();
1033 
        let named = hub.report_file(job, name, path, bytes).with_context(|| format!("uploading {path}"))?;
1034 
        if named != blob {
1035 
            bail!("the hub kept {path} as {named}, not {blob}");
1036 
        }
1037 
        total += bytes.len() as u64;
1038 
        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
1039 
    }
1040 
    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
1041 
    let blob = Repo::blob_ref(&manifest);
1042 
    let named = hub.artifact(job, name, &manifest)?;
1043 
    if named != blob.to_hex() {
1044 
        bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
1045 
    }
1046 
    Ok((blob, files.len(), total))
1047 
}
1048
1049 
fn sign_receipt(
1050 
    repo: &Repo,
1051 
    heads: &[ChangeHash],
1052 
    cfg: &RunnerConfig,
1053 
    key: &RunnerKey,
1054 
    claim: &Claim,
1055 
    receipt: &Receipt,
1056 
) -> Result<oxy_core::receipt::SignedReceipt> {
1057 
    let state = state_of(heads);
1058 
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
1059 
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
1060 
}
1061
1062 
/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
1063 
pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
1064 
    use oxy_core::hash::{hash_object, ObjectType};
1065 
    hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
1066 
}
1067
1068 
fn hex32(s: &str) -> [u8; 32] {
1069 
    let mut out = [0; 32];
1070 
    if let Ok(v) = hex::decode(s) {
1071 
        if v.len() == 32 {
1072 
            out.copy_from_slice(&v);
1073 
        }
1074 
    }
1075 
    out
1076 
}
1077
1078 
/// Whether a network error is worth another try: DNS, a refused or reset
1079 
/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
1080 
/// hub restarts (oxygenhub#111).
1081 
fn transient(msg: &str) -> bool {
1082 
    let m = msg.to_lowercase();
1083 
    [
1084 
        "dns failed",
1085 
        "name resolution",
1086 
        "connection refused",
1087 
        "connection reset",
1088 
        "timed out",
1089 
        "timeout",
1090 
        "http 502",
1091 
        "http 503",
1092 
        "http 504",
1093 
        "broken pipe",
1094 
        "unexpected eof",
1095 
    ]
1096 
    .iter()
1097 
    .any(|k| m.contains(k))
1098 
}
1099
1100 
/// Run `f` again after a transient network error, waiting `waits` between
1101 
/// tries (about a minute in all), and give up on any other error at once.
1102 
fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
1103 
    let mut tries = waits.iter();
1104 
    loop {
1105 
        match f() {
1106 
            Ok(v) => return Ok(v),
1107 
            Err(e) => match tries.next() {
1108 
                Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
1109 
                _ => return Err(e),
1110 
            },
1111 
        }
1112 
    }
1113 
}
1114
1115 
/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
1116 
const FETCH_WAITS: [Duration; 5] =
1117 
    [Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
1118
1119 
/// The runner's copy of a repository, fetched up to date with the job's token.
1120 
fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
1121 
    cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
1122 
}
1123
1124 
/// The runner's copy of `repo`, cloned or fetched with `token`.
1125 
fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
1126 
    let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
1127 
    let set_remote = |repo: &mut Repo| -> Result<()> {
1128 
        repo.config.set("remote.origin.url", clone_url).map_err(|e| anyhow::anyhow!("{e}"))?;
1129 
        repo.config.set("remote.origin.token", token).map_err(|e| anyhow::anyhow!("{e}"))?;
1130 
        if let Some(ca) = &cfg.ca {
1131 
            repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(|e| anyhow::anyhow!("{e}"))?;
1132 
        }
1133 
        repo.save_config().map_err(|e| anyhow::anyhow!("{e}"))
1134 
    };
1135 
    if dir.join(".oxy").is_dir() {
1136 
        let mut repo = Repo::open(&dir).map_err(|e| anyhow::anyhow!("{e}"))?;
1137 
        set_remote(&mut repo)?;
1138 
        with_retries(&FETCH_WAITS, || repo.fetch("origin").map(|_| ()).map_err(|e| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
1139 
        return Ok(repo);
1140 
    }
1141 
    // First job for this repository: clone with the job's token.
1142 
    std::env::set_var("OXY_TOKEN", token);
1143 
    if let Some(ca) = &cfg.ca {
1144 
        std::env::set_var("OXY_CA", ca);
1145 
    }
1146 
    let cloned = with_retries(&FETCH_WAITS, || {
1147 
        let _ = std::fs::remove_dir_all(&dir);
1148 
        Repo::clone(clone_url, &dir).map_err(|e| anyhow::anyhow!("{e}"))
1149 
    });
1150 
    std::env::remove_var("OXY_TOKEN");
1151 
    std::env::remove_var("OXY_CA");
1152 
    let mut repo = cloned.map_err(|e| {
1153 
        let _ = std::fs::remove_dir_all(&dir);
1154 
        anyhow::anyhow!("clone {}: {e}", clone_url)
1155 
    })?;
1156 
    set_remote(&mut repo)?;
1157 
    Ok(repo)
1158 
}
1159
1160 
#[cfg(test)]
1161 
mod tests {
1162
1163 
    #[test]
1164 
    fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
1165 
        let dir = tempfile::tempdir().unwrap();
1166 
        let bytes = b"the reviewed plan";
1167 
        let blob = Repo::blob_ref(bytes).to_hex();
1168 
        let input = |path: &str, blob: &str| flow_core::api::ArtifactInput {
1169 
            artifact: "tfplan".into(),
1170 
            from: "plan".into(),
1171 
            path: path.into(),
1172 
            blob: blob.into(),
1173 
        };
1174 
        let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
1175 
        assert_eq!(hex::encode(got), blob);
1176 
        assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
1177 
        let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
1178 
        assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
1179 
        assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
1180 
        for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
1181 
            assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
1182 
        }
1183 
    }
1184
1185 
    use super::*;
1186
1187 
    #[test]
1188 
    fn a_runner_names_what_stops_a_sandbox() {
1189 
        let dir = tempfile::tempdir().unwrap();
1190 
        let home = Home::locate(Some(dir.path())).unwrap();
1191 
        let bytes = b"not really ext4";
1192 
        let hash = blake3::hash(bytes).to_hex().to_string();
1193 
        let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
1194 
            "name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
1195 
        }))
1196 
        .unwrap();
1197 
        let problem = |sb: &flow_core::Sandbox, free| sandbox_problem(&home, sb, "x86_64", free);
1198 
        assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
1199 
        std::fs::write(home.rootfs(&hash), bytes).unwrap();
1200 
        sb.rootfs = hash.clone();
1201 
        assert_eq!(problem(&sb, Some(4096)), None);
1202 
        assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
1203 
        std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
1204 
        assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
1205 
        sb.arch = Some("aarch64".into());
1206 
        assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
1207 
        sb.arch = None;
1208 
        std::fs::remove_file(home.image_manifest(&hash)).unwrap();
1209 
        std::fs::write(home.rootfs(&hash), b"changed").unwrap();
1210 
        assert!(problem(&sb, None).unwrap().contains("does not verify"));
1211 
    }
1212
1213 
    #[test]
1214 
    fn transient_network_errors_are_tried_again_and_others_are_not() {
1215 
        assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
1216 
        assert!(transient("HTTP 503: the hub is restarting"));
1217 
        assert!(!transient("HTTP 404: no such repository"));
1218 
        let mut n = 0;
1219 
        let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], || {
1220 
            n += 1;
1221 
            if n < 3 {
1222 
                Err(anyhow::anyhow!("connection refused"))
1223 
            } else {
1224 
                Ok(7)
1225 
            }
1226 
        });
1227 
        assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
1228 
        let mut m = 0;
1229 
        let got: Result<u32> = with_retries(&[Duration::ZERO; 4], || {
1230 
            m += 1;
1231 
            Err(anyhow::anyhow!("HTTP 403: forbidden"))
1232 
        });
1233 
        assert!(got.is_err());
1234 
        assert_eq!(m, 1, "a refusal is not tried again");
1235 
        let mut k = 0;
1236 
        let got: Result<u32> = with_retries(&[Duration::ZERO; 2], || {
1237 
            k += 1;
1238 
            Err(anyhow::anyhow!("timed out"))
1239 
        });
1240 
        assert!(got.is_err());
1241 
        assert_eq!(k, 3, "gives up after the last wait");
1242 
    }
1243
1244 
    /// A hub that answers each request with the next of `codes`, counting them.
1245 
    fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
1246 
        fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
1247 
    }
1248
1249 
    /// A hub at `listener` answering each request with the next of `codes`.
1250 
    fn fake_hub_on(listener: std::net::TcpListener, codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
1251 
        use std::io::{Read, Write};
1252 
        let addr = listener.local_addr().unwrap();
1253 
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
1254 
        let count = seen.clone();
1255 
        std::thread::spawn(move || {
1256 
            for code in codes {
1257 
                let (mut s, _) = listener.accept().unwrap();
1258 
                let mut buf = [0u8; 8192];
1259 
                let _ = s.read(&mut buf);
1260 
                count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1261 
                let reason = if code == 204 { "No Content" } else { "Error" };
1262 
                let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
1263 
            }
1264 
        });
1265 
        (format!("http://{addr}"), seen)
1266 
    }
1267
1268 
    /// A hub that answers one request with `body` (status 200) and hands
1269 
    /// back what it was sent.
1270 
    fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
1271 
        use std::io::{Read, Write};
1272 
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
1273 
        let addr = listener.local_addr().unwrap();
1274 
        let (tx, rx) = std::sync::mpsc::channel();
1275 
        std::thread::spawn(move || {
1276 
            let (mut s, _) = listener.accept().unwrap();
1277 
            let mut req = Vec::new();
1278 
            let mut buf = [0u8; 65536];
1279 
            loop {
1280 
                let n = s.read(&mut buf).unwrap_or(0);
1281 
                if n == 0 {
1282 
                    break;
1283 
                }
1284 
                req.extend_from_slice(&buf[..n]);
1285 
                let text = String::from_utf8_lossy(&req);
1286 
                if let Some(end) = text.find("\r\n\r\n") {
1287 
                    let len = text[..end]
1288 
                        .lines()
1289 
                        .find_map(|l| {
1290 
                            l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
1291 
                        })
1292 
                        .unwrap_or(0);
1293 
                    if req.len() >= end + 4 + len {
1294 
                        break;
1295 
                    }
1296 
                }
1297 
            }
1298 
            tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
1299 
            let _ = write!(
1300 
                s,
1301 
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
1302 
                body.len()
1303 
            );
1304 
        });
1305 
        (format!("http://{addr}"), rx)
1306 
    }
1307
1308 
    #[test]
1309 
    fn registering_binds_the_key_and_keeps_the_credential_private() {
1310 
        use std::os::unix::fs::PermissionsExt;
1311 
        let dir = tempfile::tempdir().unwrap();
1312 
        let home = Home::locate(Some(dir.path())).unwrap();
1313 
        let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
1314 
        let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
1315 
        assert_eq!((got.id, got.name.as_str()), (7, "plex"));
1316 
        let req = sent.recv().unwrap();
1317 
        assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
1318 
        let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
1319 
        assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
1320 
        let key = home.runner_key("plex").unwrap();
1321 
        assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
1322 
        let path = dir.path().join(crate::home::RUNNER_TOML);
1323 
        assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
1324 
        // The runner then starts from runner.toml.
1325 
        let again = Home::locate(Some(dir.path())).unwrap();
1326 
        let cfg = again.config.runner.unwrap();
1327 
        assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
1328 
        assert_eq!(cfg.labels, ["gpu"]);
1329 
    }
1330
1331 
    fn hub_at(base: String) -> Hub {
1332 
        Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
1333 
    }
1334
1335 
    /// oxygenhub#71: a job's result is offered until the hub takes it,
1336 
    /// through server errors, and a refusal ends the trying at once.
1337 
    #[test]
1338 
    fn a_result_is_reported_until_the_hub_takes_it() {
1339 
        let f = Finish {
1340 
            exit_code: Some(0),
1341 
            duration_ms: 1,
1342 
            receipt: None,
1343 
            signed_receipt: None,
1344 
            error: None,
1345 
            peak_disk_mib: None,
1346 
            peak_memory_mib: None,
1347 
        };
1348 
        let quick = Duration::from_millis(10);
1349 
        let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
1350 
        hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
1351 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
1352 
        let (base, seen) = fake_hub(vec![409]);
1353 
        assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
1354 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1355 
    }
1356
1357 
    #[test]
1358 
    fn server_errors_are_tried_again_and_client_errors_are_not() {
1359 
        let quick = [Duration::from_millis(10); 3];
1360 
        let (base, seen) = fake_hub(vec![500, 503, 204]);
1361 
        let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
1362 
        assert!(r.is_none());
1363 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
1364
1365 
        let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
1366 
        let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
1367 
        assert!(e.contains("HTTP 500"), "{e}");
1368 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
1369
1370 
        let (base, seen) = fake_hub(vec![401]);
1371 
        assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
1372 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1373 
    }
1374
1375 
    /// oxygenflow#32: a quiet job asks the hub every few seconds whether
1376 
    /// it still runs; a 409 marks it cancelled, and nothing more is sent.
1377 
    #[test]
1378 
    fn a_quiet_job_learns_that_its_run_was_cancelled() {
1379 
        use microvm::Sink;
1380 
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
1381 
        let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
1382 
        let (base, seen) = fake_hub(vec![204]);
1383 
        let hub = hub_at(base);
1384 
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
1385 
        assert!(!sink.cancelled());
1386 
        assert!(!sink.cancelled(), "asked again only after the heartbeat");
1387 
        assert_eq!(n(&seen), 1);
1388
1389 
        let (base, seen) = fake_hub(vec![409]);
1390 
        let hub = hub_at(base);
1391 
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: quiet, gone: false };
1392 
        assert!(sink.cancelled());
1393 
        sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
1394 
        sink.flush();
1395 
        assert!(sink.cancelled());
1396 
        assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
1397 
    }
1398
1399 
    /// oxygenflow#49: a hub that is away for a while (a deploy restarting
1400 
    /// it) costs a job's uploads a wait, not the upload, and costs its log
1401 
    /// no lines.
1402 
    #[test]
1403 
    fn a_job_outlasts_a_hub_restart() {
1404 
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
1405 
        // A port nobody listens on yet: the hub is restarting.
1406 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1407 
        let base = format!("http://127.0.0.1:{port}");
1408 
        let hub = hub_at(base.clone());
1409
1410 
        // Log lines sent while it is away are kept, and go when it is back.
1411 
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false };
1412 
        for i in 0..450 {
1413 
            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
1414 
        }
1415 
        sink.send();
1416 
        assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
1417 
        let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
1418 
        sink.send();
1419 
        assert!(sink.buf.is_empty() && !sink.gone);
1420 
        assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");
1421
1422 
        // An upload waits for the hub to come back, through refused
1423 
        // connections and server errors alike.
1424 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1425 
        let url = format!("http://127.0.0.1:{port}/x");
1426 
        let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
1427 
        let later = std::thread::spawn(move || {
1428 
            std::thread::sleep(Duration::from_millis(150));
1429 
            fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
1430 
        });
1431 
        let quick = [Duration::from_millis(50); 20];
1432 
        let r = retrying(&url, &quick, true, || agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
1433 
        assert_eq!(r.status(), 204);
1434 
        assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
1435 
        // Without `connection`, a hub that cannot be reached ends it at once.
1436 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1437 
        let url = format!("http://127.0.0.1:{port}/x");
1438 
        assert!(retrying(&url, &quick, false, || agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
1439 
        assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
1440 
    }
1441
1442 
    /// oxygenflow#50: the first stop drains (the second, not called here,
1443 
    /// exits at once).
1444 
    #[test]
1445 
    fn a_stop_drains_the_runner() {
1446 
        assert!(!stopping());
1447 
        on_stop(libc::SIGTERM);
1448 
        assert!(stopping(), "told to stop, it claims nothing more");
1449 
    }
1450
1451 
    /// Lines kept for a hub that stays away are bounded, saying how many went.
1452 
    #[test]
1453 
    fn unsent_log_lines_are_bounded() {
1454 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1455 
        let hub = hub_at(format!("http://127.0.0.1:{port}"));
1456 
        let mut sink = LogSink { hub: &hub, job: 1, buf: Vec::new(), last: Instant::now(), gone: false };
1457 
        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();
1458 
        sink.send();
1459 
        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
1460 
        assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
1461 
        assert_eq!(sink.buf[1].t, 10, "the oldest go first");
1462 
    }
1463
1464 
    #[test]
1465 
    fn the_budget_admits_jobs_that_fit() {
1466 
        let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
1467 
        let used = |jobs, memory_mib, cpus| Used { jobs, memory_mib, cpus };
1468 
        assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
1469 
        assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
1470 
        assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
1471 
        assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
1472 
        let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
1473 
        assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
1474 
        assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
1475 
        let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
1476 
        assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
1477 
    }
1478 
}