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