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