Menu
AkurAI-Build
publicLatest change 47406137d5449e4389c491ac106a0d89e7f06589 - merge queue: adopt the target's current tip instead of failing when it moved by Ólafur Búi Ólafsson
use std::path::Path;
use anyhow::{Context, Result, ensure};
use crate::db::Database;
use crate::git_process;
use crate::runner::Runner;
/// One pass over every repository that currently has queue work: claim the
/// next FIFO entry (if none is already running), merge, run CI, and land or
/// fail it. Returns how many entries were processed (0 = idle tick).
pub fn tick(database: &Database, runner: &Runner, hosted_root: &Path) -> Result<usize> {
let mut processed = 0usize;
for repository_id in database.repositories_with_queued_merge()? {
if process_next(database, runner, hosted_root, repository_id)? {
processed += 1;
}
}
Ok(processed)
}
fn process_next(
database: &Database,
runner: &Runner,
hosted_root: &Path,
repository_id: i64,
) -> Result<bool> {
let Some(entry) = database.next_queued_merge(repository_id)? else {
return Ok(false);
};
if !database.claim_merge_queue_entry(entry.id)? {
// Another tick (or worker) already claimed it first.
return Ok(false);
}
let pull_request_id = pull_request_id_for(database, &entry)?;
let result = execute(database, runner, hosted_root, &entry);
// Also clean a ref left by an interrupted attempt that failed before CI.
if let Ok(bare) = crate::hosted_git::repository_path(hosted_root, &entry.repository) {
cleanup_staging_ref(&bare, entry.id)?;
}
match result {
Ok((merge_commit, run_id)) => {
database.finish_merged_pull_request(
entry.id,
pull_request_id,
&merge_commit,
run_id,
)?;
}
Err(error) => {
let message = format!("{error:#}");
let bounded: String = message.chars().take(16 * 1024).collect();
database.fail_merge_queue_pull_request(entry.id, pull_request_id, &bounded, None)?;
}
}
Ok(true)
}
fn cleanup_staging_ref(bare: &Path, entry_id: i64) -> Result<()> {
ensure!(entry_id > 0, "invalid merge queue entry");
let staging_ref = format!("refs/merge-queue/{entry_id}");
let refs = git_output(
bare,
[
"for-each-ref",
"--format=%(objectname) %(refname)",
&staging_ref,
],
)?;
for line in refs.lines() {
if let Some((sha, name)) = line.split_once(' ')
&& name == staging_ref
{
// Compare-and-swap: never remove a concurrently replaced ref.
git(bare, ["update-ref", "-d", &staging_ref, sha])?;
}
}
Ok(())
}
fn pull_request_id_for(database: &Database, entry: &crate::db::MergeQueueEntry) -> Result<i64> {
Ok(database
.pull_request(&entry.repository, entry.pull_request_number)?
.pull_request
.id)
}
/// Merge `entry.head_sha` onto the current tip of the target branch in a
/// disposable worktree, push the result to a disposable ref so the normal
/// runner can check it out in isolation, run the repository's `.akurai.yml`
/// pipeline against it, and only on CI success land it onto the target
/// branch with a compare-and-swap (`--force-with-lease`) push so a
/// concurrent direct push cannot be silently clobbered. Returns the merge
/// commit SHA on success.
fn execute(
database: &Database,
runner: &Runner,
hosted_root: &Path,
entry: &crate::db::MergeQueueEntry,
) -> Result<(String, Option<i64>)> {
let repository = database.repository(&entry.repository)?;
let detail = database.pull_request(&entry.repository, entry.pull_request_number)?;
let pull_request = &detail.pull_request;
ensure!(
detail.approved,
"pull request #{} is no longer approved",
pull_request.number
);
ensure!(
pull_request.head_sha == entry.head_sha,
"pull request #{} head changed since it was enqueued",
pull_request.number
);
let bare = crate::hosted_git::repository_path(hosted_root, &entry.repository)?;
ensure!(
bare.is_dir(),
"hosted mirror for {} not found",
entry.repository
);
// A crash can occur after Git accepted the final push but before the
// transaction recording PR completion. The persisted CI run identifies
// the exact checked merge; confirm its parents and ancestry before
// recovering completion, rather than treating it as a moved target.
if let Some(run_id) = entry.run_id {
let run = database.run(run_id)?;
if run.repository_id == entry.repository_id
&& run.git_ref == format!("refs/merge-queue/{}", entry.id)
&& run.status == "succeeded"
&& database
.jobs(run_id)?
.iter()
.any(|job| job.status == "succeeded")
&& let Some(commit) = run.commit_sha
{
let parents = git_output(&bare, ["rev-list", "--parents", "-n", "1", &commit])?;
let expected = format!("{commit} {} {}", entry.base_sha, entry.head_sha);
if parents == expected
&& git_process::command(&bare)
.args([
"merge-base",
"--is-ancestor",
&commit,
&format!("refs/heads/{}", pull_request.target_ref),
])
.status()?
.success()
{
return Ok((commit, Some(run_id)));
}
}
}
let scratch = hosted_root.join("merge-queue").join(entry.id.to_string());
if scratch.exists() {
std::fs::remove_dir_all(&scratch)?;
}
std::fs::create_dir_all(&scratch)?;
let result = execute_in(
&scratch,
database,
runner,
&repository,
&bare,
entry,
pull_request,
);
let _ = std::fs::remove_dir_all(&scratch);
result
}
fn execute_in(
scratch: &Path,
database: &Database,
runner: &Runner,
repository: &crate::db::Repository,
bare: &Path,
entry: &crate::db::MergeQueueEntry,
pull_request: &crate::db::PullRequest,
) -> Result<(String, Option<i64>)> {
let bare_url = format!("file://{}", bare.display());
git(scratch, ["init"])?;
git(scratch, ["remote", "add", "origin", &bare_url])?;
git(
scratch,
["fetch", "--depth=64", "origin", &pull_request.target_ref],
)?;
let current_target = git_output(scratch, ["rev-parse", "FETCH_HEAD"])?;
// The queue is serialized per repository and CI runs on the exact merge
// commit built below, so a target that moved since enqueue is not a
// failure: adopt the current tip as this entry's base (persisted, so the
// crash-recovery ancestry check and the lease below agree with it).
let base_sha = if current_target == entry.base_sha {
entry.base_sha.clone()
} else {
database.update_merge_queue_entry_base(entry.id, ¤t_target)?;
current_target
};
git(scratch, ["checkout", "--detach", "FETCH_HEAD"])?;
git(
scratch,
["fetch", "--depth=64", "origin", &pull_request.source_ref],
)?;
let fetched_head = git_output(scratch, ["rev-parse", "FETCH_HEAD"])?;
ensure!(
fetched_head == entry.head_sha,
"pull request source ref moved since this pull request was enqueued"
);
git(
scratch,
["config", "user.email", "merge-queue@akurai.build"],
)?;
git(scratch, ["config", "user.name", "AkurAI Build merge queue"])?;
let merge_title = format!(
"Merge pull request #{} from {}",
pull_request.number, pull_request.source_ref
);
let merge = git_process::command(scratch)
.args(["merge", "--no-ff", "-m", &merge_title, &entry.head_sha])
.output()
.context("failed to run git merge")?;
ensure!(
merge.status.success(),
"merge conflict between {} and {}: {}",
pull_request.source_ref,
pull_request.target_ref,
bounded(&merge.stderr)
);
let merge_commit = git_output(scratch, ["rev-parse", "HEAD"])?;
// Publish the merge result to a disposable ref BEFORE running CI, so the
// isolated runner (which checks out from the repository's own remote,
// never from this scratch worktree) can actually see it.
let staging_ref = format!("refs/merge-queue/{}", entry.id);
git(
scratch,
[
"push",
"origin",
"--force",
&format!("{merge_commit}:{staging_ref}"),
],
)?;
let ci_outcome = run_ci(
database,
entry.id,
runner,
repository,
&staging_ref,
&merge_commit,
);
// Whatever CI decided, the staging ref is scratch-only and must not
// linger as a pushable/discoverable branch.
let _ = git_process::command(scratch)
.args(["push", "origin", "--delete", &staging_ref])
.output();
let run_id = ci_outcome?;
// Compare-and-swap push: the queue only ever reasoned about `base_sha`
// as the target tip, so refuse to land if the branch moved underneath
// us (protected-branch pushes still route only through this path).
let lease = format!(
"--force-with-lease={}:{}",
pull_request.target_ref, base_sha
);
let push = git_process::command(scratch)
.args([
"push",
"origin",
&lease,
&format!("{merge_commit}:refs/heads/{}", pull_request.target_ref),
])
.output()
.context("failed to push merge commit")?;
ensure!(
push.status.success(),
"compare-and-swap push of merge commit was rejected: {}",
bounded(&push.stderr)
);
Ok((merge_commit, Some(run_id)))
}
/// Run the repository's `.akurai.yml` pipeline against the merged tree by
/// queuing and synchronously executing a normal Run against the merge
/// commit, reusing the existing runner/job-execution/isolation machinery
/// instead of duplicating it. An empty or entirely skipped run is not CI
/// evidence and must never authorize a merge.
fn run_ci(
database: &Database,
entry_id: i64,
runner: &Runner,
repository: &crate::db::Repository,
staging_ref: &str,
merge_commit: &str,
) -> Result<i64> {
let queued = runner.queue(
&repository.name,
Some(staging_ref),
Some(merge_commit),
"webhook",
)?;
database.attach_merge_queue_run(entry_id, queued.id)?;
let detail = runner.process(queued.id)?;
ensure!(
detail.run.status == "succeeded",
"CI run {} for merge commit {merge_commit} ended {}",
detail.run.id,
detail.run.status
);
ensure!(
detail.jobs.iter().any(|job| job.status == "succeeded"),
"CI run {} did not execute any successful checks",
detail.run.id
);
Ok(detail.run.id)
}
fn bounded(bytes: &[u8]) -> String {
String::from_utf8_lossy(bytes).chars().take(4096).collect()
}
fn git<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<()> {
let output = git_process::command(cwd).args(arguments).output()?;
ensure!(
output.status.success(),
"git failed: {}",
bounded(&output.stderr)
);
Ok(())
}
fn git_output<const N: usize>(cwd: &Path, arguments: [&str; N]) -> Result<String> {
git_process::output(cwd, arguments)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::db::Database;
use crate::runner::Runner;
use std::path::PathBuf;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
fn git_ok(dir: &Path, args: &[&str]) {
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(dir)
.args(args)
.status()
.expect("spawn git")
.success(),
"git {args:?} failed"
);
}
/// Build a disposable source repo with a `.akurai.yml` pipeline, a
/// `main` branch, and a `feature-a` branch one commit ahead, host it as
/// a bare mirror, register it, and return (state paths, repo name).
fn hosted_repo_with_pipeline(work: &Path, ci_command: &str) -> Result<(PathBuf, PathBuf)> {
let source = work.join("source");
std::fs::create_dir(&source)?;
git_ok(&source, &["init", "-b", "main"]);
git_ok(&source, &["config", "user.name", "t"]);
git_ok(&source, &["config", "user.email", "t@t"]);
std::fs::write(
source.join(".akurai.yml"),
format!(
"version: 1\njobs:\n - name: check\n executor: docker\n image: debian@sha256:0d97731c59efdde181e19c4a5ec22d16e9eefcb73175598b9b7bae712c7214eb\n run: {ci_command}\n"
),
)?;
std::fs::write(source.join("f"), "1")?;
git_ok(&source, &["add", "."]);
git_ok(&source, &["commit", "-m", "init"]);
git_ok(&source, &["checkout", "-b", "feature-a"]);
std::fs::write(source.join("f"), "2")?;
git_ok(&source, &["commit", "-am", "feature work"]);
let hosted_root = work.join("hosted-root");
std::fs::create_dir(&hosted_root)?;
crate::hosted_git::host(&hosted_root, "app", &source)?;
Ok((hosted_root, source))
}
#[test]
fn cleanup_removes_only_the_exact_interrupted_queue_ref() -> Result<()> {
let work = tempfile::tempdir()?;
let (root, _) = hosted_repo_with_pipeline(work.path(), "exit 0")?;
let bare = root.join("hosted/app.git");
let head = git_output(&bare, ["rev-parse", "refs/heads/main"])?;
git(&bare, ["update-ref", "refs/merge-queue/1", &head])?;
git(&bare, ["update-ref", "refs/merge-queue/10", &head])?;
cleanup_staging_ref(&bare, 1)?;
cleanup_staging_ref(&bare, 1)?;
assert!(git_output(&bare, ["rev-parse", "--verify", "refs/merge-queue/1"]).is_err());
assert_eq!(
git_output(&bare, ["rev-parse", "refs/merge-queue/10"])?,
head
);
assert_eq!(git_output(&bare, ["rev-parse", "refs/heads/main"])?, head);
Ok(())
}
#[test]
#[ignore = "requires a Docker daemon and pinned Debian image; run with --ignored --test-threads=1"]
fn merge_queue_lands_an_approved_pull_request_when_ci_passes() -> Result<()> {
let work = tempfile::tempdir()?;
let (hosted_root, _source) = hosted_repo_with_pipeline(work.path(), "exit 0")?;
let bare = hosted_root.join("hosted/app.git");
let database = Database::memory(KEY)?;
let repository = database.add_repository("app", bare.to_str().expect("utf8"), "main")?;
let base_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
let head_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/feature-a"])?;
let pull_request = database.create_pull_request(
&repository.name,
"Add feature",
"",
"feature-a",
"main",
&head_sha,
&base_sha,
"author-sub",
"Author",
)?;
database.add_pull_request_review(
&repository.name,
pull_request.number,
"reviewer-sub",
"Reviewer",
"approved",
"ok",
)?;
database.enqueue_merge(&repository.name, pull_request.number, "reviewer-sub")?;
let runner = Runner::new(database.clone(), work.path().join("runner-root"), true)?;
let processed = tick(&database, &runner, &hosted_root)?;
assert_eq!(processed, 1, "one queue entry should have been processed");
let detail = database.pull_request(&repository.name, pull_request.number)?;
assert_eq!(
detail.pull_request.state, "merged",
"pull request should have landed: {:?}",
detail.reviews
);
assert!(detail.pull_request.merge_commit.is_some());
let new_main = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
assert_eq!(
new_main,
detail.pull_request.merge_commit.expect("merge commit"),
"target branch must actually point at the merge commit"
);
let merged_file = crate::git_process::output(&bare, ["show", "refs/heads/main:f"])?;
assert_eq!(
merged_file, "2",
"merged tree must contain the feature's change"
);
// Staging ref must not linger after landing.
let refs = crate::git_process::output(&bare, ["for-each-ref", "refs/merge-queue"])?;
assert!(refs.is_empty(), "disposable staging ref must be cleaned up");
Ok(())
}
#[test]
#[ignore = "requires Docker; run with --ignored --test-threads=1"]
fn merge_queue_recovers_after_final_push_before_database_completion() -> Result<()> {
let work = tempfile::tempdir()?;
let (hosted_root, _source) = hosted_repo_with_pipeline(work.path(), "exit 0")?;
let bare = hosted_root.join("hosted/app.git");
let database = Database::memory(KEY)?;
let repository = database.add_repository("app", bare.to_str().expect("utf8"), "main")?;
let base_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
let head_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/feature-a"])?;
let pull_request = database.create_pull_request(
&repository.name,
"Add feature",
"",
"feature-a",
"main",
&head_sha,
&base_sha,
"author-sub",
"Author",
)?;
database.add_pull_request_review(
&repository.name,
pull_request.number,
"reviewer-sub",
"Reviewer",
"approved",
"ok",
)?;
database.enqueue_merge(&repository.name, pull_request.number, "reviewer-sub")?;
let runner = Runner::new(database.clone(), work.path().join("runner-root"), true)?;
let entry = database
.next_queued_merge(repository.id)?
.expect("queued merge");
database.claim_merge_queue_entry(entry.id)?;
execute(&database, &runner, &hosted_root, &entry)?;
assert_eq!(
database
.pull_request(&repository.name, pull_request.number)?
.pull_request
.state,
"queued"
);
database.recover_merge_queue()?;
let processed = tick(&database, &runner, &hosted_root)?;
assert_eq!(
database.runs(Some(&repository.name), 10)?.len(),
1,
"recovery must not rerun CI"
);
assert_eq!(processed, 1, "one queue entry should have been processed");
let detail = database.pull_request(&repository.name, pull_request.number)?;
assert_eq!(
detail.pull_request.state, "merged",
"pull request should have landed: {:?}",
detail.reviews
);
assert!(detail.pull_request.merge_commit.is_some());
let new_main = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
assert_eq!(
new_main,
detail.pull_request.merge_commit.expect("merge commit"),
"target branch must actually point at the merge commit"
);
let merged_file = crate::git_process::output(&bare, ["show", "refs/heads/main:f"])?;
assert_eq!(
merged_file, "2",
"merged tree must contain the feature's change"
);
// Staging ref must not linger after landing.
let refs = crate::git_process::output(&bare, ["for-each-ref", "refs/merge-queue"])?;
assert!(refs.is_empty(), "disposable staging ref must be cleaned up");
Ok(())
}
#[test]
#[ignore = "requires a Docker daemon and pinned Debian image; run with --ignored --test-threads=1"]
fn merge_queue_fails_closed_when_ci_fails_and_returns_pr_to_open() -> Result<()> {
assert_ci_rejected("exit 1")
}
#[test]
#[ignore = "requires a Docker daemon and pinned Debian image; run with --ignored --test-threads=1"]
fn merge_queue_rejects_success_when_all_checks_were_skipped() -> Result<()> {
assert_ci_rejected("exit 1\n branches: [main]")
}
fn assert_ci_rejected(ci_command: &str) -> Result<()> {
let work = tempfile::tempdir()?;
let (hosted_root, _source) = hosted_repo_with_pipeline(work.path(), ci_command)?;
let bare = hosted_root.join("hosted/app.git");
let database = Database::memory(KEY)?;
let repository = database.add_repository("app", bare.to_str().expect("utf8"), "main")?;
let base_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
let head_sha = crate::git_process::output(&bare, ["rev-parse", "refs/heads/feature-a"])?;
let pull_request = database.create_pull_request(
&repository.name,
"Add feature",
"",
"feature-a",
"main",
&head_sha,
&base_sha,
"author-sub",
"Author",
)?;
database.add_pull_request_review(
&repository.name,
pull_request.number,
"reviewer-sub",
"Reviewer",
"approved",
"ok",
)?;
database.enqueue_merge(&repository.name, pull_request.number, "reviewer-sub")?;
let runner = Runner::new(database.clone(), work.path().join("runner-root"), true)?;
let processed = tick(&database, &runner, &hosted_root)?;
assert_eq!(processed, 1);
let detail = database.pull_request(&repository.name, pull_request.number)?;
assert_eq!(
detail.pull_request.state, "open",
"a failing CI check must return the pull request to open, not land it"
);
let main_head = crate::git_process::output(&bare, ["rev-parse", "refs/heads/main"])?;
assert_eq!(
main_head, base_sha,
"target branch must be untouched when CI fails"
);
let entries = database.list_merge_queue(&repository.name)?;
assert!(
entries.is_empty(),
"failed entry leaves the active queue view (queued/running only)"
);
Ok(())
}
}