chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! Landing a repository's merge queues (`knotless::queue`), with
//! `--queue`, as its owner or a collaborator.
//!
//! Each branch's queue goes a batch at a time: up to `--batch` queued
//! pulls in order, whatever their own checks say, each squashed onto the
//! branch's tip as a commit of its own on the one before. The workflows
//! that run on a pull into the branch run once, at the last of them, filed
//! as checks keyed by the queue that name the batch. Once they all pass the
//! branch moves there, from where the batch was built and only from there,
//! and each pull is marked merged.
//!
//! A pull goes into a batch only once its pre-flights, the workflows that
//! say `queue: first`, have passed at its head, as any runner the
//! repository names filed them: it waits while one runs, holding up none
//! behind it, and is ejected without a batch once one fails on the
//! branch's tip. One nobody ran there is run here, as a pull's run.
//!
//! A pull that does not apply on what is before it is ejected, naming the
//! paths, and the batch goes on without it. A batch that fails is halved,
//! its first half run alone, until the pull that breaks the branch is found
//! and ejected: a half that passes lands, and what is left of a batch that
//! failed is then the same tree on the tip it moved to, so it fails too,
//! and a lone pull left is ejected without running again. The branch
//! moving any other way, or a pull in a batch moving or leaving, drops the
//! batches going, to be built again.
//!
//! While a batch runs, the next is built on its tip and run, as if it
//! passed (`--depth` 2, the default; 1 runs one at a time). It runs ahead
//! as a pull does: after the runs of pushes to the default branch and of
//! the batch it is on, with a job left over and the CPU share a pull's run
//! has, so it never holds up a deploy. The batch it is on failing, or
//! changing, drops it, cancelling its runs; it lands only on the commit
//! that batch landed as, and nothing lands but on the very commit it was
//! built on.
//!
//! The queue records are read from the repos of those who may write them,
//! the owner's and each collaborator's, not only as the appview has
//! indexed them, and looked at every `--notice` besides as Jetstream
//! brings them: a pull queued is taken in that long at most.
use crate::telemetry;
use crate::{
Asks, DROPPED, Pull, Runner, Subject, Watch, cancel, changes, short, vouched, workflow,
};
use anyhow::{Context, Result, bail};
use knotless::approvals;
use knotless::checks::{self, Filter};
use knotless::forge::{self, Item, Target};
use knotless::git::Git;
use knotless::keys;
use knotless::mirror::Mirror;
use knotless::queue::{self, Stack};
use knotless::records::queue::{CONFLICTS, HALVED};
use knotless::records::{Batched, FailedCheck, RepoRef, nsid, status};
use knotless::remote::Snapshot;
use knotless::session::{self, Session};
use knotless::xrpc::XrpcError;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::{Arc, Mutex};
/// Why landing was refused, when the gateway refused it: a runner that
/// lands as a runner the repository names, through the gateway, is refused
/// until the owner delegates writes to it.
fn refused(err: &anyhow::Error) -> Option<String> {
let refused = XrpcError::of(err).filter(|e| e.status == 403)?;
Some(format!(
"the gateway would not land its batch: {}. A runner lands the merge queue through the gateway the owner delegates writes to: under Delegated writes in their settings",
refused.message
))
}
/// A batch a run is the tip of: the branch, in full, its tip the batch is
/// on, the pulls in it, and how each workflow's run of it ended.
#[derive(Clone)]
pub(crate) struct Batch {
pub base: String,
pub base_commit: String,
pub pulls: Vec<Batched>,
/// Whether everyone whose code is in it may push here: then it shares
/// the pushes' caches.
pub trusted: bool,
/// Whether it was built on a batch still being checked: its runs go
/// as a pull's do.
pub ahead: bool,
/// The accounts whose shared steps its lock adds that the repository
/// does not vouch for, which make it untrusted.
pub strangers: Vec<String>,
runs: Arc<Mutex<BTreeMap<String, Option<String>>>>,
}
impl Batch {
pub fn new(base: &str, base_commit: &str, pulls: Vec<Batched>, trusted: bool) -> Batch {
Batch {
base: base.to_string(),
base_commit: base_commit.to_string(),
pulls,
trusted,
ahead: false,
strangers: Vec::new(),
runs: Arc::default(),
}
}
/// Notes a run of `workflow` started.
pub fn started(&self, workflow: &str) {
self.runs.lock().unwrap().insert(workflow.to_string(), None);
}
/// Notes how the run of `workflow` ended.
pub fn ended(&self, workflow: &str, status: &str) {
let ended = Some(status.to_string());
self.runs
.lock()
.unwrap()
.insert(workflow.to_string(), ended);
}
/// How its runs ended, once they all have.
pub fn outcome(&self) -> Option<BTreeMap<String, String>> {
let runs = self.runs.lock().unwrap();
runs.iter()
.map(|(w, s)| Some((w.clone(), s.clone()?)))
.collect()
}
}
/// A batch going: how it was built, and its runs.
struct Flight {
stack: Stack,
batch: Batch,
}
/// A branch's merge queue as the runner lands it: the batches going, the
/// lowest on the branch's tip and each after on the one before; once a
/// batch has failed, its pulls, one of which breaks the branch, and how it
/// failed; and the pulls landed or ejected, by the key of the record that
/// queued them, until the queue as read says so too; and the pulls whose
/// pre-flights it ran itself, each at what (see [`at`](crate::at)).
#[derive(Default)]
pub(crate) struct Line {
flights: Vec<Flight>,
suspects: Vec<String>,
failed: (String, Vec<FailedCheck>),
done: HashMap<String, String>,
flown: HashMap<String, String>,
}
impl Line {
/// The pulls the batches in flight hold, by URI.
fn flying(&self) -> impl Iterator<Item = &str> {
(self.flights.iter()).flat_map(|f| f.stack.stacked.iter().map(|s| s.pull.uri.as_str()))
}
}
/// How a pull's pre-flights stand at its head.
#[derive(Default)]
struct Preflight {
/// The newest run of each that did not pass on the branch's tip.
failed: Vec<checks::Check>,
running: bool,
/// Those with no run there, or only one cancelled or one that failed
/// on another tip.
missing: BTreeSet<String>,
}
/// A repository whose merge queues a runner lands, as last read, and the
/// queue and state records about it, each by URI and what it says, as
/// last looked at.
#[derive(Default)]
pub(crate) struct Noticing {
target: Option<Arc<Target>>,
seen: Option<BTreeSet<String>>,
}
/// The key of the record that queued `item`.
fn queued_at(item: &Item) -> &str {
item.queue.as_ref().map_or("", |q| q.at.as_str())
}
/// How the pre-flights `first` of a pull stand at its head, by its runs
/// there, `runs`, newest first, on the branch at `tip`. Another runner's
/// run only queued there counts as none: it may wait long for a job, so
/// this runner, `me`, runs it too, and whichever starts first runs it.
fn preflight(first: &BTreeSet<String>, runs: &[checks::Check], tip: &str, me: &str) -> Preflight {
let mut stands = Preflight::default();
let counts = |c: &&checks::Check| c.runner == me || c.record.status != status::QUEUED;
for workflow in first {
let newest = (runs.iter().filter(counts)).find(|c| c.record.workflow == *workflow);
match newest.map(|c| (c, c.record.status.as_str())) {
Some((_, status::SUCCESS)) => {}
Some((_, status::RUNNING | status::QUEUED)) => stands.running = true,
Some((c, status::FAILURE | status::ERROR))
if c.record.base_commit.as_deref() == Some(tip) =>
{
stands.failed.push(c.clone());
}
_ => {
stands.missing.insert(workflow.clone());
}
}
}
stands
}
/// Why a pull that conflicts in `files`, built on `into` at `tip`, is
/// ejected: naming the pulls ahead of it that change them, `touching`,
/// and how to take theirs in, or else the branch itself.
fn conflicted(into: &str, tip: &str, files: &[String], touching: &[&queue::Stacked]) -> String {
let files = files.join(", ");
if touching.is_empty() {
return format!(
"it does not apply on {into} at {}: {CONFLICTS} {files}. Merge {into} into it, or rebase it onto {into}, and queue it again",
short(tip)
);
}
let named: Vec<String> = touching
.iter()
.map(|s| {
let branch = s.pull.head.as_ref().map_or("", |h| h.branch.as_str());
format!("{} ({branch})", s.pull.uri)
})
.collect();
format!(
"{CONFLICTS} {files} with {}, queued ahead of it on {into} at {}. Merge that branch into it, or wait for it to land and merge {into} into it, and queue it again",
named.join(" and "),
short(tip)
)
}
impl Runner {
/// Lands what the merge queues of `watch` can, ejects what they must,
/// and starts their next batches. A runner that may not push to the
/// repository leaves its queues be.
pub(crate) async fn queues(
self: &Arc<Self>,
watch: &Watch,
repo: &RepoRef,
snap: &Snapshot,
) -> Result<()> {
let target = Arc::new(Target::open(&self.ctx, &watch.remote.url).await?);
if !self.lands(&target.record, watch) {
return Ok(());
}
{
let mut noticing = self.noticing.lock().unwrap();
noticing.entry(watch.id()).or_default().target = Some(target.clone());
}
let me = session::me(&self.ctx).await?;
let entries = target.queue_entries(&self.ctx).await?;
let items = forge::assemble(&target.record, watch.remote.did(), &entries);
let lines = queue::by_branch(&items);
let prefix = format!("{}\n", watch.id());
let mut branches: BTreeSet<String> = lines.keys().cloned().collect();
let known: Vec<String> = self.lines.lock().unwrap().keys().cloned().collect();
let mine = known.iter().filter_map(|k| k.strip_prefix(&prefix));
branches.extend(mine.map(str::to_string));
if branches.is_empty() {
return Ok(());
}
watch.at("syncing the mirror");
let mirror = Mirror::open(&self.ctx, watch.remote.did(), &watch.key).await?;
mirror.sync(&watch.remote, snap).await?;
let none = Vec::new();
for branch in branches {
let on = Queue {
watch,
repo,
snap,
target: &target,
git: &mirror.git,
me: &me,
branch: &branch,
};
self.line(&on, lines.get(&branch).unwrap_or(&none)).await?;
}
Ok(())
}
/// Lands, as the runner stops, the batches of `watch` whose runs have
/// all passed, and does nothing else: no batch is started or ejected,
/// and what failed is checked again on the next start.
///
/// Boxed: it is awaited in the task of the run that ends, and starts
/// runs itself, a cycle in which the compiler cannot see it is `Send`.
pub(crate) fn land_passed<'a>(
self: &'a Arc<Self>,
watch: &'a Watch,
) -> futures::future::BoxFuture<'a, Result<()>> {
Box::pin(async move {
let _one = self.landing.lock().await;
watch.at("reading the repository");
let snap = watch.remote.snapshot(&self.ctx).await?;
let repo = snap
.reference(watch.remote.did())
.context("a snapshot with a key has a repository")?;
self.queues(watch, &repo, &snap).await
})
}
/// Goes on with the queue of `on.branch`, whose pulls queued are
/// `queued`, in order.
async fn line(self: &Arc<Self>, on: &Queue<'_>, queued: &[Item]) -> Result<()> {
let id = format!("{}\n{}", on.watch.id(), on.branch);
let mut line = self.lines.lock().unwrap().remove(&id).unwrap_or_default();
let went = self.go_on(on, queued, &mut line).await;
let idle = line.flights.is_empty()
&& line.suspects.is_empty()
&& line.done.is_empty()
&& line.flown.is_empty();
if !idle {
self.lines.lock().unwrap().insert(id, line);
}
went
}
async fn go_on(self: &Arc<Self>, on: &Queue<'_>, all: &[Item], line: &mut Line) -> Result<()> {
// What this runner landed or ejected, until the queue as read has
// caught up.
let uris: HashSet<&str> = all.iter().map(|i| i.uri.as_str()).collect();
line.done.retain(|uri, _| uris.contains(uri.as_str()));
line.flown.retain(|uri, _| uris.contains(uri.as_str()));
let done = |i: &&Item| line.done.get(&i.uri).map(String::as_str) == Some(queued_at(i));
let queued: Vec<Item> = all.iter().filter(|i| !done(i)).cloned().collect();
let refname = format!("refs/heads/{}", on.branch);
let found = on.snap.find_ref(&refname);
let Some(mut tip) = found.and_then(|r| r.value.target.clone()) else {
self.drop_from(on.watch, line, 0);
return Ok(());
};
let generation = found.map_or(0, |r| r.value.generation);
// The branch moved some other way: what is going is of a tip it is
// no longer at, and what failed there may not fail now.
if line.flights.first().is_some_and(|f| f.stack.base != tip) {
self.drop_from(on.watch, line, 0);
line.suspects.clear();
}
// A pull in a batch, or among those suspected, left the queue.
let still: HashSet<&str> = queued.iter().map(|i| i.uri.as_str()).collect();
let stays = |uri: &String| still.contains(uri.as_str());
let left = |f: &Flight| !f.stack.stacked.iter().all(|s| stays(&s.pull.uri));
if let Some(at) = line.flights.iter().position(left) {
self.drop_from(on.watch, line, at);
}
if !line.suspects.iter().all(stays) {
line.suspects.clear();
}
while let Some(ended) = line.flights.first().and_then(|f| f.batch.outcome()) {
let flight = line.flights.remove(0);
let failed: Vec<String> = ended
.iter()
.filter(|(_, s)| *s != status::SUCCESS)
.map(|(w, s)| format!("{w} ({s})"))
.collect();
let checks = self.failed_checks(on, &flight, &ended);
if failed.is_empty() {
on.watch.at("landing a batch");
if let Err(err) = self.land(on, &flight).await {
let refused = refused(&err);
let (label, branch) = (&on.watch.label, on.branch);
telemetry::failed(label, &err.context(format!("landing on {branch}")));
self.drop_from(on.watch, line, 0);
// Refused, it would be refused again: its pulls go,
// rather than be checked and refused for ever.
if let Some(why) = &refused {
for s in &flight.stack.stacked {
self.eject(on, line, &s.pull, Some(&s.head), why, &[])
.await?;
}
line.suspects.clear();
}
break;
}
tip = flight.stack.tip().to_string();
let landed = flight.stack.stacked.len();
telemetry::batch(
"batch landed",
&on.watch.label,
on.branch,
&tip,
landed,
None,
);
for s in &flight.stack.stacked {
let at = queued_at(&s.pull).to_string();
line.done.insert(s.pull.uri.clone(), at);
line.suspects.retain(|u| *u != s.pull.uri);
}
// What is left of a batch that failed is the same tree on
// the tip it moved to, and fails as it did: one alone is
// what broke it.
if let [lone] = &line.suspects[..] {
let item = queued.iter().find(|i| &i.uri == lone);
let item = item.expect("a suspect is queued").clone();
let (why, checks) = line.failed.clone();
self.eject(on, line, &item, None, &why, &checks).await?;
line.suspects.clear();
}
continue;
}
self.drop_from(on.watch, line, 0);
let why = format!(
"its batch failed its checks at {}, on {} at {}: {}",
short(flight.stack.tip()),
on.branch,
short(&flight.stack.base),
failed.join(", "),
);
match &flight.stack.stacked[..] {
[one] => {
self.eject(on, line, &one.pull, Some(&one.head), &why, &checks)
.await?;
line.suspects.clear();
}
all => {
line.suspects = all.iter().map(|s| s.pull.uri.clone()).collect();
line.failed = (format!("{why}. {HALVED}"), checks);
}
}
}
// Stopping, it lands what passed and starts nothing.
if *self.stopping.borrow() {
return Ok(());
}
// Of the pulls waiting, only those whose pre-flights have passed go
// into a batch; the rest wait, or are ejected, holding up none
// behind them.
let going: HashSet<&str> = (line.flying())
.chain(line.suspects.iter().map(String::as_str))
.collect();
let outside: Vec<Item> = (queued.iter())
.filter(|i| !line.done.contains_key(&i.uri) && !going.contains(i.uri.as_str()))
.cloned()
.collect();
on.watch.at("checking pre-flights");
let ready = self.preflown(on, line, &tip, outside).await?;
while line.flights.len() < self.opts.depth {
// Halving is not run ahead of.
if !(line.suspects.is_empty() || line.flights.is_empty()) {
break;
}
let ahead = !line.flights.is_empty();
let on_tip = line
.flights
.last()
.map_or(tip.clone(), |f| f.stack.tip().to_string());
let taken: HashSet<&str> = line.flying().collect();
let half = &line.suspects[..line.suspects.len().div_ceil(2)];
let waiting = queued.iter().filter(|i| !line.done.contains_key(&i.uri));
let next: Vec<Item> = match half {
[] => waiting
.filter(|i| !taken.contains(i.uri.as_str()) && ready.contains(&i.uri))
.take(self.opts.batch)
.cloned()
.collect(),
half => waiting.filter(|i| half.contains(&i.uri)).cloned().collect(),
};
if next.is_empty() {
break;
}
let mut pulls = Vec::new();
for item in next {
let head = item.head.as_ref().expect("a pull has a head");
match forge::head_of(&self.ctx, on.git, head).await? {
Ok(sha) => pulls.push((item, sha)),
Err(why) => self.eject(on, line, &item, None, &why, &[]).await?,
}
}
on.watch.at("stacking a batch");
let stack = queue::stack(on.git, &on_tip, &pulls, on.me).await?;
// Ahead of a batch that may yet fail, a conflict may not stand:
// the pull waits to be built on the branch itself.
for (item, head, files) in stack.conflicts.iter().filter(|_| !ahead) {
let touching = stack.touching(on.git, files).await?;
let why = conflicted(on.branch, &on_tip, files, &touching);
self.eject(on, line, item, Some(head), &why, &[]).await?;
}
if stack.stacked.is_empty() {
if ahead {
break;
}
continue;
}
on.watch.at("launching a batch");
let flight = self.launch(on, stack, generation, ahead).await?;
line.flights.push(flight);
}
Ok(())
}
/// Whether the queue and state records about `watch` that those who
/// may write them have, the newest of each, are others than the last
/// look found: the first look, too. Nothing, until a pass has found
/// this runner lands its queues.
pub async fn noticed(&self, watch: &Watch) -> Result<bool> {
let noticing = self
.noticing
.lock()
.unwrap()
.get(&watch.id())
.map(|n| n.target.clone());
let Some(target) = noticing.flatten() else {
return Ok(false);
};
let fresh = target.fresh_entries(&self.ctx).await;
let found = fresh.iter().map(|e| format!("{} {}", e.uri, e.value));
let found: BTreeSet<String> = found.collect();
let mut noticing = self.noticing.lock().unwrap();
let seen = &mut noticing.entry(watch.id()).or_default().seen;
Ok(seen.replace(found.clone()) != Some(found))
}
/// The pulls among `waiting` that may go into a batch on the branch at
/// `tip`: each whose pre-flights, the workflows at `tip` that run
/// first in its queue, have each passed at its head, on any tip, as any
/// runner the repository names, or this one, filed it. One whose
/// newest run of one failed on `tip` is ejected; one with a pre-flight
/// that has none, or one cancelled or failed on another tip, has it
/// run here, as a pull's run, and waits. Once this runner has run them
/// at its head on `tip`, it waits no longer: it is a workflow the pull
/// takes away, say, and its batch will say what it does. Into a branch
/// whose rule asks for reviews, one not reviewed as it asks is ejected
/// first, as a merge would refuse it.
async fn preflown(
self: &Arc<Self>,
on: &Queue<'_>,
line: &mut Line,
tip: &str,
waiting: Vec<Item>,
) -> Result<HashSet<String>> {
let waiting = self.reviewed(on, line, tip, waiting).await?;
let refname = format!("refs/heads/{}", on.branch);
let loaded = workflow::load(on.git, tip).await?;
let first: BTreeSet<String> = (loaded.into_iter())
.filter_map(|l| l.workflow.ok())
.filter(|w| w.first_in_queue(&refname))
.map(|w| w.name)
.collect();
if first.is_empty() {
return Ok(waiting.into_iter().map(|i| i.uri).collect());
}
let mut ready = HashSet::new();
let (record, me) = (&on.target.record, &on.me.did);
let mut runners: Vec<String> = record.runners.iter().map(|r| r.did.clone()).collect();
runners.extend((!runners.contains(me)).then_some(me.clone()));
for item in waiting {
let head = item.head.as_ref().expect("a pull has a head");
// One whose branch is gone is ejected as a batch is built.
let found = forge::head_at(&self.ctx, on.git, head).await?;
let Ok((sha, generation)) = found else {
ready.insert(item.uri);
continue;
};
let filter = Filter {
pull: Some(&item.uri),
commit: Some(&sha),
..Filter::default()
};
let (remote, key) = (&on.target.remote, &record.key);
let runs = checks::by(&self.ctx, remote, key, &runners, filter, false).await?;
let stands = preflight(&first, &runs, tip, me);
if !stands.failed.is_empty() {
let failed: Vec<String> = (stands.failed.iter())
.map(|c| format!("{} ({})", c.record.workflow, c.record.status))
.collect();
let why = format!(
"its pre-flight failed at its head {}, on {} at {}: {}",
short(&sha),
on.branch,
short(tip),
failed.join(", "),
);
let checks: Vec<FailedCheck> = (stands.failed.iter())
.map(|c| FailedCheck {
workflow: c.record.workflow.clone(),
status: Some(c.record.status.clone()),
check: Some(c.uri.clone()),
})
.collect();
self.eject(on, line, &item, Some(&sha), &why, &checks)
.await?;
continue;
}
if stands.running {
continue;
}
let at = crate::at(&sha, Some(tip));
if !stands.missing.is_empty() && line.flown.get(&item.uri) != Some(&at) {
let only = stands.missing;
let flew = self.fly(on, &item, (&sha, generation), tip, only).await?;
line.flown.insert(item.uri.clone(), at);
// One that does not merge goes on, to be ejected as a batch
// is built, saying where it conflicts.
if flew {
continue;
}
}
ready.insert(item.uri);
}
Ok(ready)
}
/// Of `waiting`, those reviewed as their branch's rule asks, on the
/// branch at `tip`; the rest are ejected, saying what they lack. One
/// whose branch is gone goes on, to be ejected as a batch is built.
async fn reviewed(
&self,
on: &Queue<'_>,
line: &mut Line,
tip: &str,
waiting: Vec<Item>,
) -> Result<Vec<Item>> {
let rule = on.target.record.rule(on.branch);
if rule.is_none_or(|r| r.reviews.is_none()) {
return Ok(waiting);
}
let mut reviewed = Vec::new();
for item in waiting {
let head = item.head.as_ref().expect("a pull has a head");
let Ok(sha) = forge::head_of(&self.ctx, on.git, head).await? else {
reviewed.push(item);
continue;
};
let standing =
approvals::of(&self.ctx, on.target.facts(), &item, on.git, tip, &sha).await?;
let url = &on.watch.remote.url;
let why = standing.and_then(|s| approvals::refusal(url, &item.uri, on.branch, &s));
let eject = why
.as_deref()
.map(|why| self.eject(on, line, &item, Some(&sha), why, &[]));
let ejected = futures::future::OptionFuture::from(eject).await;
let ejected = ejected.transpose()?;
reviewed.extend(ejected.is_none().then_some(item));
}
Ok(reviewed)
}
/// Runs the pre-flights `only` of `item`, at its `head` (and the
/// branch's generation there) merged with the branch at `tip`, as a
/// pull's run, unless the two conflict: whether it does.
async fn fly(
self: &Arc<Self>,
on: &Queue<'_>,
item: &Item,
(head, generation): (&str, u64),
tip: &str,
only: BTreeSet<String>,
) -> Result<bool> {
let merged = forge::merge_result(on.git, tip, head).await?;
let Ok(merge) = &merged else {
return Ok(false);
};
let changed = changes(on.git, tip, merge).await;
let (record, owner) = (&on.target.record, on.watch.remote.did());
let branch = item.head.as_ref().expect("a pull has a head");
let trusted =
vouched(record, owner, &item.author) && vouched(record, owner, branch.repo.did());
let strangers = self.strangers(on.git, record, owner, tip, merge).await;
let trusted = trusted && strangers.is_empty();
// Where the pull ran, so a pass over the pulls finds it there, and
// does not cancel it as moved.
let id = on.watch.id();
if let Some(pulls) = self.state.lock().unwrap().pulls.get_mut(&id) {
pulls.insert(item.uri.clone(), crate::at(head, Some(tip)));
}
let subject = Subject {
ref_name: format!("refs/heads/{}", branch.branch),
commit: head.to_string(),
generation,
main: false,
pull: Some(Pull {
uri: item.uri.clone(),
base: format!("refs/heads/{}", on.branch),
base_commit: tip.to_string(),
// Only a --pulls-only repository's pulls warm its caches,
// and its queue is landed elsewhere.
main_generation: None,
merged,
trusted,
later: false,
strangers,
others: self.others(record),
}),
tree: None,
changed,
batch: None,
pushed: true,
asked: Asks::new(),
only: Some(only),
};
self.run_at(on.watch, on.repo, on.git, subject).await?;
Ok(true)
}
/// Starts the runs of a batch built as `stack`, `ahead` of one still
/// being checked or not.
async fn launch(
self: &Arc<Self>,
on: &Queue<'_>,
stack: Stack,
generation: u64,
ahead: bool,
) -> Result<Flight> {
let (record, owner) = (&on.target.record, on.watch.remote.did());
let trusted = stack.stacked.iter().all(|s| {
let head = s.pull.head.as_ref().map_or("", |h| h.repo.did());
vouched(record, owner, &s.pull.author) && vouched(record, owner, head)
});
let strangers = self.strangers(on.git, record, owner, &stack.base, stack.tip());
let strangers = strangers.await;
let trusted = trusted && strangers.is_empty();
let pulls = stack.stacked.iter().map(|s| Batched {
pull: s.pull.uri.clone(),
commit: s.commit.clone(),
head: Some(s.head.clone()),
});
let refname = format!("refs/heads/{}", on.branch);
let mut batch = Batch::new(&refname, &stack.base, pulls.collect(), trusted);
batch.ahead = ahead;
batch.strangers = strangers;
let changed = changes(on.git, &stack.base, stack.tip()).await;
let subject = Subject {
ref_name: refname,
commit: stack.tip().to_string(),
generation,
main: true,
pull: None,
tree: None,
changed,
batch: Some(batch.clone()),
pushed: true,
asked: Asks::new(),
only: None,
};
self.run_at(on.watch, on.repo, on.git, subject).await?;
let said = ["batch started", "batch started, ahead of another"][usize::from(ahead)];
let (tip, pulls) = (stack.tip(), stack.stacked.len());
telemetry::batch(said, &on.watch.label, on.branch, tip, pulls, None);
Ok(Flight { stack, batch })
}
/// Lands `flight`, if each pull in it is still at the head it took.
async fn land(&self, on: &Queue<'_>, flight: &Flight) -> Result<()> {
for s in &flight.stack.stacked {
let head = s.pull.head.as_ref().expect("a pull has a head");
let now = forge::head_of(&self.ctx, on.git, head).await?;
if now.as_ref() != Ok(&s.head) {
bail!("{} moved on from {}", s.pull.uri, short(&s.head));
}
}
let url = &on.watch.remote.url;
queue::land(&self.ctx, url, on.git, on.branch, &flight.stack).await
}
/// The runs of `flight` that `ended` other than passing, each with its
/// check, as this runner filed it.
fn failed_checks(
&self,
on: &Queue<'_>,
flight: &Flight,
ended: &BTreeMap<String, String>,
) -> Vec<FailedCheck> {
let (owner, key) = (on.watch.remote.did(), &on.watch.key);
let name = keys::queue_name(&flight.batch.base);
let (tip, base) = (flight.stack.tip(), Some(flight.stack.base.as_str()));
let failed = ended.iter().filter(|(_, s)| *s != status::SUCCESS);
failed
.map(|(w, s)| {
let rkey = keys::check_rkey(owner, key, &name, tip, base, w);
FailedCheck {
workflow: w.clone(),
status: Some(s.clone()),
check: Some(
knotless::aturi::AtUri::public(&on.me.did, nsid::CHECK, &rkey)
.resolved(on.watch.remote.space_ref().as_ref())
.to_string(),
),
}
})
.collect()
}
/// Ejects `item` from its queue, saying `why`, at `head` if it has one.
async fn eject(
&self,
on: &Queue<'_>,
line: &mut Line,
item: &Item,
head: Option<&str>,
why: &str,
checks: &[FailedCheck],
) -> Result<()> {
let ejected = queue::Ejected {
reason: Some(why),
head,
checks,
};
on.watch.at("ejecting a pull");
queue::eject(&self.ctx, on.target, &item.uri, ejected).await?;
telemetry::ejected(&on.watch.label, &item.uri, why);
let at = queued_at(item).to_string();
line.done.insert(item.uri.clone(), at);
Ok(())
}
/// Drops `line`'s batches from the `at`th up, cancelling their runs.
fn drop_from(&self, watch: &Watch, line: &mut Line, at: usize) {
let dropped: Vec<Flight> = line.flights.drain(at.min(line.flights.len())..).collect();
let going = self.going.lock().unwrap();
let repo = format!("{}\n", watch.id());
for flight in dropped {
let (tip, pulls) = (flight.stack.tip(), flight.stack.stacked.len());
telemetry::batch(
"batch dropped",
&watch.label,
&flight.batch.base,
tip,
pulls,
None,
);
let at = format!("{} {}", flight.stack.tip(), flight.stack.base);
let runs = going
.iter()
.filter(|(id, g)| id.starts_with(&repo) && g.at == at);
for (_, g) in runs {
cancel(&g.cancel, DROPPED);
}
}
}
}
/// A branch's merge queue, and where it is landed from.
struct Queue<'a> {
watch: &'a Watch,
repo: &'a RepoRef,
snap: &'a Snapshot,
target: &'a Target,
git: &'a Git,
me: &'a Session,
branch: &'a str,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn how_preflights_stand() {
let by = |runner: &str, workflow: &str, status: &str, base: &str| checks::Check {
uri: format!("at://{runner}/dev.knotless.check/{workflow}"),
runner: runner.into(),
record: serde_json::from_value(serde_json::json!({
"repo": {"uri": "at://did:a/dev.knotless.repo/r", "key": "k"},
"ref": "refs/heads/x", "commit": "h", "baseCommit": base,
"workflow": workflow, "status": status, "startedAt": "2026-01-01T00:00:00Z",
}))
.unwrap(),
log_text: None,
space: None,
};
let run = |workflow: &str, status: &str, base: &str| by("did:ci", workflow, status, base);
let first = |all: &[&str]| all.iter().map(|w| w.to_string()).collect::<BTreeSet<_>>();
// Passed on any tip, it is ready.
let stands = preflight(
&first(&["a"]),
&[run("a", "success", "old")],
"tip",
"did:ci",
);
assert!(!stands.running && stands.failed.is_empty() && stands.missing.is_empty());
// The newest run counts: one running now.
let runs = [run("a", "running", "tip"), run("a", "failure", "tip")];
let stands = preflight(&first(&["a"]), &runs, "tip", "did:ci");
assert!(stands.running && stands.failed.is_empty());
// Failed on the tip, it is failed; on another, or cancelled, or
// never run, it is to run again.
let runs = [
run("a", "failure", "tip"),
run("b", "error", "old"),
run("c", "cancelled", "tip"),
];
let stands = preflight(&first(&["a", "b", "c", "d"]), &runs, "tip", "did:ci");
let failed: Vec<&str> = stands
.failed
.iter()
.map(|c| c.record.workflow.as_str())
.collect();
assert_eq!(failed, ["a"]);
assert_eq!(stands.missing, first(&["b", "c", "d"]));
// Its own run queued, it waits; another runner's, it runs it too,
// unless an older run there is going.
let runs = [run("a", "queued", "tip")];
assert!(preflight(&first(&["a"]), &runs, "tip", "did:ci").running);
let runs = [by("did:other", "a", "queued", "tip")];
let stands = preflight(&first(&["a"]), &runs, "tip", "did:ci");
assert!(!stands.running && stands.missing == first(&["a"]));
let runs = [
by("did:other", "a", "queued", "tip"),
run("a", "running", "tip"),
];
assert!(preflight(&first(&["a"]), &runs, "tip", "did:ci").running);
}
#[test]
fn a_conflict_with_the_branch_itself_says_so() {
let why = conflicted("main", &"a".repeat(40), &["f".into(), "g".into()], &[]);
assert_eq!(
why,
"it does not apply on main at aaaaaaaaaaaa: it conflicts in f, g. Merge main into it, or rebase it onto main, and queue it again"
);
}
}