Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/crates/runner/src/queue.rs

245 lines10,495 bytesCodeBlame
1//! Builds one state of a merge queue: the default branch with a run of pull
2//! requests merged in, in queue order. The state is pushed to its own branch,
3//! where the repository's `merge_group` workflows check it,
4//! from which g1t lands it if it passed and everything ahead of it has
5//! landed.
6//!
7//! No agent runs here. A merge that does not apply cleanly is reported as a
8//! conflict, naming the pull request ahead whose change it collided with;
9//! resolving it is the job of the pull request's own agent once that one
10//! has landed.
11//!
12//! Configuration comes from the environment:
13//!
14//! - `G1T_API`, `QUEUE_ENTRY`, `QUEUE_TOKEN`: where and how to report.
15//! - `G1T_USER`, `G1T_TOKEN`: a member, to read the changes and push.
16//! - `BASE_REMOTE`, `BASE_COMMIT`: the repository and the commit to build on.
17//! - `QUEUE_BRANCH`: where to push the tested state.
18//! - `STACK`: the pull requests to merge, as JSON `[{number, title, remote,
19//! branch, commit}]`, the entry being tested last.
20//! - `CHECKS`: commands to run on the state, as a JSON array. g1t sends
21//! none now: the state is checked by its `merge_group` workflows.
22//! - `CONTRACT_CHECKS`: the checks of issues already completed, which the
23//! default branch has to keep passing. One that fails is run again on the
24//! base alone; if it fails there too, it was broken already and is not
25//! held against the stack.
26
27use std::path::Path;
28use std::process::Command;
29
30use anyhow::{Context, Result, bail};
31use serde::Deserialize;
32
33use crate::checks::{redact, run_command};
34use crate::{WORKDIR, auth_option, env, git};
35
36#[derive(Deserialize)]
37struct Item {
38 number: u32,
39 title: String,
40 remote: String,
41 branch: String,
42 commit: String,
43}
44
45/// What stopped a state from being built.
46enum Stopped {
47 /// Merging this pull request's change did not apply cleanly. The
48 /// pull request ahead whose change it collided with, if one did.
49 Conflict { number: u32, with: Option<u32>, files: Vec<String> },
50 Failed(anyhow::Error),
51}
52
53fn git_ok(dir: &Path, args: &[&str]) -> bool {
54 Command::new("git")
55 .current_dir(dir)
56 .args(args)
57 .output()
58 .is_ok_and(|output| output.status.success())
59}
60
61fn changed_files(dir: &Path, from: &str, to: &str) -> Vec<String> {
62 git(dir, &["diff", "--name-only", from, to])
63 .map(|out| out.lines().map(str::to_owned).collect())
64 .unwrap_or_default()
65}
66
67/// Clones the base, then merges each pull request in order. Returns the
68/// tested state's commit.
69fn build(stack: &[Item], auth: &str) -> std::result::Result<String, Stopped> {
70 let base_remote = env("BASE_REMOTE").map_err(Stopped::Failed)?;
71 let base = env("BASE_COMMIT").map_err(Stopped::Failed)?;
72 std::fs::create_dir_all("/work").map_err(|error| Stopped::Failed(error.into()))?;
73 let workdir = Path::new(WORKDIR);
74 crate::clone::clone(Path::new("/work"), auth, &[], &base_remote, WORKDIR)
75 .and_then(|_| git(workdir, &["rev-parse", "--abbrev-ref", "HEAD"]))
76 .and_then(|branch| crate::clone::ensure(workdir, auth, "origin", &branch, &base).map(|_| branch))
77 .and_then(|_| git(workdir, &["checkout", "--quiet", "-B", "g1t-queue", &base]))
78 .and_then(|_| git(workdir, &["config", "user.name", "g1t merge queue"]))
79 .and_then(|_| git(workdir, &["config", "user.email", "queue@g1t.sh"]))
80 .map_err(Stopped::Failed)?;
81
82 for (index, item) in stack.iter().enumerate() {
83 crate::clone::fetch(workdir, auth, &item.remote, &item.branch)
84 .with_context(|| format!("could not fetch #{}", item.number))
85 .map_err(Stopped::Failed)?;
86 let _ = crate::clone::ensure(workdir, auth, &item.remote, &item.branch, &item.commit);
87 // Shallow: deep enough for the change and the state so far to share
88 // a commit to merge from.
89 if crate::clone::has(workdir, &item.commit) {
90 crate::clone::share_history(workdir, auth, &[(base_remote.as_str(), "HEAD"), (item.remote.as_str(), item.branch.as_str())], "HEAD", &item.commit)
91 .map_err(Stopped::Failed)?;
92 }
93 // The branch may have moved on since the queue looked; what was
94 // queued is the commit, and it must be there.
95 if !git_ok(workdir, &["cat-file", "-e", &format!("{}^{{commit}}", item.commit)]) {
96 return Err(Stopped::Failed(anyhow::anyhow!(
97 "#{}'s commit {} is no longer on its branch",
98 item.number,
99 &item.commit[..item.commit.len().min(12)]
100 )));
101 }
102 let message = format!("Merge #{}: {}", item.number, item.title);
103 if !git_ok(workdir, &["merge", "--quiet", "--no-edit", "-m", &message, &item.commit]) {
104 let files: Vec<String> = git(workdir, &["diff", "--name-only", "--diff-filter=U"])
105 .map(|out| out.lines().map(str::to_owned).collect())
106 .unwrap_or_default();
107 let _ = git(workdir, &["merge", "--abort"]);
108 // The pull request ahead that changed one of the same files.
109 let with = stack[..index]
110 .iter()
111 .rev()
112 .find(|earlier| {
113 let theirs = changed_files(workdir, &base, &earlier.commit);
114 files.iter().any(|file| theirs.contains(file))
115 })
116 .map(|earlier| earlier.number);
117 return Err(Stopped::Conflict {
118 number: item.number,
119 with,
120 files,
121 });
122 }
123 }
124 git(workdir, &["rev-parse", "HEAD"])
125 .map(|out| out.trim().to_owned())
126 .map_err(Stopped::Failed)
127}
128
129fn report(api: &str, entry: &str, token: &str, mut body: serde_json::Value) -> Result<()> {
130 body["token"] = token.into();
131 ureq::post(&format!("{api}/queue/{entry}"))
132 .send_json(body)
133 .context("could not report the merge queue's result")?;
134 Ok(())
135}
136
137pub fn main() -> i32 {
138 let (api, entry, token) = match (env("G1T_API"), env("QUEUE_ENTRY"), env("QUEUE_TOKEN")) {
139 (Ok(api), Ok(entry), Ok(token)) => (api, entry, token),
140 _ => {
141 eprintln!("g1t-runner: G1T_API, QUEUE_ENTRY and QUEUE_TOKEN must be set");
142 return 2;
143 }
144 };
145 let secrets: Vec<String> = ["G1T_TOKEN", "QUEUE_TOKEN"]
146 .iter()
147 .filter_map(|name| std::env::var(name).ok())
148 .filter(|secret| !secret.is_empty())
149 .collect();
150 let result = (|| -> Result<serde_json::Value> {
151 let auth = auth_option(&env("G1T_USER")?, &env("G1T_TOKEN")?);
152 let stack: Vec<Item> = serde_json::from_str(&env("STACK")?).context("STACK is not valid")?;
153 if stack.is_empty() {
154 bail!("nothing to merge");
155 }
156 let commands: Vec<String> = std::env::var("CHECKS")
157 .ok()
158 .and_then(|json| serde_json::from_str(&json).ok())
159 .unwrap_or_default();
160 let combined = match build(&stack, &auth) {
161 Ok(combined) => combined,
162 Err(Stopped::Conflict { number, with, files }) => {
163 let against = with.map_or("the default branch".to_owned(), |n| format!("#{n}"));
164 return Ok(serde_json::json!({
165 "error": format!(
166 "#{number} does not merge cleanly with {against}: {} conflict.",
167 files.join(", ")
168 ),
169 "conflictWith": with,
170 "conflicts": files,
171 }));
172 }
173 Err(Stopped::Failed(error)) => {
174 return Ok(serde_json::json!({
175 "error": redact(&format!("{error:#}"), &secrets),
176 }));
177 }
178 };
179 let contract: Vec<String> = std::env::var("CONTRACT_CHECKS")
180 .ok()
181 .and_then(|json| serde_json::from_str(&json).ok())
182 .unwrap_or_default();
183 let workdir = Path::new(WORKDIR);
184 let mut results: Vec<_> = commands
185 .iter()
186 .map(|command| run_command(command, workdir, &secrets))
187 .collect();
188 let contract_results: Vec<_> = contract
189 .iter()
190 .map(|command| run_command(command, workdir, &secrets))
191 .collect();
192 // A contract check that fails here may have been failing on the base
193 // already: try those on the base alone.
194 let failing: Vec<usize> = (0..contract_results.len())
195 .filter(|&index| !contract_results[index].passed)
196 .collect();
197 let mut contract_results = contract_results;
198 if !failing.is_empty() {
199 let base = env("BASE_COMMIT")?;
200 let combined_head = git(workdir, &["rev-parse", "HEAD"])?.trim().to_owned();
201 git(workdir, &["checkout", "--quiet", "--detach", &base])?;
202 for index in failing {
203 let on_base = run_command(&contract_results[index].command, workdir, &secrets);
204 if !on_base.passed {
205 let result = &mut contract_results[index];
206 result.passed = true;
207 result.command = format!(
208 "{} (already failing on the default branch; not held against this)",
209 result.command
210 );
211 }
212 }
213 git(workdir, &["checkout", "--quiet", &combined_head])?;
214 }
215 results.extend(contract_results);
216 // Pushed whether or not it passed, so a failure can be looked at.
217 let branch = env("QUEUE_BRANCH")?;
218 let base_remote = env("BASE_REMOTE")?;
219 git(
220 Path::new(WORKDIR),
221 &[
222 "-c",
223 &auth,
224 "push",
225 "--quiet",
226 "--force",
227 &base_remote,
228 &format!("HEAD:refs/heads/{branch}"),
229 ],
230 )
231 .map_err(|error| anyhow::anyhow!("{}", redact(&format!("{error:#}"), &secrets)))
232 .context("could not push the tested state")?;
233 Ok(serde_json::json!({ "combinedCommit": combined, "results": results }))
234 })();
235 let body = result.unwrap_or_else(|error| {
236 serde_json::json!({ "error": redact(&format!("{error:#}"), &secrets) })
237 });
238 match report(&api, &entry, &token, body) {
239 Ok(()) => 0,
240 Err(error) => {
241 eprintln!("g1t-runner: {error:#}");
242 1
243 }
244 }
245}