chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
The merge queue ejects failed pre-flights sooner and holds conflicts
.changeset/queue-record-head.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# A queued pull's record names the head it was queued at | ||
| ||
`knotless merge --queue`, and the MCP `merge` tool with `queue`, write the pull's head into its `dev.knotless.queue` record, as an ejection already does, and `knotless show --json` gives it. The queue's history then says whether a pull queued again after an ejection had changed. | ||
crates/knotless/src/forge.rs+1−1
| @@ -196,7 +196,7 @@ pub struct Queued { | ||
/// Why it was ejected. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub reason: Option<String>, | ||
/// The head that was ejected. | ||
/// The head that was queued, or ejected. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub commit: Option<String>, | ||
/// For one ejected by a batch that failed its checks, those that did | ||
crates/knotless/src/queue.rs+11−6
| @@ -88,7 +88,8 @@ async fn maintainer(ctx: &Ctx, t: &Target) -> Result<Session> { | ||
Ok(me) | ||
} | ||
| ||
/// A queue record about `pull`, in `t`. | ||
/// A queue record about `pull`, in `t`: why it was ejected, or for one | ||
/// queued, only the head it was queued at. | ||
fn record(t: &Target, pull: &str, state: &str, ejected: Option<Ejected>) -> Value { | ||
let ejected = ejected.unwrap_or_default(); | ||
let record = QueueRecord { | ||
| @@ -137,8 +138,12 @@ pub async fn enqueue(ctx: &Ctx, url: &RepoUrl, pull: &str) -> Result<Place> { | ||
bail!("{}", forge::drafted(url, pull)); | ||
} | ||
if item.queue.as_ref().is_none_or(|q| q.state != QUEUED) { | ||
admit(ctx, &t, &item).await?; | ||
let value = record(&t, pull, QUEUED, None); | ||
let head = admit(ctx, &t, &item).await?; | ||
let at = Ejected { | ||
head: Some(&head), | ||
..Ejected::default() | ||
}; | ||
let value = record(&t, pull, QUEUED, Some(at)); | ||
let uri = t.write(ctx, nsid::QUEUE, value.clone()).await?; | ||
let author = me.did; | ||
entries.push(Entry { uri, author, value }); | ||
| @@ -177,8 +182,8 @@ pub async fn merge_or_queue( | ||
/// landing changes, when it is not reviewed as its base's rule asks | ||
/// ([`approvals`]), or when a check of it at its head merged with that | ||
/// base has failed. A conflict with a pull ahead of it is left for the | ||
/// runner, as that pull may yet be ejected first. | ||
async fn admit(ctx: &Ctx, t: &Target, item: &Item) -> Result<()> { | ||
/// runner, as that pull may yet be ejected first. Admitted, its head. | ||
async fn admit(ctx: &Ctx, t: &Target, item: &Item) -> Result<String> { | ||
let sides = forge::sides(ctx, t, item).await?; | ||
let (base, head) = (&sides.base, &sides.head); | ||
let into = sides.refname.trim_start_matches("refs/heads/"); | ||
| @@ -214,7 +219,7 @@ async fn admit(ctx: &Ctx, t: &Target, item: &Item) -> Result<()> { | ||
item.uri | ||
); | ||
} | ||
Ok(()) | ||
Ok(head.to_string()) | ||
} | ||
| ||
/// The workflows `found` are runs of, each once. | ||
crates/knotless-run/src/queue.rs+60−38
| @@ -155,11 +155,12 @@ impl Line { | ||
/// 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. | ||
/// The newest run of each that did not pass on the branch's tip, or | ||
/// on any tip since the pull was queued. | ||
failed: Vec<checks::Check>, | ||
running: bool, | ||
/// Those with no run there, or only one cancelled or one that failed | ||
/// on another tip. | ||
/// on another tip before the pull was queued. | ||
missing: BTreeSet<String>, | ||
} | ||
| ||
| @@ -178,10 +179,20 @@ fn queued_at(item: &Item) -> &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 { | ||
/// there, `runs`, newest first, on the branch at `tip`, the pull queued | ||
/// at `since`. 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. A run that failed since the pull was | ||
/// queued fails it on whatever tip it ran: run again on the branch as it | ||
/// moves, the same head fails the same way, and its author would hear of | ||
/// it only once the branch stood still for a run. | ||
fn preflight( | ||
first: &BTreeSet<String>, | ||
runs: &[checks::Check], | ||
tip: &str, | ||
since: &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 { | ||
| @@ -190,7 +201,8 @@ fn preflight(first: &BTreeSet<String>, runs: &[checks::Check], tip: &str, me: &s | ||
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) => | ||
if c.record.base_commit.as_deref() == Some(tip) | ||
|| c.record.started_at.as_str() >= since => | ||
{ | ||
stands.failed.push(c.clone()); | ||
} | ||
| @@ -202,27 +214,12 @@ fn preflight(first: &BTreeSet<String>, runs: &[checks::Check], tip: &str, me: &s | ||
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 { | ||
/// Why a pull that conflicts in `files` with the branch `into` itself, at | ||
/// `tip`, is ejected. | ||
fn conflicted(into: &str, tip: &str, files: &[String]) -> 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 "), | ||
"it does not apply on {into} at {}: {CONFLICTS} {files}. Merge {into} into it, or rebase it onto {into}, and queue it again", | ||
short(tip) | ||
) | ||
} | ||
| @@ -467,11 +464,15 @@ impl Runner { | ||
} | ||
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. | ||
// Ahead of a batch that may yet fail, a conflict may not stand, | ||
// and nor may one with a pull stacked ahead of it in this batch, | ||
// which may yet fail too: the pull waits to be built on the | ||
// branch itself, after them, where a conflict is the branch's. | ||
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); | ||
if !stack.touching(on.git, files).await?.is_empty() { | ||
continue; | ||
} | ||
let why = conflicted(on.branch, &on_tip, files); | ||
self.eject(on, line, item, Some(head), &why, &[]).await?; | ||
} | ||
if stack.stacked.is_empty() { | ||
| @@ -557,16 +558,22 @@ impl Runner { | ||
}; | ||
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); | ||
let since = item.queue.as_ref().map_or("", |q| q.created_at.as_str()); | ||
let stands = preflight(&first, &runs, tip, since, me); | ||
if !stands.failed.is_empty() { | ||
let failed: Vec<String> = (stands.failed.iter()) | ||
.map(|c| format!("{} ({})", c.record.workflow, c.record.status)) | ||
.collect(); | ||
// Named by the tip it failed on, which the branch may have | ||
// moved past since. | ||
let on_tip = (stands.failed.iter()) | ||
.find_map(|c| c.record.base_commit.as_deref()) | ||
.unwrap_or(tip); | ||
let why = format!( | ||
"its pre-flight failed at its head {}, on {} at {}: {}", | ||
short(&sha), | ||
on.branch, | ||
short(tip), | ||
short(on_tip), | ||
failed.join(", "), | ||
); | ||
let checks: Vec<FailedCheck> = (stands.failed.iter()) | ||
| @@ -861,17 +868,20 @@ mod tests { | ||
}; | ||
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<_>>(); | ||
// Queued after those runs, unless said otherwise. | ||
let later = "2026-02-01T00:00:00.000Z"; | ||
// Passed on any tip, it is ready. | ||
let stands = preflight( | ||
&first(&["a"]), | ||
&[run("a", "success", "old")], | ||
"tip", | ||
later, | ||
"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"); | ||
let stands = preflight(&first(&["a"]), &runs, "tip", later, "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. | ||
| @@ -880,7 +890,7 @@ mod tests { | ||
run("b", "error", "old"), | ||
run("c", "cancelled", "tip"), | ||
]; | ||
let stands = preflight(&first(&["a", "b", "c", "d"]), &runs, "tip", "did:ci"); | ||
let stands = preflight(&first(&["a", "b", "c", "d"]), &runs, "tip", later, "did:ci"); | ||
let failed: Vec<&str> = stands | ||
.failed | ||
.iter() | ||
| @@ -888,23 +898,35 @@ mod tests { | ||
.collect(); | ||
assert_eq!(failed, ["a"]); | ||
assert_eq!(stands.missing, first(&["b", "c", "d"])); | ||
// Failed since it was queued, on another tip, it is failed too: the | ||
// same head would fail the same way on the branch as it moves. | ||
let runs = [run("a", "failure", "old")]; | ||
let stands = preflight( | ||
&first(&["a"]), | ||
&runs, | ||
"tip", | ||
"2025-12-01T00:00:00.000Z", | ||
"did:ci", | ||
); | ||
assert_eq!(stands.failed.len(), 1); | ||
assert!(stands.missing.is_empty()); | ||
// 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); | ||
assert!(preflight(&first(&["a"]), &runs, "tip", later, "did:ci").running); | ||
let runs = [by("did:other", "a", "queued", "tip")]; | ||
let stands = preflight(&first(&["a"]), &runs, "tip", "did:ci"); | ||
let stands = preflight(&first(&["a"]), &runs, "tip", later, "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); | ||
assert!(preflight(&first(&["a"]), &runs, "tip", later, "did:ci").running); | ||
} | ||
| ||
#[test] | ||
fn a_conflict_with_the_branch_itself_says_so() { | ||
let why = conflicted("main", &"a".repeat(40), &["f".into(), "g".into()], &[]); | ||
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" | ||
crates/knotless-run/tests/run.rs+22−11
| @@ -2968,6 +2968,16 @@ fn a_merge_queue_lands_in_batches() { | ||
q.branch("c", "c", "c\n", false); | ||
let c = q.ok(&["pull", &url, "main", &url, "c", "c", "-m", &fixes]); | ||
q.queue(&[&a, &b, &c]); | ||
// Each queued names the head it was queued at. | ||
wait_until(|| { | ||
let listed = q.ok(&["queue", &url, "--json"]); | ||
let listed: Vec<serde_json::Value> = serde_json::from_str(&listed).unwrap(); | ||
let queued = listed.iter().find(|w| w["uri"] == a.as_str()); | ||
queued | ||
.and_then(|w| w["queue"]["commit"].as_str()) | ||
.map(str::len) | ||
== Some(40) | ||
}); | ||
rt.block_on(stranger.reconcile(&watches[0])).unwrap(); | ||
assert!(!rt.block_on(stranger.noticed(&watches[0])).unwrap()); | ||
l.pass(&rt); | ||
| @@ -3003,27 +3013,28 @@ fn a_merge_queue_lands_in_batches() { | ||
.to_string(); | ||
assert!(check_on(&q.owner, &q.owner, &key, &a, "ci").is_null()); | ||
| ||
// One that conflicts with those ahead of it is ejected at once, naming | ||
// where; one that breaks the branch is found by halving the batch. | ||
// One that conflicts with a pull ahead of it waits, as that one may | ||
// fail; one that breaks the branch is found by halving the batch. | ||
let d = q.pull("d", "f", "d\n"); | ||
let e = q.pull("e", "f", "e\n"); | ||
let x = q.pull("x", "bad", "x\n"); | ||
let g = q.pull("g", "g", "g\n"); | ||
q.queue(&[&d, &e, &x, &g]); | ||
l.pass(&rt); | ||
let shown = q.shows(&e, "merge queue: ejected"); | ||
assert!( | ||
shown.contains(&format!( | ||
"it conflicts in f with {d} (d), queued ahead of it on main at " | ||
)), | ||
"{shown}" | ||
); | ||
assert!(shown.contains(". Merge that branch into it"), "{shown}"); | ||
let shown = q.ok(&["show", &url, &e]); | ||
assert!(!shown.contains("merge queue: ejected"), "{shown}"); | ||
// [d, x, g] fails, then [d, x]; [d] passes and lands, so x alone is | ||
// what broke it, and is ejected without running again; then g lands. | ||
// what broke it, and is ejected without running again; then g lands, | ||
// and e, built on the branch with d in it, conflicts with the branch. | ||
for _ in 0..4 { | ||
l.pass(&rt); | ||
} | ||
let shown = q.shows(&e, "merge queue: ejected"); | ||
assert!(shown.contains("it does not apply on main at "), "{shown}"); | ||
assert!( | ||
shown.contains("it conflicts in f. Merge main into it"), | ||
"{shown}" | ||
); | ||
let shown = q.shows(&x, "merge queue: ejected"); | ||
assert!(shown.contains("Halving the batch found it"), "{shown}"); | ||
assert!(shown.contains("\nlogs: knotless checks "), "{shown}"); | ||
docs/design/merge-queue.md+20−11
| @@ -35,7 +35,9 @@ it goes in the writer's repo inside the repository's space; see | ||
[access.md](access.md). | ||
| ||
A record names the repository (`repo`), the pull by its AT-URI | ||
(`subject`), a `state` and `createdAt`. Its key is a TID, a time-ordered | ||
(`subject`), a `state` and `createdAt`. A `queued` one also names the | ||
pull's head as it was queued (`commit`), so the history says whether a | ||
pull queued again after an ejection had changed. Its key is a TID, a time-ordered | ||
record key, so records sort by when they were written. | ||
| ||
| State | Written by | Meaning | | ||
| @@ -163,12 +165,13 @@ whatever its protection rule. | ||
| ||
A pull whose head branch is gone is ejected. A pull that does not apply on | ||
what is stacked before it is left out, with the paths it conflicts in, and | ||
the rest of the batch goes on. When the batch is built on the branch | ||
itself, the conflicting pull is ejected at its head, with a reason that | ||
names the files and asks the author to merge or rebase and queue again. | ||
When pulls stacked ahead of it change those files, the reason names | ||
them and their branches instead: merge that branch in, or wait for it | ||
to land and merge the branch. | ||
the rest of the batch goes on. When it conflicts with the branch itself, | ||
in files no pull stacked ahead of it changes, it is ejected at its head, | ||
with a reason that names the files and asks the author to merge or rebase | ||
and queue again. When pulls stacked ahead of it change those files, it | ||
stays queued instead: those pulls may yet fail, and if they land, the | ||
next batch is built on the branch with them in it, where the conflict is | ||
the branch's and ejects it. | ||
If every pull is left out, the runner builds from the next ones. | ||
| ||
Details: | ||
| @@ -204,8 +207,8 @@ for another runner's that is only `queued`, which counts as none: | ||
| --- | --- | | ||
| `success`, on any tip | counts it passed | | ||
| `running`, or its own `queued` | waits | | ||
| `failure` or `error`, on the branch's tip | ejects the pull | | ||
| none, `cancelled`, or failed on another tip | runs it here, and waits | | ||
| `failure` or `error`, on the branch's tip or since the pull was queued | ejects the pull | | ||
| none, `cancelled`, or failed on another tip before it was queued | runs it here, and waits | | ||
| ||
A run is filed `queued` as soon as its runner takes it up, and may wait | ||
long for a job there: a runner of pulls with two jobs that takes up ten | ||
| @@ -225,8 +228,14 @@ behind it. | ||
A pass on an earlier tip is enough, because the batch checks the pull on | ||
the branch as it is anyway. Requiring the tip would run the pre-flight | ||
again each time the branch moves, the delay a pre-flight exists to cut. A | ||
failure ejects only on the current tip: what failed against an older one | ||
the branch may have fixed since. | ||
failure from before the pull was queued ejects only on the current tip: | ||
what failed against an older one the branch may have fixed since. One | ||
since it was queued ejects on whatever tip it ran on. A pre-flight takes | ||
minutes, and a busy branch moves in that time, so were it run again on | ||
each new tip, the same head would fail the same way each time, and its | ||
author would hear of it only once the branch stood still for a run. | ||
Queued again at the same head, it runs again once the branch has moved | ||
from the tip it failed on; queueing refuses it until then. | ||
| ||
A pre-flight may also check less than the batch does, for speed. | ||
knotless's own `test` checks the same: a pull's run tests against the | ||
lexicons/dev/knotless/queue.json+1−1
| @@ -15,7 +15,7 @@ | ||
"subject": { "type": "string", "format": "at-uri", "description": "The dev.knotless.pull queued, taken out or turned away." }, | ||
"state": { "type": "string", "maxLength": 64, "knownValues": ["queued", "left", "ejected"], "description": "`queued` puts the pull at the back of the queue; `left` takes it out; `ejected` is the runner turning it away, saying why in `reason`." }, | ||
"reason": { "type": "string", "maxLength": 4096, "maxGraphemes": 1000, "description": "For `ejected`, why: the paths it conflicts in, or the checks that failed and where to read their logs." }, | ||
"commit": { "type": "string", "minLength": 40, "maxLength": 64, "description": "For `ejected`, the hex object ID of the pull's head that was turned away." }, | ||
"commit": { "type": "string", "minLength": 40, "maxLength": 64, "description": "The hex object ID of the pull's head: for `queued`, the one queued; for `ejected`, the one turned away." }, | ||
"checks": { | ||
"type": "array", | ||
"maxLength": 64, | ||