OXYGENHUB
# e6949d87205b oxygenflow / crates / flow / src / main.rs
25.6 KiB · Rust · last changed flow builds for Windows, offering the WASM tier 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
                #[cfg(unix)]
313
                {
314
                    use std::os::unix::process::CommandExt;
315
                    let err = std::process::Command::new(&p.exe).args(std::env::args_os().skip(1)).exec();
316
                    bail!("starting flow {} again after the rollback: {err}", p.from);
317
                }
318
                #[cfg(not(unix))]
319
                bail!("rolled back to flow {}; start the runner again", p.from);
320
            }
321
            // The runner's oxy state (identity, user config) stays in its home.
322
            if std::env::var_os("OXY_HOME").is_none() {
323
                std::env::set_var("OXY_HOME", home.root.join("oxy"));
324
            }
325
            runner::serve(&home)
326
        }
327
        Cmd::Tiers => {
328
            let home = Home::locate(cli.home.as_deref())?;
329
            match runner::tiers(&home) {
330
                Ok((have, missing)) => {
331
                    for t in have {
332
                        println!("{t}: available");
333
                    }
334
                    for why in missing {
335
                        println!("{}", why.replacen(": ", ": unavailable: ", 1));
336
                    }
337
                }
338
                Err(why) => println!("{why}"),
339
            }
340
            Ok(())
341
        }
342
        Cmd::Wait { url, state, timeout } => {
343
            let token = std::env::var("OXY_TOKEN").ok();
344
            let ca = std::env::var_os("OXY_CA").map(PathBuf::from);
345
            let code =
346
                wait::wait(&url, &state, token.as_deref(), ca.as_deref(), Duration::from_secs(timeout * 60), Duration::from_secs(15))?;
347
            std::process::exit(code);
348
        }
349
    }
350
}
351
352
/// The hub image commands talk to: `--hub`, else the runner's; with
353
/// $OXY_TOKEN, else the runner's own token.
354
fn image_client(home: &Home, hub: Option<String>) -> Result<images::Client> {
355
    let runner = home.config.runner.clone();
356
    let hub = hub.or_else(|| runner.as_ref().map(|r| r.hub.clone())).context("name the hub with --hub https://HUB")?;
357
    let token = std::env::var("OXY_TOKEN")
358
        .ok()
359
        .or_else(|| runner.as_ref().map(|r| r.token.clone()))
360
        .context("set OXY_TOKEN to a token of yours on the hub")?;
361
    let ca = std::env::var_os("OXY_CA").map(PathBuf::from).or_else(|| runner.as_ref().and_then(|r| r.ca.clone()));
362
    images::Client::new(&hub, &token, ca.as_deref())
363
}
364
365
fn load(file: &Path) -> Result<Pipeline> {
366
    let src = std::fs::read_to_string(file).with_context(|| format!("read {}", file.display()))?;
367
    Pipeline::parse(&src).map_err(|p| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
368
}
369
370
fn check(file: &Path) -> Result<()> {
371
    let p = load(file)?;
372
    let depths = p.depths();
373
    for j in p.order() {
374
        let sb = &p.sandboxes[&j.sandbox];
375
        let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
376
        println!("{}{}  {} [{}]{after}", "  ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
377
    }
378
    println!("pipeline {}", hex::encode(p.hash()));
379
    Ok(())
380
}
381
382
/// The image environment file flow-init reads (`/etc/flow/env`): one
383
/// `NAME=value` per line, names checked as a pipeline's are.
384
fn image_env_file(env: &[String]) -> Result<String> {
385
    let mut out = String::new();
386
    for e in env {
387
        let Some((k, v)) = e.split_once('=') else { bail!("--env {e:?}: expected NAME=value") };
388
        let ok = !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
389
        if !ok || v.contains('\n') || v.contains('\0') {
390
            bail!("--env {e:?}: a name is letters, digits and _, and a value is one line");
391
        }
392
        out.push_str(&format!("{k}={v}\n"));
393
    }
394
    Ok(out)
395
}
396
397
/// The architecture an ELF file was built for, by its header.
398
fn elf_arch(path: &Path) -> Option<&'static str> {
399
    let bytes = std::fs::read(path).ok()?;
400
    if bytes.len() < 20 || &bytes[..4] != b"\x7fELF" {
401
        return None;
402
    }
403
    let little = bytes[5] == 1;
404
    let machine = if little { u16::from_le_bytes([bytes[18], bytes[19]]) } else { u16::from_be_bytes([bytes[18], bytes[19]]) };
405
    Some(match machine {
406
        0x3E => "x86_64",
407
        0xB7 => "aarch64",
408
        0xF3 => "riscv64",
409
        0x03 => "x86",
410
        0x28 => "arm",
411
        0x15 => "powerpc64",
412
        0x16 => "s390x",
413
        _ => return None,
414
    })
415
}
416
417
/// The architecture of an unpacked rootfs: its shell's, or the first
418
/// program's found under bin.
419
fn tree_arch(tree: &Path) -> Option<&'static str> {
420
    for rel in ["bin/sh", "bin/busybox", "usr/bin/env", "bin/bash", "usr/bin/sh"] {
421
        let p = tree.join(rel);
422
        let p = std::fs::canonicalize(&p).unwrap_or(p);
423
        if let Some(a) = elf_arch(&p) {
424
            return Some(a);
425
        }
426
    }
427
    None
428
}
429
430
fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>, env: &[String]) -> Result<()> {
431
    let env_file = image_env_file(env)?;
432
    if let Some(want) = sha256 {
433
        let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
434
        let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
435
        if got != want.to_lowercase() {
436
            bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
437
        }
438
    }
439
    let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
440
    let _ = std::fs::remove_dir_all(&tmp);
441
    let tree = tmp.join("tree");
442
    std::fs::create_dir_all(&tree)?;
443
    let result = (|| -> Result<String> {
444
        let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
445
        if !status.success() {
446
            bail!("tar could not unpack {}", tarball.display());
447
        }
448
        // The image's architecture must be this machine's: there is no
449
        // emulation tier (spec §11.2).
450
        let arch = tree_arch(&tree).unwrap_or("unknown");
451
        if arch != "unknown" && arch != std::env::consts::ARCH {
452
            bail!(
453
                "{} is a {arch} image; this runner is {}, and jobs here run no other architecture",
454
                tarball.display(),
455
                std::env::consts::ARCH
456
            );
457
        }
458
        // Mount points flow-init needs (docs/m0.md §2).
459
        for d in ["src", "tmp", "proc", "dev", "run"] {
460
            std::fs::create_dir_all(tree.join(d))?;
461
        }
462
        if !env_file.is_empty() {
463
            std::fs::create_dir_all(tree.join("etc/flow"))?;
464
            std::fs::write(tree.join("etc/flow/env"), &env_file)?;
465
        }
466
        let used: u64 = ignore::WalkBuilder::new(&tree)
467
            .hidden(false)
468
            .ignore(false)
469
            .git_ignore(false)
470
            .build()
471
            .flatten()
472
            .filter_map(|e| e.metadata().ok())
473
            .map(|m| m.len().max(4096))
474
            .sum();
475
        let mib = used * 13 / 10 / (1 << 20) + 64;
476
        let img = tmp.join("rootfs.ext4");
477
        let status = std::process::Command::new("mkfs.ext4")
478
            .args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
479
            .arg(&tree)
480
            .arg(&img)
481
            .arg(format!("{mib}M"))
482
            .status()
483
            .context("run mkfs.ext4")?;
484
        if !status.success() {
485
            bail!("mkfs.ext4 failed");
486
        }
487
        let hash = home::hash_file(&img)?;
488
        std::fs::rename(&img, home.rootfs(&hash))?;
489
        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) });
490
        std::fs::write(home.image_manifest(&hash), manifest.to_string())?;
491
        Ok(hash)
492
    })();
493
    let _ = std::fs::remove_dir_all(&tmp);
494
    let hash = result?;
495
    println!("rootfs \"cas://blob/{hash}\"");
496
    Ok(())
497
}
498
499
fn run_local(
500
    home: &Home,
501
    dir: &Path,
502
    only: &[String],
503
    timeout: Duration,
504
    vars: &[(String, String)],
505
    secrets: &[(String, String)],
506
) -> Result<()> {
507
    let p = load(&dir.join(PIPELINE_PATH))?;
508
    // The named jobs and everything they depend on.
509
    let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
510
    for name in only {
511
        if !p.jobs.contains_key(name) {
512
            bail!("no job {name:?} in the pipeline");
513
        }
514
    }
515
    if !only.is_empty() {
516
        loop {
517
            let more: Vec<String> = wanted.iter().flat_map(|j| p.jobs[j].depends_on.clone()).filter(|d| !wanted.contains(d)).collect();
518
            if more.is_empty() {
519
                break;
520
            }
521
            wanted.extend(more);
522
        }
523
    }
524
    let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
525
    let _ = std::fs::remove_dir_all(&scratch);
526
    std::fs::create_dir_all(scratch.join("tree"))?;
527
    workspace::copy_tree(dir, &scratch.join("tree"))?;
528
    let mut failed: BTreeSet<String> = BTreeSet::new();
529
    let mut worst = 0;
530
    for job in p.order() {
531
        if !only.is_empty() && !wanted.contains(&job.name) {
532
            continue;
533
        }
534
        if let Some(d) = job.depends_on.iter().find(|d| failed.contains(*d)) {
535
            println!("── {} skipped: {d} failed", job.name);
536
            failed.insert(job.name.clone());
537
            continue;
538
        }
539
        let sb = &p.sandboxes[&job.sandbox];
540
        println!("── {} ({})", job.name, job.exec.join(" "));
541
        let js = scratch.join(&job.name);
542
        std::fs::create_dir_all(&js)?;
543
        let job = &job
544
            .in_sandbox(sb)
545
            .with_flow_env(&[("FLOW_JOB".into(), job.name.clone()), ("FLOW_RUN".into(), "local".into())])
546
            .with_flow_env(vars);
547
        // Identities come from a hub: a local run has none.
548
        for i in &job.identities {
549
            println!("── {}: identity {:?} is not available locally; {} will be missing", job.name, i.audience, i.mount_path);
550
        }
551
        // Each secret the job declares must be given, as the hub would.
552
        let inputs = job
553
            .secrets
554
            .iter()
555
            .map(|s| {
556
                let value = secrets.iter().find(|(k, _)| *k == s.name).map(|(_, v)| v.clone());
557
                value
558
                    .map(|value| flow_core::api::SecretInput { name: s.name.clone(), path: s.mount_path.clone(), value })
559
                    .ok_or_else(|| anyhow::anyhow!("job {}: secret {:?} is not given; pass --secret {}=VALUE", job.name, s.name, s.name))
560
            })
561
            .collect::<Result<Vec<_>>>()?;
562
        // Each job starts from the same files: jobs do not share writes.
563
        let out = if sb.runtime == Runtime::Wasm {
564
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
565
            wasm::run(sb, job, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r))?
566
        } else if sb.runtime == Runtime::Syscall {
567
            workspace::copy_tree(&scratch.join("tree"), &js.join("tree"))?;
568
            let group = format!("local-{}-{}", std::process::id(), js.file_name().unwrap_or_default().to_string_lossy());
569
            syscall::run(home, sb, job, &inputs, &js.join("tree"), &js, timeout, &mut |r: Record| print_record(&r), &group)?
570
        } else {
571
            workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
572
            microvm::run(home, sb, job, &inputs, &js.join("workspace.ext4"), &js, timeout, &mut |r: Record| print_record(&r))?
573
        };
574
        let _ = workspace::remove_tree(&js);
575
        match out.exit_code {
576
            Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
577
            other => {
578
                println!(
579
                    "── {} failed ({}) in {} ms",
580
                    job.name,
581
                    other.map(|c| format!("exit {c}")).unwrap_or_else(|| "no exit code".into()),
582
                    out.duration_ms
583
                );
584
                failed.insert(job.name.clone());
585
                worst = 1;
586
            }
587
        }
588
    }
589
    let _ = workspace::remove_tree(&scratch);
590
    if worst != 0 {
591
        std::process::exit(worst);
592
    }
593
    Ok(())
594
}
595
596
fn print_record(r: &Record) {
597
    match r.stream.as_str() {
598
        "out" => println!("   {}", r.text),
599
        "err" => eprintln!("   {}", r.text),
600
        s => eprintln!("   [{s}] {}", r.text),
601
    }
602
}
603
604
#[cfg(test)]
605
mod tests {
606
    use super::image_env_file;
607
608
    #[test]
609
    fn image_environments_are_checked_lines() {
610
        let f = image_env_file(&["PATH=/opt/bin:/usr/bin".into(), "X16EMU=/opt/x16emu".into(), "EMPTY=".into()]).unwrap();
611
        assert_eq!(f, "PATH=/opt/bin:/usr/bin\nX16EMU=/opt/x16emu\nEMPTY=\n");
612
        assert!(image_env_file(&["NOEQUALS".into()]).is_err());
613
        assert!(image_env_file(&["1X=y".into()]).is_err());
614
        assert!(image_env_file(&["X=two\nlines".into()]).is_err());
615
        assert_eq!(image_env_file(&[]).unwrap(), "");
616
    }
617
}