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