chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A workflow waits for the checks it runs after holding no job
.changeset/workflows-run-after.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: minor | ||
--- | ||
| ||
# Workflows run after others | ||
| ||
A workflow can name others in `after`: the runner starts it only once their checks at the same commit have passed, and it holds no job while it waits. If one fails or is cancelled, it does not run. | ||
crates/knotless/src/workflow.rs+27−0
| @@ -83,6 +83,11 @@ pub struct Workflow { | ||
/// each pull before it may go into one. | ||
#[serde(default)] | ||
pub queue: InQueue, | ||
/// The workflows whose checks at the same commit, on the same ref, | ||
/// pull or batch, must pass before it starts: until then it waits | ||
/// without taking a job, and if one of them fails it does not run. | ||
#[serde(default)] | ||
pub after: Vec<String>, | ||
pub steps: Vec<Step>, | ||
} | ||
| ||
| @@ -468,6 +473,13 @@ pub fn parse(file: &str, text: &[u8]) -> Result<Workflow> { | ||
workflow.name | ||
); | ||
} | ||
for name in &workflow.after { | ||
if !valid_name(name) || *name == workflow.name { | ||
bail!( | ||
"{file}: `after` names other workflows, 1 to 64 letters, digits, `-` and `_` each, not {name:?}" | ||
); | ||
} | ||
} | ||
for schedule in &workflow.on.schedule { | ||
if let Err(err) = crate::cron::Cron::parse(&schedule.cron) { | ||
bail!("{file}: {err:#}"); | ||
| @@ -751,6 +763,21 @@ mod tests { | ||
assert!(err.to_string().contains("unknown variant `wait`"), "{err}"); | ||
} | ||
| ||
#[test] | ||
fn after() { | ||
let w = parse("d.yml", b"steps: []\n").unwrap(); | ||
assert!(w.after.is_empty()); | ||
let w = parse("d.yml", b"after: [check, test]\nsteps: []\n").unwrap(); | ||
assert_eq!(w.after, ["check", "test"]); | ||
for bad in ["after: [d]\n", "after: [\"a b\"]\n"] { | ||
let err = parse("d.yml", format!("{bad}steps: []\n").as_bytes()).unwrap_err(); | ||
assert!( | ||
err.to_string().contains("`after` names other workflows"), | ||
"{err}" | ||
); | ||
} | ||
} | ||
| ||
#[test] | ||
fn first_in_queue() { | ||
let main = "refs/heads/main"; | ||
crates/knotless-run/src/lib.rs+162−17
| @@ -1037,6 +1037,9 @@ pub struct Runner { | ||
woken: AtomicBool, | ||
/// The door of each private repository watched, by its space. | ||
private: Mutex<HashMap<String, Arc<Notify>>>, | ||
/// Counts the runs that have ended and filed how, for those waiting | ||
/// on what they name `after` to look again. | ||
ended: tokio::sync::watch::Sender<u64>, | ||
} | ||
| ||
/// A run going of a workflow whose runs finish, noted in | ||
| @@ -1255,6 +1258,7 @@ impl Runner { | ||
landing: tokio::sync::Mutex::new(()), | ||
finishing: tokio::sync::watch::Sender::new(Vec::new()), | ||
sessions: Mutex::new(HashSet::new()), | ||
ended: tokio::sync::watch::Sender::new(0), | ||
}; | ||
Ok((Arc::new(runner), watches)) | ||
} | ||
| @@ -2180,6 +2184,7 @@ impl Runner { | ||
}; | ||
tokio::spawn(async move { | ||
let ran = runner.finish(&watch, &git, &subject, &wf, job).await; | ||
runner.ended.send_modify(|n| *n += 1); | ||
// Stopping, a batch's run dropped is not one that failed, | ||
// and one that failed is checked again on the next start. | ||
// One that passed counts: the last of a batch to, the batch | ||
| @@ -2229,29 +2234,31 @@ impl Runner { | ||
mut cancelled, | ||
} = job; | ||
let rkey = rkey.as_str(); | ||
let turns = self.take_turns(rank, watch, on_ref, &mut cancelled); | ||
// Stopping, one not started yet is dropped, to run on the next start. | ||
let of = format!("{}\n{}\n{}", watch.id(), subject.keyed(), subject.at()); | ||
let held = tokio::select! { | ||
let waits = self.waits(rank, watch, subject, wf, on_ref, &mut cancelled); | ||
let waited = tokio::select! { | ||
biased; | ||
_ = self.dropping(&of) => { | ||
telemetry::run("run dropped, to run on the next start", watch, subject, &wf.name, None, None); | ||
return self.drop_run(watch, subject, wf, rkey, check).await; | ||
} | ||
held = turns => held, | ||
waited = waits => waited, | ||
}; | ||
let Some(Held { on_ref, job, said }) = held else { | ||
let why = cancelled.why(); | ||
let said = format!("run cancelled before it started: {why}"); | ||
telemetry::run( | ||
&said, | ||
watch, | ||
subject, | ||
&wf.name, | ||
Some(status::CANCELLED), | ||
None, | ||
); | ||
return self.never_started(watch, rkey, check, why).await; | ||
let Held { on_ref, job, said } = match waited { | ||
Waited::Held(held) => held, | ||
Waited::Not { status, line } => { | ||
let said = format!("run not started: {}", line.trim_start_matches("==> ")); | ||
telemetry::run(&said, watch, subject, &wf.name, Some(status), None); | ||
return self.unstarted(watch, rkey, check, status, &line).await; | ||
} | ||
Waited::Cancelled => { | ||
let why = cancelled.why(); | ||
let said = format!("run cancelled before it started: {why}"); | ||
let ended = Some(status::CANCELLED); | ||
telemetry::run(&said, watch, subject, &wf.name, ended, None); | ||
return self.never_started(watch, rkey, check, why).await; | ||
} | ||
}; | ||
// Its turn on the ref is kept until its check is filed as ended, | ||
// so the next run of a deploy files after it. | ||
| @@ -2505,6 +2512,32 @@ impl Runner { | ||
Ok(there.is_some_and(|c| c.status == status::SUCCESS)) | ||
} | ||
| ||
/// Waits for what a run ranked `rank` holds while its steps go, as | ||
/// [`Runner::take_turns`] does, once what `wf` names `after` has | ||
/// passed, waited for first, holding nothing; or for how it ended | ||
/// otherwise. | ||
async fn waits( | ||
&self, | ||
rank: Rank, | ||
watch: &Watch, | ||
subject: &Subject, | ||
wf: &Workflow, | ||
on_ref: Option<OnRef>, | ||
cancelled: &mut Cancelled, | ||
) -> Waited { | ||
let after = until_cancelled(cancelled, self.after(watch, subject, wf)).await; | ||
let waited = match after { | ||
Some(After::Passed(said)) => said, | ||
Some(After::Not { status, line }) => return Waited::Not { status, line }, | ||
None => return Waited::Cancelled, | ||
}; | ||
let held = self.take_turns(rank, watch, on_ref, cancelled).await; | ||
held.map_or(Waited::Cancelled, |mut held| { | ||
held.said = waited + &held.said; | ||
Waited::Held(held) | ||
}) | ||
} | ||
| ||
/// Waits for what a run ranked `rank` holds while its steps go, or | ||
/// `None` if `cancelled` first. One of a workflow whose runs finish | ||
/// first has its turn on its ref or pull, `on_ref`, and is kept from | ||
| @@ -2609,17 +2642,66 @@ impl Runner { | ||
&self, | ||
watch: &Watch, | ||
rkey: &str, | ||
mut check: Check, | ||
check: Check, | ||
why: &str, | ||
) -> Result<String> { | ||
let line = format!("==> cancelled {why} before it started\n"); | ||
check.status = status::CANCELLED.into(); | ||
self.unstarted(watch, rkey, check, status::CANCELLED, &line) | ||
.await | ||
} | ||
| ||
/// Files `check`, of a run that never started, as ended `status`, its | ||
/// log `line`, and returns how it ended. | ||
async fn unstarted( | ||
&self, | ||
watch: &Watch, | ||
rkey: &str, | ||
mut check: Check, | ||
status: &str, | ||
line: &str, | ||
) -> Result<String> { | ||
check.status = status.into(); | ||
check.log = Some(self.reporter.upload_log(line.as_bytes()).await?); | ||
check.finished_at = Some(now()); | ||
self.reporter.put(&watch.remote, rkey, &check).await?; | ||
Ok(check.status) | ||
} | ||
| ||
/// Waits until the checks of what `wf` names `after`, at `subject`, by | ||
/// this runner, have ended, looking again as each run here ends, or | ||
/// each [`AFTER_EVERY`]: [`After::Passed`] once all have passed, or | ||
/// [`After::Not`] once one has not, cancelled or failed. One with no | ||
/// check there, as a workflow that does not run on it or is another | ||
/// runner's, is not waited for, and the log says so. | ||
async fn after(&self, watch: &Watch, subject: &Subject, wf: &Workflow) -> After { | ||
if wf.after.is_empty() { | ||
return After::Passed(String::new()); | ||
} | ||
let began = std::time::Instant::now(); | ||
let mut ended = self.ended.subscribe(); | ||
loop { | ||
ended.mark_unchanged(); | ||
let mut absent = Vec::new(); | ||
let mut waiting = false; | ||
for name in &wf.after { | ||
let rkey = run_rkey(watch, subject, name); | ||
// One that cannot be read now is read again, as going. | ||
let read = self.reporter.get(&watch.remote, &rkey).await; | ||
let going = Some(status::QUEUED.to_string()); | ||
match read.map_or(going, |c| c.map(|c| c.status)) { | ||
None => absent.push(name.as_str()), | ||
Some(s) if s == status::SUCCESS || s == status::SKIPPED => {} | ||
Some(s) if status::going(&s) => waiting = true, | ||
Some(s) => return not_after(name, &s), | ||
} | ||
} | ||
if !waiting { | ||
return After::Passed(after_lines(&wf.after, &absent, began.elapsed())); | ||
} | ||
let _ = tokio::time::timeout(AFTER_EVERY, ended.changed()).await; | ||
} | ||
} | ||
| ||
/// Whether this runner lands the merge queues of `watch`, whose | ||
/// record is `record`: asked to, and the owner or a collaborator, or | ||
/// the runner it names to land them ([`RepoRecord::lander`]), which | ||
| @@ -4025,6 +4107,69 @@ struct Started<'a> { | ||
} | ||
| ||
/// `f`'s output, or `None` if `cancelled` first. | ||
/// How a run's wait before its steps ended: holding its job; not to run, | ||
/// as [`After::Not`] says; or cancelled. | ||
crates/knotless-run/src/lib.rsline 4111This line is the doc of | ||
enum Waited { | ||
Held(Held), | ||
Not { status: &'static str, line: String }, | ||
Cancelled, | ||
} | ||
| ||
/// How what a workflow names `after` ended, for it: all passed, with what | ||
/// its log says of them; or not, and so it is filed as `status`, its log | ||
/// `line`, without running. | ||
enum After { | ||
Passed(String), | ||
Not { status: &'static str, line: String }, | ||
} | ||
| ||
/// How often a run waiting on what it names `after` reads their checks | ||
/// again, besides as each run here ends: for one filed some other way. | ||
const AFTER_EVERY: Duration = Duration::from_secs(60); | ||
| ||
/// What a run of a workflow that waited for `name`, which ended `ended`, | ||
/// is: cancelled as it was, or failed for any other end. | ||
fn not_after(name: &str, ended: &str) -> After { | ||
if ended == status::CANCELLED { | ||
let line = format!( | ||
"==> cancelled before it started: {name}, which it runs after, was cancelled\n" | ||
); | ||
return After::Not { | ||
status: status::CANCELLED, | ||
line, | ||
}; | ||
} | ||
let line = format!("==> not run: {name}, which it runs after, ended {ended}\n"); | ||
After::Not { | ||
status: status::FAILURE, | ||
line, | ||
} | ||
} | ||
| ||
/// What a run's log says of what its workflow names `after`: how long it | ||
/// waited for them, and which of them, `absent`, have no check there. | ||
fn after_lines(after: &[String], absent: &[&str], took: Duration) -> String { | ||
let there: Vec<&str> = after | ||
.iter() | ||
.map(String::as_str) | ||
.filter(|n| !absent.contains(n)) | ||
.collect(); | ||
let mut lines = String::new(); | ||
if !there.is_empty() { | ||
let (names, secs) = (there.join(", "), took.as_secs()); | ||
lines += &format!( | ||
"==> after {names}, which passed: it waited {secs}s for them, holding no job\n" | ||
); | ||
} | ||
if !absent.is_empty() { | ||
let names = absent.join(", "); | ||
lines += &format!( | ||
"==> after {names}, which have no check here at this commit: not waited for\n" | ||
); | ||
} | ||
lines | ||
} | ||
| ||
async fn until_cancelled<T>(cancelled: &mut Cancelled, f: impl Future<Output = T>) -> Option<T> { | ||
tokio::select! { | ||
biased; | ||
crates/knotless-run/tests/run.rs+122−0
| @@ -1307,6 +1307,128 @@ fn a_deploy_is_never_cut_off() { | ||
); | ||
} | ||
| ||
/// A workflow that runs `after` others waits for their checks at its | ||
/// commit holding no job, so the one job goes to them; runs once they | ||
/// pass; and does not run when one fails or is cancelled. | ||
#[test] | ||
fn a_workflow_runs_after_others_holding_no_job() { | ||
let owner = Sandbox::new(); | ||
let mut runner = Sandbox::new(); | ||
let handle = runner.account.handle.clone(); | ||
runner.set("KNOTLESS_IDENTIFIER", &handle); | ||
let work = owner.work("after", 1); | ||
owner.git(&work, &["remote", "add", "origin", &owner.url("after")]); | ||
owner.git(&work, &["push", "-q", "origin", "main"]); | ||
let (frames, _) = tokio::sync::mpsc::unbounded_channel(); | ||
let url = "ws://127.0.0.1:9".to_string(); | ||
let js = Jetstream { url, frames }; | ||
let mut opts = runner_opts(&runner, &owner, &js); | ||
opts.repos = vec![format!("{}/after", owner.account.handle)]; | ||
opts.jobs = 1; | ||
opts.pull_jobs = 1; | ||
let rt = tokio::runtime::Runtime::new().unwrap(); | ||
let (run, watches) = rt | ||
.block_on(knotless_run::Runner::start(opts, common::ctx(&runner))) | ||
.unwrap(); | ||
let watch = &watches[0]; | ||
let key = watch.key.clone(); | ||
rt.block_on(run.reconcile(watch)).unwrap(); | ||
| ||
let go = runner.path("go"); | ||
std::fs::create_dir_all(&go).unwrap(); | ||
let dir = work.join(".knotless/workflows"); | ||
std::fs::create_dir_all(&dir).unwrap(); | ||
let on = "on: {push: {refs: [refs/heads/main]}}\n"; | ||
let test = format!( | ||
"{on}steps:\n - run: |\n touch {go}/started-$KNOTLESS_RUN_SHA\n until [ -f {go}/$KNOTLESS_RUN_SHA ]; do sleep 0.1; done\n [ ! -f {go}/fail-$KNOTLESS_RUN_SHA ]\n", | ||
go = go.display() | ||
); | ||
let deploy = | ||
format!("{on}superseded: finish\nafter: [test, nope]\nsteps:\n - run: \"true\"\n"); | ||
let notify = format!("{on}after: [test]\nsteps:\n - run: \"true\"\n"); | ||
std::fs::write(dir.join("test.yml"), test).unwrap(); | ||
std::fs::write(dir.join("deploy.yml"), deploy).unwrap(); | ||
std::fs::write(dir.join("notify.yml"), notify).unwrap(); | ||
owner.git(&work, &["add", "."]); | ||
owner.git(&work, &["commit", "-q", "-m", "workflows"]); | ||
let at = |sha: &str, wf: &str| { | ||
let rkey = check::rkey(&owner.account.did, &key, "refs/heads/main", sha, None, wf); | ||
runner.record(check::NSID, &rkey) | ||
}; | ||
let log = |c: &serde_json::Value| { | ||
blob( | ||
&runner.account.did, | ||
c["log"]["ref"]["$link"].as_str().unwrap(), | ||
) | ||
}; | ||
let push = |n: usize| { | ||
if n > 0 { | ||
owner.commit_file(&work, "f", &format!("{n}\n")); | ||
} | ||
owner.git(&work, &["push", "-q", "origin", "HEAD:main"]); | ||
let sha = owner.git(&work, &["rev-parse", "HEAD"]); | ||
rt.block_on(run.reconcile(watch)).unwrap(); | ||
sha | ||
}; | ||
| ||
// Filed before test, deploy and notify wait without its one job. | ||
let a = push(0); | ||
wait_until(|| go.join(format!("started-{a}")).exists()); | ||
assert_eq!(at(&a, "deploy")["status"], "queued"); | ||
std::fs::write(go.join(&a), "").unwrap(); | ||
wait_until(|| at(&a, "deploy")["finishedAt"].is_string()); | ||
wait_until(|| at(&a, "notify")["finishedAt"].is_string()); | ||
let deployed = at(&a, "deploy"); | ||
assert_eq!(deployed["status"], "success", "{deployed}"); | ||
let said = log(&deployed); | ||
assert!( | ||
said.starts_with("==> after test, which passed: it waited "), | ||
"{said}" | ||
); | ||
assert!( | ||
said.contains("==> after nope, which have no check here at this commit: not waited for\n"), | ||
"{said}" | ||
); | ||
assert_eq!(at(&a, "notify")["status"], "success"); | ||
| ||
// One that fails: neither runs. | ||
let b = push(1); | ||
std::fs::write(go.join(format!("fail-{b}")), "").unwrap(); | ||
std::fs::write(go.join(&b), "").unwrap(); | ||
wait_until(|| at(&b, "deploy")["finishedAt"].is_string()); | ||
wait_until(|| at(&b, "notify")["finishedAt"].is_string()); | ||
let (deploy, notify) = (at(&b, "deploy"), at(&b, "notify")); | ||
assert_eq!(deploy["status"], "failure", "{deploy}"); | ||
assert_eq!(notify["status"], "failure", "{notify}"); | ||
assert_eq!( | ||
log(&deploy), | ||
"==> not run: test, which it runs after, ended failure\n" | ||
); | ||
| ||
// A newer commit: its test cancelled, the deploy kept, which so ends | ||
// cancelled, and notify, cancelled for the newer, as ever. | ||
let c = push(2); | ||
wait_until(|| go.join(format!("started-{c}")).exists()); | ||
let d = push(3); | ||
wait_until(|| at(&c, "deploy")["finishedAt"].is_string()); | ||
wait_until(|| at(&c, "notify")["finishedAt"].is_string()); | ||
let (deploy, notify) = (at(&c, "deploy"), at(&c, "notify")); | ||
assert_eq!(deploy["status"], "cancelled", "{deploy}"); | ||
assert_eq!( | ||
log(&deploy), | ||
"==> cancelled before it started: test, which it runs after, was cancelled\n" | ||
); | ||
assert_eq!(notify["status"], "cancelled", "{notify}"); | ||
assert_eq!( | ||
log(¬ify), | ||
"==> cancelled for a newer commit's run before it started\n" | ||
); | ||
std::fs::write(go.join(&d), "").unwrap(); | ||
rt.block_on(run.idle()); | ||
assert_eq!(at(&d, "deploy")["status"], "success"); | ||
assert_eq!(at(&d, "notify")["status"], "success"); | ||
} | ||
| ||
/// A stop waits only for the runs whose steps have started, a deploy's | ||
/// among them: those waiting their turn are dropped, their checks filed | ||
/// as cancelled, not deleted, and run when the runner starts again, and | ||
docs/design/runner.md+6−0
| @@ -382,6 +382,12 @@ and any [turn](#deploys-superseded-finish) on a ref it must hold. It never | ||
holds one thing while waiting for another, so two runs never wait on each | ||
other. | ||
| ||
A workflow that names others in `after` | ||
([After](../reference/workflows.md#after)) waits for their checks at its | ||
commit before it asks for a job at all. A step that waited for them | ||
instead would hold a job the whole time, and with few jobs, the checks it | ||
waits for could get none. | ||
| ||
Waiting runs are ranked by class, then by the order they were started. A | ||
run ranked earlier has first claim on what it waits for and on a free job. | ||
A later one goes ahead only with what nothing before it waits for, and a | ||
docs/reference/workflows.md+24−0
| @@ -39,6 +39,7 @@ secrets: [DEPLOY_TOKEN] | ||
runners: [deploy.alice.example.com] # default: every runner named | ||
superseded: cancel # or finish | ||
queue: batch # or first | ||
after: [test, lint] # checks to pass first; default: none | ||
steps: | ||
- name: test | ||
run: cargo test | ||
| @@ -59,8 +60,24 @@ steps: | ||
| `runners` | every runner the repository names | the runners that run it, by DID or handle, of those the repository names; any other files no check of it, so a workflow naming none of them never runs, and a pull waiting on it never merges | | ||
| `superseded` | `cancel` | what a newer commit's run on the same ref or pull does to this one: `cancel` it, or, with `finish`, wait for it once it holds the workflow's turn, which it takes as soon as it is filed if no run holds it; anything else fails to parse | | ||
| `queue` | `batch` | where it runs in the [merge queue](#the-merge-queue), if it runs on pulls: `batch`, on each batch, or `first`, on each batch and first on each pull queued, which goes into a batch only once it passes; anything else fails to parse | | ||
| `after` | none | workflows whose checks at the same commit, on the same ref, pull or batch, must pass before it starts; see [After](#after) | | ||
| `steps` | required | the steps, run in order until one fails | | ||
| ||
### After | ||
| ||
A workflow that names others in `after` is filed `queued` with the rest, | ||
then waits for their checks at its commit, by the same runner, to end. It | ||
holds no job while it waits, so the checks it waits for are not kept from | ||
one. Once all of them have passed or were skipped, it takes its turn for | ||
a job as any run does. If one fails or errs it is filed `failure` without | ||
running, and if one is cancelled it is filed `cancelled`, its log saying | ||
which. A name with no check there, as a workflow that does not run on | ||
that ref or is another runner's, is not waited for. The checks are read | ||
again as each of the runner's runs ends, and every minute besides. | ||
| ||
A deploy that names its commit's checks in `after` so waits for them | ||
without a step of its own that polls, and without a job the checks need. | ||
| ||
## Triggers | ||
| ||
| Trigger | Runs on | Keys | | ||
| @@ -409,6 +426,13 @@ and the runner's own lines start `==>` too: | ||
`as the merge queue dropped its batch`, `as the runner stopped: asked | ||
for again when it starts`, or `as the runner stopped, past what a stop | ||
waits for`; | ||
- `==> after <names>, which passed: it waited <secs>s for them, holding no | ||
job`, and `==> after <names>, which have no check here at this commit: | ||
not waited for`, first in a run of a workflow that names | ||
[`after`](#after); or, as the whole log of one filed without running, | ||
`==> not run: <name>, which it runs after, ended <status>`, or `==> | ||
cancelled before it started: <name>, which it runs after, was | ||
cancelled`; | ||
- `==> asked for by name NAME=value ...`, first in a run asked for by | ||
name, or `==> on its schedule KNOTLESS_SCHEDULE=<cron>` in one its | ||
schedule ran. Its values show to whoever reads the log, as they do in the | ||