Scheduled runs: on { schedule } jobs start on the default bookmark when due 348f8c4baccf

- **What starts a job: `on` (oxygenflow#24).** A pipeline's or a job's `on { push "main" "feature/*"; tag "v*"; manual }` decides which events run it: pushing a tag now runs the jobs that name it, on the tagged state; a run has only the jobs its event starts (and what they need), and an event that starts none starts no run. A rerun repeats its run's event. A bookmark sent again unchanged starts no run. `schedule "0 3 * * *"` lines run their jobs on the default bookmark when due (UTC), checked each minute.

    start_run_at(app, row, bookmark, state, heads, actor, trigger, touched, None).await

}

/// [`start_run_with`], for a scheduled run due at `at` on the hub's clock.

#[allow(clippy::too_many_arguments)]

async fn start_run_at(

    app: &Shared,

    row: &db::repos::RepoRow,

    bookmark: &str,

    state: &str,

    heads: &[ChangeHash],

    actor: &str,

    trigger: &str,

    touched: Option<&[String]>,

    at: Option<flow_core::pipeline::When>,

) -> Result<Option<i64>, String> {

            let event = match at {

                Some(w) => flow_core::pipeline::Event::ScheduleAt(w),

                None => event_of(trigger, bookmark),

            };

}

/// Start the scheduled runs due at `when` (oxygenflow#24): on each

/// repository that runs Flow, the jobs of its default bookmark's pipeline

/// with a `schedule` line due then. Returns how many runs started.

pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {

    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };

    let mut started = 0;

    for row in rows.into_iter().filter(|r| enabled(app, r)) {

        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };

        let b = row.default_bookmark.clone();

        let found = app

            .repos

            .read(&repo, move |r, _| {

                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };

                Ok(Some((st, r.store().get_state(&st)?.heads)))

            })

            .await;

        let Ok(Some((state, heads))) = found else { continue };

        if heads.is_empty() {

            continue;

        }

        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when)).await {

            Ok(Some(_)) => started += 1,

            Ok(None) => {}

            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),

        }

    }

    started

}

/// The hub's clock as a [`When`](flow_core::pipeline::When).

pub fn when_now() -> flow_core::pipeline::When {

    let now = time::OffsetDateTime::now_utc();

    flow_core::pipeline::When {

        minute: now.minute() as u32,

        hour: now.hour() as u32,

        day: now.day() as u32,

        month: u8::from(now.month()) as u32,

        weekday: now.weekday().number_days_from_sunday() as u32,

    }

}

/// Start the scheduled runs due each minute, once a minute.

pub async fn scheduler(app: Shared) {

    let mut last = None;

    loop {

        tokio::time::sleep(std::time::Duration::from_secs(15)).await;

        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;

        if last == Some(minute) {

            continue;

        }

        last = Some(minute);

        let n = run_due(&app, when_now()).await;

        if n > 0 {

            tracing::info!("flow: {n} scheduled run(s) started");

        }

    }

    tokio::spawn(flow::scheduler(app.clone()));

    assert_eq!(

        (third.trigger_label().as_str(), third.status.as_str(), third.state.as_str()),

        ("rerun:1", "queued", first[0].state.as_str())

    );

}

/// oxygenflow#24: a job with a `schedule` line runs on the default

/// bookmark when the line comes due, and only then; other jobs stay out.

#[tokio::test(flavor = "multi_thread")]

async fn scheduled_jobs_run_when_due() {

    let h = hub().await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    create(&h.app, "andy", "nightly").await;

    let url = format!("{}/andy/nightly", h.base);

    let work = tempfile::tempdir().unwrap();

    let c = client(&url, work.path());

    // An odd minute, so the hub's own ticker cannot fire it during the test.

    let extra = "job \"nightly\" {\n    sandbox \"alpine\"\n    on {\n        schedule \"17 4 29 2 *\"\n    }\n    exec \"true\"\n}\n";

    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));

    write(work.path(), "test.sh", "echo ok\n");

    c.commit("Start", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert_eq!(runs_of(&h.app, "andy", "nightly").await.len(), 1);

    let first = runs_of(&h.app, "andy", "nightly").await;

    let names: Vec<String> = db::flow::jobs(&h.app.db, first[0].id).await.unwrap().into_iter().map(|j| j.name).collect();

    assert!(!names.contains(&"nightly".to_string()), "a push does not run a scheduled job");

    use flow_core::pipeline::When;

    let not_yet = When { minute: 16, hour: 4, day: 29, month: 2, weekday: 4 };

    assert_eq!(oxygenhub::flow::run_due(&h.app, not_yet).await, 0);

    let due = When { minute: 17, ..not_yet };

    assert_eq!(oxygenhub::flow::run_due(&h.app, due).await, 1);

    let runs = runs_of(&h.app, "andy", "nightly").await;

    assert_eq!(runs.len(), 2);

    assert_eq!(runs[0].trigger, "schedule");

    let names: Vec<String> = db::flow::jobs(&h.app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();

    assert_eq!(names, ["nightly"]);

    assert_eq!(runs[0].bookmark, "main", "on the default bookmark");