Skip to content

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

g1t.sh
600 linesCodeBlameRaw
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/// What a push moved, as far as the pull requests into it go.
54pub(crate) enum Moved<'a> {
55 /// The repository's default branch: every open pull request into it,
56 /// whose base is stored as none.
57 DefaultBranch,
58 /// Another branch: the open pull requests into that one.
59 Branch(&'a str),
60 /// A tag, or anything else nothing merges into.
61 Nothing,
62}
63
64/// The pair of commits an answer is for.
65fn key(divergence: &Divergence) -> String {
66 format!("{}..{}", divergence.head, divergence.base)
67}
68
69/// Whether the cheap first pass leaves the question open, so that a probe
70/// has to merge the two. It is settled when the branch has not moved, or
71/// when the two sides changed no file in common.
72pub(crate) fn needs_probe(divergence: &Divergence) -> bool {
73 if !divergence.behind {
74 return false;
75 }
76 if divergence.truncated {
77 return true;
78 }
79 let ours: HashSet<&str> = divergence.ours.iter().map(String::as_str).collect();
80 divergence.theirs.iter().any(|path| ours.contains(path.as_str()))
81}
82
83/// Paths as a sandbox reported them, tidied: trimmed, without duplicates,
84/// and no more than are useful.
85pub(crate) fn tidy(paths: Vec<String>) -> Vec<String> {
86 let mut tidy: Vec<String> = Vec::new();
87 for path in paths {
88 let path = path.trim().to_owned();
89 if !path.is_empty() && !tidy.contains(&path) {
90 tidy.push(path);
91 }
92 if tidy.len() == MAX_CONFLICTS {
93 break;
94 }
95 }
96 tidy
97}
98
99impl Work {
100 async fn merge_row(&self, pull_id: &str) -> Result<Option<MergeRow>> {
101 self.db
102 .prepare(
103 "SELECT mergeable, conflicts, mergeable_key, mergeable_until
104 FROM pulls WHERE id = ?",
105 )
106 .bind(&[pull_id.into()])?
107 .first::<MergeRow>(None)
108 .await
109 }
110
111 /// Works out whether a pull request merges cleanly as it and its target
112 /// are now, unless that pair of commits was already worked out or is
113 /// being. Settles it at once where it can, and asks for a probe where
114 /// it cannot.
115 pub(crate) async fn assess_mergeability(&self, pull: &Pull) -> Result<()> {
116 if !pull.status.is_active() {
117 return Ok(());
118 }
119 let divergence: Option<Divergence> = g1t_kit::call(
120 &self.repos,
121 "divergence",
122 &BehindArgs {
123 source_id: pull.fork_repo_id.clone().unwrap_or_else(|| pull.repo_id.clone()),
124 branch: pull.branch.clone(),
125 target_branch: pull.base.clone().filter(|base| !base.is_empty()),
126 },
127 )
128 .await?;
129 // Nothing pushed yet.
130 let Some(divergence) = divergence else {
131 return Ok(());
132 };
133 let key = key(&divergence);
134 let now = now_ms();
135 let behind = u32::from(divergence.behind);
136 let row = self.merge_row(&pull.id).await?;
137 if let Some(row) = &row
138 && row.mergeable_key.as_deref() == Some(key.as_str())
139 {
140 // Whether it is behind goes with the pair of commits; a row from
141 // before it was kept gets it now.
142 self.db
143 .prepare("UPDATE pulls SET behind = ?1 WHERE id = ?2 AND behind IS NOT ?1")
144 .bind(&[behind.into(), pull.id.as_str().into()])?
145 .run()
146 .await?;
147 let settled = match row.mergeable.as_deref() {
148 Some("clean" | "conflicting" | "unknown") => true,
149 // A probe that is still being waited for.
150 Some("checking") => row
151 .mergeable_until
152 .as_deref()
153 .is_some_and(|until| until > rfc3339(now).as_str()),
154 _ => false,
155 };
156 if settled {
157 return Ok(());
158 }
159 }
160 if !needs_probe(&divergence) {
161 self.db
162 .prepare(
163 "UPDATE pulls
164 SET mergeable = 'clean', conflicts = '[]', mergeable_key = ?, behind = ?,
165 mergeable_token_hash = NULL, mergeable_until = NULL
166 WHERE id = ?",
167 )
168 .bind(&[key.as_str().into(), behind.into(), pull.id.as_str().into()])?
169 .run()
170 .await?;
171 // It may have been conflicting before this push.
172 if row.is_some_and(|row| row.mergeable.as_deref() == Some("conflicting")) {
173 self.announce_mergeability(pull).await?;
174 }
175 return Ok(());
176 }
177 self.db
178 .prepare(
179 "UPDATE pulls
180 SET mergeable = 'checking', conflicts = '[]', mergeable_key = ?, behind = ?,
181 mergeable_token_hash = NULL, mergeable_until = ?
182 WHERE id = ?",
183 )
184 .bind(&[
185 key.as_str().into(),
186 behind.into(),
187 rfc3339(now + PROBE_MINUTES * 60 * 1000).into(),
188 pull.id.as_str().into(),
189 ])?
190 .run()
191 .await?;
192 self.ask_for_probe(pull, &divergence.head).await
193 }
194
195 async fn ask_for_probe(&self, pull: &Pull, head: &str) -> Result<()> {
196 self.publish_as(
197 "pull.mergecheck",
198 &pull.repo_id,
199 None,
200 g1t_contracts::events::PullEvent {
201 commit: Some(head.to_owned()),
202 ..Self::pull_event(pull)
203 },
204 )
205 .await
206 }
207
208 /// Says that a pull request's mergeability settled, so that the runner
209 /// looks again at one g1t is seeing through.
210 async fn announce_mergeability(&self, pull: &Pull) -> Result<()> {
211 self.publish_as("pull.mergeability", &pull.repo_id, None, Self::pull_event(pull))
212 .await
213 }
214
215 /// Works out mergeability again after a push: for the pull requests
216 /// whose heads it moved, and, when it moved a branch pull requests
217 /// merge into, for every open pull request into it. A failure is logged
218 /// and not passed on, so that it never holds up the rest of the push.
219 pub(crate) async fn after_push(&self, repo_id: &str, into: Moved<'_>, moved: &[String]) {
220 let mut ids: Vec<String> = moved.to_vec();
221 let base = match into {
222 Moved::DefaultBranch => Some(None),
223 Moved::Branch(branch) => Some(Some(branch)),
224 Moved::Nothing => None,
225 };
226 if let Some(base) = base {
227 match self.targeting(repo_id, base).await {
228 Ok(targeting) => {
229 for id in targeting {
230 if !ids.contains(&id) {
231 ids.push(id);
232 }
233 }
234 }
235 Err(error) => worker::console_warn!("mergeability: {error}"),
236 }
237 }
238 for id in ids {
239 let assessed = async {
240 if let Some(pull) = self.pull_by_id(&id).await? {
241 // A pull request whose head moved: its refs in the
242 // repository move with it, before its events go out.
243 if moved.contains(&id) {
244 self.keep_pull_refs(&pull).await;
245 }
246 self.assess_mergeability(&pull).await?;
247 }
248 Ok::<_, worker::Error>(())
249 };
250 if let Err(error) = assessed.await {
251 worker::console_warn!("mergeability of {id}: {error}");
252 }
253 }
254 }
255
256 /// Points `refs/pull/<n>/head` in the repository at the pull request's
257 /// head, its objects copied from a fork or working copy, and
258 /// `refs/pull/<n>/merge` at the head merged into its base, as GitHub
259 /// keeps them for every open pull request (repos `pull_merge_ref`), so
260 /// that both can be fetched from the repository itself. A failure is
261 /// logged: it never holds up the push.
262 pub(crate) async fn keep_pull_refs(&self, pull: &Pull) {
263 let (Some(head), true) = (pull.head_commit.clone(), pull.status.is_active()) else {
264 return;
265 };
266 let kept: Result<g1t_contracts::repos::PullMergeRef> = g1t_kit::call(
267 &self.repos,
268 "pull_merge_ref",
269 &g1t_contracts::repos::PullMergeRefArgs {
270 repo_id: pull.repo_id.clone(),
271 source_id: pull.fork_repo_id.clone().unwrap_or_else(|| pull.repo_id.clone()),
272 head,
273 number: pull.number,
274 target_branch: pull.base.clone().filter(|base| !base.is_empty()),
275 },
276 )
277 .await;
278 match kept {
279 Ok(g1t_contracts::repos::PullMergeRef::Unavailable { reason }) => {
280 worker::console_log!("refs of pull request {}: no merge commit: {reason}", pull.id);
281 }
282 Ok(_) => {}
283 Err(error) => worker::console_warn!("refs of pull request {} not kept: {error}", pull.id),
284 }
285 }
286
287 /// The open pull requests into a branch of a repository (`None`: its
288 /// default branch), most recently active first.
289 async fn targeting(&self, repo_id: &str, base: Option<&str>) -> Result<Vec<String>> {
290 Ok(self
291 .db
292 .prepare(
293 "SELECT id AS value FROM pulls
294 WHERE repo_id = ? AND status IN ('draft', 'open') AND base_branch IS ?
295 ORDER BY updated_at DESC LIMIT ?",
296 )
297 .bind(&[repo_id.into(), base.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), MAX_TARGETING.into()])?
298 .all()
299 .await?
300 .results::<ValueRow>()?
301 .into_iter()
302 .map(|row| row.value)
303 .collect())
304 }
305
306 /// Where a pull request's mergeability stands, for showing it. One that
307 /// was never worked out, or whose probe was given up on, is worked out
308 /// now. A conflict found for an older head is not shown as one.
309 pub(crate) async fn mergeability(&self, pull: &Pull) -> Result<(Mergeable, Vec<String>)> {
310 if !pull.status.is_active() || pull.head_commit.is_none() {
311 return Ok((Mergeable::Unknown, Vec::new()));
312 }
313 // As read for this request, if it was; read again after working it out.
314 let row = match self.prefetched_pull(&pull.id) {
315 Some(found) => found.first::<MergeRow>(crate::prefetch::Slot::Pull)?,
316 None => self.merge_row(&pull.id).await?,
317 };
318 let now = rfc3339(now_ms());
319 let stale = row.as_ref().is_none_or(|row| {
320 row.mergeable_key.is_none()
321 || (row.mergeable.as_deref() == Some("checking")
322 && row.mergeable_until.as_deref().is_none_or(|until| until < now.as_str()))
323 });
324 let row = if stale {
325 if let Err(error) = self.assess_mergeability(pull).await {
326 worker::console_warn!("mergeability of {}: {error}", pull.id);
327 }
328 // What was read for this request is out of date now.
329 self.keep_prefetched(None);
330 self.merge_row(&pull.id).await?
331 } else {
332 row
333 };
334 let Some(row) = row else {
335 return Ok((Mergeable::Unknown, Vec::new()));
336 };
337 let current = row
338 .mergeable_key
339 .as_deref()
340 .and_then(|key| key.split_once(".."))
341 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
342 let state = Mergeable::parse(row.mergeable.as_deref());
343 if state == Mergeable::Conflicting && !current {
344 return Ok((Mergeable::Checking, Vec::new()));
345 }
346 let conflicts = if state == Mergeable::Conflicting {
347 row.conflicts
348 .as_deref()
349 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
350 .unwrap_or_default()
351 } else {
352 Vec::new()
353 };
354 Ok((state, conflicts))
355 }
356
357 /// The files that conflict, if the pull request as it is now is known
358 /// to conflict with its target as it is now.
359 pub(crate) async fn conflicting_files(&self, pull: &Pull) -> Result<Option<Vec<String>>> {
360 let row = match self.prefetched_pull(&pull.id) {
361 Some(found) => found.first::<MergeRow>(crate::prefetch::Slot::Pull)?,
362 None => self.merge_row(&pull.id).await?,
363 };
364 let Some(row) = row else {
365 return Ok(None);
366 };
367 if row.mergeable.as_deref() != Some("conflicting") {
368 return Ok(None);
369 }
370 let current = row
371 .mergeable_key
372 .as_deref()
373 .and_then(|key| key.split_once(".."))
374 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
375 if !current {
376 return Ok(None);
377 }
378 Ok(Some(
379 row.conflicts
380 .as_deref()
381 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
382 .unwrap_or_default(),
383 ))
384 }
385
386 /// Claims the probe a `pull.mergecheck` event asked for, and returns
387 /// what a sandbox needs to carry it out.
388 pub(crate) async fn start_mergecheck(&self, a: StartMergecheckArgs) -> Result<Outcome<MergecheckJob>> {
389 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
390 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
391 };
392 if !pull.status.is_active() {
393 return Ok(Outcome::fail(
394 FailureCode::Conflict,
395 format!("This pull request is already {}.", pull.status.as_str()),
396 ));
397 }
398 let Some(row) = self.merge_row(&pull.id).await? else {
399 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
400 };
401 let Some((head, base)) = row
402 .mergeable_key
403 .as_deref()
404 .filter(|_| row.mergeable.as_deref() == Some("checking"))
405 .and_then(|key| key.split_once(".."))
406 .map(|(head, base)| (head.to_owned(), base.to_owned()))
407 else {
408 return Ok(Outcome::fail(
409 FailureCode::Conflict,
410 "Nothing is waiting to be checked for this pull request.",
411 ));
412 };
413 let now = now_ms();
414 let timestamp = rfc3339(now);
415 let running = self
416 .db
417 .prepare(
418 "SELECT count(*) AS n FROM pulls
419 WHERE repo_id = ? AND id != ? AND mergeable = 'checking'
420 AND mergeable_token_hash IS NOT NULL AND mergeable_until > ?",
421 )
422 .bind(&[pull.repo_id.as_str().into(), pull.id.as_str().into(), timestamp.as_str().into()])?
423 .first::<NumberRow>(None)
424 .await?
425 .map_or(0, |row| row.n);
426 if running >= MAX_PROBES {
427 // It waits, and starts when one of those reports.
428 return Ok(Outcome::fail(
429 FailureCode::Conflict,
430 "This repository has as many merge checks running as it may.",
431 ));
432 }
433 let token = new_token();
434 let claimed = self
435 .db
436 .prepare(
437 "UPDATE pulls SET mergeable_token_hash = ?, mergeable_until = ?
438 WHERE id = ? AND mergeable = 'checking' AND mergeable_key = ?
439 AND (mergeable_token_hash IS NULL OR mergeable_until < ?)
440 RETURNING id AS value",
441 )
442 .bind(&[
443 hash(&token).into(),
444 rfc3339(now + PROBE_MINUTES * 60 * 1000).into(),
445 pull.id.as_str().into(),
446 format!("{head}..{base}").into(),
447 timestamp.as_str().into(),
448 ])?
449 .first::<ValueRow>(None)
450 .await?;
451 if claimed.is_none() {
452 return Ok(Outcome::fail(
453 FailureCode::Conflict,
454 "This pull request is already being checked.",
455 ));
456 }
457 // Its owner (whoever asked g1t for it, or its author) can read both
458 // the repository and the change.
459 let repo: Outcome<Repo> = g1t_kit::call(
460 &self.repos,
461 "get_by_id",
462 &GetByIdArgs {
463 id: pull.repo_id.clone(),
464 viewer: self.owner_viewer(&pull).await?,
465 },
466 )
467 .await?;
468 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
469 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
470 };
471 let path = RepoPath {
472 namespace: repo.namespace.clone(),
473 name: repo.name.clone(),
474 };
475 Ok(Outcome::Ok(MergecheckJob {
476 pull_id: pull.id.clone(),
477 token,
478 source: pull.fork.clone().unwrap_or_else(|| path.clone()),
479 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
480 repo: path,
481 number: pull.number,
482 // The branch it merges into.
483 default_branch: pull.base_branch(&repo.default_branch).to_owned(),
484 base,
485 head,
486 author: pull.requested_by.unwrap_or(pull.author),
487 }))
488 }
489
490 /// Records what a probe found. Its token is the only credential, and a
491 /// probe for a pair of commits that has since moved on is refused.
492 pub(crate) async fn report_mergecheck(&self, a: ReportMergecheckArgs) -> Result<Outcome<Mergeable>> {
493 let row = self
494 .db
495 .prepare("SELECT id, repo_id, mergeable_key, mergeable_token_hash FROM pulls WHERE id = ?")
496 .bind(&[a.pull_id.as_str().into()])?
497 .first::<ProbeRow>(None)
498 .await?;
499 let Some(row) = row.filter(|row| row.mergeable_token_hash.as_deref() == Some(hash(&a.token).as_str())) else {
500 return Ok(Outcome::fail(FailureCode::NotFound, "Merge check not found."));
501 };
502 let conflicts = tidy(a.conflicts);
503 let state = if a.error.is_some() {
504 Mergeable::Unknown
505 } else if conflicts.is_empty() {
506 Mergeable::Clean
507 } else {
508 Mergeable::Conflicting
509 };
510 if let Some(error) = &a.error {
511 worker::console_warn!("merge check of {} could not run: {error}", row.id);
512 }
513 self.db
514 .prepare(
515 "UPDATE pulls
516 SET mergeable = ?, conflicts = ?, mergeable_token_hash = NULL, mergeable_until = NULL
517 WHERE id = ? AND mergeable_key IS ? AND mergeable_token_hash = ?",
518 )
519 .bind(&[
520 state.as_str().into(),
521 serde_json::to_string(&conflicts)?.into(),
522 row.id.as_str().into(),
523 crate::optional(&row.mergeable_key),
524 hash(&a.token).into(),
525 ])?
526 .run()
527 .await?;
528 if let Some(pull) = self.pull_by_id(&row.id).await? {
529 self.announce_mergeability(&pull).await?;
530 }
531 // The next one waiting its turn in this repository.
532 let waiting = self
533 .db
534 .prepare(
535 "SELECT id AS value FROM pulls
536 WHERE repo_id = ? AND mergeable = 'checking' AND mergeable_token_hash IS NULL
537 AND status IN ('draft', 'open')
538 ORDER BY updated_at DESC LIMIT 1",
539 )
540 .bind(&[row.repo_id.as_str().into()])?
541 .first::<ValueRow>(None)
542 .await?;
543 if let Some(waiting) = waiting
544 && let Some(pull) = self.pull_by_id(&waiting.value).await?
545 {
546 let head = pull.head_commit.clone().unwrap_or_default();
547 self.ask_for_probe(&pull, &head).await?;
548 }
549 Ok(Outcome::Ok(state))
550 }
551}
552
553#[cfg(test)]
554mod tests {
555 use super::*;
556
557 fn divergence(behind: bool, ours: &[&str], theirs: &[&str]) -> Divergence {
558 Divergence {
559 head: "h".into(),
560 base: "b".into(),
561 merge_base: Some("m".into()),
562 behind,
563 ours: ours.iter().map(|path| (*path).to_owned()).collect(),
564 theirs: theirs.iter().map(|path| (*path).to_owned()).collect(),
565 truncated: false,
566 }
567 }
568
569 #[test]
570 fn a_branch_that_has_not_moved_cannot_conflict() {
571 assert!(!needs_probe(&divergence(false, &["a.rs"], &[])));
572 }
573
574 #[test]
575 fn changes_to_different_files_cannot_conflict() {
576 assert!(!needs_probe(&divergence(true, &["a.rs", "b.rs"], &["c.rs"])));
577 }
578
579 #[test]
580 fn changes_to_the_same_file_are_probed() {
581 assert!(needs_probe(&divergence(true, &["a.rs", "b.rs"], &["b.rs"])));
582 }
583
584 #[test]
585 fn a_comparison_cut_short_is_probed() {
586 let cut = Divergence {
587 truncated: true,
588 ..divergence(true, &["a.rs"], &["c.rs"])
589 };
590 assert!(needs_probe(&cut));
591 }
592
593 #[test]
594 fn reported_paths_are_tidied() {
595 let paths = vec![" src/a.rs ".to_owned(), "src/a.rs".to_owned(), String::new(), "b.rs".to_owned()];
596 assert_eq!(tidy(paths), ["src/a.rs", "b.rs"]);
597 let many: Vec<String> = (0..150).map(|n| format!("f{n}")).collect();
598 assert_eq!(tidy(many).len(), MAX_CONFLICTS);
599 }
600}