OXYGENHUB
◆ v0.5.1 oxygenflow / crates / flow / src / runner.rs
111.0 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() {
de1d7926 Andy McCall ↶ 1413 
                note(
1414 
                    &mut sink,
1415 
                    format!(
1416 
                        "cache: {restored} of {} files and directories unchanged since the cached build, their times kept",
1417 
                        sources.len()
1418 
                    ),
1419 
                );
82cf6e91 Andy McCall ↶ 1420 
            }
1421 
            sources
1422 
        };
f201e1bc Andy McCall ↶ 1423 
        // An earlier attempt's checkpoint (oxygenflow#77): back in place,
1424 
        // over the cache, before the same exec runs again; the receipt
1425 
        // lists it among the job's inputs.
1426 
        if let Some(blob) = &claim.checkpoint {
1427 
            if dir_tier {
1428 
                let bytes = hub.checkpoint(claim.job_id).context("fetching the checkpoint")?;
1429 
                if Repo::blob_ref(&bytes).to_hex() != *blob {
1430 
                    bail!("the checkpoint fetched is not blob {}, as the claim names", &blob[..12.min(blob.len())]);
1431 
                }
1432 
                let place = |root: &str| crate::checkpoint::host_dir(root, &job, &scratch.join("tree"), &scratch, wasm_tier);
1433 
                let n = crate::checkpoint::unpack(&bytes, &job.checkpoints, place)?;
1434 
                note(
1435 
                    &mut sink,
1436 
                    format!("restored checkpoint {}: {} files, {} bytes, from the attempt handed back", &blob[..12], n.files, n.bytes),
1437 
                );
1438 
                input_blobs.push(Repo::blob_ref(&bytes).0);
1439 
            } else {
1440 
                note(&mut sink, "a checkpoint was saved, but only the WASM and syscall tiers restore one yet: starting again".into());
1441 
            }
1442 
        }
28cc603f Andy McCall ↶ 1443 
        if !dir_tier {
ea56c56e Andy McCall ↶ 1444 
            workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
1445 
            std::fs::remove_dir_all(scratch.join("tree"))?;
1446 
        }
bb643c25 Andy McCall ↶ 1447 
        preparing.take();
ea56c56e Andy McCall ↶ 1448 
        // Cancelled while it was being prepared: no sandbox.
280ef951 Andy McCall ↶ 1449 
        if microvm::Sink::cancelled(&mut sink) {
1450 
            return Ok(cancelled());
1451 
        }
28cc603f Andy McCall ↶ 1452 
        if wasm_tier {
1453 
            note(&mut sink, format!("starting WASM (wasmtime {}): {} MiB", crate::wasm::WASMTIME, claim.sandbox.memory_mib));
1454 
        } else {
1455 
            note(
1456 
                &mut sink,
1457 
                format!(
1458 
                    "{}: {} {}, {} MiB, rootfs {}",
1459 
                    if syscall_tier { "starting a syscall sandbox" } else { "booting a microVM" },
1460 
                    claim.sandbox.cpus,
1461 
                    if syscall_tier { "CPU" } else { "vCPU" },
1462 
                    claim.sandbox.memory_mib,
1463 
                    &claim.sandbox.rootfs[..12]
1464 
                ),
1465 
            );
1466 
        }
ea56c56e Andy McCall ↶ 1467 
        let timeout = Duration::from_secs(claim.timeout_s);
1215f05f Andy McCall ↶ 1468 
        // Where the time goes (oxygenflow#35): what came before the
1469 
        // sandbox, the sandbox, and what comes after.
1470 
        let prepare_ms = started.elapsed().as_millis() as u64;
1471 
        let sandbox_started = Instant::now();
28cc603f Andy McCall ↶ 1472 
        let mut outcome = if wasm_tier {
1473 
            crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?
1474 
        } else if syscall_tier {
ea56c56e Andy McCall ↶ 1475 
            let group = format!("job-{}", claim.job_id);
1476 
            crate::syscall::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("tree"), &scratch, timeout, &mut sink, &group)?
1477 
        } else {
1478 
            microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?
1479 
        };
1215f05f Andy McCall ↶ 1480 
        let run_ms = sandbox_started.elapsed().as_millis() as u64;
280ef951 Andy McCall ↶ 1481 
        if outcome.cancelled {
f201e1bc Andy McCall ↶ 1482 
            // Handed back as the runner leaves: its checkpoints first, for
1483 
            // the next attempt (oxygenflow#77).
1484 
            if leaving().is_some() && !job.checkpoints.is_empty() {
1485 
                save_checkpoint(hub, claim.job_id, &job, &scratch, dir_tier, wasm_tier, |t| note(&mut sink, t));
1486 
            }
280ef951 Andy McCall ↶ 1487 
            return Ok(cancelled());
d60aac6c Andy McCall ↶ 1488 
        }
1489 
        // The receipt says what a cached disk started from.
1490 
        let disk_mounts: Vec<&flow_core::pipeline::Mount> =
1491 
            job.mounts.iter().filter(|m| m.source == flow_core::pipeline::MountSource::Disk).collect();
1492 
        for (i, key, how) in &cached {
1493 
            outcome.environment.parts.push((format!("cache {}", disk_mounts[*i].target), format!("{key}: {how}")));
280ef951 Andy McCall ↶ 1494 
        }
a67b97e9 Andy McCall ↶ 1495 
        // A passing job's artifacts, read from its disks and uploaded; a
9a6fac2b Andy McCall ↶ 1496 
        // declared artifact that cannot be kept fails the job. A failing
1497 
        // job keeps only its `when "always"` ones (oxygenflow#37), outside
1498 
        // its receipt, and one it cannot keep is only noted.
a67b97e9 Andy McCall ↶ 1499 
        let mut outputs = vec![];
1500 
        let mut output_error = None;
9a6fac2b Andy McCall ↶ 1501 
        let passed = outcome.exit_code == Some(0);
1502 
        let wanted = claim.job.outputs.iter().filter(|o| passed || o.always);
1503 
        for o in wanted {
6d53fb7d Andy McCall ↶ 1504 
            if o.kind == flow_core::OutputKind::Report {
28cc603f Andy McCall ↶ 1505 
                let read = if dir_tier {
6d53fb7d Andy McCall ↶ 1506 
                    let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
1507 
                    crate::outputs::read_report_dir(&dir, &inner)
1508 
                } else {
1509 
                    let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
1510 
                    crate::outputs::read_report(&image, &inner, &scratch)
1511 
                };
1512 
                let kept = read.and_then(|files| upload_report(hub, claim.job_id, &o.name, &files));
1513 
                match kept {
1514 
                    Ok((blob, files, bytes)) => {
1515 
                        let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
1516 
                        let hex = blob.to_hex();
1517 
                        note(&mut sink, format!("report {}: {files} files, {bytes} bytes, manifest {}{unattested}", o.name, &hex[..12]));
1518 
                        if passed {
1519 
                            outputs.push(blob.0);
1520 
                        }
1521 
                    }
1522 
                    Err(e) if passed => {
1523 
                        output_error = Some(format!("report {}: {e:#}", o.name));
1524 
                        break;
1525 
                    }
1526 
                    Err(e) => note(&mut sink, format!("report {}: not kept: {e:#}", o.name)),
1527 
                }
1528 
                continue;
1529 
            }
28cc603f Andy McCall ↶ 1530 
            let read = if dir_tier {
9a6fac2b Andy McCall ↶ 1531 
                let (dir, inner) = crate::outputs::locate_dir(&claim.job, &scratch.join("tree"), &scratch, &o.path);
1532 
                crate::outputs::read_dir_file(&dir, &inner)
1533 
            } else {
1534 
                let (image, inner) = crate::outputs::locate(&claim.job, &scratch.join("workspace.ext4"), &scratch, &o.path);
1535 
                crate::outputs::read(&image, &inner, &scratch)
1536 
            };
1537 
            let kept = read.and_then(|bytes| {
1538 
                let blob = Repo::blob_ref(&bytes);
1539 
                let named = hub.artifact(claim.job_id, &o.name, &bytes)?;
1540 
                if named != blob.to_hex() {
1541 
                    bail!("the hub kept it as {named}, not {}", blob.to_hex());
1542 
                }
1543 
                let unattested = if passed { "" } else { " (the job failed: not in its receipt)" };
1544 
                note(&mut sink, format!("artifact {}: {} bytes, blob {}{unattested}", o.name, bytes.len(), &named[..12]));
1545 
                Ok(blob.0)
1546 
            });
1547 
            match kept {
1548 
                Ok(b) if passed => outputs.push(b),
1549 
                Ok(_) => {}
1550 
                Err(e) if passed => {
1551 
                    output_error = Some(format!("artifact {}: {e:#}", o.name));
1552 
                    break;
a67b97e9 Andy McCall ↶ 1553 
                }
9a6fac2b Andy McCall ↶ 1554 
                Err(e) => note(&mut sink, format!("artifact {}: not kept: {e:#}", o.name)),
a67b97e9 Andy McCall ↶ 1555 
            }
1556 
        }
1557 
        if let Some(e) = &output_error {
1558 
            note(&mut sink, e.clone());
1559 
        }
1560 
        let receipt = outcome.exit_code.filter(|_| output_error.is_none()).map(|code| Receipt {
e0918b88 Andy McCall 1561 
            job_name: claim.job.name.clone(),
1562 
            pipeline_hash: hex32(&claim.pipeline_hash),
1563 
            scope_tree_hash: scope_hash,
46fe2c81 Andy McCall ↶ 1564 
            inputs: input_blobs.clone(),
e0918b88 Andy McCall 1565 
            environment: outcome.environment.hash(),
a67b97e9 Andy McCall ↶ 1566 
            outputs,
e0918b88 Andy McCall 1567 
            exit_code: code as u32,
1568 
            duration_ms: outcome.duration_ms,
1569 
        });
47c1d64d Andy McCall ↶ 1570 
        let signed_receipt = match &receipt {
35f7a849 Andy McCall ↶ 1571 
            Some(r) => {
1572 
                let mut signed = sign_receipt(&repo, &heads, cfg, key, claim, r)?;
1573 
                if cfg.keyless {
1574 
                    match sign_keyless(&repo, &heads, cfg, hub, claim, r) {
1575 
                        Ok(k) => signed = k,
1576 
                        Err(e) => {
1577 
                            note(&mut sink, format!("receipt: signed with the runner's key; a short-lived key was not certified: {e:#}"))
1578 
                        }
1579 
                    }
1580 
                }
1581 
                Some(hex::encode(signed.encode()))
1582 
            }
47c1d64d Andy McCall ↶ 1583 
            None => None,
1584 
        };
d60aac6c Andy McCall ↶ 1585 
        // Measured before a cached disk leaves the scratch directory.
1586 
        let peak_disk_mib = crate::facts::allocated_mib(&scratch);
077a6524 Andy McCall ↶ 1587 
        // A passing job's cached disks are pruned and kept (#72); a
1588 
        // failing job's are thrown away, so nothing it wrote reaches a
1589 
        // later job.
2594278f Andy McCall ↶ 1590 
        let kept = passed && output_error.is_none();
1591 
        for (i, key, _) in &cached {
1592 
            let disk = crate::cache::disk_path(form, &scratch, *i);
077a6524 Andy McCall ↶ 1593 
            let m = disk_mounts[*i];
1594 
            let prune = crate::cache::Prune {
1595 
                started: cache_started,
1596 
                percent: cfg.cache_prune_percent.unwrap_or(flow_init::prune::DEFAULT_PERCENT),
1597 
                guest: outcome.pruned.iter().find(|(t, _)| *t == m.target).map(|(_, p)| p),
1598 
            };
1599 
            match pool.settle(&cache_ctx, m, key, &disk, &sources, kept, &prune) {
1600 
                Ok(Some(line)) => note(&mut sink, line),
1601 
                Ok(None) => {}
1602 
                Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),
d60aac6c Andy McCall ↶ 1603 
            }
1215f05f Andy McCall ↶ 1604 
        }
1605 
        let total_ms = started.elapsed().as_millis() as u64;
1606 
        let mut phases: std::collections::BTreeMap<String, u64> =
1607 
            [("prepare", prepare_ms), ("run", run_ms), ("finish", total_ms.saturating_sub(prepare_ms + run_ms))]
1608 
                .into_iter()
1609 
                .map(|(k, v)| (k.to_string(), v))
1610 
                .collect();
1611 
        if let Some(ms) = image_ms {
1612 
            phases.insert("image".into(), ms);
d60aac6c Andy McCall ↶ 1613 
        }
e0918b88 Andy McCall 1614 
        Ok(Finish {
1215f05f Andy McCall ↶ 1615 
            phases,
e0918b88 Andy McCall 1616 
            exit_code: outcome.exit_code,
1215f05f Andy McCall ↶ 1617 
            duration_ms: total_ms,
e0918b88 Andy McCall 1618 
            receipt,
47c1d64d Andy McCall ↶ 1619 
            signed_receipt,
a67b97e9 Andy McCall ↶ 1620 
            error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())),
8f211e20 Andy McCall ↶ 1621 
            // What the job took (oxygenflow#41): its disks are sparse files
1622 
            // in the scratch directory, so their blocks now are their peak.
d60aac6c Andy McCall ↶ 1623 
            peak_disk_mib,
8f211e20 Andy McCall ↶ 1624 
            peak_memory_mib: outcome.peak_memory_mib,
e0918b88 Andy McCall 1625 
        })
1626 
    })();
ea56c56e Andy McCall ↶ 1627 
    let _ = workspace::remove_tree(&scratch);
e0918b88 Andy McCall 1628 
    sink.flush();
1629 
    result
47c1d64d Andy McCall ↶ 1630 
}
1631
1632 
/// Bind `receipt` to the state of `heads` in `repo` and sign it as the
1633 
/// runner (Oxygen spec §14.6). The state is computed here, not taken from
1634 
/// the claim, so the runner vouches only for what it fetched.
6d53fb7d Andy McCall ↶ 1635 
/// Upload a report's files, then its manifest under the report's name, and
1636 
/// check the hub kept each as the blob it should (oxygenhub#127). Answers
1637 
/// the manifest's blob, the file count and their total size.
1638 
fn upload_report(hub: &Hub, job: i64, name: &str, files: &[(String, Vec<u8>)]) -> Result<(oxy_core::hash::ObjectRef, usize, u64)> {
1639 
    let mut listed = Vec::with_capacity(files.len());
1640 
    let mut total = 0u64;
1641 
    for (path, bytes) in files {
1642 
        let blob = Repo::blob_ref(bytes).to_hex();
1643 
        let named = hub.report_file(job, name, path, bytes).with_context(|| format!("uploading {path}"))?;
1644 
        if named != blob {
1645 
            bail!("the hub kept {path} as {named}, not {blob}");
1646 
        }
1647 
        total += bytes.len() as u64;
1648 
        listed.push(serde_json::json!({ "path": path, "blob": blob, "size": bytes.len() }));
1649 
    }
1650 
    let manifest = serde_json::to_vec(&serde_json::json!({ "version": 1, "files": listed }))?;
1651 
    let blob = Repo::blob_ref(&manifest);
1652 
    let named = hub.artifact(job, name, &manifest)?;
1653 
    if named != blob.to_hex() {
1654 
        bail!("the hub kept the manifest as {named}, not {}", blob.to_hex());
1655 
    }
1656 
    Ok((blob, files.len(), total))
1657 
}
1658
47c1d64d Andy McCall ↶ 1659 
fn sign_receipt(
1660 
    repo: &Repo,
1661 
    heads: &[ChangeHash],
1662 
    cfg: &RunnerConfig,
1663 
    key: &RunnerKey,
1664 
    claim: &Claim,
1665 
    receipt: &Receipt,
1666 
) -> Result<oxy_core::receipt::SignedReceipt> {
1667 
    let state = state_of(heads);
1668 
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
1669 
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &key.identity, &key.secret))
1670 
}
1671
35f7a849 Andy McCall ↶ 1672 
/// Sign `receipt` with a fresh key the hub certifies for this job now,
1673 
/// just before signing, so the certificate covers the moment the hub
1674 
/// accepts it (keyless, oxygenhub#241). The key is dropped with the
1675 
/// receipt signed: nothing long-lived signs.
1676 
fn sign_keyless(
1677 
    repo: &Repo,
1678 
    heads: &[ChangeHash],
1679 
    cfg: &RunnerConfig,
1680 
    hub: &Hub,
1681 
    claim: &Claim,
1682 
    receipt: &Receipt,
1683 
) -> Result<oxy_core::receipt::SignedReceipt> {
1684 
    let sk = oxy_core::identity::SecretKey::generate();
1685 
    let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
1686 
    let name = format!("{} job {}", cfg.name.as_deref().unwrap_or("runner"), claim.job_id);
1687 
    let id = sk.identity(&name, "", now);
1688 
    let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge(&claim.token)).map_err(|e| anyhow::anyhow!("{e}"))?;
1689 
    let (cert, issuer) = hub.certify(&claim.token, &id, &proof)?;
1690 
    let state = state_of(heads);
1691 
    let run = format!("{}/{}/flow/{}", cfg.hub.trim_end_matches('/'), claim.repo, claim.run_number);
1692 
    Ok(oxy_core::receipt::SignedReceipt::sign(receipt.to_oxy(), repo.id()?, state, &run, &id, &sk).with_certificate(cert, issuer))
1693 
}
1694
47c1d64d Andy McCall ↶ 1695 
/// The ObjectRef of the state whose heads are `heads` (Oxygen spec §10.1).
1696 
pub fn state_of(heads: &[ChangeHash]) -> oxy_core::hash::ObjectRef {
1697 
    use oxy_core::hash::{hash_object, ObjectType};
1698 
    hash_object(ObjectType::State, &oxy_core::objects::StateObject::new(heads.to_vec()).encode())
e0918b88 Andy McCall 1699 
}
1700
1701 
fn hex32(s: &str) -> [u8; 32] {
1702 
    let mut out = [0; 32];
1703 
    if let Ok(v) = hex::decode(s) {
1704 
        if v.len() == 32 {
1705 
            out.copy_from_slice(&v);
1706 
        }
1707 
    }
1708 
    out
1709 
}
1710
f53e6d47 Andy McCall ↶ 1711 
/// Whether a network error is worth another try: DNS, a refused or reset
1712 
/// connection, a timeout, or a gateway answering 502, 503 or 504 while the
1713 
/// hub restarts (oxygenhub#111).
1714 
fn transient(msg: &str) -> bool {
1715 
    let m = msg.to_lowercase();
1716 
    [
1717 
        "dns failed",
1718 
        "name resolution",
1719 
        "connection refused",
1720 
        "connection reset",
1721 
        "timed out",
1722 
        "timeout",
1723 
        "http 502",
1724 
        "http 503",
1725 
        "http 504",
1726 
        "broken pipe",
1727 
        "unexpected eof",
1728 
    ]
1729 
    .iter()
1730 
    .any(|k| m.contains(k))
1731 
}
1732
1733 
/// Run `f` again after a transient network error, waiting `waits` between
1734 
/// tries (about a minute in all), and give up on any other error at once.
1735 
fn with_retries<T>(waits: &[Duration], mut f: impl FnMut() -> Result<T>) -> Result<T> {
1736 
    let mut tries = waits.iter();
1737 
    loop {
1738 
        match f() {
1739 
            Ok(v) => return Ok(v),
1740 
            Err(e) => match tries.next() {
1741 
                Some(w) if transient(&format!("{e:#}")) => std::thread::sleep(*w),
1742 
                _ => return Err(e),
1743 
            },
1744 
        }
1745 
    }
1746 
}
1747
1748 
/// The waits between tries of a fetch: 2, 4, 8, 16 and 30 seconds.
1749 
const FETCH_WAITS: [Duration; 5] =
1750 
    [Duration::from_secs(2), Duration::from_secs(4), Duration::from_secs(8), Duration::from_secs(16), Duration::from_secs(30)];
1751
e0918b88 Andy McCall 1752 
/// The runner's copy of a repository, fetched up to date with the job's token.
1753 
fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
c23d5191 Andy McCall ↶ 1754 
    cached_repo_at(home, cfg, &claim.repo, &claim.clone_url, &claim.token)
1755 
}
1756
1757 
/// The runner's copy of `repo`, cloned or fetched with `token`.
1758 
fn cached_repo_at(home: &Home, cfg: &RunnerConfig, repo_name: &str, clone_url: &str, token: &str) -> Result<Repo> {
1759 
    let dir: PathBuf = home.root.join("repos").join(repo_name.replace('/', "__"));
e0918b88 Andy McCall 1760 
    let set_remote = |repo: &mut Repo| -> Result<()> {
c23d5191 Andy McCall ↶ 1761 
        repo.config.set("remote.origin.url", clone_url).map_err(|e| anyhow::anyhow!("{e}"))?;
1762 
        repo.config.set("remote.origin.token", token).map_err(|e| anyhow::anyhow!("{e}"))?;
e0918b88 Andy McCall 1763 
        if let Some(ca) = &cfg.ca {
1764 
            repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(|e| anyhow::anyhow!("{e}"))?;
1765 
        }
1766 
        repo.save_config().map_err(|e| anyhow::anyhow!("{e}"))
1767 
    };
1768 
    if dir.join(".oxy").is_dir() {
1769 
        let mut repo = Repo::open(&dir).map_err(|e| anyhow::anyhow!("{e}"))?;
1770 
        set_remote(&mut repo)?;
f53e6d47 Andy McCall ↶ 1771 
        with_retries(&FETCH_WAITS, || repo.fetch("origin").map(|_| ()).map_err(|e| anyhow::anyhow!("fetch {clone_url}: {e}")))?;
e0918b88 Andy McCall 1772 
        return Ok(repo);
1773 
    }
ed4aef8b Andy McCall ↶ 1774 
    // First job for this repository: clone with the job's token, which goes
1775 
    // through the process environment, so one clone at a time.
1776 
    let _clone = CLONE.lock().unwrap_or_else(|p| p.into_inner());
c23d5191 Andy McCall ↶ 1777 
    std::env::set_var("OXY_TOKEN", token);
e0918b88 Andy McCall 1778 
    if let Some(ca) = &cfg.ca {
1779 
        std::env::set_var("OXY_CA", ca);
1780 
    }
f53e6d47 Andy McCall ↶ 1781 
    let cloned = with_retries(&FETCH_WAITS, || {
1782 
        let _ = std::fs::remove_dir_all(&dir);
1783 
        Repo::clone(clone_url, &dir).map_err(|e| anyhow::anyhow!("{e}"))
1784 
    });
e0918b88 Andy McCall 1785 
    std::env::remove_var("OXY_TOKEN");
1786 
    std::env::remove_var("OXY_CA");
1787 
    let mut repo = cloned.map_err(|e| {
1788 
        let _ = std::fs::remove_dir_all(&dir);
c23d5191 Andy McCall ↶ 1789 
        anyhow::anyhow!("clone {}: {e}", clone_url)
e0918b88 Andy McCall 1790 
    })?;
1791 
    set_remote(&mut repo)?;
1792 
    Ok(repo)
bb643c25 Andy McCall ↶ 1793 
}
1794
1795 
#[cfg(test)]
1796 
mod tests {
0d663e91 Andy McCall ↶ 1797
1798 
    /// #98: register keeps a copy of the CA in the home, by absolute path,
1799 
    /// and keeping the home's own copy again is a no-op.
1800 
    #[test]
1801 
    fn the_ca_is_kept_in_the_home_by_absolute_path() {
1802 
        let home = tempfile::tempdir().unwrap();
1803 
        let elsewhere = tempfile::tempdir().unwrap();
1804 
        let given = elsewhere.path().join("ca.pem");
1805 
        std::fs::write(&given, "PEM").unwrap();
1806 
        let kept = super::keep_ca(home.path(), &given).unwrap();
1807 
        assert!(kept.is_absolute() && kept.ends_with(crate::home::CA_PEM), "{kept:?}");
1808 
        assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
1809 
        std::fs::remove_file(&given).unwrap();
1810 
        assert_eq!(super::keep_ca(home.path(), &kept).unwrap(), kept, "the home's copy again");
1811 
        assert_eq!(std::fs::read_to_string(&kept).unwrap(), "PEM");
1812 
    }
46fe2c81 Andy McCall ↶ 1813
1814 
    #[test]
1815 
    fn an_input_is_placed_only_when_it_is_what_the_receipt_names() {
1816 
        let dir = tempfile::tempdir().unwrap();
1817 
        let bytes = b"the reviewed plan";
1818 
        let blob = Repo::blob_ref(bytes).to_hex();
1819 
        let input = |path: &str, blob: &str| flow_core::api::ArtifactInput {
1820 
            artifact: "tfplan".into(),
1821 
            from: "plan".into(),
1822 
            path: path.into(),
1823 
            blob: blob.into(),
1824 
        };
1825 
        let got = super::place_input(dir.path(), &input("/src/plans/tfplan", &blob), bytes).unwrap();
1826 
        assert_eq!(hex::encode(got), blob);
1827 
        assert_eq!(std::fs::read(dir.path().join("plans/tfplan")).unwrap(), bytes);
1828 
        let wrong = super::place_input(dir.path(), &input("/src/other", &"0".repeat(64)), bytes).unwrap_err();
1829 
        assert!(format!("{wrong:#}").contains("not what plan's receipt names"), "{wrong:#}");
1830 
        assert!(!dir.path().join("other").exists(), "nothing written when it does not match");
1831 
        for bad in ["/etc/passwd", "/src/../x", "/src/.oxy/x", "/src/"] {
1832 
            assert!(super::place_input(dir.path(), &input(bad, &blob), bytes).is_err(), "{bad}");
1833 
        }
1834 
    }
1835
bb643c25 Andy McCall ↶ 1836 
    use super::*;
b11e30f5 Andy McCall ↶ 1837
1838 
    #[test]
1839 
    fn a_runner_names_what_stops_a_sandbox() {
1840 
        let dir = tempfile::tempdir().unwrap();
1841 
        let home = Home::locate(Some(dir.path())).unwrap();
1842 
        let bytes = b"not really ext4";
1843 
        let hash = blake3::hash(bytes).to_hex().to_string();
1844 
        let mut sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
1845 
            "name": "s", "runtime": "syscall", "rootfs": "ab".repeat(32), "memory_mib": 1024, "cpus": 1, "network": "isolated"
1846 
        }))
1847 
        .unwrap();
1848 
        let problem = |sb: &flow_core::Sandbox, free| sandbox_problem(&home, sb, "x86_64", free);
1849 
        assert_eq!(problem(&sb, None).unwrap(), "rootfs abababababab is not on this runner");
1850 
        std::fs::write(home.rootfs(&hash), bytes).unwrap();
1851 
        sb.rootfs = hash.clone();
1852 
        assert_eq!(problem(&sb, Some(4096)), None);
1853 
        assert_eq!(problem(&sb, Some(512)).unwrap(), "only 512 MiB of memory is free here; the sandbox needs 1024 MiB");
1854 
        std::fs::write(home.image_manifest(&hash), r#"{"arch":"riscv64"}"#).unwrap();
1855 
        assert_eq!(problem(&sb, None).unwrap(), format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12]));
1856 
        sb.arch = Some("aarch64".into());
1857 
        assert_eq!(problem(&sb, None).unwrap(), "the sandbox needs aarch64; this runner is x86_64");
1858 
        sb.arch = None;
1859 
        std::fs::remove_file(home.image_manifest(&hash)).unwrap();
1860 
        std::fs::write(home.rootfs(&hash), b"changed").unwrap();
1861 
        assert!(problem(&sb, None).unwrap().contains("does not verify"));
1862 
    }
f53e6d47 Andy McCall ↶ 1863
1864 
    #[test]
0077c748 Andy McCall ↶ 1865 
    fn a_runner_reads_an_unrecorded_images_architecture_and_declines_another() {
1866 
        // An image fetched or imported with no recorded architecture, built
1867 
        // for riscv64: the runner reads its shell, records it, and declines
1868 
        // instead of running it and failing (spec §11.2, §11.5).
1869 
        let dir = tempfile::tempdir().unwrap();
1870 
        let home = Home::locate(Some(dir.path().join("home").as_path())).unwrap();
1871 
        let tree = dir.path().join("tree");
1872 
        std::fs::create_dir_all(tree.join("bin")).unwrap();
1873 
        let mut elf = vec![0u8; 64];
1874 
        elf[..7].copy_from_slice(b"\x7fELF\x02\x01\x01");
1875 
        elf[18..20].copy_from_slice(&0xF3u16.to_le_bytes());
1876 
        std::fs::write(tree.join("bin/sh"), &elf).unwrap();
1877 
        let img = dir.path().join("rootfs.ext4");
1878 
        let made = std::process::Command::new("mkfs.ext4").args(["-q", "-F", "-d"]).arg(&tree).arg(&img).arg("8M").status();
1879 
        if !made.is_ok_and(|s| s.success()) {
1880 
            eprintln!("mkfs.ext4 not found: skipped");
1881 
            return;
1882 
        }
1883 
        let hash = crate::home::hash_file(&img).unwrap();
1884 
        std::fs::create_dir_all(home.rootfs(&hash).parent().unwrap()).unwrap();
1885 
        std::fs::rename(&img, home.rootfs(&hash)).unwrap();
1886 
        assert!(!home.image_manifest(&hash).exists());
1887 
        let sb: flow_core::Sandbox = serde_json::from_value(serde_json::json!({
1888 
            "name": "s", "runtime": "syscall", "rootfs": hash, "memory_mib": 64, "cpus": 1, "network": "isolated"
1889 
        }))
1890 
        .unwrap();
1891 
        assert_eq!(
1892 
            sandbox_problem(&home, &sb, "x86_64", None).unwrap(),
1893 
            format!("rootfs {} is for riscv64; this runner is x86_64", &hash[..12])
1894 
        );
1895 
        assert_eq!(sandbox_problem(&home, &sb, "riscv64", None), None, "a riscv64 runner takes it");
1896 
        // Learned: recorded once, and reported with the runner's claims.
1897 
        let manifest: serde_json::Value = serde_json::from_slice(&std::fs::read(home.image_manifest(&hash)).unwrap()).unwrap();
1898 
        assert_eq!(manifest["arch"], "riscv64");
1899 
        assert_eq!(home.image_archs().get(&hash).map(String::as_str), Some("riscv64"));
1900 
    }
1901
1902 
    #[test]
f53e6d47 Andy McCall ↶ 1903 
    fn transient_network_errors_are_tried_again_and_others_are_not() {
1904 
        assert!(transient("fetch x: Dns Failed: resolve dns name 'oxygenhub.io:443': Temporary failure in name resolution"));
1905 
        assert!(transient("HTTP 503: the hub is restarting"));
1906 
        assert!(!transient("HTTP 404: no such repository"));
1907 
        let mut n = 0;
1908 
        let got: Result<u32> = with_retries(&[Duration::ZERO, Duration::ZERO], || {
1909 
            n += 1;
1910 
            if n < 3 {
1911 
                Err(anyhow::anyhow!("connection refused"))
1912 
            } else {
1913 
                Ok(7)
1914 
            }
1915 
        });
1916 
        assert_eq!((got.unwrap(), n), (7, 3), "two retries, then success");
1917 
        let mut m = 0;
1918 
        let got: Result<u32> = with_retries(&[Duration::ZERO; 4], || {
1919 
            m += 1;
1920 
            Err(anyhow::anyhow!("HTTP 403: forbidden"))
1921 
        });
1922 
        assert!(got.is_err());
1923 
        assert_eq!(m, 1, "a refusal is not tried again");
1924 
        let mut k = 0;
1925 
        let got: Result<u32> = with_retries(&[Duration::ZERO; 2], || {
1926 
            k += 1;
1927 
            Err(anyhow::anyhow!("timed out"))
1928 
        });
1929 
        assert!(got.is_err());
1930 
        assert_eq!(k, 3, "gives up after the last wait");
1931 
    }
1b2229da Andy McCall ↶ 1932
1933 
    /// A hub that answers each request with the next of `codes`, counting them.
1934 
    fn fake_hub(codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
89171fca Andy McCall ↶ 1935 
        fake_hub_on(std::net::TcpListener::bind("127.0.0.1:0").unwrap(), codes)
1936 
    }
1937
1938 
    /// A hub at `listener` answering each request with the next of `codes`.
1939 
    fn fake_hub_on(listener: std::net::TcpListener, codes: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
1b2229da Andy McCall ↶ 1940 
        use std::io::{Read, Write};
1941 
        let addr = listener.local_addr().unwrap();
1942 
        let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
1943 
        let count = seen.clone();
1944 
        std::thread::spawn(move || {
1945 
            for code in codes {
1946 
                let (mut s, _) = listener.accept().unwrap();
f3b4fdeb Andy McCall ↶ 1947 
                // Read the whole request (headers, then Content-Length bytes
1948 
                // of body) before answering: closing early would reset the
1949 
                // connection while the client is still sending.
1950 
                let mut req = Vec::new();
1b2229da Andy McCall ↶ 1951 
                let mut buf = [0u8; 8192];
f3b4fdeb Andy McCall ↶ 1952 
                loop {
1953 
                    let n = s.read(&mut buf).unwrap_or(0);
1954 
                    if n == 0 {
1955 
                        break;
1956 
                    }
1957 
                    req.extend_from_slice(&buf[..n]);
1958 
                    let text = String::from_utf8_lossy(&req);
1959 
                    if let Some(end) = text.find("\r\n\r\n") {
1960 
                        let len = text[..end]
1961 
                            .lines()
1962 
                            .find_map(|l| {
1963 
                                l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
1964 
                            })
1965 
                            .unwrap_or(0);
1966 
                        if req.len() >= end + 4 + len {
1967 
                            break;
1968 
                        }
1969 
                    }
1970 
                }
1b2229da Andy McCall ↶ 1971 
                count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1972 
                let reason = if code == 204 { "No Content" } else { "Error" };
1973 
                let _ = write!(s, "HTTP/1.1 {code} {reason}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
1974 
            }
1975 
        });
1976 
        (format!("http://{addr}"), seen)
f98a8f78 Andy McCall ↶ 1977 
    }
1978
1979 
    /// A hub that answers one request with `body` (status 200) and hands
1980 
    /// back what it was sent.
1981 
    fn fake_json_hub(body: &'static str) -> (String, std::sync::mpsc::Receiver<String>) {
1982 
        use std::io::{Read, Write};
1983 
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
1984 
        let addr = listener.local_addr().unwrap();
1985 
        let (tx, rx) = std::sync::mpsc::channel();
1986 
        std::thread::spawn(move || {
1987 
            let (mut s, _) = listener.accept().unwrap();
1988 
            let mut req = Vec::new();
1989 
            let mut buf = [0u8; 65536];
1990 
            loop {
1991 
                let n = s.read(&mut buf).unwrap_or(0);
1992 
                if n == 0 {
1993 
                    break;
1994 
                }
1995 
                req.extend_from_slice(&buf[..n]);
1996 
                let text = String::from_utf8_lossy(&req);
1997 
                if let Some(end) = text.find("\r\n\r\n") {
1998 
                    let len = text[..end]
1999 
                        .lines()
2000 
                        .find_map(|l| {
2001 
                            l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
2002 
                        })
2003 
                        .unwrap_or(0);
2004 
                    if req.len() >= end + 4 + len {
2005 
                        break;
2006 
                    }
2007 
                }
2008 
            }
2009 
            tx.send(String::from_utf8_lossy(&req).into_owned()).unwrap();
2010 
            let _ = write!(
2011 
                s,
2012 
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
2013 
                body.len()
2014 
            );
2015 
        });
2016 
        (format!("http://{addr}"), rx)
2017 
    }
2018
2019 
    #[test]
2020 
    fn registering_binds_the_key_and_keeps_the_credential_private() {
2021 
        use std::os::unix::fs::PermissionsExt;
2022 
        let dir = tempfile::tempdir().unwrap();
2023 
        let home = Home::locate(Some(dir.path())).unwrap();
2024 
        let (hub, sent) = fake_json_hub(r#"{"id":7,"name":"plex","credential":"flowrunner_secret"}"#);
2025 
        let got = register(&home, &hub, " reg-token ", "plex", None, vec!["gpu".into()]).unwrap();
2026 
        assert_eq!((got.id, got.name.as_str()), (7, "plex"));
2027 
        let req = sent.recv().unwrap();
2028 
        assert!(req.starts_with("POST /api/v1/flow/runners/register"), "{req}");
2029 
        let body: Register = serde_json::from_str(&req[req.find("\r\n\r\n").unwrap() + 4..]).unwrap();
2030 
        assert_eq!((body.token.as_str(), body.info.labels.clone()), ("reg-token", vec!["gpu".to_string()]));
2031 
        let key = home.runner_key("plex").unwrap();
2032 
        assert_eq!(body.identity, hex::encode(key.identity.encode()), "the runner's signing identity");
2033 
        let path = dir.path().join(crate::home::RUNNER_TOML);
2034 
        assert_eq!(std::fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o600);
2035 
        // The runner then starts from runner.toml.
2036 
        let again = Home::locate(Some(dir.path())).unwrap();
2037 
        let cfg = again.config.runner.unwrap();
2038 
        assert_eq!((cfg.hub.as_str(), cfg.token.as_str(), cfg.name.as_deref()), (hub.as_str(), "flowrunner_secret", Some("plex")));
2039 
        assert_eq!(cfg.labels, ["gpu"]);
35f7a849 Andy McCall ↶ 2040 
    }
2041
2042 
    /// A hub that answers certify with a certificate `sign_as` signs for
2043 
    /// `cert_key`, and publishes `published` as its key: two requests.
2044 
    fn fake_certify_hub(
2045 
        sign_as: (oxy_core::identity::Identity, oxy_core::identity::SecretKey),
2046 
        cert_key: oxy_core::identity::Identity,
2047 
        published: oxy_core::identity::Identity,
2048 
    ) -> String {
2049 
        use std::io::{Read, Write};
2050 
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
2051 
        let addr = listener.local_addr().unwrap();
2052 
        std::thread::spawn(move || {
2053 
            for _ in 0..2 {
2054 
                let Ok((mut s, _)) = listener.accept() else { return };
2055 
                let mut req = Vec::new();
2056 
                let mut buf = [0u8; 65536];
2057 
                loop {
2058 
                    let n = s.read(&mut buf).unwrap_or(0);
2059 
                    if n == 0 {
2060 
                        break;
2061 
                    }
2062 
                    req.extend_from_slice(&buf[..n]);
2063 
                    let text = String::from_utf8_lossy(&req);
2064 
                    if let Some(end) = text.find("\r\n\r\n") {
2065 
                        let len = text[..end]
2066 
                            .lines()
2067 
                            .find_map(|l| {
2068 
                                l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))
2069 
                            })
2070 
                            .unwrap_or(0);
2071 
                        if req.len() >= end + 4 + len {
2072 
                            break;
2073 
                        }
2074 
                    }
2075 
                }
2076 
                let line = String::from_utf8_lossy(&req).lines().next().unwrap_or("").to_string();
2077 
                let body: Vec<u8> = if line.contains("/keyless/certify") {
2078 
                    let cert = oxy_core::cert::SignedCertificate::sign(
2079 
                        oxy_core::cert::Certificate {
2080 
                            key: cert_key.clone(),
2081 
                            issuer: sign_as.0.hash(),
2082 
                            subject: "repo:andy/demo:bookmark:main".into(),
2083 
                            claims: vec![],
2084 
                            scope: oxy_core::identity::Scope::SEAL,
2085 
                            not_before: 1,
2086 
                            expires: 600_001,
fc2a6d18 Andy McCall ↶ 2087 
                            grant: None,
35f7a849 Andy McCall ↶ 2088 
                        },
2089 
                        &sign_as.0,
2090 
                        &sign_as.1,
2091 
                    );
2092 
                    serde_json::json!({ "certificate": oxy_core::dsse::base64(&cert.encode()) }).to_string().into_bytes()
2093 
                } else {
2094 
                    published.encode()
2095 
                };
2096 
                let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len());
2097 
                let _ = s.write_all(&body);
2098 
            }
2099 
        });
2100 
        format!("http://{addr}")
2101 
    }
2102
2103 
    /// keyless (oxygenhub#241): the runner takes a certificate only when it
2104 
    /// is for its fresh key and signed by the key the hub publishes.
2105 
    #[test]
2106 
    fn a_certificate_is_taken_only_for_this_key_from_the_hubs_key() {
2107 
        use oxy_core::identity::SecretKey;
2108 
        let hub_sk = SecretKey::generate();
2109 
        let hub_id = hub_sk.identity("hub", "", 1);
2110 
        let sk = SecretKey::generate();
2111 
        let id = sk.identity("earth-1 job 7", "", 1);
2112 
        let proof = sk.sign_challenge(&id.hash(), &oxy_core::cert::certify_challenge("flowjob_x")).unwrap();
2113
2114 
        let base = fake_certify_hub((hub_id.clone(), hub_sk.clone()), id.clone(), hub_id.clone());
2115 
        let (cert, issuer) = hub_at(base).certify("flowjob_x", &id, &proof).unwrap();
2116 
        assert_eq!((cert.cert.key_id(), issuer.hash()), (id.hash(), hub_id.hash()));
2117
2118 
        let mallory_sk = SecretKey::generate();
2119 
        let mallory = mallory_sk.identity("mallory", "", 1);
2120 
        let base = fake_certify_hub((mallory.clone(), mallory_sk), id.clone(), hub_id.clone());
2121 
        let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
2122 
        assert!(e.contains("not signed by the hub's published key"), "{e}");
2123
2124 
        let other = SecretKey::generate().identity("other", "", 1);
2125 
        let base = fake_certify_hub((hub_id.clone(), hub_sk), other, hub_id);
2126 
        let e = hub_at(base).certify("flowjob_x", &id, &proof).unwrap_err().to_string();
2127 
        assert!(e.contains("another key"), "{e}");
1b2229da Andy McCall ↶ 2128 
    }
2129
2130 
    fn hub_at(base: String) -> Hub {
2131 
        Hub { base, token: "t".into(), agent: ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build() }
cedd1234 Andy McCall ↶ 2132 
    }
2133
2134 
    /// oxygenhub#322: an offered build is kept only if it is the bytes the
2135 
    /// hub attests; a peer or a proxy cannot change what is installed.
2136 
    #[test]
2137 
    fn a_build_is_kept_only_if_it_matches_the_hubs_blake3() {
2138 
        let (body, other) = (b"#!flow build 0.3.2".to_vec(), b"#!something else!!".to_vec());
2139 
        let serve = |bytes: Vec<u8>| {
2140 
            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
2141 
            let addr = listener.local_addr().unwrap();
2142 
            std::thread::spawn(move || {
2143 
                use std::io::{Read, Write};
2144 
                let (mut s, _) = listener.accept().unwrap();
2145 
                let mut buf = [0u8; 4096];
2146 
                let _ = s.read(&mut buf);
2147 
                let _ = write!(s, "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", bytes.len());
2148 
                let _ = s.write_all(&bytes);
2149 
            });
2150 
            format!("http://{addr}")
2151 
        };
2152 
        let offer = flow_core::api::UpdateOffer {
2153 
            version: "0.3.2".into(),
2154 
            asset: "flow-linux-x86_64".into(),
2155 
            url: "/andy/oxygenflow/asset/1/flow-linux-x86_64".into(),
2156 
            blake3: blake3::hash(&body).to_hex().to_string(),
2157 
            size: body.len() as u64,
2158 
            ..Default::default()
2159 
        };
2160 
        let dir = tempfile::tempdir().unwrap();
2161 
        let to = dir.path().join("flow-0.3.2");
2162 
        hub_at(serve(body.clone())).download_build(&offer, &to).unwrap();
2163 
        assert_eq!(std::fs::read(&to).unwrap(), body);
2164 
        std::fs::remove_file(&to).unwrap();
2165 
        let e = hub_at(serve(other)).download_build(&offer, &to).unwrap_err();
2166 
        assert!(e.to_string().contains("does not match what the hub attests"), "{e:#}");
2167 
        assert!(!to.exists() && !to.with_extension("partial").exists(), "nothing of it is left");
3007d8f0 Andy McCall ↶ 2168 
    }
2169
2170 
    /// oxygenhub#71: a job's result is offered until the hub takes it,
2171 
    /// through server errors, and a refusal ends the trying at once.
2172 
    #[test]
2173 
    fn a_result_is_reported_until_the_hub_takes_it() {
8f211e20 Andy McCall ↶ 2174 
        let f = Finish {
2175 
            exit_code: Some(0),
2176 
            duration_ms: 1,
2177 
            receipt: None,
2178 
            signed_receipt: None,
2179 
            error: None,
2180 
            peak_disk_mib: None,
2181 
            peak_memory_mib: None,
1215f05f Andy McCall ↶ 2182 
            phases: Default::default(),
8f211e20 Andy McCall ↶ 2183 
        };
3007d8f0 Andy McCall ↶ 2184 
        let quick = Duration::from_millis(10);
2185 
        let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
2186 
        hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
2187 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");
2188 
        let (base, seen) = fake_hub(vec![409]);
2189 
        assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());
2190 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
1b2229da Andy McCall ↶ 2191 
    }
2192
2193 
    #[test]
2194 
    fn server_errors_are_tried_again_and_client_errors_are_not() {
2195 
        let quick = [Duration::from_millis(10); 3];
2196 
        let (base, seen) = fake_hub(vec![500, 503, 204]);
2197 
        let r = hub_at(base).post_with("/flow/jobs/1/log", &serde_json::json!({}), &quick).unwrap();
2198 
        assert!(r.is_none());
2199 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 3, "two retries, then accepted");
2200
2201 
        let (base, seen) = fake_hub(vec![500, 500, 500, 500]);
2202 
        let e = hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).unwrap_err().to_string();
2203 
        assert!(e.contains("HTTP 500"), "{e}");
2204 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4, "gives up after the last delay");
2205
2206 
        let (base, seen) = fake_hub(vec![401]);
2207 
        assert!(hub_at(base).post_with("/flow/claim", &serde_json::json!({}), &quick).is_err());
2208 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");
06a4c472 Andy McCall ↶ 2209 
    }
2210
1b160d37 Andy McCall ↶ 2211 
    static NOT_LEAVING: std::sync::Mutex<Option<Leave>> = std::sync::Mutex::new(None);
06a4c472 Andy McCall ↶ 2212
2213 
    /// oxygenflow#76: once the runner leaves, a job's sink says it is
2214 
    /// cancelled, having sent what the job said so far, and the job is
2215 
    /// handed back to the hub.
2216 
    #[test]
2217 
    fn a_leaving_runner_stops_its_job_and_hands_it_back() {
2218 
        use microvm::Sink;
2219 
        let leaving = std::sync::Mutex::new(None);
2220 
        let (base, seen) = fake_hub(vec![204, 204]);
2221 
        let hub = hub_at(base);
1b160d37 Andy McCall ↶ 2222 
        let mut sink = LogSink {
2223 
            hub: &hub,
2224 
            job: 7,
2225 
            buf: Vec::new(),
2226 
            last: Instant::now(),
2227 
            gone: false,
2228 
            leaving: &leaving,
2229 
            started: Instant::now(),
2230 
            expected: None,
2231 
        };
06a4c472 Andy McCall ↶ 2232 
        sink.buf.push(Record { t: 1, stream: "out".into(), text: "step 1".into() });
2233 
        assert!(!sink.cancelled());
1b160d37 Andy McCall ↶ 2234 
        *leaving.lock().unwrap() = Some(Leave { reason: "off Wi-Fi".into(), by: Instant::now() });
06a4c472 Andy McCall ↶ 2235 
        assert!(sink.cancelled());
2236 
        assert!(sink.buf.is_empty(), "flushed before the job stops");
2237 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
2238 
        hub.release(7, "off Wi-Fi").unwrap();
2239 
        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 2);
1b160d37 Andy McCall ↶ 2240 
    }
2241
2242 
    /// oxygenflow#76: with notice, a job the hub expects to finish within
2243 
    /// it runs on; one that will not, or whose time is unknown, is handed
2244 
    /// back at once, and any job still running when the notice is up.
2245 
    #[test]
2246 
    fn a_job_that_fits_the_notice_finishes_and_the_others_go_back() {
2247 
        use microvm::Sink;
2248 
        let (base, _) = fake_hub(vec![204; 8]);
2249 
        let hub = hub_at(base);
2250 
        let leaving = std::sync::Mutex::new(Some(Leave { reason: "off charge".into(), by: Instant::now() + Duration::from_secs(60) }));
2251 
        let sink = |expected: Option<u64>, ran: u64| LogSink {
2252 
            hub: &hub,
2253 
            job: 1,
2254 
            buf: Vec::new(),
2255 
            last: Instant::now(),
2256 
            gone: false,
2257 
            leaving: &leaving,
2258 
            started: Instant::now() - Duration::from_secs(ran),
2259 
            expected: expected.map(Duration::from_secs),
2260 
        };
2261 
        assert!(!sink(Some(70), 30).cancelled(), "40 s left of it, within a minute: it finishes");
2262 
        assert!(sink(Some(600), 30).cancelled(), "nearly ten minutes left: handed back");
2263 
        assert!(sink(None, 0).cancelled(), "no idea how long: handed back");
2264 
        *leaving.lock().unwrap() = Some(Leave { reason: "off charge".into(), by: Instant::now() - Duration::from_secs(1) });
2265 
        assert!(sink(Some(70), 30).cancelled(), "the notice is up: handed back whatever was expected");
280ef951 Andy McCall ↶ 2266 
    }
2267
2268 
    /// oxygenflow#32: a quiet job asks the hub every few seconds whether
2269 
    /// it still runs; a 409 marks it cancelled, and nothing more is sent.
2270 
    #[test]
2271 
    fn a_quiet_job_learns_that_its_run_was_cancelled() {
2272 
        use microvm::Sink;
2273 
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
2274 
        let quiet = Instant::now() - HEARTBEAT - Duration::from_secs(1);
2275 
        let (base, seen) = fake_hub(vec![204]);
2276 
        let hub = hub_at(base);
1b160d37 Andy McCall ↶ 2277 
        let mut sink = LogSink {
2278 
            hub: &hub,
2279 
            job: 1,
2280 
            buf: Vec::new(),
2281 
            last: quiet,
2282 
            gone: false,
2283 
            leaving: &NOT_LEAVING,
2284 
            started: Instant::now(),
2285 
            expected: None,
2286 
        };
280ef951 Andy McCall ↶ 2287 
        assert!(!sink.cancelled());
2288 
        assert!(!sink.cancelled(), "asked again only after the heartbeat");
2289 
        assert_eq!(n(&seen), 1);
2290
2291 
        let (base, seen) = fake_hub(vec![409]);
2292 
        let hub = hub_at(base);
1b160d37 Andy McCall ↶ 2293 
        let mut sink = LogSink {
2294 
            hub: &hub,
2295 
            job: 1,
2296 
            buf: Vec::new(),
2297 
            last: quiet,
2298 
            gone: false,
2299 
            leaving: &NOT_LEAVING,
2300 
            started: Instant::now(),
2301 
            expected: None,
2302 
        };
280ef951 Andy McCall ↶ 2303 
        assert!(sink.cancelled());
2304 
        sink.record(Record { t: 0, stream: "out".into(), text: "late".into() });
2305 
        sink.flush();
2306 
        assert!(sink.cancelled());
2307 
        assert_eq!(n(&seen), 1, "nothing is sent for a cancelled job");
89171fca Andy McCall ↶ 2308 
    }
2309
2310 
    /// oxygenflow#49: a hub that is away for a while (a deploy restarting
2311 
    /// it) costs a job's uploads a wait, not the upload, and costs its log
2312 
    /// no lines.
2313 
    #[test]
2314 
    fn a_job_outlasts_a_hub_restart() {
2315 
        let n = |seen: &std::sync::Arc<std::sync::atomic::AtomicUsize>| seen.load(std::sync::atomic::Ordering::SeqCst);
2316 
        // A port nobody listens on yet: the hub is restarting.
2317 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
2318 
        let base = format!("http://127.0.0.1:{port}");
2319 
        let hub = hub_at(base.clone());
2320
2321 
        // Log lines sent while it is away are kept, and go when it is back.
1b160d37 Andy McCall ↶ 2322 
        let mut sink = LogSink {
2323 
            hub: &hub,
2324 
            job: 1,
2325 
            buf: Vec::new(),
2326 
            last: Instant::now(),
2327 
            gone: false,
2328 
            leaving: &NOT_LEAVING,
2329 
            started: Instant::now(),
2330 
            expected: None,
2331 
        };
89171fca Andy McCall ↶ 2332 
        for i in 0..450 {
2333 
            sink.buf.push(Record { t: i, stream: "out".into(), text: format!("line {i}") });
2334 
        }
2335 
        sink.send();
2336 
        assert_eq!(sink.buf.len(), 450, "nothing lost while the hub is away");
2337 
        let (_, seen) = fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![204, 204, 204]);
2338 
        sink.send();
2339 
        assert!(sink.buf.is_empty() && !sink.gone);
2340 
        assert_eq!(n(&seen), 3, "in batches of 200: 200, 200 and 50");
2341
2342 
        // An upload waits for the hub to come back, through refused
2343 
        // connections and server errors alike.
2344 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
2345 
        let url = format!("http://127.0.0.1:{port}/x");
2346 
        let agent = ureq::AgentBuilder::new().timeout(Duration::from_secs(5)).build();
2347 
        let later = std::thread::spawn(move || {
2348 
            std::thread::sleep(Duration::from_millis(150));
2349 
            fake_hub_on(std::net::TcpListener::bind(("127.0.0.1", port)).unwrap(), vec![502, 204])
2350 
        });
2351 
        let quick = [Duration::from_millis(50); 20];
2352 
        let r = retrying(&url, &quick, true, || agent.post(&url).send_string("{}").map_err(Box::new)).unwrap();
2353 
        assert_eq!(r.status(), 204);
2354 
        assert_eq!(n(&later.join().unwrap().1), 2, "a 502 from the proxy, then the hub");
2355 
        // Without `connection`, a hub that cannot be reached ends it at once.
2356 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
2357 
        let url = format!("http://127.0.0.1:{port}/x");
2358 
        assert!(retrying(&url, &quick, false, || agent.post(&url).send_string("{}").map_err(Box::new)).is_err());
2359 
        assert!(PATIENT_DELAYS.iter().sum::<Duration>() >= Duration::from_secs(180), "longer than a hub takes to restart");
2360 
    }
2361
2362 
    /// oxygenflow#50: the first stop drains (the second, not called here,
2363 
    /// exits at once).
2364 
    #[test]
2365 
    fn a_stop_drains_the_runner() {
2366 
        assert!(!stopping());
2367 
        on_stop(libc::SIGTERM);
2368 
        assert!(stopping(), "told to stop, it claims nothing more");
5cf1e8ae Andy McCall ↶ 2369 
        // An embedded runner starts afresh, and stop() drains it the same way.
2370 
        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
2371 
        stop();
2372 
        assert!(stopping());
2373 
        STOPS.store(0, std::sync::atomic::Ordering::SeqCst);
89171fca Andy McCall ↶ 2374 
    }
2375
2376 
    /// Lines kept for a hub that stays away are bounded, saying how many went.
2377 
    #[test]
2378 
    fn unsent_log_lines_are_bounded() {
2379 
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
2380 
        let hub = hub_at(format!("http://127.0.0.1:{port}"));
1b160d37 Andy McCall ↶ 2381 
        let mut sink = LogSink {
2382 
            hub: &hub,
2383 
            job: 1,
2384 
            buf: Vec::new(),
2385 
            last: Instant::now(),
2386 
            gone: false,
2387 
            leaving: &NOT_LEAVING,
2388 
            started: Instant::now(),
2389 
            expected: None,
2390 
        };
89171fca Andy McCall ↶ 2391 
        sink.buf = (0..MAX_UNSENT as u64 + 10).map(|i| Record { t: i, stream: "out".into(), text: String::new() }).collect();
2392 
        sink.send();
2393 
        assert_eq!(sink.buf.len(), MAX_UNSENT + 1);
2394 
        assert_eq!(sink.buf[0].text, "(10 log lines lost while the hub could not be reached)");
2395 
        assert_eq!(sink.buf[1].t, 10, "the oldest go first");
1b2229da Andy McCall ↶ 2396 
    }
bb643c25 Andy McCall ↶ 2397
2398 
    #[test]
2399 
    fn the_budget_admits_jobs_that_fit() {
2400 
        let b = Budget { max_jobs: 3, memory_mib: Some(24_576), cpus: Some(8) };
2401 
        let used = |jobs, memory_mib, cpus| Used { jobs, memory_mib, cpus };
2402 
        assert_eq!(b.room(used(0, 0, 0)), Some((None, None)), "a lone job may take everything");
2403 
        assert_eq!(b.room(used(1, 8192, 4)), Some((Some(16_384), Some(4))));
2404 
        assert_eq!(b.room(used(2, 16_384, 8)), None, "no vCPU left");
2405 
        assert_eq!(b.room(used(3, 1024, 1)), None, "at max_jobs");
2406 
        let one = Budget { max_jobs: 1, memory_mib: None, cpus: None };
2407 
        assert_eq!(one.room(used(0, 0, 0)), Some((None, None)));
2408 
        assert_eq!(one.room(used(1, 0, 0)), None, "the default runs one job at a time");
2409 
        let open = Budget { max_jobs: 2, memory_mib: None, cpus: None };
2410 
        assert_eq!(open.room(used(1, 8192, 4)), Some((None, None)), "no limits configured");
77c8c240 Andy McCall ↶ 2411 
    }
2412
2413 
    #[test]
2414 
    fn a_claim_never_asks_for_more_memory_than_the_host_has_free() {
2415 
        assert_eq!(claim_memory(None, Some(6541)), Some(6541), "idle: the host's free memory, not unlimited");
2416 
        assert_eq!(claim_memory(Some(16_384), Some(6541)), Some(6541));
2417 
        assert_eq!(claim_memory(Some(4096), Some(6541)), Some(4096), "the budget when it is smaller");
2418 
        assert_eq!(claim_memory(None, None), None, "nothing known: no limit, as before");
2419 
        assert_eq!(claim_memory(Some(4096), None), Some(4096));
bb643c25 Andy McCall ↶ 2420 
    }
e0918b88 Andy McCall 2421 
}