OXYGENHUB
# 72772aa01d6b oxygenflow / crates / flow / src / main.rs
25.4 KiB · Rust · last changed flow runner update --from FILE: the runner's side of runner updates 10 days ago source blame history raw
1
//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
2
//! (docs/spec.md, docs/m0.md).
3
4
use flow::{facts, home, images, microvm, runner, syscall, wait, wasm, workspace};
5
6
use anyhow::{bail, Context, Result};
7
use clap::{Parser, Subcommand};
8
use flow_core::api::Record;
9
use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
10
use home::Home;
11
use std::collections::BTreeSet;
12
use std::path::{Path, PathBuf};
13
use std::time::Duration;
14
15
#[derive(Parser)]
16
#[command(name = "flow", version = env!("OXY_VERSION_LONG"), about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
17
struct Cli {
18
    /// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
19
    #[arg(long, global = true)]
20
    home: Option<PathBuf>,
21
    #[command(subcommand)]
22
    cmd: Cmd,
23
}
24
25
#[derive(Subcommand)]
26
enum RunnerCmd {
27
    /// Register this runner with a hub using a one-time token from its Admin → Runners page
28
    Register {
29
        /// The hub, e.g. https://earth.example:7343
30
        hub: String,
31
        /// The registration token (read from stdin when not given)
32
        #[arg(long)]
33
        token: Option<String>,
34
        /// The runner's name (default: the host name)
35
        #[arg(long)]
36
        name: Option<String>,
37
        /// The hub's CA certificate, when self-signed
38
        #[arg(long)]
39
        ca: Option<PathBuf>,
40
        /// A label the runner offers (repeatable)
41
        #[arg(long)]
42
        label: Vec<String>,
43
    },
44
    /// Measure this machine as a runner does: its speed score (100 on the
45
    /// reference machine) and class, free disk and load (spec §11.1)
46
    Bench,
47
    /// Install a new flow build in place of this one (docs/jetstream.md
48
    /// §8.2): check it, keep this one as flow.prev, and have the running
49
    /// runner leave with notice so its unit starts the new build. A build
50
    /// that does not reach the hub goes back to flow.prev by itself
51
    Update {
52
        /// The new flow binary
53
        #[arg(long)]
54
        from: PathBuf,
55
        /// Seconds the running runner has to finish what it holds; jobs
56
        /// that will not finish in time are handed back with their checkpoints
57
        #[arg(long, default_value_t = 600)]
58
        notice: u64,
59
        /// Check the build and say what would happen, changing nothing
60
        #[arg(long)]
61
        dry_run: bool,
62
        /// Install a build older than this one
63
        #[arg(long)]
64
        allow_older: bool,
65
        /// The installed binary to replace (default: this one)
66
        #[arg(long)]
67
        exe: Option<PathBuf>,
68
    },
69
}
70
71
#[derive(Subcommand)]
72
enum Cmd {
73
    /// Validate a pipeline and print its jobs and hash
74
    Check {
75
        /// The pipeline file (default: .flow/pipeline.kdl)
76
        file: Option<PathBuf>,
77
    },
78
    /// Print a file's BLAKE3, for pinning in flow.toml
79
    Hash { file: PathBuf },
80
    /// Rootfs images for sandboxes
81
    Rootfs {
82
        #[command(subcommand)]
83
        cmd: RootfsCmd,
84
    },
85
    /// Run the pipeline on this directory's files, locally
86
    Run {
87
        /// Only these jobs (and what they depend on)
88
        jobs: Vec<String>,
89
        /// The directory to build (default: the current one)
90
        #[arg(long, default_value = ".")]
91
        dir: PathBuf,
92
        /// Minutes before a job is stopped
93
        #[arg(long, default_value_t = 60)]
94
        timeout: u64,
95
        /// A variable for every job, as the hub's settings would give it;
96
        /// the pipeline's own `env` wins
97
        #[arg(long = "var", value_name = "NAME=VALUE")]
98
        vars: Vec<String>,
99
        /// A secret a job's `secret` block names, as the hub would supply
100
        /// it; mounted as a file, never in the environment
101
        #[arg(long = "secret", value_name = "NAME=VALUE")]
102
        secrets: Vec<String>,
103
    },
104
    /// Claim and run jobs for the hub in flow.toml's [runner] (or the
105
    /// runner.toml `flow runner register` wrote)
106
    Runner {
107
        #[command(subcommand)]
108
        cmd: Option<RunnerCmd>,
109
    },
110
    /// Follow the checks of a pushed state on OxygenHub until they finish:
111
    /// exit 0 when all passed, 1 when one failed, 2 at the time limit.
112
    /// Uses $OXY_TOKEN and $OXY_CA as oxy does.
113
    Wait {
114
        /// The repository's URL on the hub (https://hub/owner/name)
115
        url: String,
116
        /// The state hash the push moved the bookmark to (`oxy push --json` prints it)
117
        state: String,
118
        /// Minutes to wait at most
119
        #[arg(long, default_value_t = 60)]
120
        timeout: u64,
121
    },
122
    /// Show which isolation tiers this machine can run jobs in, and why
123
    /// the others cannot
124
    Tiers,
125
    /// The syscall tier's sandbox helper, started by the runner for each job
126
    #[command(name = "__sandbox", hide = true)]
127
    Sandbox { spec: PathBuf },
128
    /// Check that this machine can run the syscall tier
129
    #[command(name = "__sandbox-probe", hide = true)]
130
    SandboxProbe,
131
    /// The WASM tier's helper, started by the runner for each job
132
    #[command(name = "__wasm", hide = true)]
133
    Wasm { spec: PathBuf },
134
}
135
136
#[derive(Subcommand)]
137
enum RootfsCmd {
138
    /// Make an ext4 rootfs image from a root file system tarball and store it by hash
139
    Import {
140
        tarball: PathBuf,
141
        /// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
142
        #[arg(long)]
143
        sha256: Option<String>,
144
        /// The image's own environment, NAME=value (repeatable): what a
145
        /// Dockerfile's ENV sets and `docker export` loses. Jobs start from
146
        /// it; their sandbox's and their own `env` override it
147
        #[arg(long = "env", value_name = "NAME=VALUE")]
148
        env: Vec<String>,
149
    },
150
    /// List the images this home holds
151
    List,
152
    /// Send an image to a hub, so its runners fetch it instead of having it
153
    /// copied to them: only the chunks the hub lacks travel. Uses
154
    /// $OXY_TOKEN (a token that may write repositories) and $OXY_CA
155
    Push {
156
        /// The image's hash, as pipelines pin it (cas://blob/HASH)
157
        hash: String,
158
        /// The hub (default: the runner's, from runner.toml or flow.toml)
159
        #[arg(long)]
160
        hub: Option<String>,
161
    },
162
    /// Fetch an image a hub holds into this home, as a runner does for a
163
    /// job whose image it lacks. Uses $OXY_TOKEN and $OXY_CA
164
    Pull {
165
        hash: String,
166
        #[arg(long)]
167
        hub: Option<String>,
168
    },
169
}
170
171
fn main() {
172
    let cli = Cli::parse();
173
    if let Err(e) = run(cli) {
174
        eprintln!("flow: {e:#}");
175
        std::process::exit(1);
176
    }
177
}
178
179
fn run(cli: Cli) -> Result<()> {
180
    match cli.cmd {
181
        Cmd::Sandbox { spec } => std::process::exit(syscall::sandbox_main(&spec)),
182
        Cmd::SandboxProbe => std::process::exit(syscall::probe_main()),
183
        Cmd::Wasm { spec } => std::process::exit(wasm::wasm_main(&spec)),
184
        Cmd::Check { file } => check(&file.unwrap_or_else(|| PathBuf::from(PIPELINE_PATH))),
185
        Cmd::Hash { file } => {
186
            println!("{}", home::hash_file(&file)?);
187
            Ok(())
188
        }
189
        Cmd::Rootfs { cmd } => {
190
            let home = Home::locate(cli.home.as_deref())?;
191
            match cmd {
192
                RootfsCmd::Import { tarball, sha256, env } => rootfs_import(&home, &tarball, sha256.as_deref(), &env),
193
                RootfsCmd::Push { hash, hub } => {
194
                    let client = image_client(&home, hub)?;
195
                    eprintln!("Sending image {} …", &hash[..12.min(hash.len())]);
196
                    let m = client.push(&home, &hash)?;
197
                    println!(
198
                        "Sent image {}: {} chunks the hub lacked, {} in {:.1} s",
199
                        &hash[..12],
200
                        m.chunks,
201
                        images::human(m.bytes),
202
                        m.elapsed.as_secs_f64()
203
                    );
204
                    Ok(())
205
                }
206
                RootfsCmd::Pull { hash, hub } => {
207
                    let client = image_client(&home, hub)?;
208
                    match client.fetch(&home, &hash)? {
209
                        Some(m) => {
210
                            println!(
211
                                "Fetched image {}: {} chunks, {} in {:.1} s",
212
                                &hash[..12],
213
                                m.chunks,
214
                                images::human(m.bytes),
215
                                m.elapsed.as_secs_f64()
216
                            );
217
                            Ok(())
218
                        }
219
                        None => anyhow::bail!("the hub does not hold image {}; send it with flow rootfs push", &hash[..12.min(hash.len())]),
220
                    }
221
                }
222
                RootfsCmd::List => {
223
                    for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
224
                        let size = e.metadata().map(|m| m.len()).unwrap_or(0);
225
                        if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
226
                            println!("cas://blob/{h}  {} MiB", size >> 20);
227
                        }
228
                    }
229
                    Ok(())
230
                }
231
            }
232
        }
233
        Cmd::Run { jobs, dir, timeout, vars, secrets } => {
234
            let home = Home::locate(cli.home.as_deref())?;
235
            let pairs = |what: &str, list: &[String]| {
236
                list.iter()
237
                    .map(|e| {
238
                        e.split_once('=')
239
                            .map(|(k, v)| (k.to_string(), v.to_string()))
240
                            .ok_or_else(|| anyhow::anyhow!("--{what} {e:?}: expected NAME=VALUE"))
241
                    })
242
                    .collect::<Result<Vec<_>>>()
243
            };
244
            let (vars, secrets) = (pairs("var", &vars)?, pairs("secret", &secrets)?);
245
            run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60), &vars, &secrets)
246
        }
247
        Cmd::Runner { cmd: Some(RunnerCmd::Bench) } => {
248
            let home = Home::locate(cli.home.as_deref())?;
249
            let (speed, ms) = facts::bench();
250
            let f = facts::now(&home, Some(speed));
251
            println!("speed {speed} (class {}), benchmark {ms} ms", flow_core::api::Facts::class_of(speed));
252
            println!("free disk {} MiB at {}", f.free_disk_mib.unwrap_or(0), home.root.display());
253
            println!("load {:.2}", f.load_x100.unwrap_or(0) as f64 / 100.0);
254
            if let Some(k) = f.kernel {
255
                println!("kernel {k}");
256
            }
257
            Ok(())
258
        }
259
        Cmd::Runner { cmd: Some(RunnerCmd::Register { hub, token, name, ca, label }) } => {
260
            let home = Home::locate(cli.home.as_deref())?;
261
            if std::env::var_os("OXY_HOME").is_none() {
262
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
263
            }
264
            let name = name.unwrap_or_else(|| {
265
                std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_else(|_| "flow".into())
266
            });
267
            let token = match token {
268
                Some(t) => t,
269
                None => {
270
                    eprintln!("Paste the registration token from the hub's Admin → Runners, then press Enter:");
271
                    let mut line = String::new();
272
                    std::io::stdin().read_line(&mut line)?;
273
                    line.trim().to_string()
274
                }
275
            };
276
            let got = runner::register(&home, &hub, &token, &name, ca, label)?;
277
            println!("registered as {} (runner {}); start it with `flow runner`", got.name, got.id);
278
            Ok(())
279
        }
280
        Cmd::Runner { cmd: Some(RunnerCmd::Update { from, notice, dry_run, allow_older, exe }) } => {
281
            let home = Home::locate(cli.home.as_deref())?;
282
            let exe = match exe {
283
                Some(e) => e,
284
                None => std::env::current_exe().context("finding this flow binary")?,
285
            };
286
            let done = flow::update::install(&home, &exe, &from, Duration::from_secs(notice), dry_run, allow_older)?;
287
            if dry_run {
288
                println!(
289
                    "flow {} checks out: it would replace flow {} at {} (dry run: nothing changed)",
290
                    done.to,
291
                    done.from,
292
                    done.exe.display()
293
                );
294
                return Ok(());
295
            }
296
            println!("flow {} installed at {}; flow {} kept as {}", done.to, done.exe.display(), done.from, flow::update::PREV);
297
            match done.told {
298
                Some(pid) => println!("the runner (pid {pid}) leaves within {notice} s, then its unit starts flow {}", done.to),
299
                None => println!(
300
                    "no runner of this home was told (none runs, or one from before runner updates, which does not listen): \
301
                     restart its unit, e.g. `systemctl restart --no-block flow-runner`; it drains and starts flow {}",
302
                    done.to
303
                ),
304
            }
305
            Ok(())
306
        }
307
        Cmd::Runner { cmd: None } => {
308
            let home = Home::locate(cli.home.as_deref())?;
309
            // A new build on trial that keeps failing to start goes back to
310
            // the one before, which then runs in its place (#85).
311
            if let flow::update::Start::RolledBack(p) = flow::update::on_start(&home.root)? {
312
                use std::os::unix::process::CommandExt;
313
                let err = std::process::Command::new(&p.exe).args(std::env::args_os().skip(1)).exec();
314
                bail!("starting flow {} again after the rollback: {err}", p.from);
315
            }
316
            // The runner's oxy state (identity, user config) stays in its home.
317
            if std::env::var_os("OXY_HOME").is_none() {
318
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
319
            }
320
            runner::serve(&home)
321
        }
322
        Cmd::Tiers => {
323
            let home = Home::locate(cli.home.as_deref())?;
324
            match runner::tiers(&home) {
325
                Ok((have, missing)) => {
326
                    for t in have {
327
                        println!("{t}: available");
328
                    }
329
                    for why in missing {
330
                        println!("{}", why.replacen(": ", ": unavailable: ", 1));
331
                    }
332
                }
333
                Err(why) => println!("{why}"),
334
            }
335
            Ok(())
336
        }
337
        Cmd::Wait { url, state, timeout } => {
338
            let token = std::env::var("OXY_TOKEN").ok();
339
            let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
340
            let code =
341
                wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
342
            std::process::exit(code);
343
        }
344
    }
345
}
346
347
/// The hub image commands talk to: `--hub`, else the runner's; with
348
/// $OXY_TOKEN, else the runner's own token.
349
fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
350
    let runner = home.config.runner.clone();
351
    let hub = hub.or_else(|| runner.as_ref().map(|r| r.hub.clone())).context("name the hub with --hub https://HUB")?;
352
    let token = std::env::var("OXY_TOKEN")
353
        .ok()
354
        .or_else(|| runner.as_ref().map(|r| r.token.clone()))
355
        .context("set OXY_TOKEN to a token of yours on the hub")?;
356
    let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or_else(|| runner.as_ref().and_then(|r| r.ca.clone()));
357
    images::Client::new(&hub, &token, ca.as_deref())
358
}
359
360
fn load(file: &Path) -> Result<Pipeline> {
361
    let src = std::fs::read_to_string(file).with_context(|| format!("read {}", file.display()))?;
362
    Pipeline::parse(&src).map_err(|p| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
363
}
364
365
fn check(file: &Path) -> Result<()> {
366
    let p = load(file)?;
367
    let depths = p.depths();
368
    for j in p.order() {
369
        let sb = &p.sandboxes[&j.sandbox];
370
        let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
371
        println!("{}{}  {} [{}]{after}", "  ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
372
    }
373
    println!("pipeline {}", hex::encode(p.hash()));
374
    Ok(())
375
}
376
377
/// The image environment file flow-init reads (`/etc/flow/env`): one
378
/// `NAME=value` per line, names checked as a pipeline's are.
379
fn image_env_file(env: &[String]) -> Result<String> {
380
    let mut out = String::new();
381
    for e in env {
382
        let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
383
        let ok = !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
384
        if !ok || v.contains('\n') || v.contains('\0') {
385
            bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
386
        }
387
        out.push_str(&format!("{k}={v}\n"));
388
    }
389
    Ok(out)
390
}
391
392
/// The architecture an ELF file was built for, by its header.
393
fn elf_arch(path: &Path) -> Option<&'static str> {
394
    let bytes = std::fs::read(path).ok()?;
395
    if bytes.len() < 20 || &bytes[..4] != b"\x7fELF" {
396
        return None;
397
    }
398
    let little = bytes[5] == 1;
399
    let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
400
    Some(match machine {
401
        0x3E => "x86_64",
402
        0xB7 => "aarch64",
403
        0xF3 => "riscv64",
404
        0x03 => "x86",
405
        0x28 => "arm",
406
        0x15 => "powerpc64",
407
        0x16 => "s390x",
408
        _ => return None,
409
    })
410
}
411
412
/// The architecture of an unpacked rootfs: its shell's, or the first
413
/// program's found under bin.
414
fn tree_arch(tree: &Path) -> Option<&'static str> {
415
    for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
416
        let p = tree.join(rel);
417
        let p = std::fs::canonicalize(&p).unwrap_or(p);
418
        if let Some(a) = elf_arch(&p) {
419
            return Some(a);
420
        }
421
    }
422
    None
423
}
424
425
fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
426
    let env_file = image_env_file(env)?;
427
    if let Some(want) = sha256 {
428
        let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
429
        let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
430
        if got != want.to_lowercase() {
431
            bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
432
        }
433
    }
434
    let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
435
    let _ = std::fs::remove_dir_all(&tmp);
436
    let tree = tmp.join("tree");
437
    std::fs::create_dir_all(&tree)?;
438
    let result = (|| -> Result<String> {
439
        let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
440
        if !status.success() {
441
            bail!("tar could not unpack {}", tarball.display());
442
        }
443
        // The image's architecture must be this machine's: there is no
444
        // emulation tier (spec §11.2).
445
        let arch = tree_arch(&tree).unwrap_or("unknown");
446
        if arch != "unknown" && arch != std::env::consts::ARCH {
447
            bail!(
448
                "{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
449
                tarball.display(),
450
                std::env::consts::ARCH
451
            );
452
        }
453
        // Mount points flow-init needs (docs/m0.md §2).
454
        for d in ["src", "tmp", "proc", "dev", "run"] {
455
            std::fs::create_dir_all(tree.join(d))?;
456
        }
457
        if !env_file.is_empty() {
458
            std::fs::create_dir_all(tree.join("etc/flow"))?;
459
            std::fs::write(tree.join("etc/flow/env"), &env_file)?;
460
        }
461
        let used: u64 = ignore::WalkBuilder::new(&tree)
462
            .hidden(false)
463
            .ignore(false)
464
            .git_ignore(false)
465
            .build()
466
            .flatten()
467
            .filter_map(|e| e.metadata().ok())
468
            .map(|m| m.len().max(4096))
469
            .sum();
470
        let mib = used * 13 / 10 / (1 << 20) + 64;
471
        let img = tmp.join("rootfs.ext4");
472
        let status = std::process::Command::new("mkfs.ext4")
473
            .args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
474
            .arg(&tree)
475
            .arg(&img)
476
            .arg(format!("{mib}M"))
477
            .status()
478
            .context("run mkfs.ext4")?;
479
        if !status.success() {
480
            bail!("mkfs.ext4 failed");
481
        }
482
        let hash = home::hash_file(&img)?;
483
        std::fs::rename(&img, home.rootfs(&hash))?;
484
        let manifest = serde_json::json!({ "arch": if arch == "unknown" { std::env::consts::ARCH } else { arch }, "imported_ms": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0) });
485
        std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
486
        Ok(hash)
487
    })();
488
    let _ = std::fs::remove_dir_all(&tmp);
489
    let hash = result?;
490
    println!("rootfs \"cas://blob/{hash}\"");
491
    Ok(())
492
}
493
494
fn run_local(
495
    home: &Home,
496
    dir: &Path,
497
    only: &[String],
498
    timeout: Duration,
499
    vars: &[(String, String)],
500
    secrets: &[(String, String)],
501
) -> Result<()> {
502
    let p = load(&dir.join(PIPELINE_PATH))?;
503
    // The named jobs and everything they depend on.
504
    let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
505
    for name in only {
506
        if !p.jobs.contains_key(name) {
507
            bail!("no job {name:?} in the pipeline");
508
        }
509
    }
510
    if !only.is_empty() {
511
        loop {
512
            let more: Vec<String> = wanted.iter().flat_map(|j| p.jobs[j].depends_on.clone()).filter(|d| !wanted.contains(d)).collect();
513
            if more.is_empty() {
514
                break;
515
            }
516
            wanted.extend(more);
517
        }
518
    }
519
    let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
520
    let _ = std::fs::remove_dir_all(&scratch);
521
    std::fs::create_dir_all(scratch.join("tree"))?;
522
    workspace::copy_tree(dir, &scratch.join("tree"))?;
523
    let mut failed: BTreeSet<String> = BTreeSet::new();
524
    let mut worst = 0;
525
    for job in p.order() {
526
        if !only.is_empty() && !wanted.contains(&job.name) {
527
            continue;
528
        }
529
        if let Some(d) = job.depends_on.iter().find(|d| failed.contains(*d)) {
530
            println!("── {} skipped: {d} failed", job.name);
531
            failed.insert(job.name.clone());
532
            continue;
533
        }
534
        let sb = &p.sandboxes[&job.sandbox];
535
        println!("── {} ({})", job.name, job.exec.join(" "));
536
        let js = scratch.join(&job.name);
537
        std::fs::create_dir_all(&js)?;
538
        let job = &job
539
            .in_sandbox(sb)
540
            .with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
541
            .with_flow_env(vars);
542
        // Identities come from a hub: a local run has none.
543
        for i in &job.identities {
544
            println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
545
        }
546
        // Each secret the job declares must be given, as the hub would.
547
        let inputs = job
548
            .secrets
549
            .iter()
550
            .map(|s| {
551
                let value = secrets.iter().find(|(k, _)| *k == s.name).map(|(_, v)| v.clone());
552
                value
553
                    .map(|value| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
554
                    .ok_or_else(|| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
555
            })
556
            .collect::<Result<Vec<_>>>()?;
557
        // Each job starts from the same files: jobs do not share writes.
558
        let out = if sb.runtime == Runtime::Wasm {
559
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
560
            wasm::run(sb, job, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r))?
561
        } else if sb.runtime == Runtime::Syscall {
562
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
563
            let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
564
            syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r), &group)?
565
        } else {
566
            workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
567
            microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut |r: Record| print_record(&r))?
568
        };
569
        let _ = workspace::remove_tree(&js);
570
        match out.exit_code {
571
            Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
572
            other => {
573
                println!(
574
                    "── {} failed ({}) in {} ms",
575
                    job.name,
576
                    other.map(|c| format!("exit {c}")).unwrap_or_else(|| "no exit code".into()),
577
                    out.duration_ms
578
                );
579
                failed.insert(job.name.clone());
580
                worst = 1;
581
            }
582
        }
583
    }
584
    let _ = workspace::remove_tree(&scratch);
585
    if worst != 0 {
586
        std::process::exit(worst);
587
    }
588
    Ok(())
589
}
590
591
fn print_record(r: &Record) {
592
    match r.stream.as_str() {
593
        "out" => println!("   {}", r.text),
594
        "err" => eprintln!("   {}", r.text),
595
        s => eprintln!("   [{s}] {}", r.text),
596
    }
597
}
598
599
#[cfg(test)]
600
mod tests {
601
    use super::image_env_file;
602
603
    #[test]
604
    fn image_environments_are_checked_lines() {
605
        let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
606
        assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
607
        assert!(image_env_file(&["NOEQUALS".into()]).is_err());
608
        assert!(image_env_file(&["1X=y".into()]).is_err());
609
        assert!(image_env_file(&["X=two\nlines".into()]).is_err());
610
        assert_eq!(image_env_file(&[]).unwrap(), "");
611
    }
612
}