chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
knotless-run: a queue batches a pull once its pre-flights pass
.changeset/queue-wakes-on-preflight.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# A merge queue batches a pull as soon as its pre-flights pass | ||
| ||
A runner landing a merge queue now looks again the moment a pull's run ends, on it or on another runner the repository names. A pull whose last pre-flight passed used to wait for the runner's next sweep, up to 10 minutes, before going into a batch. | ||
crates/knotless-run/src/lib.rs+33−0
| @@ -2311,6 +2311,10 @@ impl Runner { | ||
if stopping && batch.outcome().is_some() { | ||
report(&watch, runner.land_passed(&watch).await); | ||
} | ||
} else if subject.pull.is_some() && !stopping { | ||
// A pull's run may be a pre-flight its queue waits on: | ||
// the queue looks again now, not at the next sweep. | ||
runner.ring(&watch); | ||
} | ||
report(&watch, ran.map(drop)); | ||
let mut going = runner.going.lock().unwrap(); | ||
| @@ -4325,6 +4329,14 @@ fn ring(doors: &Doors, heads: &Heads, event: &jetstream::Event) { | ||
nsid::PULL | nsid::STATE | nsid::QUEUE => { | ||
op.record.as_ref().and_then(pull_into).into_iter().collect() | ||
} | ||
// A pull's run another runner ended may be a pre-flight the | ||
// queue waits on. Only its end rings, not each step it files. | ||
nsid::CHECK => (op.record.as_ref()) | ||
.filter(|r| r["pull"].is_string()) | ||
.filter(|r| r["status"].as_str().is_some_and(|s| !status::going(s))) | ||
.and_then(pull_into) | ||
.into_iter() | ||
.collect(), | ||
// A repository's record rings each of its owner's watched, | ||
// for one that no longer names this runner, or is gone. | ||
nsid::REPO => doors.keys().filter(|(d, _)| d == did).cloned().collect(), | ||
| @@ -5367,6 +5379,27 @@ mod tests { | ||
let queue = vec![op(nsid::QUEUE, "q", Some(queued))]; | ||
ring(&doors, &heads, &event(Some("did:plc:b"), queue)); | ||
assert!(rung()); | ||
// A pull's run that ended rings, as the queue may wait on it; one | ||
// still going, or a run not of a pull, does not. | ||
let check = |status: &str, of_pull: Option<&str>| { | ||
let mut record = pull("at://did:plc:a/dev.knotless.repo/r", "k1"); | ||
record["status"] = status.into(); | ||
if let Some(uri) = of_pull { | ||
record["pull"] = uri.into(); | ||
} | ||
event(Some("did:plc:r"), vec![op(nsid::CHECK, "c", Some(record))]) | ||
}; | ||
let of = Some("at://did:plc:b/dev.knotless.pull/p"); | ||
for quiet in [ | ||
check(status::QUEUED, of), | ||
check(status::RUNNING, of), | ||
check(status::SUCCESS, None), | ||
] { | ||
ring(&doors, &heads, &quiet); | ||
} | ||
assert!(!rung()); | ||
ring(&doors, &heads, &check(status::FAILURE, of)); | ||
assert!(rung()); | ||
// A repository record rings its owner's, and no one else's. | ||
let repo = |did| event(Some(did), vec![op(nsid::REPO, "r", None)]); | ||
ring(&doors, &heads, &repo("did:plc:b")); | ||
docs/design/merge-queue.md+4−1
| @@ -446,7 +446,10 @@ A pass starts when something rings the repository's door: | ||
- Every `--notice` seconds (30 by default) the runner makes the fresh | ||
reads and rings if anything changed. A queued pull is taken in within | ||
`--notice`, whatever Jetstream brings. | ||
- Each batch run's end rings, so the queue goes on without an event. | ||
- Each batch run's end rings, so the queue goes on without an event. So | ||
does each end of a pull's run, here or, as a `dev.knotless.check` | ||
record that Jetstream brings, on another runner: a pull whose last | ||
pre-flight passes goes into a batch at once, not at the next sweep. | ||
- A failed pass rings again after `--sweep` seconds or 60, whichever is | ||
less, and `--sweep` rings every | ||
door regardless. | ||
docs/design/runner.md+5−0
| @@ -83,6 +83,8 @@ A frame rings a watched repository's door when it changes: | ||
- a `dev.knotless.pull` or `dev.knotless.state` record naming the | ||
repository, so a new, reopened or retargeted pull runs; | ||
- a `dev.knotless.queue` record, for the merge queue; | ||
- a `dev.knotless.check` record of a pull's run that has ended, on any | ||
runner, for a merge queue waiting on that pull's pre-flights; | ||
- a `dev.knotless.repo` record of the owner's, which may no longer name | ||
the runner (see [Repositories that name it](#repositories-that-name-it)). | ||
| ||
| @@ -101,6 +103,8 @@ rings doors in other ways too: | ||
not runs. | ||
- A pass that fails, say because the PDS is down, rings its own door | ||
again after `min(--sweep, 60 s)`. | ||
- A run of its own ends: a batch's, or a pull's, which may be a | ||
pre-flight the merge queue waits on. | ||
- With `--queue`, a change to the merge queue's records, re-read every | ||
`--notice` seconds (30), rings the door. | ||
- A private repository's space written to rings its door, as the | ||
| @@ -117,6 +121,7 @@ flowchart TD | ||
notice[queue records changed, every --notice seconds] --> ring | ||
poll[a private repo's space written to: the appview says, or every --poll seconds] --> ring | ||
retry[a failed pass, after at most 60 s] --> ring | ||
ended[a batch's or a pull's run ended] --> ring | ||
ring[ring the repo's door] --> pass[reconcile pass] | ||
pds[(owner's PDS: repo record, refs, reruns, cancels)] --> pass | ||
appview[(appview: open pulls, merge queue)] --> pass | ||