crates/flow-core/src/pipeline.rs at 448568561f2a · andymccall/oxygenflow

# 448568561f2a oxygenflow / crates / flow-core / src / pipeline.rs

20.5 KiB · Rust · last changed OxygenFlow M0: pipeline parser and canonical job graph, receipts, the runner API types, flow-init, the Firecracker tier, flow check, run, rootfs import and runner 1 month ago source blame history raw

//! `.flow/pipeline.kdl`: the M0 subset of spec §3 (docs/m0.md §1), parsed
//! from KDL 2.0 and validated before anything runs. Every problem is
//! reported with its line, and all of them at once.

use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;

/// Where a pipeline lives in the state being built.
pub const PIPELINE_PATH: &str = ".flow/pipeline.kdl";

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Pipeline {
    pub schema_version: u32,
    /// By name.
    pub sandboxes: BTreeMap<String, Sandbox>,
    /// By name.
    pub jobs: BTreeMap<String, Job>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Runtime {
    /// Tier 0 (spec §4.1); refused in M0.
    Wasm,
    /// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
    Syscall,
    /// Tier 2 (spec §4.3), the default.
    Microvm,
}

impl Runtime {
    pub fn as_str(self) -> &'static str {
        match self {
            Runtime::Wasm => "wasm",
            Runtime::Syscall => "syscall",
            Runtime::Microvm => "microvm",
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Network {
    Isolated,
    Loopback,
    Egress,
}

impl Network {
    pub fn as_str(self) -> &'static str {
        match self {
            Network::Isolated => "isolated",
            Network::Loopback => "loopback",
            Network::Egress => "egress",
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Sandbox {
    pub name: String,
    pub runtime: Runtime,
    /// BLAKE3 of the rootfs image, hex.
    pub rootfs: String,
    pub memory_mib: u32,
    pub cpus: u32,
    pub network: Network,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Job {
    pub name: String,
    pub sandbox: String,
    /// Glob patterns over repository paths (spec §6.2).
    pub scope: Vec<String>,
    /// Sorted.
    pub depends_on: Vec<String>,
    pub env: BTreeMap<String, String>,
    /// Executed as is: no shell, no expansion (spec §2.2 I2).
    pub exec: Vec<String>,
}

/// One problem in a pipeline, at a 1-based line.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Problem {
    pub line: usize,
    pub message: String,
}

impl fmt::Display for Problem {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "line {}: {}", self.line, self.message)
    }
}

/// Every problem found, in line order.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
pub struct Problems(pub Vec<Problem>);

impl fmt::Display for Problems {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let lines: Vec<String> = self.0.iter().map(|p| p.to_string()).collect();
        f.write_str(&lines.join("\n"))
    }
}

const MIN_MEMORY_MIB: u64 = 64;
const MAX_MEMORY_MIB: u64 = 16 * 1024;
const MAX_CPUS: i128 = 16;

/// Nodes of the full spec that M0 recognises and refuses.
const LATER_TOP: &[&str] = &["publish", "secret"];
const LATER_JOB: &[&str] = &["mount", "output", "secret", "attest"];

struct Ctx<'a> {
    src: &'a str,
    problems: Vec<Problem>,
}

impl Ctx<'_> {
    fn line(&self, span: SourceSpan) -> usize {
        let at = span.offset().min(self.src.len());
        self.src[..at].bytes().filter(|b| *b == b'\n').count() + 1
    }

    fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
        let line = self.line(span);
        self.problems.push(Problem { line, message: message.into() });
    }

    /// The node's arguments as strings; properties and other types are errors.
    fn strings(&mut self, node: &KdlNode) -> Vec<String> {
        let mut out = Vec::new();
        for e in node.entries() {
            if let Some(k) = e.name() {
                self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
                continue;
            }
            match e.value() {
                KdlValue::String(s) => out.push(s.clone()),
                other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
            }
        }
        out
    }

    /// Exactly one string argument.
    fn one_string(&mut self, node: &KdlNode) -> Option<String> {
        let v = self.strings(node);
        if v.len() != 1 {
            self.err(node.span(), format!("`{}` takes one string", node.name().value()));
            return None;
        }
        v.into_iter().next()
    }

    fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
        match node.entries() {
            [e] if e.name().is_none() => match e.value() {
                KdlValue::Integer(n) => Some(*n),
                other => {
                    self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
                    None
                }
            },
            _ => {
                self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
                None
            }
        }
    }

    fn no_children(&mut self, node: &KdlNode) {
        if node.children().is_some() {
            self.err(node.span(), format!("`{}` takes no block", node.name().value()));
        }
    }

    fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
        let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b));
        if !ok {
            self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
        }
        ok
    }
}

use kdl::{KdlDocument, KdlNode, KdlValue};
use miette::SourceSpan;

fn kind_of(v: &KdlValue) -> &'static str {
    match v {
        KdlValue::String(_) => "a string",
        KdlValue::Integer(_) => "an integer",
        KdlValue::Float(_) => "a number",
        KdlValue::Bool(_) => "a boolean",
        KdlValue::Null => "null",
    }
}

/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
fn parse_memory(s: &str) -> Option<u64> {
    let (num, unit) = s.split_at(s.find(|c: char| !c.is_ascii_digit())?);
    let n: u64 = num.parse().ok()?;
    let mib = match unit {
        "KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
        "MiB" => n,
        "GiB" => n.checked_mul(1024)?,
        _ => return None,
    };
    Some(mib)
}

impl Pipeline {
    /// Parse and validate a pipeline document.
    pub fn parse(src: &str) -> Result<Pipeline, Problems> {
        let doc = match KdlDocument::parse_v2(src) {
            Ok(d) => d,
            Err(e) => {
                let mut ctx = Ctx { src, problems: Vec::new() };
                for d in &e.diagnostics {
                    let msg = d.message.clone().or_else(|| d.label.clone()).unwrap_or_else(|| "not valid KDL 2.0".into());
                    ctx.err(d.span, msg);
                }
                if ctx.problems.is_empty() {
                    ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
                }
                return Err(Problems(ctx.problems));
            }
        };
        let mut ctx = Ctx { src, problems: Vec::new() };
        let mut schema = None;
        let mut sandboxes = BTreeMap::new();
        let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
        for node in doc.nodes() {
            match node.name().value() {
                "schema-version" => {
                    ctx.no_children(node);
                    match ctx.one_integer(node) {
                        Some(1) => schema = Some(1),
                        Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
                        None => {}
                    }
                }
                "sandbox" => {
                    if let Some(sb) = parse_sandbox(&mut ctx, node) {
                        if sandboxes.contains_key(&sb.name) {
                            ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
                        } else {
                            sandboxes.insert(sb.name.clone(), sb);
                        }
                    }
                }
                "job" => {
                    if let Some(job) = parse_job(&mut ctx, node) {
                        if jobs.contains_key(&job.name) {
                            ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
                        } else {
                            jobs.insert(job.name.clone(), (job, node.span()));
                        }
                    }
                }
                n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
                n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, sandbox or job")),
            }
        }
        if schema.is_none() && !ctx.problems.iter().any(|p| p.message.contains("schema-version")) {
            ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
        }
        if jobs.is_empty() && ctx.problems.is_empty() {
            ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
        }
        // References and cycles.
        for (job, span) in jobs.values() {
            if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
                ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
            }
            for d in &job.depends_on {
                if !jobs.contains_key(d) {
                    ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
                } else if d == &job.name {
                    ctx.err(*span, format!("job {:?} depends on itself", job.name));
                }
            }
        }
        let graph: BTreeMap<&str, Vec<&str>> =
            jobs.values().map(|(j, _)| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
        if let Some(cycle) = find_cycle(&graph) {
            let span = jobs[cycle[0]].1;
            ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
        }
        if !ctx.problems.is_empty() {
            ctx.problems.sort_by_key(|p| p.line);
            return Err(Problems(ctx.problems));
        }
        Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect() })
    }

    /// Jobs in an order where each comes after what it depends on (ties by name).
    pub fn order(&self) -> Vec<&Job> {
        let mut done: BTreeSet<&str> = BTreeSet::new();
        let mut out = Vec::with_capacity(self.jobs.len());
        while out.len() < self.jobs.len() {
            let before = out.len();
            for j in self.jobs.values() {
                if !done.contains(j.name.as_str()) && j.depends_on.iter().all(|d| done.contains(d.as_str())) {
                    out.push(j);
                }
            }
            for j in &out[before..] {
                done.insert(j.name.as_str());
            }
            if out.len() == before {
                break; // validated pipelines have no cycles
            }
        }
        out
    }

    /// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
    pub fn depths(&self) -> BTreeMap<&str, usize> {
        let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
        for j in self.order() {
            let d = j.depends_on.iter().map(|x| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
            depth.insert(j.name.as_str(), d);
        }
        depth
    }
}

fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
    #[derive(Clone, Copy, PartialEq)]
    enum Mark {
        Visiting,
        Done,
    }
    fn visit<'a>(
        n: &'a str,
        graph: &BTreeMap<&'a str, Vec<&'a str>>,
        marks: &mut BTreeMap<&'a str, Mark>,
        stack: &mut Vec<&'a str>,
    ) -> Option<Vec<&'a str>> {
        match marks.get(n) {
            Some(Mark::Done) => return None,
            Some(Mark::Visiting) => {
                let at = stack.iter().position(|s| *s == n).unwrap_or(0);
                let mut c = stack[at..].to_vec();
                c.push(n);
                return Some(c);
            }
            None => {}
        }
        marks.insert(n, Mark::Visiting);
        stack.push(n);
        for d in graph.get(n).into_iter().flatten() {
            if graph.contains_key(d) && *d != n {
                if let Some(c) = visit(d, graph, marks, stack) {
                    return Some(c);
                }
            }
        }
        stack.pop();
        marks.insert(n, Mark::Done);
        None
    }
    let mut marks = BTreeMap::new();
    for n in graph.keys() {
        let mut stack = Vec::new();
        if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
            return Some(c);
        }
    }
    None
}

fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
    match node.children() {
        Some(d) => d.nodes(),
        None => {
            ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
            &[]
        }
    }
}

fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
    let name = ctx.one_string(node)?;
    let named = ctx.name_ok(node.span(), "sandbox", &name);
    let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
        (Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
    let mut seen = BTreeSet::new();
    for c in block(ctx, node) {
        let key = c.name().value();
        if !seen.insert(key.to_string()) {
            ctx.err(c.span(), format!("`{key}` is given twice"));
        }
        ctx.no_children(c);
        match key {
            "runtime" => {
                runtime = match ctx.one_string(c).as_deref() {
                    Some("microvm") => Some(Runtime::Microvm),
                    Some("syscall") | Some("gvisor") => Some(Runtime::Syscall),
                    Some("wasm") => {
                        ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
                        None
                    }
                    Some(other) => {
                        ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
                        None
                    }
                    None => None,
                }
            }
            "rootfs" => {
                rootfs = match ctx.one_string(c) {
                    Some(r) => match r.strip_prefix("cas://blob/") {
                        Some(h) if h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
                            Some(h.to_string())
                        }
                        _ => {
                            ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
                            None
                        }
                    },
                    None => None,
                }
            }
            "memory-limit" => {
                memory = match ctx.one_string(c) {
                    Some(m) => match parse_memory(&m) {
                        Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
                        Some(_) => {
                            ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
                            None
                        }
                        None => {
                            ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
                            None
                        }
                    },
                    None => None,
                }
            }
            "cpu-limit" => {
                cpus = match ctx.one_integer(c) {
                    Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
                    Some(n) => {
                        ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
                        None
                    }
                    None => None,
                }
            }
            "network" => {
                network = match ctx.one_string(c).as_deref() {
                    Some("isolated") => Some(Network::Isolated),
                    Some("loopback") | Some("egress") => {
                        ctx.err(c.span(), "only network \"isolated\" is in M0");
                        None
                    }
                    Some(other) => {
                        ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
                        None
                    }
                    None => None,
                }
            }
            other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
        }
    }
    if rootfs.is_none() && !seen.contains("rootfs") {
        ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
    }
    if !named {
        return None;
    }
    Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network? })
}

fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
    let name = ctx.one_string(node)?;
    let named = ctx.name_ok(node.span(), "job", &name);
    let mut job = Job { name: name.clone(), sandbox: String::new(), scope: vec![], depends_on: vec![], env: BTreeMap::new(), exec: vec![] };
    let mut seen = BTreeSet::new();
    for c in block(ctx, node) {
        let key = c.name().value();
        if key != "scope" && key != "depends-on" && !seen.insert(key.to_string()) {
            ctx.err(c.span(), format!("`{key}` is given twice"));
        }
        match key {
            "sandbox" => {
                ctx.no_children(c);
                job.sandbox = ctx.one_string(c).unwrap_or_default();
            }
            "scope" => {
                ctx.no_children(c);
                for g in ctx.strings(c) {
                    match globset::Glob::new(&g) {
                        Ok(_) => job.scope.push(g),
                        Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
                    }
                }
            }
            "depends-on" => {
                ctx.no_children(c);
                job.depends_on.extend(ctx.strings(c));
            }
            "env" => {
                if !c.entries().is_empty() {
                    ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
                }
                for e in block(ctx, c) {
                    let k = e.name().value().to_string();
                    let valid = !k.is_empty()
                        && !k.starts_with(|c: char| c.is_ascii_digit())
                        && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
                    if !valid {
                        ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
                        continue;
                    }
                    ctx.no_children(e);
                    if let Some(v) = ctx.one_string(e) {
                        if job.env.insert(k.clone(), v).is_some() {
                            ctx.err(e.span(), format!("environment {k} is set twice"));
                        }
                    }
                }
            }
            "exec" => {
                ctx.no_children(c);
                job.exec = ctx.strings(c);
                if job.exec.is_empty() {
                    ctx.err(c.span(), "`exec` needs a program and its arguments");
                } else if job.exec.iter().any(|a| a.contains('\0')) {
                    ctx.err(c.span(), "`exec` arguments cannot contain NUL");
                }
            }
            n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
            other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
        }
    }
    if job.sandbox.is_empty() && !seen.contains("sandbox") {
        ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
    }
    if job.exec.is_empty() && !seen.contains("exec") {
        ctx.err(node.span(), format!("job {name:?} needs an exec"));
    }
    job.depends_on.sort();
    job.depends_on.dedup();
    // Kept even when incomplete, so its references are checked too; the
    // problems above already make the pipeline invalid.
    named.then_some(job)
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! `.flow/pipeline.kdl`: the M0 subset of spec §3 (docs/m0.md §1), parsed
2	//! from KDL 2.0 and validated before anything runs. Every problem is
3	//! reported with its line, and all of them at once.
4
5	use serde::{Deserialize, Serialize};
6	use std::collections::{BTreeMap, BTreeSet};
7	use std::fmt;
8
9	/// Where a pipeline lives in the state being built.
10	pub const PIPELINE_PATH: &str = ".flow/pipeline.kdl";
11
12	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
13	pub struct Pipeline {
14	pub schema_version: u32,
15	/// By name.
16	pub sandboxes: BTreeMap<String, Sandbox>,
17	/// By name.
18	pub jobs: BTreeMap<String, Job>,
19	}
20
21	#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
22	#[serde(rename_all = "lowercase")]
23	pub enum Runtime {
24	/// Tier 0 (spec §4.1); refused in M0.
25	Wasm,
26	/// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
27	Syscall,
28	/// Tier 2 (spec §4.3), the default.
29	Microvm,
30	}
31
32	impl Runtime {
33	pub fn as_str(self) -> &'static str {
34	match self {
35	Runtime::Wasm => "wasm",
36	Runtime::Syscall => "syscall",
37	Runtime::Microvm => "microvm",
38	}
39	}
40	}
41
42	#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
43	#[serde(rename_all = "lowercase")]
44	pub enum Network {
45	Isolated,
46	Loopback,
47	Egress,
48	}
49
50	impl Network {
51	pub fn as_str(self) -> &'static str {
52	match self {
53	Network::Isolated => "isolated",
54	Network::Loopback => "loopback",
55	Network::Egress => "egress",
56	}
57	}
58	}
59
60	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
61	pub struct Sandbox {
62	pub name: String,
63	pub runtime: Runtime,
64	/// BLAKE3 of the rootfs image, hex.
65	pub rootfs: String,
66	pub memory_mib: u32,
67	pub cpus: u32,
68	pub network: Network,
69	}
70
71	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
72	pub struct Job {
73	pub name: String,
74	pub sandbox: String,
75	/// Glob patterns over repository paths (spec §6.2).
76	pub scope: Vec<String>,
77	/// Sorted.
78	pub depends_on: Vec<String>,
79	pub env: BTreeMap<String, String>,
80	/// Executed as is: no shell, no expansion (spec §2.2 I2).
81	pub exec: Vec<String>,
82	}
83
84	/// One problem in a pipeline, at a 1-based line.
85	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
86	pub struct Problem {
87	pub line: usize,
88	pub message: String,
89	}
90
91	impl fmt::Display for Problem {
92	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
93	write!(f, "line {}: {}", self.line, self.message)
94	}
95	}
96
97	/// Every problem found, in line order.
98	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
99	pub struct Problems(pub Vec<Problem>);
100
101	impl fmt::Display for Problems {
102	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
103	let lines: Vec<String> = self.0.iter().map(\|p\| p.to_string()).collect();
104	f.write_str(&lines.join("\n"))
105	}
106	}
107
108	const MIN_MEMORY_MIB: u64 = 64;
109	const MAX_MEMORY_MIB: u64 = 16 * 1024;
110	const MAX_CPUS: i128 = 16;
111
112	/// Nodes of the full spec that M0 recognises and refuses.
113	const LATER_TOP: &[&str] = &["publish", "secret"];
114	const LATER_JOB: &[&str] = &["mount", "output", "secret", "attest"];
115
116	struct Ctx<'a> {
117	src: &'a str,
118	problems: Vec<Problem>,
119	}
120
121	impl Ctx<'_> {
122	fn line(&self, span: SourceSpan) -> usize {
123	let at = span.offset().min(self.src.len());
124	self.src[..at].bytes().filter(\|b\| *b == b'\n').count() + 1
125	}
126
127	fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
128	let line = self.line(span);
129	self.problems.push(Problem { line, message: message.into() });
130	}
131
132	/// The node's arguments as strings; properties and other types are errors.
133	fn strings(&mut self, node: &KdlNode) -> Vec<String> {
134	let mut out = Vec::new();
135	for e in node.entries() {
136	if let Some(k) = e.name() {
137	self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
138	continue;
139	}
140	match e.value() {
141	KdlValue::String(s) => out.push(s.clone()),
142	other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
143	}
144	}
145	out
146	}
147
148	/// Exactly one string argument.
149	fn one_string(&mut self, node: &KdlNode) -> Option<String> {
150	let v = self.strings(node);
151	if v.len() != 1 {
152	self.err(node.span(), format!("`{}` takes one string", node.name().value()));
153	return None;
154	}
155	v.into_iter().next()
156	}
157
158	fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
159	match node.entries() {
160	[e] if e.name().is_none() => match e.value() {
161	KdlValue::Integer(n) => Some(*n),
162	other => {
163	self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
164	None
165	}
166	},
167	_ => {
168	self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
169	None
170	}
171	}
172	}
173
174	fn no_children(&mut self, node: &KdlNode) {
175	if node.children().is_some() {
176	self.err(node.span(), format!("`{}` takes no block", node.name().value()));
177	}
178	}
179
180	fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
181	let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b"._-".contains(&b));
182	if !ok {
183	self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
184	}
185	ok
186	}
187	}
188
189	use kdl::{KdlDocument, KdlNode, KdlValue};
190	use miette::SourceSpan;
191
192	fn kind_of(v: &KdlValue) -> &'static str {
193	match v {
194	KdlValue::String(_) => "a string",
195	KdlValue::Integer(_) => "an integer",
196	KdlValue::Float(_) => "a number",
197	KdlValue::Bool(_) => "a boolean",
198	KdlValue::Null => "null",
199	}
200	}
201
202	/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
203	fn parse_memory(s: &str) -> Option<u64> {
204	let (num, unit) = s.split_at(s.find(\|c: char\| !c.is_ascii_digit())?);
205	let n: u64 = num.parse().ok()?;
206	let mib = match unit {
207	"KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
208	"MiB" => n,
209	"GiB" => n.checked_mul(1024)?,
210	_ => return None,
211	};
212	Some(mib)
213	}
214
215	impl Pipeline {
216	/// Parse and validate a pipeline document.
217	pub fn parse(src: &str) -> Result<Pipeline, Problems> {
218	let doc = match KdlDocument::parse_v2(src) {
219	Ok(d) => d,
220	Err(e) => {
221	let mut ctx = Ctx { src, problems: Vec::new() };
222	for d in &e.diagnostics {
223	let msg = d.message.clone().or_else(\|\| d.label.clone()).unwrap_or_else(\|\| "not valid KDL 2.0".into());
224	ctx.err(d.span, msg);
225	}
226	if ctx.problems.is_empty() {
227	ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
228	}
229	return Err(Problems(ctx.problems));
230	}
231	};
232	let mut ctx = Ctx { src, problems: Vec::new() };
233	let mut schema = None;
234	let mut sandboxes = BTreeMap::new();
235	let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
236	for node in doc.nodes() {
237	match node.name().value() {
238	"schema-version" => {
239	ctx.no_children(node);
240	match ctx.one_integer(node) {
241	Some(1) => schema = Some(1),
242	Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
243	None => {}
244	}
245	}
246	"sandbox" => {
247	if let Some(sb) = parse_sandbox(&mut ctx, node) {
248	if sandboxes.contains_key(&sb.name) {
249	ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
250	} else {
251	sandboxes.insert(sb.name.clone(), sb);
252	}
253	}
254	}
255	"job" => {
256	if let Some(job) = parse_job(&mut ctx, node) {
257	if jobs.contains_key(&job.name) {
258	ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
259	} else {
260	jobs.insert(job.name.clone(), (job, node.span()));
261	}
262	}
263	}
264	n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
265	n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, sandbox or job")),
266	}
267	}
268	if schema.is_none() && !ctx.problems.iter().any(\|p\| p.message.contains("schema-version")) {
269	ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
270	}
271	if jobs.is_empty() && ctx.problems.is_empty() {
272	ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
273	}
274	// References and cycles.
275	for (job, span) in jobs.values() {
276	if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
277	ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
278	}
279	for d in &job.depends_on {
280	if !jobs.contains_key(d) {
281	ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
282	} else if d == &job.name {
283	ctx.err(*span, format!("job {:?} depends on itself", job.name));
284	}
285	}
286	}
287	let graph: BTreeMap<&str, Vec<&str>> =
288	jobs.values().map(\|(j, _)\| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
289	if let Some(cycle) = find_cycle(&graph) {
290	let span = jobs[cycle[0]].1;
291	ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
292	}
293	if !ctx.problems.is_empty() {
294	ctx.problems.sort_by_key(\|p\| p.line);
295	return Err(Problems(ctx.problems));
296	}
297	Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(\|(k, (j, _))\| (k, j)).collect() })
298	}
299
300	/// Jobs in an order where each comes after what it depends on (ties by name).
301	pub fn order(&self) -> Vec<&Job> {
302	let mut done: BTreeSet<&str> = BTreeSet::new();
303	let mut out = Vec::with_capacity(self.jobs.len());
304	while out.len() < self.jobs.len() {
305	let before = out.len();
306	for j in self.jobs.values() {
307	if !done.contains(j.name.as_str()) && j.depends_on.iter().all(\|d\| done.contains(d.as_str())) {
308	out.push(j);
309	}
310	}
311	for j in &out[before..] {
312	done.insert(j.name.as_str());
313	}
314	if out.len() == before {
315	break; // validated pipelines have no cycles
316	}
317	}
318	out
319	}
320
321	/// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
322	pub fn depths(&self) -> BTreeMap<&str, usize> {
323	let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
324	for j in self.order() {
325	let d = j.depends_on.iter().map(\|x\| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
326	depth.insert(j.name.as_str(), d);
327	}
328	depth
329	}
330	}
331
332	fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
333	#[derive(Clone, Copy, PartialEq)]
334	enum Mark {
335	Visiting,
336	Done,
337	}
338	fn visit<'a>(
339	n: &'a str,
340	graph: &BTreeMap<&'a str, Vec<&'a str>>,
341	marks: &mut BTreeMap<&'a str, Mark>,
342	stack: &mut Vec<&'a str>,
343	) -> Option<Vec<&'a str>> {
344	match marks.get(n) {
345	Some(Mark::Done) => return None,
346	Some(Mark::Visiting) => {
347	let at = stack.iter().position(\|s\| *s == n).unwrap_or(0);
348	let mut c = stack[at..].to_vec();
349	c.push(n);
350	return Some(c);
351	}
352	None => {}
353	}
354	marks.insert(n, Mark::Visiting);
355	stack.push(n);
356	for d in graph.get(n).into_iter().flatten() {
357	if graph.contains_key(d) && *d != n {
358	if let Some(c) = visit(d, graph, marks, stack) {
359	return Some(c);
360	}
361	}
362	}
363	stack.pop();
364	marks.insert(n, Mark::Done);
365	None
366	}
367	let mut marks = BTreeMap::new();
368	for n in graph.keys() {
369	let mut stack = Vec::new();
370	if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
371	return Some(c);
372	}
373	}
374	None
375	}
376
377	fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
378	match node.children() {
379	Some(d) => d.nodes(),
380	None => {
381	ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
382	&[]
383	}
384	}
385	}
386
387	fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
388	let name = ctx.one_string(node)?;
389	let named = ctx.name_ok(node.span(), "sandbox", &name);
390	let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
391	(Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
392	let mut seen = BTreeSet::new();
393	for c in block(ctx, node) {
394	let key = c.name().value();
395	if !seen.insert(key.to_string()) {
396	ctx.err(c.span(), format!("`{key}` is given twice"));
397	}
398	ctx.no_children(c);
399	match key {
400	"runtime" => {
401	runtime = match ctx.one_string(c).as_deref() {
402	Some("microvm") => Some(Runtime::Microvm),
403	Some("syscall") \| Some("gvisor") => Some(Runtime::Syscall),
404	Some("wasm") => {
405	ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
406	None
407	}
408	Some(other) => {
409	ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
410	None
411	}
412	None => None,
413	}
414	}
415	"rootfs" => {
416	rootfs = match ctx.one_string(c) {
417	Some(r) => match r.strip_prefix("cas://blob/") {
418	Some(h) if h.len() == 64 && h.bytes().all(\|b\| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
419	Some(h.to_string())
420	}
421	_ => {
422	ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
423	None
424	}
425	},
426	None => None,
427	}
428	}
429	"memory-limit" => {
430	memory = match ctx.one_string(c) {
431	Some(m) => match parse_memory(&m) {
432	Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
433	Some(_) => {
434	ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
435	None
436	}
437	None => {
438	ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
439	None
440	}
441	},
442	None => None,
443	}
444	}
445	"cpu-limit" => {
446	cpus = match ctx.one_integer(c) {
447	Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
448	Some(n) => {
449	ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
450	None
451	}
452	None => None,
453	}
454	}
455	"network" => {
456	network = match ctx.one_string(c).as_deref() {
457	Some("isolated") => Some(Network::Isolated),
458	Some("loopback") \| Some("egress") => {
459	ctx.err(c.span(), "only network \"isolated\" is in M0");
460	None
461	}
462	Some(other) => {
463	ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
464	None
465	}
466	None => None,
467	}
468	}
469	other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
470	}
471	}
472	if rootfs.is_none() && !seen.contains("rootfs") {
473	ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
474	}
475	if !named {
476	return None;
477	}
478	Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network? })
479	}
480
481	fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
482	let name = ctx.one_string(node)?;
483	let named = ctx.name_ok(node.span(), "job", &name);
484	let mut job = Job { name: name.clone(), sandbox: String::new(), scope: vec![], depends_on: vec![], env: BTreeMap::new(), exec: vec![] };
485	let mut seen = BTreeSet::new();
486	for c in block(ctx, node) {
487	let key = c.name().value();
488	if key != "scope" && key != "depends-on" && !seen.insert(key.to_string()) {
489	ctx.err(c.span(), format!("`{key}` is given twice"));
490	}
491	match key {
492	"sandbox" => {
493	ctx.no_children(c);
494	job.sandbox = ctx.one_string(c).unwrap_or_default();
495	}
496	"scope" => {
497	ctx.no_children(c);
498	for g in ctx.strings(c) {
499	match globset::Glob::new(&g) {
500	Ok(_) => job.scope.push(g),
501	Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
502	}
503	}
504	}
505	"depends-on" => {
506	ctx.no_children(c);
507	job.depends_on.extend(ctx.strings(c));
508	}
509	"env" => {
510	if !c.entries().is_empty() {
511	ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
512	}
513	for e in block(ctx, c) {
514	let k = e.name().value().to_string();
515	let valid = !k.is_empty()
516	&& !k.starts_with(\|c: char\| c.is_ascii_digit())
517	&& k.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b == b'_');
518	if !valid {
519	ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
520	continue;
521	}
522	ctx.no_children(e);
523	if let Some(v) = ctx.one_string(e) {
524	if job.env.insert(k.clone(), v).is_some() {
525	ctx.err(e.span(), format!("environment {k} is set twice"));
526	}
527	}
528	}
529	}
530	"exec" => {
531	ctx.no_children(c);
532	job.exec = ctx.strings(c);
533	if job.exec.is_empty() {
534	ctx.err(c.span(), "`exec` needs a program and its arguments");
535	} else if job.exec.iter().any(\|a\| a.contains('\0')) {
536	ctx.err(c.span(), "`exec` arguments cannot contain NUL");
537	}
538	}
539	n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
540	other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
541	}
542	}
543	if job.sandbox.is_empty() && !seen.contains("sandbox") {
544	ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
545	}
546	if job.exec.is_empty() && !seen.contains("exec") {
547	ctx.err(node.span(), format!("job {name:?} needs an exec"));
548	}
549	job.depends_on.sort();
550	job.depends_on.dedup();
551	// Kept even when incomplete, so its references are checked too; the
552	// problems above already make the pipeline invalid.
553	named.then_some(job)
554	}

Keyboard