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