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