Skip to content

g1t/services/work/src/mergeability.rs

565 lines21,648 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/// 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: a conflict is the agent's
210 /// to resolve.
211 async fn announce_mergeability(&self, pull: &Pull) -> Result<()> {
212 self.publish_as("pull.mergeability", &pull.repo_id, None, Self::pull_event(pull))
213 .await
214 }
215
216 /// Works out mergeability again after a push: for the pull requests
217 /// whose heads it moved, and, when it moved a branch pull requests
218 /// merge into, for every open pull request into it. A failure is logged
219 /// and not passed on, so that it never holds up the rest of the push.
220 pub(crate) async fn after_push(&self, repo_id: &str, into: Moved<'_>, moved: &[String]) {
221 let mut ids: Vec<String> = moved.to_vec();
222 let base = match into {
223 Moved::DefaultBranch => Some(None),
224 Moved::Branch(branch) => Some(Some(branch)),
225 Moved::Nothing => None,
226 };
227 if let Some(base) = base {
228 match self.targeting(repo_id, base).await {
229 Ok(targeting) => {
230 for id in targeting {
231 if !ids.contains(&id) {
232 ids.push(id);
233 }
234 }
235 }
236 Err(error) => worker::console_warn!("mergeability: {error}"),
237 }
238 }
239 for id in ids {
240 let assessed = async {
241 if let Some(pull) = self.pull_by_id(&id).await? {
242 self.assess_mergeability(&pull).await?;
243 }
244 Ok::<_, worker::Error>(())
245 };
246 if let Err(error) = assessed.await {
247 worker::console_warn!("mergeability of {id}: {error}");
248 }
249 }
250 }
251
252 /// The open pull requests into a branch of a repository (`None`: its
253 /// default branch), most recently active first.
254 async fn targeting(&self, repo_id: &str, base: Option<&str>) -> Result<Vec<String>> {
255 Ok(self
256 .db
257 .prepare(
258 "SELECT id AS value FROM pulls
259 WHERE repo_id = ? AND status IN ('draft', 'open') AND base_branch IS ?
260 ORDER BY updated_at DESC LIMIT ?",
261 )
262 .bind(&[repo_id.into(), base.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), MAX_TARGETING.into()])?
263 .all()
264 .await?
265 .results::<ValueRow>()?
266 .into_iter()
267 .map(|row| row.value)
268 .collect())
269 }
270
271 /// Where a pull request's mergeability stands, for showing it. One that
272 /// was never worked out, or whose probe was given up on, is worked out
273 /// now. A conflict found for an older head is not shown as one.
274 pub(crate) async fn mergeability(&self, pull: &Pull) -> Result<(Mergeable, Vec<String>)> {
275 if !pull.status.is_active() || pull.head_commit.is_none() {
276 return Ok((Mergeable::Unknown, Vec::new()));
277 }
278 // As read for this request, if it was; read again after working it out.
279 let row = match self.prefetched_pull(&pull.id) {
280 Some(found) => found.first::<MergeRow>(crate::prefetch::Slot::Pull)?,
281 None => self.merge_row(&pull.id).await?,
282 };
283 let now = rfc3339(now_ms());
284 let stale = row.as_ref().is_none_or(|row| {
285 row.mergeable_key.is_none()
286 || (row.mergeable.as_deref() == Some("checking")
287 && row.mergeable_until.as_deref().is_none_or(|until| until < now.as_str()))
288 });
289 let row = if stale {
290 if let Err(error) = self.assess_mergeability(pull).await {
291 worker::console_warn!("mergeability of {}: {error}", pull.id);
292 }
293 // What was read for this request is out of date now.
294 self.keep_prefetched(None);
295 self.merge_row(&pull.id).await?
296 } else {
297 row
298 };
299 let Some(row) = row else {
300 return Ok((Mergeable::Unknown, Vec::new()));
301 };
302 let current = row
303 .mergeable_key
304 .as_deref()
305 .and_then(|key| key.split_once(".."))
306 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
307 let state = Mergeable::parse(row.mergeable.as_deref());
308 if state == Mergeable::Conflicting && !current {
309 return Ok((Mergeable::Checking, Vec::new()));
310 }
311 let conflicts = if state == Mergeable::Conflicting {
312 row.conflicts
313 .as_deref()
314 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
315 .unwrap_or_default()
316 } else {
317 Vec::new()
318 };
319 Ok((state, conflicts))
320 }
321
322 /// The files that conflict, if the pull request as it is now is known
323 /// to conflict with its target as it is now.
324 pub(crate) async fn conflicting_files(&self, pull: &Pull) -> Result<Option<Vec<String>>> {
325 let row = match self.prefetched_pull(&pull.id) {
326 Some(found) => found.first::<MergeRow>(crate::prefetch::Slot::Pull)?,
327 None => self.merge_row(&pull.id).await?,
328 };
329 let Some(row) = row else {
330 return Ok(None);
331 };
332 if row.mergeable.as_deref() != Some("conflicting") {
333 return Ok(None);
334 }
335 let current = row
336 .mergeable_key
337 .as_deref()
338 .and_then(|key| key.split_once(".."))
339 .is_some_and(|(head, _)| Some(head) == pull.head_commit.as_deref());
340 if !current {
341 return Ok(None);
342 }
343 Ok(Some(
344 row.conflicts
345 .as_deref()
346 .and_then(|conflicts| serde_json::from_str(conflicts).ok())
347 .unwrap_or_default(),
348 ))
349 }
350
351 /// Claims the probe a `pull.mergecheck` event asked for, and returns
352 /// what a sandbox needs to carry it out.
353 pub(crate) async fn start_mergecheck(&self, a: StartMergecheckArgs) -> Result<Outcome<MergecheckJob>> {
354 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
355 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
356 };
357 if !pull.status.is_active() {
358 return Ok(Outcome::fail(
359 FailureCode::Conflict,
360 format!("This pull request is already {}.", pull.status.as_str()),
361 ));
362 }
363 let Some(row) = self.merge_row(&pull.id).await? else {
364 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
365 };
366 let Some((head, base)) = row
367 .mergeable_key
368 .as_deref()
369 .filter(|_| row.mergeable.as_deref() == Some("checking"))
370 .and_then(|key| key.split_once(".."))
371 .map(|(head, base)| (head.to_owned(), base.to_owned()))
372 else {
373 return Ok(Outcome::fail(
374 FailureCode::Conflict,
375 "Nothing is waiting to be checked for this pull request.",
376 ));
377 };
378 let now = now_ms();
379 let timestamp = rfc3339(now);
380 let running = self
381 .db
382 .prepare(
383 "SELECT count(*) AS n FROM pulls
384 WHERE repo_id = ? AND id != ? AND mergeable = 'checking'
385 AND mergeable_token_hash IS NOT NULL AND mergeable_until > ?",
386 )
387 .bind(&[pull.repo_id.as_str().into(), pull.id.as_str().into(), timestamp.as_str().into()])?
388 .first::<NumberRow>(None)
389 .await?
390 .map_or(0, |row| row.n);
391 if running >= MAX_PROBES {
392 // It waits, and starts when one of those reports.
393 return Ok(Outcome::fail(
394 FailureCode::Conflict,
395 "This repository has as many merge checks running as it may.",
396 ));
397 }
398 let token = new_token();
399 let claimed = self
400 .db
401 .prepare(
402 "UPDATE pulls SET mergeable_token_hash = ?, mergeable_until = ?
403 WHERE id = ? AND mergeable = 'checking' AND mergeable_key = ?
404 AND (mergeable_token_hash IS NULL OR mergeable_until < ?)
405 RETURNING id AS value",
406 )
407 .bind(&[
408 hash(&token).into(),
409 rfc3339(now + PROBE_MINUTES * 60 * 1000).into(),
410 pull.id.as_str().into(),
411 format!("{head}..{base}").into(),
412 timestamp.as_str().into(),
413 ])?
414 .first::<ValueRow>(None)
415 .await?;
416 if claimed.is_none() {
417 return Ok(Outcome::fail(
418 FailureCode::Conflict,
419 "This pull request is already being checked.",
420 ));
421 }
422 // Its owner (whoever asked g1t for it, or its author) can read both
423 // the repository and the change.
424 let repo: Outcome<Repo> = g1t_kit::call(
425 &self.repos,
426 "get_by_id",
427 &GetByIdArgs {
428 id: pull.repo_id.clone(),
429 viewer: self.owner_viewer(&pull).await?,
430 },
431 )
432 .await?;
433 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
434 return Ok(Outcome::fail(FailureCode::NotFound, "Pull request not found."));
435 };
436 let path = RepoPath {
437 namespace: repo.namespace.clone(),
438 name: repo.name.clone(),
439 };
440 Ok(Outcome::Ok(MergecheckJob {
441 pull_id: pull.id.clone(),
442 token,
443 source: pull.fork.clone().unwrap_or_else(|| path.clone()),
444 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
445 repo: path,
446 number: pull.number,
447 // The branch it merges into.
448 default_branch: pull.base_branch(&repo.default_branch).to_owned(),
449 base,
450 head,
451 author: pull.requested_by.unwrap_or(pull.author),
452 }))
453 }
454
455 /// Records what a probe found. Its token is the only credential, and a
456 /// probe for a pair of commits that has since moved on is refused.
457 pub(crate) async fn report_mergecheck(&self, a: ReportMergecheckArgs) -> Result<Outcome<Mergeable>> {
458 let row = self
459 .db
460 .prepare("SELECT id, repo_id, mergeable_key, mergeable_token_hash FROM pulls WHERE id = ?")
461 .bind(&[a.pull_id.as_str().into()])?
462 .first::<ProbeRow>(None)
463 .await?;
464 let Some(row) = row.filter(|row| row.mergeable_token_hash.as_deref() == Some(hash(&a.token).as_str())) else {
465 return Ok(Outcome::fail(FailureCode::NotFound, "Merge check not found."));
466 };
467 let conflicts = tidy(a.conflicts);
468 let state = if a.error.is_some() {
469 Mergeable::Unknown
470 } else if conflicts.is_empty() {
471 Mergeable::Clean
472 } else {
473 Mergeable::Conflicting
474 };
475 if let Some(error) = &a.error {
476 worker::console_warn!("merge check of {} could not run: {error}", row.id);
477 }
478 self.db
479 .prepare(
480 "UPDATE pulls
481 SET mergeable = ?, conflicts = ?, mergeable_token_hash = NULL, mergeable_until = NULL
482 WHERE id = ? AND mergeable_key IS ? AND mergeable_token_hash = ?",
483 )
484 .bind(&[
485 state.as_str().into(),
486 serde_json::to_string(&conflicts)?.into(),
487 row.id.as_str().into(),
488 crate::optional(&row.mergeable_key),
489 hash(&a.token).into(),
490 ])?
491 .run()
492 .await?;
493 if let Some(pull) = self.pull_by_id(&row.id).await? {
494 self.announce_mergeability(&pull).await?;
495 }
496 // The next one waiting its turn in this repository.
497 let waiting = self
498 .db
499 .prepare(
500 "SELECT id AS value FROM pulls
501 WHERE repo_id = ? AND mergeable = 'checking' AND mergeable_token_hash IS NULL
502 AND status IN ('draft', 'open')
503 ORDER BY updated_at DESC LIMIT 1",
504 )
505 .bind(&[row.repo_id.as_str().into()])?
506 .first::<ValueRow>(None)
507 .await?;
508 if let Some(waiting) = waiting
509 && let Some(pull) = self.pull_by_id(&waiting.value).await?
510 {
511 let head = pull.head_commit.clone().unwrap_or_default();
512 self.ask_for_probe(&pull, &head).await?;
513 }
514 Ok(Outcome::Ok(state))
515 }
516}
517
518#[cfg(test)]
519mod tests {
520 use super::*;
521
522 fn divergence(behind: bool, ours: &[&str], theirs: &[&str]) -> Divergence {
523 Divergence {
524 head: "h".into(),
525 base: "b".into(),
526 merge_base: Some("m".into()),
527 behind,
528 ours: ours.iter().map(|path| (*path).to_owned()).collect(),
529 theirs: theirs.iter().map(|path| (*path).to_owned()).collect(),
530 truncated: false,
531 }
532 }
533
534 #[test]
535 fn a_branch_that_has_not_moved_cannot_conflict() {
536 assert!(!needs_probe(&divergence(false, &["a.rs"], &[])));
537 }
538
539 #[test]
540 fn changes_to_different_files_cannot_conflict() {
541 assert!(!needs_probe(&divergence(true, &["a.rs", "b.rs"], &["c.rs"])));
542 }
543
544 #[test]
545 fn changes_to_the_same_file_are_probed() {
546 assert!(needs_probe(&divergence(true, &["a.rs", "b.rs"], &["b.rs"])));
547 }
548
549 #[test]
550 fn a_comparison_cut_short_is_probed() {
551 let cut = Divergence {
552 truncated: true,
553 ..divergence(true, &["a.rs"], &["c.rs"])
554 };
555 assert!(needs_probe(&cut));
556 }
557
558 #[test]
559 fn reported_paths_are_tidied() {
560 let paths = vec![" src/a.rs ".to_owned(), "src/a.rs".to_owned(), String::new(), "b.rs".to_owned()];
561 assert_eq!(tidy(paths), ["src/a.rs", "b.rs"]);
562 let many: Vec<String> = (0..150).map(|n| format!("f{n}")).collect();
563 assert_eq!(tidy(many).len(), MAX_CONFLICTS);
564 }
565}