~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

Workflows run on a schedule, at the default branch's tip

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:run/schedule
.changeset/workflow-schedules.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# Workflows run on a schedule
A workflow's `on: {schedule: [{cron: '30 3 * * *'}]}` runs it at the default branch's tip at the minutes its cron expressions name, in UTC, at most every five minutes. Its steps see `KNOTLESS_SCHEDULE`, the expression that ran it, and its log says `==> on its schedule`. The runner reads schedules from the default branch's tip as it starts and whenever that tip moves. A time that comes while the last run is still going is skipped. `knotless workflows` lists each workflow's schedules at the default branch.
crates/knotless/src/cli.rs+3−0
View file
@@ -1345,6 +1345,9 @@ async fn command(
let runs: Vec<&str> = runs.iter().filter(|r| r.1).map(|r| r.0).collect();
let error = w.error.map(|e| format!("\t{e}")).unwrap_or_default();
writeln!(out, "{}\t{}{error}", w.name, runs.join(","))?;
for s in w.schedule {
writeln!(out, "\tschedule {}", s.cron)?;
}
}
}
"labels" => {
crates/knotless/src/cron.rsnew+268−0
View file
@@ -0,0 +1,268 @@
//! When a workflow's `schedule` runs it: a cron expression, five fields
//! of the time in UTC, `minute hour day-of-month month day-of-week`, or
//! one of `@hourly`, `@daily`, `@weekly`, `@monthly` and `@yearly`.
//!
//! A field is `*`, a number, a range `a-b`, any of those with a step
//! (`*/15`, `1-30/2`), or a list of them, `,` between. Days of the week
//! run from 0, Sunday, to 6, and 7 is Sunday too. As cron has it, when
//! both days are named, a day either names runs it. Names, as `mon` or
//! `jan`, are not taken: numbers only.
//!
//! A schedule runs at most every five minutes: one with two minutes of
//! the hour closer than that is refused, so that a schedule costs a
//! runner what a push does at most.
use anyhow::{Result, bail};
/// The fewest minutes between two runs of a schedule.
pub const EVERY_MIN: u32 = 5;
/// A schedule, as the minutes, hours, days, months and weekdays it runs
/// at, each a bit set.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cron {
minutes: u64,
hours: u64,
days: u64,
months: u64,
weekdays: u64,
/// Whether both days were named, rather than one left `*`.
both_days: bool,
}
/// A time to the minute, in UTC: what a schedule is matched against.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Minute {
pub minute: u32,
pub hour: u32,
pub day: u32,
pub month: u32,
/// 0, Sunday, to 6.
pub weekday: u32,
}
impl Minute {
/// The minute `secs` seconds after the epoch falls in.
pub fn at(secs: u64) -> Minute {
let days = secs / 86_400;
let into = secs % 86_400;
// Howard Hinnant's days-to-civil, for the date.
let z = days as i64 + 719_468;
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
Minute {
minute: (into / 60 % 60) as u32,
hour: (into / 3600) as u32,
day: (doy - (153 * mp + 2) / 5 + 1) as u32,
month: (if mp < 10 { mp + 3 } else { mp - 9 }) as u32,
// 1970-01-01 was a Thursday.
weekday: ((days + 4) % 7) as u32,
}
}
}
impl Cron {
pub fn parse(expr: &str) -> Result<Cron> {
let expr = match expr.trim() {
"@hourly" => "0 * * * *",
"@daily" => "0 0 * * *",
"@weekly" => "0 0 * * 0",
"@monthly" => "0 0 1 * *",
"@yearly" => "0 0 1 1 *",
other => other,
};
let fields: Vec<&str> = expr.split_whitespace().collect();
let [minute, hour, day, month, weekday] = fields[..] else {
bail!("a schedule is five fields, minute hour day month weekday, not {expr:?}");
};
let mut weekdays = field(weekday, 0, 7, "weekday")?;
// 7 is Sunday too.
weekdays = (weekdays | weekdays >> 7) & 0x7f;
let cron = Cron {
minutes: field(minute, 0, 59, "minute")?,
hours: field(hour, 0, 23, "hour")?,
days: field(day, 1, 31, "day")?,
months: field(month, 1, 12, "month")?,
weekdays,
both_days: day != "*" && weekday != "*",
};
if let Some(gap) = cron.closest() {
bail!(
"{expr:?} runs {gap} minute{} apart: a schedule runs at most every {EVERY_MIN} minutes",
if gap == 1 { "" } else { "s" }
);
}
Ok(cron)
}
/// Whether it runs at `at`.
pub fn runs_at(&self, at: Minute) -> bool {
let has = |set: u64, n: u32| set >> n & 1 == 1;
let day = has(self.days, at.day);
let weekday = has(self.weekdays, at.weekday);
let days = if self.both_days {
day || weekday
} else {
day && weekday
};
has(self.minutes, at.minute)
&& has(self.hours, at.hour)
&& has(self.months, at.month)
&& days
}
/// The fewest minutes between two of its minutes of the hour, the
/// last and the next hour's first among them, if fewer than
/// [`EVERY_MIN`].
fn closest(&self) -> Option<u32> {
let minutes: Vec<u32> = (0..60).filter(|m| self.minutes >> m & 1 == 1).collect();
let wrapped = minutes.first().map(|m| m + 60);
let next = minutes.iter().skip(1).chain(wrapped.as_ref());
let gaps = minutes.iter().zip(next).map(|(a, b)| b - a);
gaps.filter(|&gap| gap < EVERY_MIN).min()
}
}
/// The values a field names, from `low` to `high`, as a bit set.
fn field(text: &str, low: u32, high: u32, what: &str) -> Result<u64> {
let mut set = 0u64;
for part in text.split(',') {
let (range, step) = match part.split_once('/') {
Some((range, step)) => (range, number(step, what)?),
None => (part, 1),
};
let (from, to) = match range.split_once('-') {
_ if range == "*" => (low, high),
Some((a, b)) => (number(a, what)?, number(b, what)?),
None if step > 1 => (number(range, what)?, high),
None => (number(range, what)?, number(range, what)?),
};
if from < low || to > high || from > to || step == 0 {
bail!("a schedule's {what} is {low} to {high}, not {part:?}");
}
for n in (from..=to).step_by(step as usize) {
set |= 1 << n;
}
}
Ok(set)
}
fn number(text: &str, what: &str) -> Result<u32> {
match text.parse() {
Ok(n) => Ok(n),
Err(_) => bail!("a schedule's {what} is a number, not {text:?}"),
}
}
#[cfg(test)]
mod tests {
use super::*;
/// 2026-11-02, a Monday, at `hour:minute` UTC.
fn monday(hour: u64, minute: u64) -> Minute {
Minute::at(1_793_577_600 + hour * 3600 + minute * 60)
}
#[test]
fn minutes_are_told_in_utc() {
let at = monday(13, 45);
assert_eq!(
at,
Minute {
minute: 45,
hour: 13,
day: 2,
month: 11,
weekday: 1
}
);
assert_eq!(
Minute::at(0),
Minute {
minute: 0,
hour: 0,
day: 1,
month: 1,
weekday: 4
}
);
// A leap day, and the year's last minute.
assert_eq!(
(Minute::at(951_782_400).day, Minute::at(951_782_400).month),
(29, 2)
);
let last = Minute::at(1_798_761_599);
assert_eq!(
(last.month, last.day, last.hour, last.minute),
(12, 31, 23, 59)
);
}
#[test]
fn schedules_run_when_they_say() {
let nightly = Cron::parse("30 3 * * *").unwrap();
assert!(nightly.runs_at(monday(3, 30)));
assert!(!nightly.runs_at(monday(3, 31)));
let quarter = Cron::parse("*/15 9-17 * * 1-5").unwrap();
assert!(quarter.runs_at(monday(9, 45)));
assert!(!quarter.runs_at(monday(18, 0)));
assert!(!quarter.runs_at(Minute {
weekday: 0,
..monday(9, 0)
}));
// Sunday as 7; a step from a number runs to the field's end.
let sunday = Cron::parse("0 0 * * 7").unwrap();
assert!(sunday.runs_at(Minute {
weekday: 0,
..monday(0, 0)
}));
assert_eq!(
Cron::parse("10/20 * * * *").unwrap(),
Cron::parse("10,30,50 * * * *").unwrap()
);
// Both days named: either runs it.
let either = Cron::parse("0 0 1 * 1").unwrap();
assert!(either.runs_at(monday(0, 0)));
assert!(either.runs_at(Minute {
day: 1,
weekday: 3,
..monday(0, 0)
}));
assert!(!either.runs_at(Minute {
day: 5,
weekday: 3,
..monday(0, 0)
}));
// One named: that one alone.
let first = Cron::parse("0 0 1 * *").unwrap();
assert!(!first.runs_at(monday(0, 0)));
assert_eq!(
Cron::parse("@daily").unwrap(),
Cron::parse("0 0 * * *").unwrap()
);
for named in ["@hourly", "@weekly", "@monthly", "@yearly"] {
assert!(Cron::parse(named).is_ok(), "{named}");
}
}
#[test]
fn schedules_say_what_is_wrong() {
let said = |expr: &str| Cron::parse(expr).unwrap_err().to_string();
assert!(said("* * * *").contains("five fields"));
assert!(said("60 * * * *").contains("minute is 0 to 59"));
assert!(said("0 24 * * *").contains("hour is 0 to 23"));
assert!(said("0 0 0 * *").contains("day is 1 to 31"));
assert!(said("0 0 * 13 *").contains("month is 1 to 12"));
assert!(said("0 0 * * 8").contains("weekday is 0 to 7"));
assert!(said("0 0 * * mon").contains("weekday is a number"));
assert!(said("0 0 5-1 * *").contains("not \"5-1\""));
assert!(said("*/0 * * * *").contains("not \"*/0\""));
assert!(said("* * * * *").contains("1 minute apart"));
assert!(said("0,2 * * * *").contains("2 minutes apart"));
// The last of the hour and the next hour's first count too.
assert!(said("1,58 * * * *").contains("3 minutes apart"));
assert!(Cron::parse("*/5 * * * *").is_ok());
}
}
crates/knotless/src/landing.rs+22−11
View file
@@ -378,30 +378,40 @@ pub struct Listed {
/// A pull into the ref, a branch, runs it.
pub pull: bool,
/// Its owner may ask for it to run at the ref by name, with variables:
/// it runs on a push there, or its `asked` trigger names the ref.
/// it runs on a push there, its `asked` trigger names the ref, or the
/// ref is the default branch and it has a schedule.
pub asked: bool,
/// The schedules it runs on at the ref, the default branch: none at
/// any other.
#[serde(skip_serializing_if = "Vec::is_empty")]
pub schedule: Vec<workflow::Schedule>,
/// Why it does not load, when it does not: nothing runs it then.
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
/// The workflows at `sha`, in file name order, and what runs each at the
/// ref `refname`.
pub async fn listed(git: &Git, sha: &str, refname: &str) -> Result<Vec<Listed>> {
/// ref `refname`, the default branch if `main`.
pub async fn listed(git: &Git, sha: &str, refname: &str, main: bool) -> Result<Vec<Listed>> {
let loaded = workflow::load(git, sha).await?;
let listed = loaded.into_iter().map(|l| match l.workflow {
Ok(w) => Listed {
push: w.runs_on_push(refname),
pull: w.runs_on_pull(refname),
asked: w.runs_when_asked(refname),
error: None,
name: l.name,
},
Ok(w) => {
let schedule = if main { w.on.schedule.clone() } else { vec![] };
Listed {
push: w.runs_on_push(refname),
pull: w.runs_on_pull(refname),
asked: w.runs_when_asked(refname) || !schedule.is_empty(),
schedule,
error: None,
name: l.name,
}
}
Err(e) => Listed {
name: l.name,
push: false,
pull: false,
asked: false,
schedule: vec![],
error: Some(format!("{e:#}")),
},
});
@@ -429,6 +439,7 @@ pub async fn workflows(
let key = t.snap.key().expect("an opened repository has a key");
let mirror = Mirror::of(ctx, &t.remote, key).await?;
mirror.sync(&t.remote, &t.snap).await?;
let listed = listed(&mirror.git, sha, refname).await?;
let main = *refname == format!("refs/heads/{default}");
let listed = listed(&mirror.git, sha, refname, main).await?;
Ok((refname.clone(), listed))
}
crates/knotless/src/lib.rs+1−0
View file
@@ -21,6 +21,7 @@ pub mod closing;
pub mod codeowners;
pub mod commit;
pub mod compact;
pub mod cron;
pub mod crypto;
pub mod drisl;
pub mod emails;
crates/knotless/src/workflow.rs+87−1
View file
@@ -24,6 +24,14 @@
//! commits what it changed among them onto that branch and pushes it, as
//! screenshots taken again in the image their baselines come from.
//!
//! A `schedule` trigger runs a workflow on its own, at the default
//! branch's tip, at the times its cron expressions name ([`crate::cron`]).
//! Its runner learns them from each push to the default branch, as it
//! runs that push, so a schedule added or changed holds from then on. A
//! scheduled run is one asked for by name, with no variables of its own
//! but `KNOTLESS_SCHEDULE`, the cron expression that ran it: one of an
//! older time still going is cancelled for it, unless its runs finish.
//!
//! A step may name `on`, the refs it runs at, matched as a push trigger's
//! are, against the ref a push moved or a run was asked for at. A pull's
//! run and a merge queue batch's are at no ref of the repository's, as
@@ -34,8 +42,8 @@
use crate::git::Git;
use crate::read;
use anyhow::{Result, bail};
use serde::Deserialize;
use serde::de::{self, Deserializer, MapAccess, SeqAccess, Visitor};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fmt;
@@ -117,6 +125,18 @@ pub struct On {
/// The refs its owner may ask for it to run at, by name.
#[serde(default)]
pub asked: Option<Asked>,
/// When it runs at the default branch's tip, as its runner's clock
/// says ([`crate::cron`]).
#[serde(default)]
pub schedule: Vec<Schedule>,
}
/// One time a workflow runs on its own.
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct Schedule {
/// A cron expression, in UTC.
pub cron: String,
}
impl On {
@@ -125,6 +145,7 @@ impl On {
push: Some(Push::default()),
pull: None,
asked: None,
schedule: Vec::new(),
}
}
}
@@ -353,6 +374,16 @@ impl Workflow {
self.runs_on_push(name) || asked.is_some_and(|a| a.refs.iter().any(|p| matches(p, name)))
}
/// The schedules it runs on, at the default branch.
pub fn crons(&self) -> Vec<crate::cron::Cron> {
let parsed = self
.on
.schedule
.iter()
.map(|s| crate::cron::Cron::parse(&s.cron));
parsed.filter_map(Result::ok).collect()
}
/// The globs of the files a run asked for by name writes back.
pub fn commits(&self) -> &[String] {
self.on.asked.as_ref().map_or(&[], |a| &a.commit)
@@ -437,6 +468,11 @@ pub fn parse(file: &str, text: &[u8]) -> Result<Workflow> {
workflow.name
);
}
for schedule in &workflow.on.schedule {
if let Err(err) = crate::cron::Cron::parse(&schedule.cron) {
bail!("{file}: {err:#}");
}
}
for path in &workflow.cache {
if mount_path(path).is_none() {
bail!(
@@ -499,6 +535,21 @@ pub struct Loaded {
pub workflow: Result<Workflow>,
}
/// The cron expressions of each of `loaded` that has a schedule, by its
/// name: what a runner asks for on its own at the default branch. One
/// that does not load has none.
pub fn scheduled(loaded: &[Loaded]) -> std::collections::BTreeMap<String, Vec<String>> {
let mut all = std::collections::BTreeMap::new();
for l in loaded {
let Ok(wf) = &l.workflow else { continue };
let crons: Vec<String> = wf.on.schedule.iter().map(|s| s.cron.clone()).collect();
if !crons.is_empty() {
all.insert(l.name.clone(), crons);
}
}
all
}
/// The workflows at `sha`, in file name order. A commit without the
/// directory has none.
pub async fn load(git: &Git, sha: &str) -> Result<Vec<Loaded>> {
@@ -565,6 +616,41 @@ mod tests {
assert_eq!(w.superseded, Superseded::Cancel);
}
#[test]
fn schedules() {
let text = "on:\n schedule:\n - cron: '30 3 * * *'\n - cron: '@weekly'\nsteps:\n - run: x\n";
let w = parse("nightly.yml", text.as_bytes()).unwrap();
assert_eq!(w.crons().len(), 2);
let loaded = |name: &str, workflow| Loaded {
name: name.into(),
workflow,
};
let all = scheduled(&[
loaded("nightly", Ok(w.clone())),
loaded("ci", parse("ci.yml", b"steps:\n - run: x\n")),
loaded("bad", Err(anyhow::anyhow!("no"))),
]);
let crons = vec!["30 3 * * *".to_string(), "@weekly".to_string()];
assert_eq!(all, [("nightly".to_string(), crons)].into());
// Only on its schedule: no push runs it.
assert!(!w.runs_on_push("refs/heads/main"));
assert!(
parse("ci.yml", b"steps:\n - run: x\n")
.unwrap()
.crons()
.is_empty()
);
let wrong = parse(
"n.yml",
b"on: {schedule: [{cron: '* * * * *'}]}\nsteps:\n - run: x\n",
);
let wrong = wrong.unwrap_err().to_string();
assert!(
wrong.starts_with("n.yml: ") && wrong.contains("every 5 minutes"),
"{wrong}"
);
}
#[test]
fn a_step_runs_or_uses() {
let text = "secrets: [KEY]\nsteps:\n - uses: a.test/steps/wait@1\n with: {workflows: test}\n secrets: [KEY]\n - {name: own, uses: a.test/steps/wait@1}\n";
crates/knotless-mcp/src/lib.rs+1−1
View file
@@ -2907,7 +2907,7 @@ impl Server {
}
#[tool(
description = "The workflows of a repository at a ref (the default branch if none), in file name order, each with what runs it there: push, a push to it; pull, a pull into it; asked, whether its owner may ask for it by name with rerun, ref, workflow and env, as one that runs on a push there or whose `on: {asked: {refs: [...]}}` names it; and error, why it does not load. Gives the ref in full too."
description = "The workflows of a repository at a ref (the default branch if none), in file name order, each with what runs it there: push, a push to it; pull, a pull into it; asked, whether its owner may ask for it by name with rerun, ref, workflow and env, as one that runs on a push there, whose `on: {asked: {refs: [...]}}` names it, or that has a schedule, at the default branch; schedule, the `[{cron}]` it runs on there, at the default branch only; and error, why it does not load. Gives the ref in full too."
)]
async fn workflows(
&self,
crates/knotless-run/src/lib.rs+147−3
View file
@@ -69,6 +69,15 @@
//! its default branch that passes makes its caches' bases instead, as a
//! push's would, so its pulls start warm (see [`caches`]).
//!
//! A workflow with a `schedule` runs at the default branch's tip at the
//! minutes its cron expressions name, in UTC, as a run asked for by name
//! is, its steps seeing `KNOTLESS_SCHEDULE`, the expression that ran it.
//! Each pass over a repository reads its schedules at the default
//! branch's tip, when it moved, and keeps them with its state; each minute
//! the runner asks for those due, but not one whose last run is still
//! going, and rings their repositories. A minute the runner was down for
//! is not made up.
//!
//! A workflow whose trigger names `paths` none of which a push or pull
//! changes is not run: its check is filed at once as passed, its steps
//! skipped, noting why, so it holds no cache volume and waits for nothing.
@@ -528,8 +537,30 @@ struct State {
/// Where the invites read so far end, to read on from.
#[serde(default, skip_serializing_if = "Option::is_none")]
invited: Option<String>,
/// Each repository's scheduled workflows, as its default branch had
/// them when it was last read.
#[serde(default)]
schedules: BTreeMap<String, Scheduled>,
/// The minute, since the epoch, each scheduled workflow was last asked
/// for, by repository and workflow: once a minute at most.
#[serde(default)]
fired: BTreeMap<String, BTreeMap<String, u64>>,
}
/// A repository's scheduled workflows: the default branch they run at,
/// the commit there they were read at, and each one's cron expressions,
/// by its name.
#[derive(Serialize, Deserialize, Default, Clone, Debug)]
struct Scheduled {
ref_name: String,
#[serde(default)]
commit: String,
crons: BTreeMap<String, Vec<String>>,
}
/// What a scheduled run's steps see as the cron expression that ran it.
pub const SCHEDULE_VAR: &str = "KNOTLESS_SCHEDULE";
impl State {
/// The usage of the public repositories, or the private ones.
fn usage(&mut self, private: bool) -> &mut BTreeMap<String, Used> {
@@ -1246,6 +1277,10 @@ impl Runner {
watch.at("running refs");
self.refs(watch, &repo, &snap, moved).await?;
}
if !watch.pulls_only {
watch.at("reading schedules");
self.learn(watch, &snap).await?;
}
if self.opts.pulls {
watch.at("running pulls");
self.pulls(watch, &repo, &snap).await?;
@@ -1748,6 +1783,89 @@ impl Runner {
dropped.is_some_and(|d| d.contains_key(at))
}
/// Reads the schedules of the workflows at the tip of the default
/// branch of `watch`, when it is not where they were last read: when
/// the repository is first watched, after a push there, and at a start
/// once its tip moved meanwhile, whichever runner ran the push.
async fn learn(&self, watch: &Watch, snap: &Snapshot) -> Result<()> {
let main = main_ref(snap);
let tip = snap.refs.iter().find(|r| r.value.name == main);
let tip = tip.and_then(|r| r.value.target.clone());
let known = self.known_schedule(watch);
// An empty repository has nothing to read.
let Some(tip) = tip.filter(|tip| known.as_ref() != Some(tip)) else {
return Ok(());
};
let mirror = Mirror::open(&self.ctx, watch.remote.did(), &watch.key).await?;
mirror.sync(&watch.remote, snap).await?;
let loaded = workflow::load(&mirror.git, &tip).await?;
let scheduled = Scheduled {
ref_name: main,
commit: tip,
crons: workflow::scheduled(&loaded),
};
let mut state = self.state.lock().unwrap();
// What a schedule no longer there last fired at goes with it.
let fired = state.fired.entry(watch.id()).or_default();
fired.retain(|name, _| scheduled.crons.contains_key(name));
state.schedules.insert(watch.id(), scheduled);
store::write(&self.state_path, &*state, false)
}
/// The commit `watch`'s schedules were last read at.
fn known_schedule(&self, watch: &Watch) -> Option<String> {
let state = self.state.lock().unwrap();
state.schedules.get(&watch.id()).map(|s| s.commit.clone())
}
/// Asks for each scheduled workflow due at `at`, in minutes since the
/// epoch, at its repository's default branch, once, and rings each
/// repository that has one.
pub fn on_schedule(&self, at: u64) -> Result<()> {
let minute = knotless::cron::Minute::at(at * 60);
let due = |cron: &&String| {
let cron = knotless::cron::Cron::parse(cron);
cron.is_ok_and(|c| c.runs_at(minute))
};
// A run at the default branch still going finishes, rather than
// being cancelled for this one: this time is passed over.
let going: HashSet<String> = self.going.lock().unwrap().keys().cloned().collect();
let mut rung = Vec::new();
let mut state = self.state.lock().unwrap();
let State {
schedules,
fired,
wanted,
..
} = &mut *state;
for (id, scheduled) in schedules.iter() {
let last = fired.entry(id.clone()).or_default();
for (name, crons) in &scheduled.crons {
let cron = crons.iter().find(due);
let run = format!("{id}\n{}\n{name}", scheduled.ref_name);
let fresh = |_: &&String| last.get(name) != Some(&at) && !going.contains(&run);
let Some(cron) = cron.filter(fresh) else {
continue;
};
last.insert(name.clone(), at);
let refs = wanted.entry(id.clone()).or_default();
let asks = refs.entry(scheduled.ref_name.clone()).or_default();
asks.insert(name.clone(), vec![(SCHEDULE_VAR.into(), cron.clone())]);
rung.push(id.clone());
}
}
if !rung.is_empty() {
store::write(&self.state_path, &*state, false)?;
}
drop(state);
let doors = self.doors.lock().unwrap();
for (did, key) in rung.iter().filter_map(|id| id.split_once('/')) {
let door = doors.get(&(did.to_string(), key.to_string()));
door.inspect(|d| d.notify_one());
}
Ok(())
}
/// The workflows asked for by name at the ref `name` of `watch`.
fn wanted(&self, watch: &Watch, name: &str) -> Asks {
let state = self.state.lock().unwrap();
@@ -1906,7 +2024,11 @@ impl Runner {
None => loaded.workflow,
};
let wf = match workflow {
Ok(wf) if asked && !wf.runs_when_asked(&subject.ref_name) => {
Ok(wf)
if asked
&& !wf.runs_when_asked(&subject.ref_name)
&& !(subject.main && !wf.on.schedule.is_empty()) =>
{
check.status = status::ERROR.into();
check.error = Some(format!(
"asked for, but it runs neither on a push to {} nor when asked there",
@@ -2502,6 +2624,7 @@ impl Runner {
}
let mut state = self.state.lock().unwrap();
state.discovered.remove(&watch.label);
state.schedules.remove(&watch.id());
store::write(&self.state_path, &*state, false)?;
telemetry::left(&watch.label);
Ok(())
@@ -2777,8 +2900,14 @@ impl Runner {
let mut log = std::fs::File::create(&log_path)?;
log.write_all(said.as_bytes())?;
if subject.asked.contains_key(&wf.name) {
let scheduled = given(subject, wf).any(|(k, _)| k == SCHEDULE_VAR);
let why = if scheduled {
"on its schedule"
} else {
"asked for by name"
};
let given = given(subject, wf).map(|(k, v)| format!(" {k}={v}"));
writeln!(log, "==> asked for by name{}", given.collect::<String>())?;
writeln!(log, "==> {why}{}", given.collect::<String>())?;
}
log.write_all(not_granted(&ungranted, watch, subject).as_bytes())?;
let pull = subject.pull.as_ref().map(|p| &p.strangers);
@@ -2829,7 +2958,10 @@ impl Runner {
let timeout = self.opts.step_timeout;
let _tracked = Tracked::new(self, setup.owner.run);
let session = self.opts.executor.start(&setup, &log, timeout).await?;
let asked = subject.asked.contains_key(&wf.name);
// A run on its schedule is held to the runner's limits, as one
// that recurs on its own: only its owner's own ask goes past them.
let scheduled = given(subject, wf).any(|(k, _)| k == SCHEDULE_VAR);
let asked = subject.asked.contains_key(&wf.name) && !scheduled;
let steps = Steps {
log,
progress: &progress,
@@ -3445,6 +3577,7 @@ pub async fn watch(runner: Arc<Runner>, watches: Vec<Watch>) -> Result<()> {
}
tokio::spawn(sweep(runner.clone()));
tokio::spawn(tidy(runner.clone()));
tokio::spawn(schedules(runner.clone()));
if let Some(appview) = runner.appview.clone() {
tokio::spawn(wakes(runner.clone(), appview));
}
@@ -3551,6 +3684,17 @@ async fn sweep(runner: Arc<Runner>) {
}
}
/// Asks for the scheduled workflows due as each minute begins.
async fn schedules(runner: Arc<Runner>) {
loop {
let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
let now = since.unwrap_or_default();
let next = now.as_secs() / 60 + 1;
tokio::time::sleep(Duration::from_secs(next * 60) - now).await;
noted("schedules", runner.on_schedule(next));
}
}
/// Watches each repository that names this runner, as the appview lists
/// them and invites to private ones say, at the start and every
/// `--sweep`: Jetstream brings public ones named meanwhile
crates/knotless-run/tests/run.rs+146−0
View file
@@ -199,6 +199,152 @@ fn blob(did: &str, cid: &str) -> String {
.unwrap()
}
/// A workflow on a schedule runs at the default branch's tip at the
/// minute it names, its steps told which schedule ran it, once the runner
/// has learned it from a push there; one whose minute has not come waits.
#[test]
fn a_schedule_runs_its_workflow() {
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let work = owner.work("proj", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let key = owner.record("dev.knotless.repo", "proj")["key"]
.as_str()
.unwrap()
.to_string();
let ref_key = ref_rkey(&key, "refs/heads/main");
let rt = tokio::runtime::Runtime::new().unwrap();
let js = jetstream(&rt);
let opts = runner_opts(&runner, &owner, &js);
rt.spawn(knotless_run::run(opts, common::ctx(&runner)));
wait_until(|| runner.path("state.json").exists());
// Due at the first minute at least half of one away, for the push to
// be run first.
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
let at = (now.unwrap().as_secs() + 30) / 60 + 1;
let due = knotless::cron::Minute::at(at * 60);
let cron = format!("{} {} * * *", due.minute, due.hour);
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let nightly = format!(
"on: {{schedule: [{{cron: '{cron}'}}]}}\nsteps: [{{run: 'echo \"ran on $KNOTLESS_SCHEDULE\"'}}]\n"
);
for (file, text) in [
("nightly.yml", nightly.as_str()),
(
"yearly.yml",
"on: {schedule: [{cron: '@yearly'}]}\nsteps: [{run: 'true'}]\n",
),
(
"ci.yml",
"on: {push: {refs: [refs/heads/main]}}\nsteps: [{run: 'true'}]\n",
),
] {
std::fs::write(dir.join(file), text).unwrap();
}
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "schedules"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let sha = owner.git(&work, &["rev-parse", "HEAD"]);
js.ring(&owner.account.did, &ref_key);
wait_until(|| check_of(&runner, &owner, &key, "ci")["finishedAt"].is_string());
// A push runs neither.
assert!(check_of(&runner, &owner, &key, "nightly").is_null());
let until = std::time::Instant::now() + Duration::from_secs(180);
while !check_of(&runner, &owner, &key, "nightly")["finishedAt"].is_string() {
assert!(std::time::Instant::now() < until, "{cron} never ran");
std::thread::sleep(Duration::from_millis(500));
}
let ran = check_of(&runner, &owner, &key, "nightly");
assert_eq!(ran["status"], "success", "{ran}");
assert_eq!(ran["commit"], sha.as_str());
let out = blob(
&runner.account.did,
ran["log"]["ref"]["$link"].as_str().unwrap(),
);
assert!(
out.contains(&format!("==> on its schedule KNOTLESS_SCHEDULE={cron}")),
"{out}"
);
assert!(out.contains(&format!("ran on {cron}")), "{out}");
assert!(check_of(&runner, &owner, &key, "yearly").is_null());
// And lists them, at the default branch.
let (code, out) = owner.cli(&["workflows", &owner.url("proj")], b"");
assert_eq!(code, 0, "{out}");
assert!(
out.contains(&format!("nightly\tasked\n\tschedule {cron}\n")),
"{out}"
);
assert!(out.contains("yearly\tasked\n\tschedule @yearly"), "{out}");
}
#[test]
fn a_schedule_is_read_at_start_and_waits_for_its_last_run() {
let rt = tokio::runtime::Runtime::new().unwrap();
let owner = Sandbox::new();
let mut runner = Sandbox::new();
let handle = runner.account.handle.clone();
runner.set("KNOTLESS_IDENTIFIER", &handle);
let go = runner.path("go");
std::fs::create_dir_all(&go).unwrap();
// Pushed before the runner starts: no push teaches it the schedule.
let work = owner.work("proj", 1);
owner.git(&work, &["remote", "add", "origin", &owner.url("proj")]);
let dir = work.join(".knotless/workflows");
std::fs::create_dir_all(&dir).unwrap();
let slow = format!(
"on: {{schedule: [{{cron: '@daily'}}]}}\nsteps:\n - run: |\n touch {go}/started\n while [ ! -e {go}/done ]; do sleep 0.1; done\n",
go = go.display()
);
std::fs::write(dir.join("slow.yml"), slow).unwrap();
owner.git(&work, &["add", "."]);
owner.git(&work, &["commit", "-q", "-m", "schedule"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let sha = owner.git(&work, &["rev-parse", "HEAD"]);
let js = jetstream(&rt);
let opts = runner_opts(&runner, &owner, &js);
let ctx = common::ctx(&runner);
let reporter = rt.block_on(check::Reporter::new(&ctx)).unwrap();
let started = knotless_run::Runner::start_as(opts, ctx, reporter);
let (run, watches) = rt.block_on(started).unwrap();
let key = watches[0].key.clone();
rt.spawn(knotless_run::watch(run.clone(), watches));
let state = || {
let text = std::fs::read_to_string(runner.path("state.json")).unwrap_or_default();
serde_json::from_str::<serde_json::Value>(&text).unwrap_or_default()
};
let id = format!("{}/{key}", owner.account.did);
wait_until(|| state()["schedules"][&id]["commit"] == sha.as_str());
// Midnight, two days running: the first run is still going at the
// second, which is passed over rather than cutting it off.
let day = 1_793_577_600 / 60;
run.on_schedule(day).unwrap();
wait_until(|| go.join("started").exists());
run.on_schedule(day + 24 * 60).unwrap();
assert_eq!(state()["fired"][&id]["slow"], day);
std::fs::write(go.join("done"), "").unwrap();
wait_until(|| check_of(&runner, &owner, &key, "slow")["finishedAt"].is_string());
let ran = check_of(&runner, &owner, &key, "slow");
assert_eq!(ran["status"], "success", "{ran}");
assert_eq!(ran["commit"], sha.as_str());
// Its schedule taken out, what it last fired at goes too.
owner.git(&work, &["rm", "-q", ".knotless/workflows/slow.yml"]);
owner.git(&work, &["commit", "-q", "-m", "no schedule"]);
owner.git(&work, &["push", "-q", "origin", "main"]);
let gone = owner.git(&work, &["rev-parse", "HEAD"]);
js.ring(&owner.account.did, &ref_rkey(&key, "refs/heads/main"));
wait_until(|| state()["schedules"][&id]["commit"] == gone.as_str());
assert!(state()["fired"][&id]["slow"].is_null(), "{}", state());
}
#[test]
fn a_push_runs_its_workflows() {
let owner = Sandbox::new();
docs/design/runner.md+23−0
View file
@@ -252,6 +252,27 @@ A workflow left out is filed at once as `success`, every step `skipped`,
with a note naming the paths. It takes no job and no cache, and the merge
gate and deploys read it as passed. Runs asked for by name ignore `paths`.
### Schedules
A workflow's `schedule` names minutes, in UTC, at which it runs at the
default branch's tip ([syntax](../reference/workflows.md#schedules)).
Nothing rings for a schedule, so the runner keeps them itself: each pass
over a repository reads the workflows at its default branch's tip, when
that tip is not the one last read, and notes each workflow's cron
expressions in its state. As each minute begins, it asks for every
workflow due then, as a run asked for by name is asked for, with
`KNOTLESS_SCHEDULE` the expression that matched, and rings each
repository with one. A workflow whose last run there is still going is
not asked for: that time is passed over, so a run longer than its
interval is never cut off by the next. Unlike a run its owner asks for,
a scheduled run is held to `--step-timeout` and `--run-timeout`, so one
that hangs frees its job and lets the schedule run again. The pass that follows runs it as
it runs any run asked for by name, so it waits behind every other run,
and its check replaces the last one of the same commit's. A schedule runs at
most every five minutes, so it costs a runner no more than pushes do. A
minute the runner was down for is not made up, and each workflow is asked
for at most once a minute, even across a restart.
## What a pull runs
With `--pulls`, the runner checks each open pull into a watched
@@ -520,6 +541,8 @@ then watched as if for the first time. Each field is keyed by repository:
| `passed` | trusted pull and batch runs that passed, for reuse, at most 256 |
| `dropped` | the workflows dropped at a stop, per ref, pull or queue, and commit and base |
| `wanted` | the workflows asked for by name and not yet run, per ref, with their variables |
| `schedules` | the default branch, the commit there last read, and each workflow's cron expressions at it |
| `fired` | the minute each scheduled workflow was last asked for |
| `used` | the milliseconds each owner's public repositories' runs' steps took, per day, for the last 30 days |
| `used_private` | the same for private repositories |
docs/reference/cli.md+5−3
View file
docs/reference/mcp.md+1−1
View file
docs/reference/runner-internals.md+2−0
View file
docs/reference/workflows.md+39−2
View file
lexicons/dev/knotless/listWorkflows.json+14−1
View file