~ 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

A private repository's merge queue lands through the gateway

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:queue/private-landing
crates/knotless-run/src/queue.rs+45−5
View file
@@ -24,6 +24,10 @@
//! moving any other way, or a pull in a batch leaving the queue, drops the
//! batches going, to be built again. A pull whose head moves while its
//! batch runs is caught as the batch lands, which then drops what is going.
//! A batch that passes and still fails to land is built again, and run, to
//! land on the next try; the same batch failing to land [`LANDING_TRIES`]
//! times in a row has its pulls ejected, saying why, rather than run for
//! ever; and refused by the gateway, at once.
//!
//! 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
@@ -71,6 +75,18 @@ fn refused(err: &anyhow::Error) -> Option<String> {
))
}
/// How many times in a row the same batch, on the same tip with the same
/// heads, fails to land before its pulls are ejected.
pub(crate) const LANDING_TRIES: u32 = 3;
/// Why the pulls of a batch that failed to land `tries` times in a row,
/// last with `err`, are ejected, once that is [`LANDING_TRIES`].
fn stuck(tries: u32, err: &anyhow::Error) -> Option<String> {
(tries >= LANDING_TRIES).then(|| {
format!("its batch passed its checks but failed to land {tries} times in a row: {err:#}")
})
}
/// 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)]
@@ -148,9 +164,28 @@ pub(crate) struct Line {
failed_heads: HashMap<String, String>,
done: HashMap<String, String>,
flown: HashMap<String, String>,
/// The batch that last failed to land, by its tip and each pull's
/// head, and how many times in a row it has.
unlanded: Option<(Vec<String>, u32)>,
}
impl Line {
/// Counts a failure of `stack` to land: how many times in a row it
/// has failed, the same batch on the same tip.
fn unlanded(&mut self, stack: &Stack) -> u32 {
let heads = stack
.stacked
.iter()
.map(|s| format!("{} {}", s.pull.uri, s.head));
let batch: Vec<String> = std::iter::once(stack.base.clone()).chain(heads).collect();
let tries = match &self.unlanded {
Some((last, tries)) if *last == batch => tries + 1,
_ => 1,
};
self.unlanded = Some((batch, tries));
tries
}
/// 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()))
@@ -317,7 +352,8 @@ impl Runner {
let idle = line.flights.is_empty()
&& line.suspects.is_empty()
&& line.done.is_empty()
&& line.flown.is_empty();
&& line.flown.is_empty()
&& line.unlanded.is_none();
if !idle {
self.lines.lock().unwrap().insert(id, line);
}
@@ -366,21 +402,25 @@ impl Runner {
if failed.is_empty() {
on.watch.at("landing a batch");
if let Err(err) = self.land(on, &flight).await {
let refused = refused(&err);
let tries = line.unlanded(&flight.stack);
let why = refused(&err).or_else(|| stuck(tries, &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 {
// Refused, it would be refused again, and failing as
// often, it may fail for ever: its pulls go, rather
// than be checked and fail to land for ever.
if let Some(why) = &why {
for s in &flight.stack.stacked {
self.eject(on, line, &s.pull, Some(&s.head), why, &[])
.await?;
}
line.suspects.clear();
line.unlanded = None;
}
break;
}
line.unlanded = None;
tip = flight.stack.tip().to_string();
let landed = flight.stack.stacked.len();
telemetry::batch(
crates/knotless-run/tests/run.rs+34−2
View file
@@ -3339,6 +3339,8 @@ struct Line {
/// The appview, and its following of the PDS.
base: String,
ears: tokio::task::JoinHandle<()>,
/// Where the appview keeps its config: the sessions owners delegate.
server: Sandbox,
}
impl Line {
@@ -3349,7 +3351,8 @@ impl Line {
/// A [`Line`] whose appview is a gateway too if it names `runner`.
fn served(rt: &tokio::runtime::Runtime, runner: Option<&str>) -> Line {
let mut owner = Sandbox::new();
let (base, ears) = serving(rt, &Sandbox::new(), runner);
let server = Sandbox::new();
let (base, ears) = serving(rt, &server, runner);
owner.set("KNOTLESS_APPVIEW", &base);
let handle = owner.account.handle.clone();
owner.set("KNOTLESS_IDENTIFIER", &handle);
@@ -3374,6 +3377,7 @@ impl Line {
go,
base,
ears,
server,
};
line.ok(&["runner", "add", &line.url, &handle]);
line
@@ -3882,7 +3886,9 @@ fn a_stack_lands_in_turn() {
/// A runner the repository names that may not push lands its queue
/// through the appview's gateway, which takes a batch from it only once the
/// owner has delegated writes there: until then, each batch it would land
/// is ejected, saying so, rather than checked and refused for ever.
/// is ejected, saying so, rather than checked and refused for ever. One
/// the gateway takes but fails to land, the same batch three times in a
/// row, is ejected too, saying why.
#[test]
fn a_runner_named_lands_through_the_gateway() {
let rt = tokio::runtime::Runtime::new().unwrap();
@@ -3925,6 +3931,32 @@ fn a_runner_named_lands_through_the_gateway() {
shown.contains("under Delegated writes in their settings"),
"{shown}"
);
// Delegated, as the gateway keeps the owner's session; but it cannot
// read the queue, having no appview of its own, so each landing fails.
let server = common::ctx(&q.server);
let (who, password) = (&q.owner.account.handle, &q.owner.account.password);
let held = rt.block_on(knotless::session::login(&server, who, password));
knotless::session::save(&server, &held.unwrap()).unwrap();
q.queue(&[&a]);
let ejected = || {
q.ok(&["pull", "show", &url, &a])
.contains("merge queue: ejected")
};
wait_until(|| !ejected());
let mut passes = 0;
while !ejected() {
assert!(passes < 12, "never ejected");
l.pass(&rt);
passes += 1;
}
assert_eq!(q.main(), start);
let shown = q.ok(&["pull", "show", &url, &a]);
assert!(
shown.contains("its batch passed its checks but failed to land 3 times in a row: "),
"{shown}"
);
assert!(shown.contains("KNOTLESS_APPVIEW is set empty"), "{shown}");
}
#[test]
crates/knotless-serve/src/gateway.rs+15−4
View file
@@ -481,6 +481,16 @@ pub(crate) fn grant_behind(app: &App, did: &str) -> bool {
kept.is_some_and(|s| oauth::behind(&s, gateway::SCOPE))
}
/// The gateway's own context, reading as `owner`, whose session it keeps
/// as they delegated: what decides a push reads a private repository's
/// space as its owner, since nobody is logged in here.
fn as_owner(app: &App, owner: &str) -> knotless::env::Ctx {
let mut ctx = app.ctx.clone();
ctx.env.identifier = Some(owner.to_string());
ctx.env.password = None;
ctx
}
/// The repository, as its owner, for a caller who may write to it.
async fn writable(
app: &App,
@@ -616,19 +626,20 @@ pub async fn push(
// past the gateway. A collaborator's merge is held to the merge gate,
// read as the repository is now: as `writer` read it, under this
// push's turn, so it is not read again, for the gate or a runner's
// landing.
// landing. Read as the owner, whose space a private one is in.
let ctx = as_owner(&app, remote.did());
let record = snap.repo.as_ref().map(|r| &r.value);
let reviewed = |r: &RepoRecord| r.rules.iter().any(|x| x.reviews.is_some());
let gated = record.is_some_and(|r| !r.runners.is_empty() || reviewed(r));
let gated = gated && caller != remote.did();
let runner = matches!(writer, Writer::Runner);
let opened = match runner || gated {
true => Some(Target::read(&app.ctx, remote.clone(), snap.clone()).await),
true => Some(Target::read(&ctx, remote.clone(), snap.clone()).await),
false => None,
};
let opened = opened.transpose().map_err(AppError::upstream)?;
if let (true, Some(target)) = (runner, &opened) {
let refusal = queue::landing_refusal(&app.ctx, target, &mirror.git, &updates).await;
let refusal = queue::landing_refusal(&ctx, target, &mirror.git, &updates).await;
if let Some(why) = refusal.map_err(AppError::internal)? {
return Err(forbidden(format!(
"{caller} runs {}'s checks, and lands its merge queue here, nothing else: {why}",
@@ -642,7 +653,7 @@ pub async fn push(
let refusal = gateway::refusal(&mirror.git, &snap, &update, remote.did(), &caller).await;
let mut refusal = refusal.map_err(AppError::internal)?;
if let (None, Some(t)) = (&refusal, &target) {
let held = landing::gate_refusal(&app.ctx, t, &mirror.git, &update, &caller).await;
let held = landing::gate_refusal(&ctx, t, &mirror.git, &update, &caller).await;
refusal = held.map_err(AppError::upstream)?;
}
match refusal {
crates/knotless-serve/tests/gateway.rs+92−0
View file
@@ -1167,6 +1167,98 @@ fn runners_land_the_merge_queue_through_a_gateway() {
}
}
/// Lands, as `runner` does in the merge queue, the pulls queued in the
/// repository at `url`, through the gateway: each squashed onto the one
/// before, and the batch landed. Answers how many it landed.
fn land_queued(runner: &Sandbox, url: &str) -> usize {
let ctx = ctx(runner);
let repo = knotless::url::RepoUrl::parse(url).unwrap();
tokio::runtime::Runtime::new().unwrap().block_on(async {
let t = knotless::forge::Target::open(&ctx, &repo).await.unwrap();
let entries = t.queue_entries(&ctx).await.unwrap();
let items = knotless::forge::assemble(&t.record, t.remote.did(), &entries);
let key = t.reference().key;
let mirror = knotless::mirror::Mirror::open(&ctx, t.remote.did(), &key)
.await
.unwrap();
mirror.sync(&t.remote, &t.snap).await.unwrap();
let mut pulls = Vec::new();
for item in knotless::queue::queued(&items) {
let head = item.head.as_ref().unwrap();
let at = knotless::forge::head_of(&ctx, &mirror.git, head).await;
pulls.push((item.clone(), at.unwrap().unwrap()));
}
let me = knotless::session::me(&ctx).await.unwrap();
let tip = t.snap.find_ref("refs/heads/main").unwrap();
let base = tip.value.target.clone().unwrap();
let stack = knotless::queue::stack(&mirror.git, &base, &pulls, &me)
.await
.unwrap();
knotless::queue::land(&ctx, &repo, &mirror.git, "main", &stack)
.await
.unwrap();
stack.stacked.len()
})
}
/// A private repository's runner that may not push lands its merge queue
/// through the gateway its owner delegated to, which reads the queue in
/// the repository's space as the owner, nobody being logged in there.
/// These need a PDS on the spaces alpha, and pass trivially without one.
#[test]
fn runners_land_a_private_merge_queue_through_a_gateway() {
if !net().spaces {
eprintln!("skipped: the PDS is not on the spaces alpha");
return;
}
let gw = gateway_with(None, true);
let [mut owner, mut runner] = people();
owner.set("KNOTLESS_APPVIEW", &gw.base);
runner.set("KNOTLESS_APPVIEW", &gw.base);
let ok = |s: &Sandbox, args: &[&str]| {
let (code, out) = s.cli(args, b"");
assert_eq!(code, 0, "{args:?}: {out}");
out.trim().to_string()
};
let url = owner.url("hushed");
let handle = runner.account.handle.clone();
ok(
&owner,
&["repo", "create", &url, "--private", "--runner", &handle],
);
ok(
&owner,
&["account", "delegate", &owner.account.handle, &gw.did],
);
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
for b in ["a", "b"] {
owner.git(&work, &["checkout", "-q", "-B", b, "main"]);
owner.commit_file(&work, b, &format!("{b}\n"));
owner.git(&work, &["push", "-q", &url, b]);
}
let a = ok(&owner, &["pull", "create", &url, "main", &url, "a", "a"]);
let b = ok(&owner, &["pull", "create", &url, "main", &url, "b", "b"]);
for pull in [&a, &b] {
ok(&owner, &["pull", "merge", &url, pull, "--queue"]);
}
wait_until(|| {
let line = owner.cli(&["queue", "list", &url], b"").1;
line.contains(&a) && line.contains(&b)
});
assert_eq!(land_queued(&runner, &url), 2);
let refs = ok(&owner, &["repo", "refs", &url]);
let has = |branch: &str| {
refs.lines()
.any(|l| l.ends_with(&format!("\trefs/heads/{branch}")))
};
assert!(!has("a") && !has("b") && has("main"), "{refs}");
for pull in [&a, &b] {
wait_until(|| ok(&owner, &["pull", "show", &url, pull]).contains("merged pull"));
}
}
#[test]
fn a_server_that_is_not_a_gateway() {
let s = Sandbox::new();
docs/design/gateway.md+6−0
View file
@@ -185,6 +185,12 @@ nothing else: <why>`. It checks, against the queue as the records say:
- each branch deleted is the head of one of those pulls, in this
repository (`<name> is not the branch of a pull landed`).
The gateway reads the queue, and each pull's head, as the owner, with the
session they delegated: a private repository's are in its space, which
takes a member's credential to read, and nobody is logged in to the
gateway. The [merge gate](#protected-branches) a collaborator's push is
held to reads the repository as the owner too.
So a runner, or whoever holds its account, can land what a maintainer
queued, before or without its checks passing, and nothing more.
docs/design/merge-queue.md+9−1
View file
@@ -337,7 +337,15 @@ Each close is a `closed` state naming the pull, and within a batch an issue
is closed once, by the first pull naming it. The keyword rules are those
of any merge; see [collaboration.md](collaboration.md).
A landing that fails drops every batch, and the queue is built again.
A landing that fails drops every batch, and the queue is built again, so
a branch that moved, or a gateway down for a moment, costs a run. The same
batch, on the same tip with the same heads, failing to land three times in
a row has its pulls ejected, saying `its batch passed its checks but failed
to land 3 times in a row` and the last error, rather than checked and
failed for ever: whatever fails it is not the pulls', and each is queued
again once it is fixed. A landing the gateway refuses (403) ejects them at
once ([Who lands a queue](#who-lands-a-queue)). The count is kept in the
runner's memory, so a restart starts it again.
## When a batch fails