Skip to content
Sign in

flagon-io/g1t

public
0

g1t/services/work/src/mergeability.rs

565 lines21,648 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Agents and memory, checks and conflicts, profiles, slug renames, custom domains1//! 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
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar53/// 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
Agents and memory, checks and conflicts, profiles, slug renames, custom domains64/// 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(),
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar125 target_branch: pull.base.clone().filter(|base| !base.is_empty()),
Agents and memory, checks and conflicts, profiles, slug renames, custom domains126 },
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();
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily135 let behind = u32::from(divergence.behind);
Agents and memory, checks and conflicts, profiles, slug renames, custom domains136 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 {
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily140 // 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?;
Agents and memory, checks and conflicts, profiles, slug renames, custom domains147 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
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily164 SET mergeable = 'clean', conflicts = '[]', mergeable_key = ?, behind = ?,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains165 mergeable_token_hash = NULL, mergeable_until = NULL
166 WHERE id = ?",
167 )
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily168 .bind(&[key.as_str().into(), behind.into(), pull.id.as_str().into()])?
Agents and memory, checks and conflicts, profiles, slug renames, custom domains169 .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
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily180 SET mergeable = 'checking', conflicts = '[]', mergeable_key = ?, behind = ?,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains181 mergeable_token_hash = NULL, mergeable_until = ?
182 WHERE id = ?",
183 )
184 .bind(&[
185 key.as_str().into(),
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily186 behind.into(),
Agents and memory, checks and conflicts, profiles, slug renames, custom domains187 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
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar217 /// 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]) {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains221 let mut ids: Vec<String> = moved.to_vec();
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar222 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 {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains229 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
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar252 /// 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>> {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains255 Ok(self
256 .db
257 .prepare(
258 "SELECT id AS value FROM pulls
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar259 WHERE repo_id = ? AND status IN ('draft', 'open') AND base_branch IS ?
Agents and memory, checks and conflicts, profiles, slug renames, custom domains260 ORDER BY updated_at DESC LIMIT ?",
261 )
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar262 .bind(&[repo_id.into(), base.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into), MAX_TARGETING.into()])?
Agents and memory, checks and conflicts, profiles, slug renames, custom domains263 .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 }
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily278 // 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 };
Agents and memory, checks and conflicts, profiles, slug renames, custom domains283 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 }
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily293 // What was read for this request is out of date now.
294 self.keep_prefetched(None);
Agents and memory, checks and conflicts, profiles, slug renames, custom domains295 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>>> {
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily325 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 {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains330 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 }
g1t is the stored author of what it opens; the person who asked is requested_by and keeps the author's rights422 // Its owner (whoever asked g1t for it, or its author) can read both
423 // the repository and the change.
Agents and memory, checks and conflicts, profiles, slug renames, custom domains424 let repo: Outcome<Repo> = g1t_kit::call(
425 &self.repos,
426 "get_by_id",
427 &GetByIdArgs {
428 id: pull.repo_id.clone(),
g1t is the stored author of what it opens; the person who asked is requested_by and keeps the author's rights429 viewer: self.owner_viewer(&pull).await?,
Agents and memory, checks and conflicts, profiles, slug renames, custom domains430 },
431 )
432 .await?;
Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look433 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
Agents and memory, checks and conflicts, profiles, slug renames, custom domains434 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,
Teams and CODEOWNERS, labels and milestones, dependency updates, the security suite, and a clearer top bar447 // The branch it merges into.
448 default_branch: pull.base_branch(&repo.default_branch).to_owned(),
Agents and memory, checks and conflicts, profiles, slug renames, custom domains449 base,
450 head,
g1t is the stored author of what it opens; the person who asked is requested_by and keeps the author's rights451 author: pull.requested_by.unwrap_or(pull.author),
Agents and memory, checks and conflicts, profiles, slug renames, custom domains452 }))
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}