Skip to content
Sign up

flagon-io/g1t

public
0
601 linesCodeBlameRaw

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.

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

This file's history is long; its oldest lines are credited to the oldest commit read.