OXYGENHUB
# bf7bb1b80fa3 oxygenflow / crates / flow-core / src / pipeline.rs
40.0 KiB · Rust · last changed pipeline.kdl: on { push; tag; manual; schedule } for the pipeline and each job 1 month ago source blame history raw
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
    /// When the pipeline's jobs run, unless a job says (oxygenflow#24);
20
    /// absent: on every push and from "Run pipeline".
21
    #[serde(default, skip_serializing_if = "Option::is_none")]
22
    pub on: Option<On>,
23
}
24
25
/// What starts a job (`on { push "main"; tag "v*"; manual; schedule "0 3 * * *" }`).
26
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
27
pub struct On {
28
    /// Bookmark globs a push runs it for; `Some(vec![])`: any bookmark.
29
    #[serde(default, skip_serializing_if = "Option::is_none")]
30
    pub push: Option<Vec<String>>,
31
    /// Tag globs a pushed tag runs it for; `Some(vec![])`: any tag.
32
    #[serde(default, skip_serializing_if = "Option::is_none")]
33
    pub tag: Option<Vec<String>>,
34
    /// From "Run pipeline" (and `oh run start`).
35
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
36
    pub manual: bool,
37
    /// Five-field cron lines, on the hub's clock, for the default bookmark.
38
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
39
    pub schedule: Vec<String>,
40
}
41
42
/// Why a run started, for [`On::matches`].
43
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
44
pub enum Event<'a> {
45
    /// A push moved this bookmark.
46
    Push(&'a str),
47
    /// This tag was pushed.
48
    Tag(&'a str),
49
    /// "Run pipeline", `oh run start`.
50
    Manual,
51
    /// A `schedule` line came due.
52
    Schedule,
53
}
54
55
impl On {
56
    /// What a job without any `on` runs for: every push, and by hand.
57
    pub fn default_on() -> On {
58
        On { push: Some(vec![]), tag: None, manual: true, schedule: vec![] }
59
    }
60
61
    pub fn matches(&self, event: Event<'_>) -> bool {
62
        let any = |globs: &Option<Vec<String>>, name: &str| match globs {
63
            None => false,
64
            Some(g) if g.is_empty() => true,
65
            Some(g) => g.iter().any(|p| globset::Glob::new(p).is_ok_and(|p| p.compile_matcher().is_match(name))),
66
        };
67
        match event {
68
            Event::Push(b) => any(&self.push, b),
69
            Event::Tag(t) => any(&self.tag, t),
70
            Event::Manual => self.manual,
71
            Event::Schedule => !self.schedule.is_empty(),
72
        }
73
    }
74
}
75
76
impl Pipeline {
77
    /// When `job` runs: its own `on`, else the pipeline's, else the default.
78
    pub fn on_of(&self, job: &Job) -> On {
79
        job.on.clone().or_else(|| self.on.clone()).unwrap_or_else(On::default_on)
80
    }
81
}
82
83
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
84
#[serde(rename_all = "lowercase")]
85
pub enum Runtime {
86
    /// Tier 0 (spec §4.1); refused in M0.
87
    Wasm,
88
    /// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
89
    Syscall,
90
    /// Tier 2 (spec §4.3), the default.
91
    Microvm,
92
}
93
94
impl Runtime {
95
    pub fn as_str(self) -> &'static str {
96
        match self {
97
            Runtime::Wasm => "wasm",
98
            Runtime::Syscall => "syscall",
99
            Runtime::Microvm => "microvm",
100
        }
101
    }
102
}
103
104
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
105
#[serde(rename_all = "lowercase")]
106
pub enum Network {
107
    Isolated,
108
    Loopback,
109
    Egress,
110
}
111
112
impl Network {
113
    pub fn as_str(self) -> &'static str {
114
        match self {
115
            Network::Isolated => "isolated",
116
            Network::Loopback => "loopback",
117
            Network::Egress => "egress",
118
        }
119
    }
120
}
121
122
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
123
pub struct Sandbox {
124
    pub name: String,
125
    pub runtime: Runtime,
126
    /// BLAKE3 of the rootfs image, hex.
127
    pub rootfs: String,
128
    pub memory_mib: u32,
129
    pub cpus: u32,
130
    pub network: Network,
131
    /// Environment every job in this sandbox starts from; a job's own `env`
132
    /// overrides it.
133
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
134
    pub env: BTreeMap<String, String>,
135
    /// Labels a runner must have to run jobs here (`runs-on`), sorted.
136
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
137
    pub runs_on: Vec<String>,
138
    /// How long each job here may run, in seconds (`timeout "2h"`), when
139
    /// the job does not say; the hub caps it at the site's maximum.
140
    #[serde(default, skip_serializing_if = "Option::is_none")]
141
    pub timeout_s: Option<u64>,
142
}
143
144
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
145
pub struct Job {
146
    pub name: String,
147
    pub sandbox: String,
148
    /// Glob patterns over repository paths (spec §6.2).
149
    pub scope: Vec<String>,
150
    /// Sorted.
151
    pub depends_on: Vec<String>,
152
    pub env: BTreeMap<String, String>,
153
    /// Executed as is: no shell, no expansion (spec §2.2 I2).
154
    pub exec: Vec<String>,
155
    /// Ephemeral scratch mounts, sorted by target (spec §3.1 `mount`).
156
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
157
    pub mounts: Vec<Mount>,
158
    /// Files the job produces, sorted by name (spec §5.2 `output`).
159
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
160
    pub outputs: Vec<Output>,
161
    /// Labels a runner must have to take this job (`runs-on`), sorted;
162
    /// with its sandbox's, what the hub matches against a runner's.
163
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
164
    pub runs_on: Vec<String>,
165
    /// How long the job may run, in seconds (`timeout "90m"`); its
166
    /// sandbox's when absent, else the site's default (oxygenflow#23).
167
    #[serde(default, skip_serializing_if = "Option::is_none")]
168
    pub timeout_s: Option<u64>,
169
    /// When it runs, over the pipeline's `on` (oxygenflow#24).
170
    #[serde(default, skip_serializing_if = "Option::is_none")]
171
    pub on: Option<On>,
172
}
173
174
impl Job {
175
    /// The job's environment on top of its sandbox's: what it runs with.
176
    pub fn in_sandbox(&self, sandbox: &Sandbox) -> Job {
177
        let mut job = self.clone();
178
        for (k, v) in &sandbox.env {
179
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
180
        }
181
        job.runs_on.extend(sandbox.runs_on.iter().cloned());
182
        job.runs_on.sort();
183
        job.runs_on.dedup();
184
        job.timeout_s = job.timeout_s.or(sandbox.timeout_s);
185
        job
186
    }
187
188
    /// The job with variables that tell it what it builds (`FLOW_STATE`,
189
    /// `FLOW_BOOKMARK` …) added to its environment; a variable the pipeline
190
    /// sets itself keeps the pipeline's value.
191
    pub fn with_flow_env(&self, vars: &[(String, String)]) -> Job {
192
        let mut job = self.clone();
193
        for (k, v) in vars {
194
            job.env.entry(k.clone()).or_insert_with(|| v.clone());
195
        }
196
        job
197
    }
198
}
199
200
/// An ephemeral scratch mount: empty at the start of every job and gone
201
/// after it, so it cannot carry anything between jobs (spec §2.2 I4).
202
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
203
pub struct Mount {
204
    pub source: MountSource,
205
    /// An absolute path under `/src` or `/tmp`.
206
    pub target: String,
207
    pub size_mib: u32,
208
}
209
210
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
211
#[serde(rename_all = "lowercase")]
212
pub enum MountSource {
213
    /// `ephemeral://tmpfs`: memory in the sandbox, counted against its memory-limit.
214
    Tmpfs,
215
    /// `ephemeral://disk`: a fresh sparse disk on the runner, for large build trees.
216
    Disk,
217
}
218
219
impl MountSource {
220
    pub fn as_str(self) -> &'static str {
221
        match self {
222
            MountSource::Tmpfs => "ephemeral://tmpfs",
223
            MountSource::Disk => "ephemeral://disk",
224
        }
225
    }
226
}
227
228
/// A file a job produces (spec §5.2): read from the job's disks when it
229
/// ends, kept in the repository as a blob, and named by its receipt.
230
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
231
pub struct Output {
232
    /// The artifact's name, unique in the job.
233
    pub name: String,
234
    /// A clean absolute path under `/src`, not on a tmpfs mount.
235
    pub path: String,
236
}
237
238
/// Artifacts a job may declare.
239
pub const MAX_OUTPUTS: usize = 16;
240
241
/// Whether `p` is a clean absolute path strictly under `/src`, with nothing
242
/// a debugfs command line would have to quote.
243
fn output_path_ok(p: &str) -> bool {
244
    p.strip_prefix("/src/").is_some_and(|rest| {
245
        !rest.is_empty()
246
            && rest.split('/').all(|c| !c.is_empty() && c != "." && c != "..")
247
            && p.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\')
248
    })
249
}
250
251
/// Mounts a job may declare.
252
pub const MAX_MOUNTS: usize = 4;
253
/// Sizes a mount may have: 16 MiB to 64 GiB.
254
pub const MIN_MOUNT_MIB: u64 = 16;
255
pub const MAX_MOUNT_MIB: u64 = 64 * 1024;
256
257
/// Whether `t` is a clean absolute path strictly under `/src` or `/tmp`.
258
fn mount_target_ok(t: &str) -> bool {
259
    let under = t.strip_prefix("/src/").or_else(|| t.strip_prefix("/tmp/"));
260
    under.is_some_and(|rest| !rest.is_empty() && rest.split('/').all(|c| !c.is_empty() && c != "." && c != ".."))
261
}
262
263
/// One problem in a pipeline, at a 1-based line.
264
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
265
pub struct Problem {
266
    pub line: usize,
267
    pub message: String,
268
}
269
270
impl fmt::Display for Problem {
271
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
272
        write!(f, "line {}: {}", self.line, self.message)
273
    }
274
}
275
276
/// Every problem found, in line order.
277
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
278
pub struct Problems(pub Vec<Problem>);
279
280
impl fmt::Display for Problems {
281
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
282
        let lines: Vec<String> = self.0.iter().map(|p| p.to_string()).collect();
283
        f.write_str(&lines.join("\n"))
284
    }
285
}
286
287
const MIN_MEMORY_MIB: u64 = 64;
288
const MAX_MEMORY_MIB: u64 = 16 * 1024;
289
const MAX_CPUS: i128 = 16;
290
291
/// Nodes of the full spec that M0 recognises and refuses.
292
const LATER_TOP: &[&str] = &["publish", "secret"];
293
const LATER_JOB: &[&str] = &["secret", "attest"];
294
295
struct Ctx<'a> {
296
    src: &'a str,
297
    problems: Vec<Problem>,
298
}
299
300
impl Ctx<'_> {
301
    fn line(&self, span: SourceSpan) -> usize {
302
        let at = span.offset().min(self.src.len());
303
        self.src[..at].bytes().filter(|b| *b == b'\n').count() + 1
304
    }
305
306
    fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
307
        let line = self.line(span);
308
        self.problems.push(Problem { line, message: message.into() });
309
    }
310
311
    /// The node's arguments as strings; properties and other types are errors.
312
    fn strings(&mut self, node: &KdlNode) -> Vec<String> {
313
        let mut out = Vec::new();
314
        for e in node.entries() {
315
            if let Some(k) = e.name() {
316
                self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
317
                continue;
318
            }
319
            match e.value() {
320
                KdlValue::String(s) => out.push(s.clone()),
321
                other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
322
            }
323
        }
324
        out
325
    }
326
327
    /// Exactly one string argument.
328
    fn one_string(&mut self, node: &KdlNode) -> Option<String> {
329
        let v = self.strings(node);
330
        if v.len() != 1 {
331
            self.err(node.span(), format!("`{}` takes one string", node.name().value()));
332
            return None;
333
        }
334
        v.into_iter().next()
335
    }
336
337
    fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
338
        match node.entries() {
339
            [e] if e.name().is_none() => match e.value() {
340
                KdlValue::Integer(n) => Some(*n),
341
                other => {
342
                    self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
343
                    None
344
                }
345
            },
346
            _ => {
347
                self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
348
                None
349
            }
350
        }
351
    }
352
353
    fn no_children(&mut self, node: &KdlNode) {
354
        if node.children().is_some() {
355
            self.err(node.span(), format!("`{}` takes no block", node.name().value()));
356
        }
357
    }
358
359
    fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
360
        let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(|b| b.is_ascii_alphanumeric() || b"._-".contains(&b));
361
        if !ok {
362
            self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
363
        }
364
        ok
365
    }
366
}
367
368
use kdl::{KdlDocument, KdlNode, KdlValue};
369
use miette::SourceSpan;
370
371
fn kind_of(v: &KdlValue) -> &'static str {
372
    match v {
373
        KdlValue::String(_) => "a string",
374
        KdlValue::Integer(_) => "an integer",
375
        KdlValue::Float(_) => "a number",
376
        KdlValue::Bool(_) => "a boolean",
377
        KdlValue::Null => "null",
378
    }
379
}
380
381
/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
382
fn parse_memory(s: &str) -> Option<u64> {
383
    let (num, unit) = s.split_at(s.find(|c: char| !c.is_ascii_digit())?);
384
    let n: u64 = num.parse().ok()?;
385
    let mib = match unit {
386
        "KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
387
        "MiB" => n,
388
        "GiB" => n.checked_mul(1024)?,
389
        _ => return None,
390
    };
391
    Some(mib)
392
}
393
394
/// The line of the first escape KDL 2.0 does not have but shells and C
395
/// do (`\0NN`, `\e`, `\xNN`), as someone colouring output writes them.
396
fn foreign_escape_line(src: &str) -> Option<usize> {
397
    src.lines()
398
        .position(|l| {
399
            let b = l.as_bytes();
400
            b.windows(2).enumerate().any(|(i, w)| {
401
                w[0] == b'\\'
402
                    && (i == 0 || b[i - 1] != b'\\')
403
                    && (w[1] == b'e' || w[1] == b'x' || (w[1] == b'0' && b.get(i + 2).is_some_and(u8::is_ascii_digit)))
404
            })
405
        })
406
        .map(|i| i + 1)
407
}
408
409
impl Pipeline {
410
    /// Parse and validate a pipeline document.
411
    pub fn parse(src: &str) -> Result<Pipeline, Problems> {
412
        let doc = match KdlDocument::parse_v2(src) {
413
            Ok(d) => d,
414
            Err(e) => {
415
                let mut ctx = Ctx { src, problems: Vec::new() };
416
                for d in &e.diagnostics {
417
                    let msg = d.message.clone().or_else(|| d.label.clone()).unwrap_or_else(|| "not valid KDL 2.0".into());
418
                    ctx.err(d.span, msg);
419
                }
420
                if ctx.problems.is_empty() {
421
                    ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
422
                }
423
                if let Some(line) = foreign_escape_line(src) {
424
                    ctx.problems.push(Problem {
425
                        line,
426
                        message: "KDL strings have no \\033, \\e or \\x escapes: write ESC (for ANSI colours) as \\u{1b}".into(),
427
                    });
428
                }
429
                return Err(Problems(ctx.problems));
430
            }
431
        };
432
        let mut ctx = Ctx { src, problems: Vec::new() };
433
        let mut schema = None;
434
        let mut sandboxes = BTreeMap::new();
435
        let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
436
        let mut on = None;
437
        for node in doc.nodes() {
438
            match node.name().value() {
439
                "on" => {
440
                    if on.is_some() {
441
                        ctx.err(node.span(), "`on` is given twice");
442
                    }
443
                    on = Some(parse_on(&mut ctx, node));
444
                }
445
                "schema-version" => {
446
                    ctx.no_children(node);
447
                    match ctx.one_integer(node) {
448
                        Some(1) => schema = Some(1),
449
                        Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
450
                        None => {}
451
                    }
452
                }
453
                "sandbox" => {
454
                    if let Some(sb) = parse_sandbox(&mut ctx, node) {
455
                        if sandboxes.contains_key(&sb.name) {
456
                            ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
457
                        } else {
458
                            sandboxes.insert(sb.name.clone(), sb);
459
                        }
460
                    }
461
                }
462
                "job" => {
463
                    if let Some(job) = parse_job(&mut ctx, node) {
464
                        if jobs.contains_key(&job.name) {
465
                            ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
466
                        } else {
467
                            jobs.insert(job.name.clone(), (job, node.span()));
468
                        }
469
                    }
470
                }
471
                n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
472
                n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, on, sandbox or job")),
473
            }
474
        }
475
        if schema.is_none() && !ctx.problems.iter().any(|p| p.message.contains("schema-version")) {
476
            ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
477
        }
478
        if jobs.is_empty() && ctx.problems.is_empty() {
479
            ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
480
        }
481
        // References and cycles.
482
        for (job, span) in jobs.values() {
483
            if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
484
                ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
485
            }
486
            for d in &job.depends_on {
487
                if !jobs.contains_key(d) {
488
                    ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
489
                } else if d == &job.name {
490
                    ctx.err(*span, format!("job {:?} depends on itself", job.name));
491
                }
492
            }
493
        }
494
        let graph: BTreeMap<&str, Vec<&str>> =
495
            jobs.values().map(|(j, _)| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
496
        if let Some(cycle) = find_cycle(&graph) {
497
            let span = jobs[cycle[0]].1;
498
            ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
499
        }
500
        if !ctx.problems.is_empty() {
501
            ctx.problems.sort_by_key(|p| p.line);
502
            return Err(Problems(ctx.problems));
503
        }
504
        Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect(), on })
505
    }
506
507
    /// Jobs in an order where each comes after what it depends on (ties by name).
508
    pub fn order(&self) -> Vec<&Job> {
509
        let mut done: BTreeSet<&str> = BTreeSet::new();
510
        let mut out = Vec::with_capacity(self.jobs.len());
511
        while out.len() < self.jobs.len() {
512
            let before = out.len();
513
            for j in self.jobs.values() {
514
                if !done.contains(j.name.as_str()) && j.depends_on.iter().all(|d| done.contains(d.as_str())) {
515
                    out.push(j);
516
                }
517
            }
518
            for j in &out[before..] {
519
                done.insert(j.name.as_str());
520
            }
521
            if out.len() == before {
522
                break; // validated pipelines have no cycles
523
            }
524
        }
525
        out
526
    }
527
528
    /// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
529
    pub fn depths(&self) -> BTreeMap<&str, usize> {
530
        let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
531
        for j in self.order() {
532
            let d = j.depends_on.iter().map(|x| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
533
            depth.insert(j.name.as_str(), d);
534
        }
535
        depth
536
    }
537
}
538
539
fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
540
    #[derive(Clone, Copy, PartialEq)]
541
    enum Mark {
542
        Visiting,
543
        Done,
544
    }
545
    fn visit<'a>(
546
        n: &'a str,
547
        graph: &BTreeMap<&'a str, Vec<&'a str>>,
548
        marks: &mut BTreeMap<&'a str, Mark>,
549
        stack: &mut Vec<&'a str>,
550
    ) -> Option<Vec<&'a str>> {
551
        match marks.get(n) {
552
            Some(Mark::Done) => return None,
553
            Some(Mark::Visiting) => {
554
                let at = stack.iter().position(|s| *s == n).unwrap_or(0);
555
                let mut c = stack[at..].to_vec();
556
                c.push(n);
557
                return Some(c);
558
            }
559
            None => {}
560
        }
561
        marks.insert(n, Mark::Visiting);
562
        stack.push(n);
563
        for d in graph.get(n).into_iter().flatten() {
564
            if graph.contains_key(d) && *d != n {
565
                if let Some(c) = visit(d, graph, marks, stack) {
566
                    return Some(c);
567
                }
568
            }
569
        }
570
        stack.pop();
571
        marks.insert(n, Mark::Done);
572
        None
573
    }
574
    let mut marks = BTreeMap::new();
575
    for n in graph.keys() {
576
        let mut stack = Vec::new();
577
        if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
578
            return Some(c);
579
        }
580
    }
581
    None
582
}
583
584
fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
585
    match node.children() {
586
        Some(d) => d.nodes(),
587
        None => {
588
            ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
589
            &[]
590
        }
591
    }
592
}
593
594
fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
595
    let name = ctx.one_string(node)?;
596
    let named = ctx.name_ok(node.span(), "sandbox", &name);
597
    let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
598
        (Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
599
    let mut seen = BTreeSet::new();
600
    let mut env = BTreeMap::new();
601
    let mut runs_on = Vec::new();
602
    let mut timeout_s = None;
603
    for c in block(ctx, node) {
604
        let key = c.name().value();
605
        if key != "runs-on" && !seen.insert(key.to_string()) {
606
            ctx.err(c.span(), format!("`{key}` is given twice"));
607
        }
608
        if key == "env" {
609
            parse_env(ctx, c, &mut env);
610
            continue;
611
        }
612
        if key == "runs-on" {
613
            ctx.no_children(c);
614
            parse_labels(ctx, c, &mut runs_on);
615
            continue;
616
        }
617
        ctx.no_children(c);
618
        match key {
619
            "runtime" => {
620
                runtime = match ctx.one_string(c).as_deref() {
621
                    Some("microvm") => Some(Runtime::Microvm),
622
                    Some("syscall") | Some("gvisor") => Some(Runtime::Syscall),
623
                    Some("wasm") => {
624
                        ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
625
                        None
626
                    }
627
                    Some(other) => {
628
                        ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
629
                        None
630
                    }
631
                    None => None,
632
                }
633
            }
634
            "rootfs" => {
635
                rootfs = match ctx.one_string(c) {
636
                    Some(r) => match r.strip_prefix("cas://blob/") {
637
                        Some(h) if h.len() == 64 && h.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
638
                            Some(h.to_string())
639
                        }
640
                        _ => {
641
                            ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
642
                            None
643
                        }
644
                    },
645
                    None => None,
646
                }
647
            }
648
            "memory-limit" => {
649
                memory = match ctx.one_string(c) {
650
                    Some(m) => match parse_memory(&m) {
651
                        Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
652
                        Some(_) => {
653
                            ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
654
                            None
655
                        }
656
                        None => {
657
                            ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
658
                            None
659
                        }
660
                    },
661
                    None => None,
662
                }
663
            }
664
            "cpu-limit" => {
665
                cpus = match ctx.one_integer(c) {
666
                    Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
667
                    Some(n) => {
668
                        ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
669
                        None
670
                    }
671
                    None => None,
672
                }
673
            }
674
            "network" => {
675
                network = match ctx.one_string(c).as_deref() {
676
                    Some("isolated") => Some(Network::Isolated),
677
                    // Loopback only: the guest's `lo` comes up, and it still has
678
                    // no device that reaches anything outside it.
679
                    Some("loopback") => Some(Network::Loopback),
680
                    Some("egress") => {
681
                        ctx.err(c.span(), "network \"egress\" is not in M0; M0 has \"isolated\" and \"loopback\"");
682
                        None
683
                    }
684
                    Some(other) => {
685
                        ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
686
                        None
687
                    }
688
                    None => None,
689
                }
690
            }
691
            "timeout" => timeout_s = parse_timeout(ctx, c),
692
            other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
693
        }
694
    }
695
    if rootfs.is_none() && !seen.contains("rootfs") {
696
        ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
697
    }
698
    if !named {
699
        return None;
700
    }
701
    Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network?, env, runs_on, timeout_s })
702
}
703
704
/// An `on` block: `push` and `tag` with globs (none: any), `manual`,
705
/// `schedule` with five-field cron lines.
706
fn parse_on(ctx: &mut Ctx, node: &KdlNode) -> On {
707
    let mut on = On::default();
708
    if !node.entries().is_empty() {
709
        ctx.err(node.span(), "`on` takes a block: on { push \"main\"; tag \"v*\"; manual }");
710
    }
711
    let globs = |ctx: &mut Ctx, c: &KdlNode| -> Vec<String> {
712
        let g = ctx.strings(c);
713
        for p in &g {
714
            if let Err(e) = globset::Glob::new(p) {
715
                ctx.err(c.span(), format!("{p:?}: {e}"));
716
            }
717
        }
718
        g
719
    };
720
    for c in block(ctx, node) {
721
        ctx.no_children(c);
722
        match c.name().value() {
723
            "push" => on.push.get_or_insert_with(Vec::new).extend(globs(ctx, c)),
724
            "tag" => on.tag.get_or_insert_with(Vec::new).extend(globs(ctx, c)),
725
            "manual" => on.manual = true,
726
            "schedule" => {
727
                for line in ctx.strings(c) {
728
                    if crate::cron::Cron::parse(&line).is_some() {
729
                        on.schedule.push(line);
730
                    } else {
731
                        ctx.err(
732
                            c.span(),
733
                            format!("schedule {line:?}: expected five cron fields (minute hour day month weekday), as \"0 3 * * *\""),
734
                        );
735
                    }
736
                }
737
            }
738
            other => ctx.err(c.span(), format!("unknown `on` event `{other}`; expected push, tag, manual or schedule")),
739
        }
740
    }
741
    if on.push.is_none() && on.tag.is_none() && !on.manual && on.schedule.is_empty() {
742
        ctx.err(node.span(), "`on` names no event, so nothing would run it");
743
    }
744
    on
745
}
746
747
/// The longest a pipeline may ask a job to run.
748
pub const MAX_TIMEOUT_S: u64 = 24 * 3600;
749
750
/// A duration: whole hours, minutes and seconds in that order, each
751
/// optional, as "90m", "2h", "1h30m" or "45s".
752
pub fn parse_duration(s: &str) -> Option<u64> {
753
    let (mut total, mut num, mut last) = (0u64, String::new(), 0u8);
754
    for ch in s.chars() {
755
        if ch.is_ascii_digit() {
756
            num.push(ch);
757
            continue;
758
        }
759
        let (unit, mult) = match ch {
760
            'h' => (1, 3600),
761
            'm' => (2, 60),
762
            's' => (3, 1),
763
            _ => return None,
764
        };
765
        // Each unit once, largest first, each with its number.
766
        if num.is_empty() || unit <= last {
767
            return None;
768
        }
769
        last = unit;
770
        total = total.checked_add(num.parse::<u64>().ok()?.checked_mul(mult)?)?;
771
        num.clear();
772
    }
773
    (num.is_empty() && total > 0).then_some(total)
774
}
775
776
/// A `timeout` line: one duration, from a minute to a day.
777
fn parse_timeout(ctx: &mut Ctx, c: &KdlNode) -> Option<u64> {
778
    let t = ctx.one_string(c)?;
779
    match parse_duration(&t) {
780
        Some(secs) if (60..=MAX_TIMEOUT_S).contains(&secs) => Some(secs),
781
        Some(_) => {
782
            ctx.err(c.span(), format!("timeout {t:?} must be from 1m to 24h"));
783
            None
784
        }
785
        None => {
786
            ctx.err(c.span(), format!("timeout {t:?}: expected hours, minutes and seconds, as \"90m\" or \"1h30m\""));
787
            None
788
        }
789
    }
790
}
791
792
/// The labels of a `runs-on` line: lower-case letters, digits, `-`, `_`
793
/// and `.`, at most 64 bytes each; into `into`, sorted, once each.
794
fn parse_labels(ctx: &mut Ctx, c: &KdlNode, into: &mut Vec<String>) {
795
    let labels = ctx.strings(c);
796
    if labels.is_empty() {
797
        ctx.err(c.span(), "`runs-on` needs one or more labels, as runs-on \"riscv64\"");
798
    }
799
    for l in labels {
800
        let ok = !l.is_empty()
801
            && l.len() <= 64
802
            && l.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_' | b'.'));
803
        if ok {
804
            into.push(l);
805
        } else {
806
            ctx.err(c.span(), format!("runs-on {l:?}: a label is lower-case letters, digits, -, _ and ., at most 64"));
807
        }
808
    }
809
    into.sort();
810
    into.dedup();
811
}
812
813
/// An `env` block of `NAME "value"` lines, into `env`.
814
fn parse_env(ctx: &mut Ctx, c: &KdlNode, env: &mut BTreeMap<String, String>) {
815
    if !c.entries().is_empty() {
816
        ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
817
    }
818
    for e in block(ctx, c) {
819
        let k = e.name().value().to_string();
820
        let valid =
821
            !k.is_empty() && !k.starts_with(|c: char| c.is_ascii_digit()) && k.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_');
822
        if !valid {
823
            ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
824
            continue;
825
        }
826
        ctx.no_children(e);
827
        if let Some(v) = ctx.one_string(e) {
828
            if env.insert(k.clone(), v).is_some() {
829
                ctx.err(e.span(), format!("environment {k} is set twice"));
830
            }
831
        }
832
    }
833
}
834
835
/// `mount { source "ephemeral://disk"; target "/src/target"; size "8GiB" }`.
836
fn parse_mount(ctx: &mut Ctx, node: &KdlNode) -> Option<Mount> {
837
    let (mut source, mut target, mut size) = (None, None, None);
838
    let mut seen = BTreeSet::new();
839
    for c in block(ctx, node) {
840
        let key = c.name().value();
841
        if !seen.insert(key.to_string()) {
842
            ctx.err(c.span(), format!("`{key}` is given twice"));
843
            continue;
844
        }
845
        ctx.no_children(c);
846
        match key {
847
            "source" => {
848
                source = match ctx.one_string(c).as_deref() {
849
                    Some("ephemeral://tmpfs") => Some(MountSource::Tmpfs),
850
                    Some("ephemeral://disk") => Some(MountSource::Disk),
851
                    Some(other) => {
852
                        ctx.err(
853
                            c.span(),
854
                            format!("mount source {other:?} is not in M0 yet; M0 mounts \"ephemeral://tmpfs\" or \"ephemeral://disk\""),
855
                        );
856
                        None
857
                    }
858
                    None => None,
859
                }
860
            }
861
            "target" => {
862
                target = match ctx.one_string(c) {
863
                    Some(t) if mount_target_ok(&t) => Some(t),
864
                    Some(t) => {
865
                        ctx.err(
866
                            c.span(),
867
                            format!("mount target {t:?} must be an absolute path under /src or /tmp, without . or .. or a trailing /"),
868
                        );
869
                        None
870
                    }
871
                    None => None,
872
                }
873
            }
874
            "size" => {
875
                size = match ctx.one_string(c) {
876
                    Some(m) => match parse_memory(&m) {
877
                        Some(mib) if (MIN_MOUNT_MIB..=MAX_MOUNT_MIB).contains(&mib) => Some(mib as u32),
878
                        Some(_) => {
879
                            ctx.err(c.span(), format!("mount size {m:?} must be from 16MiB to 64GiB"));
880
                            None
881
                        }
882
                        None => {
883
                            ctx.err(c.span(), format!("mount size {m:?}: expected a whole number of KiB, MiB or GiB, as \"8GiB\""));
884
                            None
885
                        }
886
                    },
887
                    None => None,
888
                }
889
            }
890
            "readonly" | "include" => ctx.err(c.span(), format!("mount `{key}` is not in M0 yet")),
891
            other => ctx.err(c.span(), format!("unknown mount setting `{other}`")),
892
        }
893
    }
894
    for (name, missing) in [("source", source.is_none()), ("target", target.is_none()), ("size", size.is_none())] {
895
        if missing && !seen.contains(name) {
896
            ctx.err(node.span(), format!("a mount needs a {name}"));
897
        }
898
    }
899
    Some(Mount { source: source?, target: target?, size_mib: size? })
900
}
901
902
/// `output { artifact "name" { path "/src/target/release/tool" } }`.
903
fn parse_output(ctx: &mut Ctx, node: &KdlNode, job: &mut Job) {
904
    if !node.entries().is_empty() {
905
        ctx.err(node.span(), "`output` takes a block of artifacts");
906
    }
907
    for a in block(ctx, node) {
908
        if a.name().value() != "artifact" {
909
            ctx.err(a.span(), format!("unknown output `{}`; outputs are `artifact \"name\" {{ path \"/src/…\" }}`", a.name().value()));
910
            continue;
911
        }
912
        let Some(name) = ctx.one_string(a) else { continue };
913
        if !ctx.name_ok(a.span(), "artifact", &name) {
914
            continue;
915
        }
916
        let mut path = None;
917
        for c in block(ctx, a) {
918
            ctx.no_children(c);
919
            match c.name().value() {
920
                "path" => match ctx.one_string(c) {
921
                    Some(p) if output_path_ok(&p) => path = Some(p),
922
                    Some(p) => ctx.err(
923
                        c.span(),
924
                        format!("artifact path {p:?} must be an absolute path under /src, without . or .., quotes or spaces"),
925
                    ),
926
                    None => {}
927
                },
928
                "format" => {
929
                    if ctx.one_string(c).as_deref() != Some("blob") {
930
                        ctx.err(c.span(), "artifact format is \"blob\"");
931
                    }
932
                }
933
                other => ctx.err(c.span(), format!("unknown artifact setting `{other}`")),
934
            }
935
        }
936
        let Some(path) = path else {
937
            ctx.err(a.span(), format!("artifact {name:?} needs a path"));
938
            continue;
939
        };
940
        if job.outputs.iter().any(|o| o.name == name) {
941
            ctx.err(a.span(), format!("artifact {name:?} is given twice"));
942
        } else if job.outputs.len() == MAX_OUTPUTS {
943
            ctx.err(a.span(), format!("a job has at most {MAX_OUTPUTS} artifacts"));
944
        } else {
945
            job.outputs.push(Output { name, path });
946
        }
947
    }
948
}
949
950
fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
951
    let name = ctx.one_string(node)?;
952
    let named = ctx.name_ok(node.span(), "job", &name);
953
    let mut job = Job {
954
        name: name.clone(),
955
        sandbox: String::new(),
956
        scope: vec![],
957
        depends_on: vec![],
958
        env: BTreeMap::new(),
959
        exec: vec![],
960
        mounts: vec![],
961
        outputs: vec![],
962
        runs_on: vec![],
963
        timeout_s: None,
964
        on: None,
965
    };
966
    let mut seen = BTreeSet::new();
967
    let mut outputs_at = None;
968
    for c in block(ctx, node) {
969
        let key = c.name().value();
970
        if key != "scope" && key != "depends-on" && key != "mount" && key != "runs-on" && !seen.insert(key.to_string()) {
971
            ctx.err(c.span(), format!("`{key}` is given twice"));
972
        }
973
        match key {
974
            "sandbox" => {
975
                ctx.no_children(c);
976
                job.sandbox = ctx.one_string(c).unwrap_or_default();
977
            }
978
            "scope" => {
979
                ctx.no_children(c);
980
                for g in ctx.strings(c) {
981
                    match globset::Glob::new(&g) {
982
                        Ok(_) => job.scope.push(g),
983
                        Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
984
                    }
985
                }
986
            }
987
            "depends-on" => {
988
                ctx.no_children(c);
989
                job.depends_on.extend(ctx.strings(c));
990
            }
991
            "runs-on" => {
992
                ctx.no_children(c);
993
                parse_labels(ctx, c, &mut job.runs_on);
994
            }
995
            "timeout" => {
996
                ctx.no_children(c);
997
                job.timeout_s = parse_timeout(ctx, c);
998
            }
999
            "on" => job.on = Some(parse_on(ctx, c)),
1000
            "env" => parse_env(ctx, c, &mut job.env),
1001
            "exec" => {
1002
                ctx.no_children(c);
1003
                job.exec = ctx.strings(c);
1004
                if job.exec.is_empty() {
1005
                    ctx.err(c.span(), "`exec` needs a program and its arguments");
1006
                } else if job.exec.iter().any(|a| a.contains('\0')) {
1007
                    ctx.err(c.span(), "`exec` arguments cannot contain NUL");
1008
                }
1009
            }
1010
            "mount" => {
1011
                if !c.entries().is_empty() {
1012
                    ctx.err(c.span(), "`mount` takes a block of source, target and size");
1013
                }
1014
                if let Some(m) = parse_mount(ctx, c) {
1015
                    if job.mounts.len() == MAX_MOUNTS {
1016
                        ctx.err(c.span(), format!("a job has at most {MAX_MOUNTS} mounts"));
1017
                    } else if let Some(other) = job.mounts.iter().find(|o| {
1018
                        o.target == m.target
1019
                            || m.target.starts_with(&format!("{}/", o.target))
1020
                            || o.target.starts_with(&format!("{}/", m.target))
1021
                    }) {
1022
                        ctx.err(c.span(), format!("mount {:?} overlaps mount {:?}", m.target, other.target));
1023
                    } else {
1024
                        job.mounts.push(m);
1025
                    }
1026
                }
1027
            }
1028
            "output" => {
1029
                outputs_at = Some(c.span());
1030
                parse_output(ctx, c, &mut job);
1031
            }
1032
            n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
1033
            other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
1034
        }
1035
    }
1036
    job.mounts.sort_by(|a, b| a.target.cmp(&b.target));
1037
    job.outputs.sort_by(|a, b| a.name.cmp(&b.name));
1038
    // A tmpfs is gone when the job ends; nothing on it can be kept.
1039
    for o in &job.outputs {
1040
        if let Some(m) = job.mounts.iter().find(|m| m.source == MountSource::Tmpfs && o.path.starts_with(&format!("{}/", m.target))) {
1041
            ctx.err(
1042
                outputs_at.unwrap_or(node.span()),
1043
                format!(
1044
                    "artifact {:?} is on the tmpfs mount {:?}, which is gone when the job ends; use ephemeral://disk",
1045
                    o.name, m.target
1046
                ),
1047
            );
1048
        }
1049
    }
1050
    if job.sandbox.is_empty() && !seen.contains("sandbox") {
1051
        ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
1052
    }
1053
    if job.exec.is_empty() && !seen.contains("exec") {
1054
        ctx.err(node.span(), format!("job {name:?} needs an exec"));
1055
    }
1056
    job.depends_on.sort();
1057
    job.depends_on.dedup();
1058
    // Kept even when incomplete, so its references are checked too; the
1059
    // problems above already make the pipeline invalid.
1060
    named.then_some(job)
1061
}