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/services/work/src/mergeability.rs

526 lines19,673 bytesCodeBlame
1//! Whether a pull request merges cleanly into the branch it targets, worked
2//! out ahead of time, as GitHub does, rather than discovered when a merge
3//! or the merge queue trips over it.
4//!
5//! It is worked out whenever the pull request's head or its target branch
6//! moves, once per pair of commits. The first pass needs no sandbox: if the
7//! files the pull request changed since it and the branch last agreed share
8//! none with the files the branch changed since then, the merge cannot
9//! conflict. Otherwise a `pull.mergecheck` event asks the runner for a short
10//! probe in a sandbox, which merges the two without an agent and reports the
11//! files that conflict. Only so many probes run at once per repository; the
12//! rest wait their turn and start as earlier ones report.
13
14use std::collections::HashSet;
15
16use g1t_contracts::repos::{BehindArgs, Divergence, GetByIdArgs, Repo, RepoPath};
17use g1t_contracts::time::rfc3339;
18use g1t_contracts::work::*;
19use g1t_contracts::{FailureCode, Outcome};
20use g1t_kit::now_ms;
21use serde::Deserialize;
22use worker::Result;
23
24use crate::Work;
25use crate::checks::{hash, new_token};
26use crate::rows::{NumberRow, ValueRow};
27
28/// How long a probe is waited for before it is asked for again.
29const PROBE_MINUTES: u64 = 10;
30/// How many probes one repository may have running at once.
31const MAX_PROBES: u32 = 3;
32/// How many conflicting files are kept.
33const MAX_CONFLICTS: usize = 100;
34/// How many open pull requests are looked at when their target moves.
35const MAX_TARGETING: u32 = 100;
36
37#[derive(Deserialize)]
38struct MergeRow {
39 mergeable: Option<String>,
40 conflicts: Option<String>,
41 mergeable_key: Option<String>,
42 mergeable_until: Option<String>,
43}
44
45#[derive(Deserialize)]
46struct ProbeRow {
47 id: String,
48 repo_id: String,
49 mergeable_key: Option<String>,
50 mergeable_token_hash: Option<String>,
51}
52
53/// The pair of commits an answer is for.
54fn key(divergence: &Divergence) -> String {
55 format!("{}..{}", divergence.head, divergence.base)
56}
57
58/// Whether the cheap first pass leaves the question open, so that a probe
59/// has to merge the two. It is settled when the branch has not moved, or
60/// when the two sides changed no file in common.
61pub(crate) fn needs_probe(divergence: &Divergence) -> bool {
62 if !divergence.behind {
63 return false;
64 }
65 if divergence.truncated {
66 return true;
67 }
68 let ours: HashSet<&str> = divergence.ours.iter().map(String::as_str).collect();
69 divergence.theirs.iter().any(|path| ours.contains(path.as_str()))
70}
71
72/// Paths as a sandbox reported them, tidied: trimmed, without duplicates,
73/// and no more than are useful.
74pub(crate) fn tidy(paths: Vec<String>) -> Vec<String> {
75 let mut tidy: Vec<String> = Vec::new();
76 for path in paths {
77 let path = path.trim().to_owned();
78 if !path.is_empty() && !tidy.contains(&path) {
79 tidy.push(path);
80 }
81 if tidy.len() == MAX_CONFLICTS {
82 break;
83 }
84 }
85 tidy
86}
87
88impl Work {
89 async fn merge_row(&self, pull_id: &str) -> Result<Option<MergeRow>> {
90 self.db
91 .prepare(
92 "SELECT mergeable, conflicts, mergeable_key, mergeable_until
93 FROM pulls WHERE id = ?",
94 )
95 .bind(&[pull_id.into()])?
96 .first::<MergeRow>(None)
97 .await
98 }
99
100 /// Works out whether a pull request merges cleanly as it and its target
101 /// are now, unless that pair of commits was already worked out or is
102 /// being. Settles it at once where it can, and asks for a probe where
103 /// it cannot.
104 pub(crate) async fn assess_mergeability(&self, pull: &Pull) -> Result<()> {
105 if !pull.status.is_active() {
106 return Ok(());
107 }
108 let divergence: Option<Divergence> = g1t_kit::call(
109 &self.repos,
110 "divergence",
111 &BehindArgs {
112 source_id: pull.fork_repo_id.clone().unwrap_or_else(|| pull.repo_id.clone()),
113 branch: pull.branch.clone(),
114 },
115 )
116 .await?;
117 // Nothing pushed yet.
118 let Some(divergence) = divergence else {
119 return Ok(());
120 };
121 let key = key(&divergence);
122 let now = now_ms();
123 let row = self.merge_row(&pull.id).await?;
124 if let Some(row) = &row
125 && row.mergeable_key.as_deref() == Some(key.as_str())
126 {
127 let settled = match row.mergeable.as_deref() {
128 Some("clean" | "conflicting" | "unknown") => true,
129 // A probe that is still being waited for.
130 Some("checking") => row
131 .mergeable_until
132 .as_deref()
133 .is_some_and(|until| until > rfc3339(now).as_str()),
134 _ => false,
135 };
136 if settled {
137 return Ok(());
138 }
139 }
140 if !needs_probe(&divergence) {
141 self.db
142 .prepare(
143 "UPDATE pulls
144 SET mergeable = 'clean', conflicts = '[]', mergeable_key = ?,
145 mergeable_token_hash = NULL, mergeable_until = NULL
146 WHERE id = ?",
147 )
148 .bind(&[key.as_str().into(), pull.id.as_str().into()])?
149 .run()
150 .await?;
151 // It may have been conflicting before this push.
152 if row.is_some_and(|row| row.mergeable.as_deref() == Some("conflicting")) {
153 self.announce_mergeability(pull).await?;
154 }
155 return Ok(());
156 }
157 self.db
158 .prepare(
159 "UPDATE pulls
160 SET mergeable = 'checking', conflicts = '[]', mergeable_key = ?,
161 mergeable_token_hash = NULL, mergeable_until = ?
162 WHERE id = ?",
163 )
164 .bind(&[
165 key.as_str().into(),
166 rfc3339(now + PROBE_MINUTES * 60 * 1000).into(),
167 pull.id.as_str().into(),
168 ])?
169 .run()
170 .await?;
171 self.ask_for_probe(pull, &divergence.head).await
172 }
173
174 async fn ask_for_probe(&self, pull: &Pull, head: &str) -> Result<()> {
175 self.publish_as(
176 "pull.mergecheck",
177 &pull.repo_id,
178 None,
179 g1t_contracts::events::PullEvent {
180 commit: Some(head.to_owned()),
181 ..Self::pull_event(pull)
182 },
183 )
184 .await
185 }
186
187 /// Says that a pull request's mergeability settled, so that the runner
188 /// looks again at one g1t is seeing through: a conflict is the agent's
189 /// to resolve.
190 async fn announce_mergeability(&self, pull: &Pull) -> Result<()> {
191 self.publish_as("pull.mergeability", &pull.repo_id, None, Self::pull_event(pull))
192 .await
193 }
194
195 /// Works out mergeability again after a push: for the pull requests
196 /// whose heads it moved, and, when it moved a repository's default
197 /// branch, for every open pull request into it. A failure is logged and
198 /// not passed on, so that it never holds up the rest of the push.
199 pub(crate) async fn after_push(&self, repo_id: &str, default_branch: bool, moved: &[String]) {
200 let mut ids: Vec<String> = moved.to_vec();
201 if default_branch {
202 match self.targeting(repo_id).await {
203 Ok(targeting) => {
204 for id in targeting {
205 if !ids.contains(&id) {
206 ids.push(id);
207 }
208 }
209 }
210 Err(error) => worker::console_warn!("mergeability: {error}"),
211 }
212 }
213 for id in ids {
214 let assessed = async {
215 if let Some(pull) = self.pull_by_id(&id).await? {
216 self.assess_mergeability(&pull).await?;
217 }
218 Ok::<_, worker::Error>(())
219 };
220 if let Err(error) = assessed.await {
221 worker::console_warn!("mergeability of {id}: {error}");
222 }
223 }
224 }
225
226 /// The open pull requests into a repository, most recently active first.
227 async fn targeting(&self, repo_id: &str) -> Result<Vec<String>> {
228 Ok(self
229 .db
230 .prepare(
231 "SELECT id AS value FROM pulls
232 WHERE repo_id = ? AND status IN ('draft', 'open')
233 ORDER BY updated_at DESC LIMIT ?",
234 )
235 .bind(&[repo_id.into(), MAX_TARGETING.into()])?
236 .all()
237 .await?
238 .results::<ValueRow>()?
239 .into_iter()
240 .map(|row| row.value)
241 .collect())
242 }
243
244 /// Where a pull request's mergeability stands, for showing it. One that
245 /// was never worked out, or whose probe was given up on, is worked out
246 /// now. A conflict found for an older head is not shown as one.
247 pub(crate) async fn mergeability(&self, pull: &Pull) -> Result<(Mergeable, Vec<String>)> {
248 if !pull.status.is_active() || pull.head_commit.is_none() {
249 return Ok((Mergeable::Unknown, Vec::new()));
250 }
251 let row = self.merge_row(&pull.id).await?;
252 let now = rfc3339(now_ms());
253 let stale = row.as_ref().is_none_or(|row| {
254 row.mergeable_key.is_none()
255 || (row.mergeable.as_deref() == Some("checking")
256 && row.mergeable_until.as_deref().is_none_or(|until| until < now.as_str()))
257 });
258 let row = if stale {
259 if let Err(error) = self.assess_mergeability(pull).await {
260 worker::console_warn!("mergeability of {}: {error}", pull.id);
261 }
262 self.merge_row(&pull.id).await?
263 } else {
264 row
265 };
266 let Some(row) = row else {
267 return Ok((Mergeable::Unknown, Vec::new()));
268 };
269 let current = row
270 .mergeable_key
271 .as_deref()
272 .and_then(|key| key.split_once(".."))
273 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
274 let state = Mergeable::parse(row.mergeable.as_deref());
275 if state == Mergeable::Conflicting && !current {
276 return Ok((Mergeable::Checking, Vec::new()));
277 }
278 let conflicts = if state == Mergeable::Conflicting {
279 row.conflicts
280 .as_deref()
281 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
282 .unwrap_or_default()
283 } else {
284 Vec::new()
285 };
286 Ok((state, conflicts))
287 }
288
289 /// The files that conflict, if the pull request as it is now is known
290 /// to conflict with its target as it is now.
291 pub(crate) async fn conflicting_files(&self, pull: &Pull) -> Result<Option<Vec<String>>> {
292 let Some(row) = self.merge_row(&pull.id).await? else {
293 return Ok(None);
294 };
295 if row.mergeable.as_deref() != Some("conflicting") {
296 return Ok(None);
297 }
298 let current = row
299 .mergeable_key
300 .as_deref()
301 .and_then(|key| key.split_once(".."))
302 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
303 if !current {
304 return Ok(None);
305 }
306 Ok(Some(
307 row.conflicts
308 .as_deref()
309 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
310 .unwrap_or_default(),
311 ))
312 }
313
314 /// Claims the probe a `pull.mergecheck` event asked for, and returns
315 /// what a sandbox needs to carry it out.
316 pub(crate) async fn start_mergecheck(&self, a: StartMergecheckArgs) -> Result<Outcome<MergecheckJob>> {
317 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
318 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
319 };
320 if !pull.status.is_active() {
321 return Ok(Outcome::fail(
322 FailureCode::Conflict,
323 format!("This pull request is already {}.", pull.status.as_str()),
324 ));
325 }
326 let Some(row) = self.merge_row(&pull.id).await? else {
327 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
328 };
329 let Some((head, base)) = row
330 .mergeable_key
331 .as_deref()
332 .filter(|_| row.mergeable.as_deref() == Some("checking"))
333 .and_then(|key| key.split_once(".."))
334 .map(|(head, base)| (head.to_owned(), base.to_owned()))
335 else {
336 return Ok(Outcome::fail(
337 FailureCode::Conflict,
338 "Nothing is waiting to be checked for this pull request.",
339 ));
340 };
341 let now = now_ms();
342 let timestamp = rfc3339(now);
343 let running = self
344 .db
345 .prepare(
346 "SELECT count(*) AS n FROM pulls
347 WHERE repo_id = ? AND id != ? AND mergeable = 'checking'
348 AND mergeable_token_hash IS NOT NULL AND mergeable_until > ?",
349 )
350 .bind(&[pull.repo_id.as_str().into(), pull.id.as_str().into(), timestamp.as_str().into()])?
351 .first::<NumberRow>(None)
352 .await?
353 .map_or(0, |row| row.n);
354 if running >= MAX_PROBES {
355 // It waits, and starts when one of those reports.
356 return Ok(Outcome::fail(
357 FailureCode::Conflict,
358 "This repository has as many merge checks running as it may.",
359 ));
360 }
361 let token = new_token();
362 let claimed = self
363 .db
364 .prepare(
365 "UPDATE pulls SET mergeable_token_hash = ?, mergeable_until = ?
366 WHERE id = ? AND mergeable = 'checking' AND mergeable_key = ?
367 AND (mergeable_token_hash IS NULL OR mergeable_until < ?)
368 RETURNING id AS value",
369 )
370 .bind(&[
371 hash(&token).into(),
372 rfc3339(now + PROBE_MINUTES * 60 * 1000).into(),
373 pull.id.as_str().into(),
374 format!("{head}..{base}").into(),
375 timestamp.as_str().into(),
376 ])?
377 .first::<ValueRow>(None)
378 .await?;
379 if claimed.is_none() {
380 return Ok(Outcome::fail(
381 FailureCode::Conflict,
382 "This pull request is already being checked.",
383 ));
384 }
385 // Its author can read both the repository and the change.
386 let repo: Outcome<Repo> = g1t_kit::call(
387 &self.repos,
388 "get_by_id",
389 &GetByIdArgs {
390 id: pull.repo_id.clone(),
391 viewer: self.author_viewer(&pull).await?,
392 },
393 )
394 .await?;
395 let Outcome::Ok(repo) = repo else {
396 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
397 };
398 let path = RepoPath {
399 namespace: repo.namespace.clone(),
400 name: repo.name.clone(),
401 };
402 Ok(Outcome::Ok(MergecheckJob {
403 pull_id: pull.id.clone(),
404 token,
405 source: pull.fork.clone().unwrap_or_else(|| path.clone()),
406 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
407 repo: path,
408 number: pull.number,
409 default_branch: repo.default_branch,
410 base,
411 head,
412 author: pull.author,
413 }))
414 }
415
416 /// Records what a probe found. Its token is the only credential, and a
417 /// probe for a pair of commits that has since moved on is refused.
418 pub(crate) async fn report_mergecheck(&self, a: ReportMergecheckArgs) -> Result<Outcome<Mergeable>> {
419 let row = self
420 .db
421 .prepare("SELECT id, repo_id, mergeable_key, mergeable_token_hash FROM pulls WHERE id = ?")
422 .bind(&[a.pull_id.as_str().into()])?
423 .first::<ProbeRow>(None)
424 .await?;
425 let Some(row) = row.filter(|row| row.mergeable_token_hash.as_deref() == Some(hash(&a.token).as_str())) else {
426 return Ok(Outcome::fail(FailureCode::NotFound, "Merge check not found."));
427 };
428 let conflicts = tidy(a.conflicts);
429 let state = if a.error.is_some() {
430 Mergeable::Unknown
431 } else if conflicts.is_empty() {
432 Mergeable::Clean
433 } else {
434 Mergeable::Conflicting
435 };
436 if let Some(error) = &a.error {
437 worker::console_warn!("merge check of {} could not run: {error}", row.id);
438 }
439 self.db
440 .prepare(
441 "UPDATE pulls
442 SET mergeable = ?, conflicts = ?, mergeable_token_hash = NULL, mergeable_until = NULL
443 WHERE id = ? AND mergeable_key IS ? AND mergeable_token_hash = ?",
444 )
445 .bind(&[
446 state.as_str().into(),
447 serde_json::to_string(&conflicts)?.into(),
448 row.id.as_str().into(),
449 crate::optional(&row.mergeable_key),
450 hash(&a.token).into(),
451 ])?
452 .run()
453 .await?;
454 if let Some(pull) = self.pull_by_id(&row.id).await? {
455 self.announce_mergeability(&pull).await?;
456 }
457 // The next one waiting its turn in this repository.
458 let waiting = self
459 .db
460 .prepare(
461 "SELECT id AS value FROM pulls
462 WHERE repo_id = ? AND mergeable = 'checking' AND mergeable_token_hash IS NULL
463 AND status IN ('draft', 'open')
464 ORDER BY updated_at DESC LIMIT 1",
465 )
466 .bind(&[row.repo_id.as_str().into()])?
467 .first::<ValueRow>(None)
468 .await?;
469 if let Some(waiting) = waiting
470 && let Some(pull) = self.pull_by_id(&waiting.value).await?
471 {
472 let head = pull.head_commit.clone().unwrap_or_default();
473 self.ask_for_probe(&pull, &head).await?;
474 }
475 Ok(Outcome::Ok(state))
476 }
477}
478
479#[cfg(test)]
480mod tests {
481 use super::*;
482
483 fn divergence(behind: bool, ours: &[&str], theirs: &[&str]) -> Divergence {
484 Divergence {
485 head: "h".into(),
486 base: "b".into(),
487 merge_base: Some("m".into()),
488 behind,
489 ours: ours.iter().map(|path| (*path).to_owned()).collect(),
490 theirs: theirs.iter().map(|path| (*path).to_owned()).collect(),
491 truncated: false,
492 }
493 }
494
495 #[test]
496 fn a_branch_that_has_not_moved_cannot_conflict() {
497 assert!(!needs_probe(&divergence(false, &["a.rs"], &[])));
498 }
499
500 #[test]
501 fn changes_to_different_files_cannot_conflict() {
502 assert!(!needs_probe(&divergence(true, &["a.rs", "b.rs"], &["c.rs"])));
503 }
504
505 #[test]
506 fn changes_to_the_same_file_are_probed() {
507 assert!(needs_probe(&divergence(true, &["a.rs", "b.rs"], &["b.rs"])));
508 }
509
510 #[test]
511 fn a_comparison_cut_short_is_probed() {
512 let cut = Divergence {
513 truncated: true,
514 ..divergence(true, &["a.rs"], &["c.rs"])
515 };
516 assert!(needs_probe(&cut));
517 }
518
519 #[test]
520 fn reported_paths_are_tidied() {
521 let paths = vec![" src/a.rs ".to_owned(), "src/a.rs".to_owned(), String::new(), "b.rs".to_owned()];
522 assert_eq!(tidy(paths), ["src/a.rs", "b.rs"]);
523 let many: Vec<String> = (0..150).map(|n| format!("f{n}")).collect();
524 assert_eq!(tidy(many).len(), MAX_CONFLICTS);
525 }
526}