johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! A repository's merge queues (`knotless::queue`): what is queued on each
//! branch, in the order it lands, the batch being checked, what was ejected
//! and why, and the batches landed last. The pull page shows where its pull
//! stands from the same.
use super::repo::found;
use super::*;
use knotless::checks::Check as Run;
use knotless::queue::{self, Batch, Typical, Waiting};
/// Batches landed that the page shows, newest first.
const LANDED: usize = 5;
/// `dev.knotless.getQueue`'s answer: a repository's merge queues.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Queue {
/// Each branch with pulls queued, by name, its pulls in the order they
/// land.
branches: Vec<QueueBranch>,
/// The open pulls the queue turned away, newest first.
ejected: Vec<EjectedPull>,
/// The last few batches landed, newest first.
landed: Vec<LandedBatch>,
/// How long its merge queue typically takes.
#[serde(skip_serializing_if = "Option::is_none")]
typical: Option<Typical>,
}
/// A branch's merge queue.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct QueueBranch {
branch: String,
pulls: Vec<QueuedPull>,
/// The batches being checked or landing, in the order they land: a
/// pull names the one it is in.
batches: Vec<QueueBatch>,
}
/// A pull in a merge queue.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct QueuedPull {
pull: ItemView,
queued_at: String,
/// Where it stands, in words: `Queued, 3rd of 5, running its checks`,
/// `In a batch of 4 being checked` or `Landing`.
said: String,
/// Where it stands, as `knotless::queue::stage` says.
stage: String,
/// Until it is in a batch, the newest run of each of its pre-flights.
checks: Vec<CheckView>,
/// The batch it is in, by its place in its branch's.
#[serde(skip_serializing_if = "Option::is_none")]
batch: Option<usize>,
}
/// A batch being checked, or landing once its checks have passed.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct QueueBatch {
/// The commit it would leave the branch at.
commit: String,
/// `checking` or `landing`.
state: String,
/// How many pulls are in it.
size: usize,
started_at: String,
checks: Vec<CheckView>,
}
/// A pull the merge queue turned away.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EjectedPull {
pull: ItemView,
ejected_at: String,
/// Why, as the runner said.
#[serde(skip_serializing_if = "Option::is_none")]
reason: Option<String>,
/// The checks of its batch that did not pass, if they are why.
checks: Vec<CheckView>,
}
/// A batch the merge queue landed.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct LandedBatch {
branch: String,
/// The commit it left the branch at.
commit: String,
#[serde(skip_serializing_if = "Option::is_none")]
landed_at: Option<String>,
/// Its pulls, in the order they were squashed onto the branch.
pulls: Vec<ItemView>,
}
/// `dev.knotless.defs#queueStateView`: where an open pull stands in its
/// branch's merge queue.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct InQueue {
/// `queued` or `ejected`.
state: String,
/// Why it was ejected.
#[serde(skip_serializing_if = "Option::is_none")]
reason: Option<String>,
created_at: String,
/// Where it stands, in words: `Queued, 3rd of 5, running its checks`,
/// `In a batch of 4 being checked`, `Landing`, or why it was ejected.
said: String,
/// While queued, where it stands, as `knotless::queue::stage` says.
#[serde(skip_serializing_if = "String::is_empty")]
stage: String,
/// While queued, how long its merge queue typically takes.
#[serde(skip_serializing_if = "Option::is_none")]
typical: Option<Typical>,
/// While its batch is being checked or landing, when that started.
#[serde(skip_serializing_if = "Option::is_none")]
batch_started_at: Option<String>,
/// Queued in no batch, when the newest batch being checked ahead of it
/// started.
#[serde(skip_serializing_if = "Option::is_none")]
ahead_started_at: Option<String>,
/// While its batch is being checked or landing, the batch's checks;
/// queued in no batch, its pre-flights' newest runs; once ejected by
/// checks that failed, those.
checks: Vec<CheckView>,
}
/// The runs of the batches of the merge queues `items` were ever in, and
/// of the heads of the pulls queued, as the index has them from `repo`'s
/// runners, newest first: none from a server with no index.
fn runs(app: &App, repo: &Repo, items: &[Item]) -> Vec<Run> {
let refnames: Vec<String> = queue::branches(items)
.iter()
.map(|base| format!("refs/heads/{base}"))
.collect();
let batches = refnames.iter().map(|refname| CheckFilter {
refname: Some(refname),
batch: true,
..CheckFilter::default()
});
let heads = queue::queued(items).into_iter().map(|i| CheckFilter {
pull: Some(&i.uri),
..CheckFilter::default()
});
let mut all = Vec::new();
for filter in batches.chain(heads) {
let found = app.index.as_ref().and_then(|index| {
let record = repo.record(app)?;
let runners: Vec<String> = record.runners.into_iter().map(|r| r.did).collect();
index
.checks(&repo.did, &repo.key, &runners, filter, 100, None)
.ok()
});
let entries = found.map(|(entries, _)| entries).unwrap_or_default();
all.extend(entries.into_iter().filter_map(|e| {
Some(Run {
record: serde_json::from_value(e.value).ok()?,
uri: e.uri,
runner: e.author,
log_text: None,
space: None,
})
}));
}
all.sort_by(|a, b| b.record.started_at.cmp(&a.record.started_at));
all
}
/// The checks among `runs` that `uris` name, as the pages show them.
fn shown_checks<'a>(
repo: &RepoViewBasic,
runs: &[Run],
uris: impl IntoIterator<Item = &'a str>,
) -> Vec<CheckView> {
let found = uris.into_iter().filter_map(|uri| {
let run = runs.iter().find(|r| r.uri == uri)?;
check_view(repo, uri.to_string(), run.record.clone())
});
newest(found.collect())
}
fn batch_checks(repo: &RepoViewBasic, runs: &[Run], batch: &Batch) -> Vec<CheckView> {
shown_checks(repo, runs, batch.checks.iter().map(|c| c.uri.as_str()))
}
fn preflight_checks(repo: &RepoViewBasic, runs: &[Run], w: &Waiting) -> Vec<CheckView> {
shown_checks(repo, runs, w.preflights.iter().map(|c| c.uri.as_str()))
}
fn failed_checks(repo: &RepoViewBasic, runs: &[Run], item: &Item) -> Vec<CheckView> {
let named = item.queue.iter().flat_map(|q| &q.checks);
shown_checks(repo, runs, named.filter_map(|c| c.check.as_deref()))
}
/// The merge queues of `repo`, with everything said of its pulls.
struct Queues {
items: Vec<Item>,
runs: Vec<Run>,
waiting: Vec<Waiting>,
typical: Option<Typical>,
}
impl Queues {
async fn of(app: &App, repo: &Repo) -> Result<Queues> {
let items = assembled(app, repo).await?.items.clone();
let runs = runs(app, repo, &items);
let record = repo.record(app).unwrap_or_default();
let waiting = queue::waiting(&items, &runs, &record, &repo.did);
let typical = queue::typical(&runs);
Ok(Queues {
items,
runs,
waiting,
typical,
})
}
}
#[derive(Deserialize)]
pub struct QueueParams {
repo: String,
}
/// `dev.knotless.getQueue`.
pub(super) async fn get_queue(
State(app): State<Arc<App>>,
Query(p): Query<QueueParams>,
) -> Result<Response> {
let repo = found(&app, &p.repo).await?;
let Queues {
items,
runs,
waiting,
typical,
} = Queues::of(&app, &repo).await?;
let shown = repo.basic(&app).await;
let record = repo.record(&app).unwrap_or_default();
let mut branches: Vec<QueueBranch> = Vec::new();
let mut ejected = Vec::new();
for w in &waiting {
let pull = item_view(&app, &record, &w.item).await;
let q = w.item.queue.clone().expect("a pull waiting was queued");
let base = match &w.place {
Some(place) => place.base.clone(),
None if w.stage == queue::stage::STACKED => queue::lands_on(&items, &w.item),
None => {
ejected.push(EjectedPull {
pull,
ejected_at: q.created_at,
reason: q.reason,
checks: failed_checks(&shown, &runs, &w.item),
});
continue;
}
};
// Pulls into different branches interleave in the order they were
// queued, so each joins its branch's line wherever that is.
let at = branches.iter().position(|b| b.branch == base);
let at = at.unwrap_or_else(|| {
branches.push(QueueBranch {
branch: base.clone(),
pulls: Vec::new(),
batches: Vec::new(),
});
branches.len() - 1
});
let line = &mut branches[at];
let batch = w.batch.as_ref().map(|b| {
let at = line.batches.iter().position(|s| s.commit == b.commit);
at.unwrap_or_else(|| {
line.batches.push(QueueBatch {
commit: b.commit.clone(),
state: b.state.clone(),
size: b.pulls.len(),
started_at: b.started_at.clone(),
checks: batch_checks(&shown, &runs, b),
});
line.batches.len() - 1
})
});
line.pulls.push(QueuedPull {
pull,
queued_at: q.created_at,
said: w.said(),
stage: w.stage.clone(),
checks: preflight_checks(&shown, &runs, w),
batch,
});
}
branches.sort_by(|a, b| a.branch.cmp(&b.branch));
ejected.sort_by(|a, b| b.ejected_at.cmp(&a.ejected_at));
let batches = queue::batches(&runs);
let mut landed = Vec::new();
for b in queue::landed(&items, &batches).into_iter().take(LANDED) {
let mut pulls = Vec::new();
for p in &b.pulls {
let item = items.iter().find(|i| i.uri == p.pull);
pulls.push(item_view(&app, &record, item.expect("a pull landed is known")).await);
}
landed.push(LandedBatch {
branch: b.branch.trim_start_matches("refs/heads/").to_string(),
commit: b.commit.clone(),
landed_at: b.finished_at.clone(),
pulls,
});
}
Ok(answer(Queue {
branches,
ejected,
landed,
typical,
}))
}
/// Where `item` stands in `repo`'s merge queue, if it is open and was
/// queued or ejected: as the queue is read again, which may have moved on
/// since `item` was.
pub(super) async fn in_queue(app: &App, repo: &Repo, item: &Item) -> Result<Option<InQueue>> {
let queued = item.queue.as_ref().filter(|q| q.state != "left");
if queued.is_none() || item.state != "open" {
return Ok(None);
}
let Queues {
runs,
waiting,
typical,
..
} = Queues::of(app, repo).await?;
let shown = repo.basic(app).await;
let w = waiting.into_iter().find(|w| w.item.uri == item.uri);
Ok(w.map(|w| {
let checks = match (&w.batch, &w.place) {
(Some(b), _) => batch_checks(&shown, &runs, b),
(None, Some(_)) => preflight_checks(&shown, &runs, &w),
(None, None) => failed_checks(&shown, &runs, &w.item),
};
let q = w.item.queue.clone().expect("a pull waiting was queued");
InQueue {
said: w.said(),
typical: typical.filter(|_| w.place.is_some()),
batch_started_at: w.batch.as_ref().map(|b| b.started_at.clone()),
ahead_started_at: w.ahead.clone(),
stage: w.stage,
state: q.state,
reason: q.reason,
created_at: q.created_at,
checks,
}
}))
}