pipeline.kdl: on { push; tag; manual; schedule } for the pipeline and each job bf7bb1b80fa3

//! Five-field cron lines for `on { schedule "…" }` (oxygenflow#24):

//! minute, hour, day of month, month, day of week (0 or 7 is Sunday), each

//! `*`, a number, a range `a-b`, a list `a,b`, or any of those `/step`.

//! The hub checks each minute which lines are due.

/// A parsed line: the values each field allows.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Cron {

    minute: Vec<bool>,

    hour: Vec<bool>,

    day: Vec<bool>,

    month: Vec<bool>,

    weekday: Vec<bool>,

    /// Both day fields restricted: either may match, as cron has it.

    either_day: bool,

}

fn field(text: &str, lo: u32, hi: u32) -> Option<(Vec<bool>, bool)> {

    let mut allowed = vec![false; hi as usize + 1];

    let star = text == "*";

    for part in text.split(',') {

        let (range, step) = match part.split_once('/') {

            Some((r, s)) => (r, s.parse::<u32>().ok().filter(|s| *s > 0)?),

            None => (part, 1),

        };

        let (a, b) = if range == "*" {

            (lo, hi)

        } else if let Some((a, b)) = range.split_once('-') {

            (a.parse().ok()?, b.parse().ok()?)

        } else {

            let a: u32 = range.parse().ok()?;

            // `5/15` is from 5 to the end, every 15.

            (a, if part.contains('/') { hi } else { a })

        };

        if a < lo || b > hi || a > b {

            return None;

        }

        let mut v = a;

        while v <= b {

            allowed[v as usize] = true;

            v += step;

        }

    }

    Some((allowed, star))

}

impl Cron {

    pub fn parse(line: &str) -> Option<Cron> {

        let f: Vec<&str> = line.split_whitespace().collect();

        let [m, h, d, mo, w] = f.as_slice() else { return None };

        let (minute, _) = field(m, 0, 59)?;

        let (hour, _) = field(h, 0, 23)?;

        let (day, day_star) = field(d, 1, 31)?;

        let (month, _) = field(mo, 1, 12)?;

        let (mut weekday, week_star) = field(w, 0, 7)?;

        if weekday[7] {

            weekday[0] = true;

        }

        Some(Cron { minute, hour, day, month, weekday, either_day: !day_star && !week_star })

    }

    /// Whether the line is due at this minute (UTC): `weekday` 0 is Sunday.

    pub fn matches(&self, minute: u32, hour: u32, day: u32, month: u32, weekday: u32) -> bool {

        let get = |v: &Vec<bool>, i: u32| v.get(i as usize).copied().unwrap_or(false);

        let (d, w) = (get(&self.day, day), get(&self.weekday, weekday % 7));

        let days = if self.either_day { d || w } else { d && w };

        get(&self.minute, minute) && get(&self.hour, hour) && get(&self.month, month) && days

    }

}

#[cfg(test)]

mod tests {

    use super::Cron;

    #[test]

    fn lines_parse_and_match_as_cron_does() {

        let nightly = Cron::parse("0 3 * * *").unwrap();

        assert!(nightly.matches(0, 3, 15, 6, 2));

        assert!(!nightly.matches(1, 3, 15, 6, 2));

        let quarter = Cron::parse("*/15 9-17 * * 1-5").unwrap();

        assert!(quarter.matches(45, 17, 2, 10, 5));

        assert!(!quarter.matches(50, 17, 2, 10, 5));

        assert!(!quarter.matches(0, 12, 3, 10, 6), "Saturday");

        let sunday = Cron::parse("30 4 * * 7").unwrap();

        assert!(sunday.matches(30, 4, 4, 10, 0));

        let either = Cron::parse("0 0 1 * 1").unwrap();

        assert!(either.matches(0, 0, 1, 10, 3) && either.matches(0, 0, 5, 10, 1), "the 1st or a Monday");

        for bad in ["", "* * * *", "60 * * * *", "* 24 * * *", "* * 0 * *", "*/0 * * * *", "a * * * *", "5-1 * * * *"] {

            assert!(Cron::parse(bad).is_none(), "{bad:?}");

        }

    }

}

pub mod cron;

    /// When the pipeline's jobs run, unless a job says (oxygenflow#24);

    /// absent: on every push and from "Run pipeline".

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub on: Option<On>,

}

/// What starts a job (`on { push "main"; tag "v*"; manual; schedule "0 3 * * *" }`).

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]

pub struct On {

    /// Bookmark globs a push runs it for; `Some(vec![])`: any bookmark.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub push: Option<Vec<String>>,

    /// Tag globs a pushed tag runs it for; `Some(vec![])`: any tag.

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub tag: Option<Vec<String>>,

    /// From "Run pipeline" (and `oh run start`).

    #[serde(default, skip_serializing_if = "std::ops::Not::not")]

    pub manual: bool,

    /// Five-field cron lines, on the hub's clock, for the default bookmark.

    #[serde(default, skip_serializing_if = "Vec::is_empty")]

    pub schedule: Vec<String>,

/// Why a run started, for [`On::matches`].

#[derive(Clone, Copy, Debug, PartialEq, Eq)]

pub enum Event<'a> {

    /// A push moved this bookmark.

    Push(&'a str),

    /// This tag was pushed.

    Tag(&'a str),

    /// "Run pipeline", `oh run start`.

    Manual,

    /// A `schedule` line came due.

    Schedule,

}

impl On {

    /// What a job without any `on` runs for: every push, and by hand.

    pub fn default_on() -> On {

        On { push: Some(vec![]), tag: None, manual: true, schedule: vec![] }

    }

    pub fn matches(&self, event: Event<'_>) -> bool {

        let any = |globs: &Option<Vec<String>>, name: &str| match globs {

            None => false,

            Some(g) if g.is_empty() => true,

            Some(g) => g.iter().any(|p| globset::Glob::new(p).is_ok_and(|p| p.compile_matcher().is_match(name))),

        };

        match event {

            Event::Push(b) => any(&self.push, b),

            Event::Tag(t) => any(&self.tag, t),

            Event::Manual => self.manual,

            Event::Schedule => !self.schedule.is_empty(),

        }

    }

}

impl Pipeline {

    /// When `job` runs: its own `on`, else the pipeline's, else the default.

    pub fn on_of(&self, job: &Job) -> On {

        job.on.clone().or_else(|| self.on.clone()).unwrap_or_else(On::default_on)

    }

}

    /// When it runs, over the pipeline's `on` (oxygenflow#24).

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub on: Option<On>,

        let mut on = None;

                "on" => {

                    if on.is_some() {

                        ctx.err(node.span(), "`on` is given twice");

                    }

                    on = Some(parse_on(&mut ctx, node));

                }

                n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, on, sandbox or job")),

        Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(|(k, (j, _))| (k, j)).collect(), on })

}

/// An `on` block: `push` and `tag` with globs (none: any), `manual`,

/// `schedule` with five-field cron lines.

fn parse_on(ctx: &mut Ctx, node: &KdlNode) -> On {

    let mut on = On::default();

    if !node.entries().is_empty() {

        ctx.err(node.span(), "`on` takes a block: on { push \"main\"; tag \"v*\"; manual }");

    }

    let globs = |ctx: &mut Ctx, c: &KdlNode| -> Vec<String> {

        let g = ctx.strings(c);

        for p in &g {

            if let Err(e) = globset::Glob::new(p) {

                ctx.err(c.span(), format!("{p:?}: {e}"));

            }

        }

        g

    };

    for c in block(ctx, node) {

        ctx.no_children(c);

        match c.name().value() {

            "push" => on.push.get_or_insert_with(Vec::new).extend(globs(ctx, c)),

            "tag" => on.tag.get_or_insert_with(Vec::new).extend(globs(ctx, c)),

            "manual" => on.manual = true,

            "schedule" => {

                for line in ctx.strings(c) {

                    if crate::cron::Cron::parse(&line).is_some() {

                        on.schedule.push(line);

                    } else {

                        ctx.err(

                            c.span(),

                            format!("schedule {line:?}: expected five cron fields (minute hour day month weekday), as \"0 3 * * *\""),

                        );

                    }

                }

            }

            other => ctx.err(c.span(), format!("unknown `on` event `{other}`; expected push, tag, manual or schedule")),

        }

    }

    if on.push.is_none() && on.tag.is_none() && !on.manual && on.schedule.is_empty() {

        ctx.err(node.span(), "`on` names no event, so nothing would run it");

    }

    on

        on: None,

            "on" => job.on = Some(parse_on(ctx, c)),

/// oxygenflow#24: `on` says what starts a job, at pipeline or job level;

/// a job without one runs on every push and by hand, as before.

#[test]

fn on_says_what_starts_a_job() {

    use crate::pipeline::{Event, Pipeline};

    let root = "a".repeat(64);

    let text = format!(

        "schema-version 1\non {{\n    push \"main\" \"feature/*\"\n    manual\n}}\nsandbox \"s\" {{\n    runtime \"microvm\"\n    rootfs \"cas://blob/{root}\"\n    memory-limit \"1GiB\"\n    cpu-limit 1\n    network \"isolated\"\n}}\njob \"test\" {{\n    sandbox \"s\"\n    exec \"true\"\n}}\njob \"dist\" {{\n    sandbox \"s\"\n    on {{\n        tag \"v*\"\n        manual\n    }}\n    exec \"true\"\n}}\njob \"nightly\" {{\n    sandbox \"s\"\n    on {{\n        schedule \"0 3 * * *\"\n    }}\n    exec \"true\"\n}}\n"

    );

    let p = Pipeline::parse(&text).unwrap();

    let runs = |job: &str, e: Event| p.on_of(&p.jobs[job]).matches(e);

    assert!(runs("test", Event::Push("main")) && runs("test", Event::Push("feature/9-x")));

    assert!(!runs("test", Event::Push("develop")), "not a bookmark the pipeline names");

    assert!(!runs("test", Event::Tag("v1.0.0")) && runs("test", Event::Manual));

    assert!(runs("dist", Event::Tag("v0.3.1")) && !runs("dist", Event::Tag("nightly")));

    assert!(!runs("dist", Event::Push("main")) && runs("dist", Event::Manual));

    assert!(runs("nightly", Event::Schedule) && !runs("nightly", Event::Manual));

    // Without any `on`, as before.

    let plain = text.replace("on {\n    push \"main\" \"feature/*\"\n    manual\n}\n", "");

    let p = Pipeline::parse(&plain).unwrap();

    assert!(p.on_of(&p.jobs["test"]).matches(Event::Push("anything")) && p.on_of(&p.jobs["test"]).matches(Event::Manual));

    for (bad, why) in [

        ("on {\n}\n", "names no event"),

        ("on {\n    when \"x\"\n}\n", "unknown `on` event"),

        ("on {\n    schedule \"0 3 * *\"\n}\n", "five cron fields"),

    ] {

        let t = format!("{bad}{plain}");

        let e = Pipeline::parse(&t).unwrap_err().to_string();

        assert!(e.contains(why), "{bad:?}: {e}");

    }

}

            on: None,

**What starts a job.** An `on` block, at the top of the pipeline or in a job (the job's wins), names the events that run it: `push` with bookmark globs (none: any bookmark), `tag` with tag globs (none: any tag), `manual` ("Run pipeline", `oh run start`) and `schedule` with five-field cron lines on the hub's clock (UTC), run on the default bookmark. Without `on`, a job runs on every push and by hand, as before. A run has only the jobs its event starts (and what they depend on); an event that starts no job starts no run. A rerun repeats its run's event.

```kdl

on {

    push "main" "feature/*"

    manual

}

job "dist" {

    on {

        tag "v*"

        manual

    }

    …

}

```