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
1,090 linesCodeBlameRaw
1//! Where a pull request stands, and merging it on the repository's say.
2//!
3//! A pull request g1t's agent once made (`made_by_g1t`, `managed = 1`)
4//! still shows where it stands: its checks, the approvals the branch
5//! wants, the merge queue, and whether a person has to step in. Where the
6//! repository's settings say ready pull requests land by themselves
7//! (`auto_merge`), such a pull request is merged here once everything its
8//! branch's rules ask for is met, attributed to `g1t` (`POLICY_ACTOR_ID`).
9//!
10//! A merge asked for while a pull request is behind is recorded here
11//! (`request_landing`). Once the push that brings it up to date arrives
12//! (`land_if_requested`), it lands when the checks the branch requires
13//! pass on that new commit, as they must for any merge (`land_when_checked`,
14//! also called as each of the commit's checks finishes).
15
16use g1t_contracts::repos::{GetByIdArgs, Repo, RepoPath};
17use g1t_contracts::time::rfc3339;
18use g1t_contracts::work::*;
19use g1t_contracts::{FailureCode, Membership, Outcome, User, Viewer};
20use g1t_kit::now_ms;
21use std::collections::HashMap;
22
23use serde::Deserialize;
24use worker::Result;
25use worker::wasm_bindgen::JsValue;
26
27use crate::Work;
28use crate::prefetch::Slot;
29use crate::reviews::{AGENT_ID, AGENT_NAME};
30use crate::statuses::{self, WorkflowFacts};
31use crate::rows::ValueRow;
32
33/// How long a claimed step is waited for before it may be taken again.
34const CATCH_UP_MINUTES: u64 = 30;
35const MERGE_MINUTES: u64 = 2;
36/// How long a merge that was brought up to date waits for the required
37/// checks on the new commit; after that it is no longer pending.
38const CHECKS_MINUTES: u64 = 180;
39/// The step under way while it does (`working_on`).
40const WAITING_FOR_CHECKS: &str = "checks";
41
42/// How long a merge waits on the repository service for one write (the
43/// catch-up push, the landing): long enough for a large merge, short
44/// enough that the person who clicked hears back well before the site
45/// stops waiting on the merge (apps/web/app/lib/deadline.ts, 60 s).
46pub(crate) const REPOS_WRITE_WITHIN: std::time::Duration = std::time::Duration::from_secs(25);
47
48/// What a merge says when the repository service did not answer in time.
49pub(crate) fn repository_unreachable<T>() -> Outcome<T> {
50 Outcome::fail(
51 FailureCode::Unavailable,
52 "g1t couldn't reach the repository to merge this. Try again in a moment.",
53 )
54}
55
56/// What becomes of a merge that was brought up to date, by the checks the
57/// branch requires on the commit that did it.
58#[derive(Debug, PartialEq)]
59pub(crate) enum Landing {
60 /// None are required, or all of them passed.
61 Now,
62 /// Some are still running or have not reported yet.
63 Wait,
64 /// One failed: a person has to push a fix.
65 Stall(String),
66}
67
68/// Whether a merge that was brought up to date lands now, by its new
69/// commit's checks. The same checks hold it as hold any merge
70/// (`merge_gate`): landing it before they report would only be refused.
71pub(crate) fn landing_step(facts: &WorkflowFacts) -> Landing {
72 let failed = facts.required_failed();
73 if !failed.is_empty() {
74 let one = failed.len() == 1;
75 return Landing::Stall(format!(
76 "It was brought up to date, but the required {} {} failed on that commit, so it was not merged. Push a fix, then merge again.",
77 if one { "check" } else { "checks" },
78 statuses::list(&failed),
79 ));
80 }
81 if facts.required.iter().any(|check| check.state != RequiredState::Success) {
82 return Landing::Wait;
83 }
84 Landing::Now
85}
86
87/// Who a merge made by a repository's settings is attributed to. Not an
88/// account: `g1t` cannot be registered.
89pub(crate) const POLICY_ACTOR_ID: &str = "g1t_policy";
90pub(crate) const POLICY_ACTOR_NAME: &str = "g1t";
91const MANAGED_PAGE: u32 = 200;
92
93/// The part of a pull request's row that tracks its lifecycle.
94#[derive(Deserialize)]
95struct Progress {
96 managed: u8,
97 revisions: u32,
98 revised_at: Option<String>,
99 working_on: Option<String>,
100 working_until: Option<String>,
101 stalled: Option<String>,
102}
103
104/// A merge asked for while the pull request was behind, as stored: a
105/// fast-forward, which needs it up to date first (merging.rs).
106#[derive(serde::Serialize, Deserialize)]
107#[serde(rename_all = "camelCase")]
108struct LandRequest {
109 actor: User,
110 keep_issue_open: bool,
111 /// Delete its branch once merged, as asked; the repository's setting
112 /// when absent.
113 #[serde(default)]
114 delete_branch: Option<bool>,
115 /// The merger asked to merge past the checks, and the repository
116 /// allows that: it lands without waiting for them.
117 #[serde(default)]
118 ignore_checks: bool,
119}
120
121#[derive(Deserialize)]
122struct LandRow {
123 land_requested: Option<String>,
124 land_requested_at: Option<String>,
125 stalled: Option<String>,
126 #[serde(default)]
127 working_on: Option<String>,
128}
129
130/// What should happen next, if it is g1t's turn.
131pub(crate) enum Next {
132 /// A step is under way, or it is a person's turn.
133 Wait,
134 /// Land it, because the repository says ready pull requests land.
135 Merge,
136}
137
138/// Whether g1t's agent made this pull request, and so g1t shows where it
139/// stands. Only pull requests from before g1t stopped hosting agents.
140pub(crate) fn made_by_g1t(pull: &Pull) -> bool {
141 pull.runtime == Runtime::Hosted && pull.agent == AGENT_NAME && pull.fork.is_some()
142}
143
144/// Whether a verdict by `reviewer` is a person's other than the pull
145/// request's owner (whoever asked g1t for it, or its author): neither
146/// theirs nor g1t's agent's.
147pub(crate) fn from_someone_else(pull: &Pull, reviewer: &str) -> bool {
148 !pull.is_owned_by(reviewer) && reviewer != AGENT_ID
149}
150
151fn at(stage: Stage, detail: impl Into<String>, revisions: u32) -> Lifecycle {
152 Lifecycle {
153 stage,
154 detail: detail.into(),
155 revisions,
156 }
157}
158
159/// Everything the next step depends on.
160struct Facts {
161 /// Still being made: not yet marked ready for review.
162 draft: bool,
163 /// Why g1t stopped, if it has.
164 stalled: Option<String>,
165 /// The step under way, if one was claimed and is still being waited for.
166 working_on: Option<String>,
167 /// `failed` when the merge queue took it out.
168 check_status: Option<CheckStatus>,
169 revisions: u32,
170 /// Whether the branch it would land on has moved without it.
171 behind: bool,
172 /// Whether it is known to conflict with the branch it would land on.
173 conflicting: bool,
174 /// Whether the repository lands a ready pull request by itself.
175 auto_merge: bool,
176 /// Whether the repository refuses to merge one that is behind.
177 require_up_to_date: bool,
178 /// What the repository's approval rule still wants, if anything.
179 approvals_missing: Option<String>,
180 /// The person who asked for changes since the last revision, if one did.
181 person_request: Option<String>,
182 /// Its place in the merge queue, and what is ahead of it there.
183 queued: Option<(QueueState, Vec<u32>)>,
184 /// What the checks on its head say, against the required ones.
185 workflows: WorkflowFacts,
186}
187
188/// Where a pull request stands, and whether to merge it.
189///
190/// The order is: nothing while a step is under way; a person asking for
191/// changes holds it; its checks must finish, and the required ones pass;
192/// it must not conflict, and where the repository insists, it must be up
193/// to date; then the approvals the branch wants. Once everything is met,
194/// it merges where the repository says so.
195fn decide(facts: Facts) -> (Lifecycle, Next) {
196 let revisions = facts.revisions;
197 let wait = |stage, detail: &str| (at(stage, detail, revisions), Next::Wait);
198
199 if facts.draft {
200 return wait(Stage::Working, "It is still a draft.");
201 }
202 if let Some(reason) = &facts.stalled {
203 return wait(Stage::NeedsYou, reason);
204 }
205 if let Some((state, ahead)) = &facts.queued {
206 let named = ahead.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", ");
207 let detail = match (state, ahead.is_empty()) {
208 (QueueState::Testing, true) => "In the merge queue: being tested on the default branch as it is.".to_owned(),
209 (QueueState::Testing, false) => {
210 format!("In the merge queue: being tested together with {named}, ahead of it.")
211 }
212 (QueueState::Passed, true) => "Passed in the merge queue. Landing.".to_owned(),
213 (QueueState::Passed, false) => {
214 format!("Passed in the merge queue together with {named}. It lands once they have.")
215 }
216 _ => "In the merge queue, waiting for its turn to be tested.".to_owned(),
217 };
218 return wait(Stage::Queued, &detail);
219 }
220 match facts.working_on.as_deref() {
221 Some("catch_up") => {
222 return wait(Stage::CatchingUp, "Bringing it up to date with the branch it will land on.");
223 }
224 Some("merge") => return wait(Stage::Ready, "Merging."),
225 Some(WAITING_FOR_CHECKS) => {
226 return wait(
227 Stage::Checking,
228 "Brought up to date. It lands once the required checks pass on that commit.",
229 );
230 }
231 _ => {}
232 }
233 if let Some(name) = &facts.person_request {
234 return wait(Stage::NeedsYou, &format!("{name} asked for changes."));
235 }
236 match facts.check_status {
237 Some(CheckStatus::Failed) => return wait(Stage::NeedsYou, "It failed in the merge queue."),
238 Some(CheckStatus::Errored) => return wait(Stage::NeedsYou, "Its checks could not be run."),
239 _ => {}
240 }
241 // Only the checks the branch requires hold it.
242 let required = facts.workflows.required_failed();
243 if !required.is_empty() {
244 return wait(
245 Stage::NeedsYou,
246 &format!(
247 "The required {} {} {}.",
248 if required.len() == 1 { "check" } else { "checks" },
249 statuses::list(&required),
250 if required.len() == 1 { "fails" } else { "fail" },
251 ),
252 );
253 }
254 if !facts.workflows.pending.is_empty() {
255 return wait(
256 Stage::Checking,
257 &format!("Waiting for {} to finish.", statuses::list(&facts.workflows.pending)),
258 );
259 }
260 let expected = facts.workflows.expected();
261 if !expected.is_empty() {
262 return wait(
263 Stage::Checking,
264 &format!(
265 "Waiting for the required {} {} to report on its latest commit.",
266 if expected.len() == 1 { "check" } else { "checks" },
267 statuses::list(&expected)
268 ),
269 );
270 }
271 if facts.conflicting {
272 return wait(
273 Stage::NeedsYou,
274 "It conflicts with the branch it will land on. Merge that branch in, resolve the conflicts and push.",
275 );
276 }
277 if facts.behind && facts.require_up_to_date {
278 return wait(
279 Stage::NeedsYou,
280 "The branch it will land on has moved, and this repository requires pull requests to be up to date. Catch up with it first.",
281 );
282 }
283 if let Some(missing) = &facts.approvals_missing {
284 return wait(Stage::NeedsYou, missing);
285 }
286 if facts.auto_merge {
287 return (
288 at(
289 Stage::Ready,
290 "Everything this repository asks for is met. Merging, as its settings say.",
291 revisions,
292 ),
293 Next::Merge,
294 );
295 }
296 wait(
297 Stage::Ready,
298 if facts.behind {
299 "Ready to merge. Merging brings it up to date with the default branch first."
300 } else {
301 "Everything this repository asks for is met. Ready to merge."
302 },
303 )
304}
305
306impl Work {
307 /// Where a pull request stands and whether g1t merges it, remembered
308 /// so lists can show it without working it out again. `None` for one
309 /// g1t is not seeing through.
310 pub(crate) async fn assess(
311 &self,
312 pull: &Pull,
313 _issue: &Option<Issue>,
314 behind: bool,
315 ) -> Result<Option<(Lifecycle, Next)>> {
316 let assessed = self.assess_now(pull, behind).await?;
317 if let Some((lifecycle, _)) = &assessed
318 && !self.remembered_as(&pull.id, lifecycle)?
319 {
320 self.remember(&pull.id, lifecycle).await?;
321 }
322 Ok(assessed)
323 }
324
325 /// Whether the row read for this request already says `lifecycle`, so
326 /// that showing a pull request does not write it again unchanged.
327 fn remembered_as(&self, pull_id: &str, lifecycle: &Lifecycle) -> Result<bool> {
328 let Some(found) = self.prefetched_pull(pull_id) else {
329 return Ok(false);
330 };
331 Ok(found
332 .first::<crate::rows::Snapshot>(Slot::Pull)?
333 .is_some_and(|stored| {
334 stored.stage == Some(lifecycle.stage) && stored.stage_detail.as_deref() == Some(lifecycle.detail.as_str())
335 }))
336 }
337
338 /// Saves where a pull request stands.
339 pub(crate) async fn remember(&self, pull_id: &str, lifecycle: &Lifecycle) -> Result<()> {
340 let stage = serde_json::to_value(lifecycle.stage)?;
341 self.db
342 .prepare("UPDATE pulls SET stage = ?, stage_detail = ? WHERE id = ?")
343 .bind(&[
344 stage.as_str().unwrap_or_default().into(),
345 lifecycle.detail.as_str().into(),
346 pull_id.into(),
347 ])?
348 .run()
349 .await?;
350 Ok(())
351 }
352
353 async fn assess_now(&self, pull: &Pull, behind: bool) -> Result<Option<(Lifecycle, Next)>> {
354 if !pull.status.is_active() {
355 return Ok(None);
356 }
357 // As read for this request (prefetch.rs), or now.
358 let prefetched = self.prefetched_pull(&pull.id);
359 let progress = match &prefetched {
360 Some(found) => found.first::<Progress>(Slot::Pull)?,
361 None => {
362 self.db
363 .prepare(
364 "SELECT managed, revisions, revised_at, working_on, working_until, stalled
365 FROM pulls WHERE id = ?",
366 )
367 .bind(&[pull.id.as_str().into()])?
368 .first::<Progress>(None)
369 .await?
370 }
371 };
372 let Some(progress) = progress.filter(|progress| progress.managed != 0) else {
373 return Ok(None);
374 };
375 let now = rfc3339(now_ms());
376 let working_on = progress
377 .working_until
378 .as_deref()
379 .is_some_and(|until| until > now.as_str())
380 .then(|| progress.working_on.clone().unwrap_or_default());
381 let settings = self.settings_for(pull).await?;
382 let statuses = self.statuses(&pull.repo_id, pull.head_commit.as_deref()).await?;
383 let workflows = self.workflow_facts(&pull.repo_id, pull.head_commit.as_deref(), &statuses, &settings.required_checks).await?;
384 // The rules of the branch it merges into, as g1t sees them when it
385 // merges by itself: what people must still do, and whether an
386 // agent's change may land here unattended.
387 let gate = match self.repo_for(pull).await? {
388 Some(repo) => Some(self.merge_gate(&repo, pull, Some(&User::system(&repo.namespace)), false, true).await?),
389 None => None,
390 };
391 let rules_allow_auto_merge = gate
392 .as_ref()
393 .is_none_or(|gate| g1t_rules::merge::auto_merge_refusal(&gate.requirements).is_none());
394 let (lifecycle, next) = decide(Facts {
395 draft: pull.status == PullStatus::Draft,
396 stalled: progress.stalled,
397 working_on,
398 check_status: pull.check_status,
399 revisions: progress.revisions,
400 behind,
401 conflicting: self.conflicting_files(pull).await?.is_some(),
402 auto_merge: settings.auto_merge && rules_allow_auto_merge,
403 require_up_to_date: settings.require_up_to_date,
404 approvals_missing: gate.as_ref().and_then(|gate| gate.people_gap()),
405 person_request: self.person_request(pull, progress.revised_at.as_deref()).await?,
406 queued: self.queued_entry(&pull.id).await?,
407 workflows,
408 });
409 Ok(Some((lifecycle, next)))
410 }
411
412 /// Who last asked for changes, if it was a person other than its owner
413 /// (whoever asked g1t for it, or its author), that person's latest
414 /// verdict, and after the last revision.
415 async fn person_request(&self, pull: &Pull, revised_at: Option<&str>) -> Result<Option<String>> {
416 #[derive(Deserialize)]
417 struct Verdicts {
418 author_id: String,
419 author_name: String,
420 verdict: Verdict,
421 created_at: String,
422 }
423 let rows = match self.prefetched_pull(&pull.id) {
424 Some(found) => found
425 .rows::<Verdicts>(Slot::Verdicts)?
426 .into_iter()
427 .filter(|row| from_someone_else(pull, &row.author_id))
428 .collect(),
429 None => self
430 .db
431 .prepare(
432 "SELECT author_id, author_name, verdict, created_at FROM comments
433 WHERE repo_id = ? AND number = ? AND verdict IS NOT NULL
434 AND author_id != ? AND author_id != ?
435 ORDER BY id",
436 )
437 .bind(&[
438 pull.repo_id.as_str().into(),
439 pull.number.into(),
440 pull.owner().id.as_str().into(),
441 AGENT_ID.into(),
442 ])?
443 .all()
444 .await?
445 .results::<Verdicts>()?,
446 };
447 // Each person's latest verdict is the one that stands.
448 let mut latest: HashMap<String, Verdicts> = HashMap::new();
449 for row in rows {
450 latest.insert(row.author_id.clone(), row);
451 }
452 Ok(latest
453 .into_values()
454 .filter(|row| row.verdict == Verdict::RequestChanges)
455 .filter(|row| revised_at.is_none_or(|revised| row.created_at.as_str() > revised))
456 .max_by(|a, b| a.created_at.cmp(&b.created_at))
457 .map(|row| row.author_name))
458 }
459
460 /// Takes a step for a pull request, if nobody else has. One statement,
461 /// so that two callers cannot both take it.
462 pub(crate) async fn claim(&self, pull_id: &str, step: &str, minutes: u64) -> Result<bool> {
463 let now = now_ms();
464 Ok(self
465 .db
466 .prepare(
467 "UPDATE pulls SET working_on = ?2, working_until = ?3
468 WHERE id = ?4 AND status = 'open' AND stalled IS NULL
469 AND (working_until IS NULL OR working_until < ?1)
470 RETURNING id AS value",
471 )
472 .bind(&[
473 rfc3339(now).into(),
474 step.into(),
475 rfc3339(now + minutes * 60 * 1000).into(),
476 pull_id.into(),
477 ])?
478 .first::<ValueRow>(None)
479 .await?
480 .is_some())
481 }
482
483 /// Merges a pull request g1t is seeing through once it is ready, where
484 /// the repository's settings say ready pull requests land by
485 /// themselves. Returns what the runner is to do, which is now always
486 /// nothing: no step is carried out in a sandbox any more.
487 pub(crate) async fn advance(&self, a: AdvanceArgs) -> Result<Advance> {
488 let Some(pull) = self.pull_by_id(&a.pull_id).await? else {
489 return Ok(Advance::None);
490 };
491 if pull.status != PullStatus::Open {
492 return Ok(Advance::None);
493 }
494 // A person asked for it to merge on its own (automerge.rs).
495 if pull.auto_merge.is_some() {
496 self.try_auto_merge(&pull).await?;
497 return Ok(Advance::None);
498 }
499 // As a member: a private repository would look missing otherwise,
500 // and the pull request would never move.
501 let viewer: Viewer = self.owner_viewer(&pull).await?;
502 let repo: Outcome<Repo> = g1t_kit::call(
503 &self.repos,
504 "get_by_id",
505 &GetByIdArgs {
506 id: pull.repo_id.clone(),
507 viewer,
508 },
509 )
510 .await?;
511 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
512 return Ok(Advance::None);
513 };
514 let behind = self.is_behind(&pull).await?;
515 if let Some((_, Next::Merge)) = self.assess(&pull, &None, behind).await? {
516 self.merge_by_policy(&repo, &pull).await?;
517 }
518 Ok(Advance::None)
519 }
520
521 /// Lands a pull request that is ready, on the authority of the
522 /// repository's settings instead of a person's click.
523 async fn merge_by_policy(&self, repo: &Repo, pull: &Pull) -> Result<()> {
524 if !self.claim(&pull.id, "merge", MERGE_MINUTES).await? {
525 return Ok(());
526 }
527 // g1t acts for the workspace whose members turned this on, with the
528 // default base permission (Write), which merging needs.
529 let actor = User {
530 id: POLICY_ACTOR_ID.to_owned(),
531 username: POLICY_ACTOR_NAME.to_owned(),
532 verified: true,
533 workspaces: vec![Membership::member(repo.namespace.to_lowercase())],
534 ..User::default()
535 };
536 let merged = self
537 .merge_pull(PullActionArgs {
538 actor,
539 repo: RepoPath {
540 namespace: repo.namespace.clone(),
541 name: repo.name.clone(),
542 },
543 number: pull.number,
544 summary: String::new(),
545 keep_issue_open: false,
546 ignore_checks: false,
547 bypass_rules: false,
548 // The repository's default method, message and branch
549 // setting.
550 merge_method: None,
551 commit_title: None,
552 commit_message: None,
553 expected_head: pull.head_commit.clone(),
554 delete_branch: None,
555 update_method: None,
556 })
557 .await?;
558 match merged {
559 Outcome::Ok(_) => Ok(()),
560 // Most likely the branch moved in the moment between: let go, and
561 // the next look at it tries again.
562 Outcome::Fail(failure) if failure.code == FailureCode::Conflict => {
563 self.db
564 .prepare(
565 "UPDATE pulls SET working_on = NULL, working_until = NULL
566 WHERE id = ? AND working_on = 'merge'",
567 )
568 .bind(&[pull.id.as_str().into()])?
569 .run()
570 .await?;
571 Ok(())
572 }
573 Outcome::Fail(failure) => {
574 self.stall(StallArgs {
575 pull_id: pull.id.clone(),
576 reason: format!("g1t could not merge this: {}", failure.message),
577 by: None,
578 })
579 .await?;
580 Ok(())
581 }
582 }
583 }
584
585 /// Records that a merge was asked for while the pull request was
586 /// behind, so that it lands once the push that brings it up to date
587 /// arrives, and announces it.
588 pub(crate) async fn request_landing(
589 &self,
590 pull: &Pull,
591 actor: &User,
592 keep_issue_open: bool,
593 delete_branch: Option<bool>,
594 ignore_checks: bool,
595 ) -> Result<()> {
596 let now = now_ms();
597 let request = serde_json::to_string(&LandRequest {
598 actor: actor.clone(),
599 keep_issue_open,
600 delete_branch,
601 ignore_checks,
602 })?;
603 self.db
604 .prepare(
605 "UPDATE pulls
606 SET land_requested = ?, land_requested_at = ?, stalled = NULL,
607 working_on = 'catch_up', working_until = ?
608 WHERE id = ?",
609 )
610 .bind(&[
611 request.into(),
612 rfc3339(now).into(),
613 rfc3339(now + CATCH_UP_MINUTES * 60 * 1000).into(),
614 pull.id.as_str().into(),
615 ])?
616 .run()
617 .await?;
618 self.publish(
619 "pull.merge_requested",
620 &pull.repo_id,
621 actor,
622 Self::pull_event(pull),
623 )
624 .await
625 }
626
627 /// The merge waiting on a pull request, if one was asked for recently
628 /// enough to still stand.
629 async fn land_request(&self, pull_id: &str) -> Result<Option<LandRequest>> {
630 let row = self.land_row(pull_id).await?;
631 let now = now_ms();
632 Ok(row
633 .filter(|row| {
634 // Waiting for checks on the caught-up commit stands longer
635 // than waiting for the push.
636 let minutes = if row.working_on.as_deref() == Some(WAITING_FOR_CHECKS) {
637 CHECKS_MINUTES
638 } else {
639 CATCH_UP_MINUTES
640 };
641 let oldest = rfc3339(now.saturating_sub(minutes * 60 * 1000));
642 row.land_requested_at
643 .as_deref()
644 .is_some_and(|at| at >= oldest.as_str())
645 })
646 .and_then(|row| row.land_requested)
647 .and_then(|request| serde_json::from_str(&request).ok()))
648 }
649
650 /// Whether a merge is waiting on a pull request, and why g1t stopped
651 /// working on it if it did.
652 pub(crate) async fn landing_state(&self, pull_id: &str) -> Result<(bool, Option<String>)> {
653 let stalled = self.land_row(pull_id).await?.and_then(|row| row.stalled);
654 Ok((self.land_request(pull_id).await?.is_some(), stalled))
655 }
656
657 /// The merge waiting on a pull request and why g1t stopped, as read
658 /// for this request (prefetch.rs) or now.
659 async fn land_row(&self, pull_id: &str) -> Result<Option<LandRow>> {
660 if let Some(found) = self.prefetched_pull(pull_id) {
661 return found.first::<LandRow>(Slot::Pull);
662 }
663 self.db
664 .prepare("SELECT land_requested, land_requested_at, stalled, working_on FROM pulls WHERE id = ?")
665 .bind(&[pull_id.into()])?
666 .first::<LandRow>(None)
667 .await
668 }
669
670 /// Forgets a merge that was waiting on a pull request, and the step
671 /// that was bringing it up to date or waiting for its checks.
672 pub(crate) async fn forget_landing(&self, pull_id: &str) -> Result<()> {
673 self.take_landing(pull_id).await?;
674 Ok(())
675 }
676
677 /// Forgets a merge that was waiting on a pull request and returns it,
678 /// in one statement, so that of two callers deciding to land it at
679 /// once (two checks finishing together) only one does.
680 async fn take_landing(&self, pull_id: &str) -> Result<Option<LandRequest>> {
681 // SET sees the row as it was: the CASEs read the old `working_on`.
682 let taken = self
683 .db
684 .prepare(
685 "UPDATE pulls SET land_requested = NULL, land_requested_at = NULL,
686 working_on = CASE WHEN working_on IN ('catch_up', 'checks') THEN NULL ELSE working_on END,
687 working_until = CASE WHEN working_on IN ('catch_up', 'checks') THEN NULL ELSE working_until END
688 WHERE id = ? AND land_requested IS NOT NULL
689 RETURNING land_requested AS value",
690 )
691 .bind(&[pull_id.into()])?
692 .first::<ValueRow>(None)
693 .await?;
694 Ok(taken.and_then(|row| serde_json::from_str(&row.value).ok()))
695 }
696
697 /// Notes that a merge which was brought up to date waits for the
698 /// required checks on its new commit, from now: on the first look at
699 /// this commit (a later push clears the step and starts it again).
700 async fn wait_for_checks(&self, pull_id: &str) -> Result<()> {
701 let now = now_ms();
702 self.db
703 .prepare(
704 "UPDATE pulls SET working_on = 'checks', working_until = ?1, land_requested_at = ?2
705 WHERE id = ?3 AND land_requested IS NOT NULL AND status = 'open'
706 AND (working_on IS NULL OR working_on <> 'checks')",
707 )
708 .bind(&[
709 rfc3339(now + CHECKS_MINUTES * 60 * 1000).into(),
710 rfc3339(now).into(),
711 pull_id.into(),
712 ])?
713 .run()
714 .await?;
715 Ok(())
716 }
717
718 /// Moves on a merge that was waiting for a pull request's head to move
719 /// (the push that brought it up to date): it lands once the checks the
720 /// branch requires pass on that commit ([`Work::land_when_checked`]).
721 /// The commit is new, so they are waited for, as for any merge: a
722 /// ruleset that requires them refuses a merge that skips them.
723 pub(crate) async fn land_if_requested(&self, pull_id: &str) -> Result<()> {
724 if self.land_request(pull_id).await?.is_none() {
725 return Ok(());
726 }
727 self.land_when_checked(pull_id).await
728 }
729
730 /// Lands a pull request that was brought up to date for a merge, once
731 /// the checks the branch requires pass on its head: at once when none
732 /// are required, never when one failed (it stops and says which), and
733 /// otherwise when the last of them reports (statuses.rs `store_status`
734 /// calls this as the commit's checks finish).
735 pub(crate) async fn land_when_checked(&self, pull_id: &str) -> Result<()> {
736 let Some(request) = self.land_request(pull_id).await? else {
737 return Ok(());
738 };
739 let Some(pull) = self.pull_by_id(pull_id).await? else {
740 return Ok(());
741 };
742 if pull.status != PullStatus::Open {
743 return self.forget_landing(pull_id).await;
744 }
745 let repo: Outcome<Repo> = g1t_kit::call(
746 &self.repos,
747 "get_by_id",
748 &GetByIdArgs {
749 id: pull.repo_id.clone(),
750 viewer: Some(request.actor.clone()),
751 },
752 )
753 .await?;
754 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
755 return self.forget_landing(pull_id).await;
756 };
757 let step = if request.ignore_checks {
758 Landing::Now
759 } else {
760 let settings = self.settings_on(&repo, &pull).await?;
761 let statuses = self.statuses(&pull.repo_id, pull.head_commit.as_deref()).await?;
762 let facts = self
763 .workflow_facts(&pull.repo_id, pull.head_commit.as_deref(), &statuses, &settings.required_checks)
764 .await?;
765 landing_step(&facts)
766 };
767 match step {
768 Landing::Wait => return self.wait_for_checks(pull_id).await,
769 Landing::Stall(reason) => {
770 self.stall(StallArgs {
771 pull_id: pull.id,
772 reason,
773 by: None,
774 })
775 .await?;
776 return Ok(());
777 }
778 Landing::Now => {}
779 }
780 // Taken by one caller: two checks finishing together land it once.
781 let Some(request) = self.take_landing(pull_id).await? else {
782 return Ok(());
783 };
784 let merged = self
785 .merge_pull(PullActionArgs {
786 actor: request.actor,
787 repo: RepoPath {
788 namespace: repo.namespace,
789 name: repo.name,
790 },
791 number: pull.number,
792 summary: String::new(),
793 keep_issue_open: request.keep_issue_open,
794 ignore_checks: request.ignore_checks,
795 bypass_rules: false,
796 // Only a fast-forward waits to be brought up to date.
797 merge_method: Some(g1t_contracts::rules::MergeMethod::FastForward),
798 commit_title: None,
799 commit_message: None,
800 expected_head: None,
801 delete_branch: request.delete_branch,
802 update_method: None,
803 })
804 .await?;
805 if let Outcome::Fail(failure) = merged {
806 self.stall(StallArgs {
807 pull_id: pull.id,
808 reason: format!(
809 "It was brought up to date but could not be merged: {}",
810 failure.message
811 ),
812 by: None,
813 })
814 .await?;
815 }
816 Ok(())
817 }
818
819 /// Stops seeing a pull request through until a person steps in. The
820 /// first stop is published (`pull.stalled`), which tells its people
821 /// that it needs them; a stop on one already stopped only says why.
822 pub(crate) async fn stall(&self, a: StallArgs) -> Result<bool> {
823 let before = self.pull_by_id(&a.pull_id).await?;
824 let was_stalled = self.is_stalled(&a.pull_id).await?;
825 self.db
826 .prepare(
827 "UPDATE pulls
828 SET stalled = ?, working_on = NULL, working_until = NULL,
829 land_requested = NULL, land_requested_at = NULL
830 WHERE id = ? AND status = 'open'",
831 )
832 .bind(&[a.reason.trim().into(), a.pull_id.as_str().into()])?
833 .run()
834 .await?;
835 self.db
836 .prepare("UPDATE pulls SET stage = 'needs_you', stage_detail = ? WHERE id = ?")
837 .bind(&[a.reason.trim().into(), a.pull_id.as_str().into()])?
838 .run()
839 .await?;
840 if let Some(pull) = before.filter(|pull| !was_stalled && pull.status == PullStatus::Open) {
841 self.publish_as(
842 "pull.stalled",
843 &pull.repo_id,
844 a.by.clone(),
845 g1t_contracts::events::PullEvent {
846 detail: Some(a.reason.trim().to_owned()),
847 ..Self::pull_event(&pull)
848 },
849 )
850 .await?;
851 }
852 Ok(true)
853 }
854
855 /// Whether g1t has stopped seeing the pull request through.
856 pub(crate) async fn is_stalled(&self, pull_id: &str) -> Result<bool> {
857 Ok(self
858 .db
859 .prepare("SELECT 1 AS value FROM pulls WHERE id = ? AND stalled IS NOT NULL")
860 .bind(&[pull_id.into()])?
861 .first::<u32>(Some("value"))
862 .await?
863 .is_some())
864 }
865
866 /// Says that a pull request g1t had stopped on is going again, which
867 /// closes what it was waiting on a person for.
868 pub(crate) async fn announce_resumed(&self, pull_id: &str, actor: Option<String>) -> Result<()> {
869 if let Some(pull) = self.pull_by_id(pull_id).await? {
870 self.publish_as("pull.resumed", &pull.repo_id, actor, Self::pull_event(&pull)).await?;
871 }
872 Ok(())
873 }
874
875 pub(crate) async fn managed_pulls(&self, a: ManagedPullsArgs) -> Result<Vec<String>> {
876 let rows = self
877 .db
878 .prepare(
879 "SELECT id AS value FROM pulls
880 WHERE status = 'open' AND ((managed = 1 AND stalled IS NULL) OR auto_merge IS NOT NULL)
881 AND (?1 IS NULL OR repo_id = ?1)
882 ORDER BY updated_at DESC LIMIT ?2",
883 )
884 .bind(&[
885 a.repo_id.map_or(JsValue::NULL, JsValue::from),
886 MANAGED_PAGE.into(),
887 ])?
888 .all()
889 .await?
890 .results::<ValueRow>()?;
891 Ok(rows.into_iter().map(|row| row.value).collect())
892 }
893}
894
895#[cfg(test)]
896mod tests {
897 use super::*;
898
899 #[test]
900 fn a_verdict_from_whoever_asked_for_g1t_s_change_is_not_someone_else_s() {
901 use crate::rows::stored::{ASKER, G1T, pull};
902 let made = pull(G1T, Some(ASKER));
903 assert!(made_by_g1t(&made));
904 assert!(!from_someone_else(&made, ASKER.0));
905 assert!(!from_someone_else(&made, AGENT_ID));
906 assert!(from_someone_else(&made, "usr_reviewer"));
907 let own = pull(ASKER, None);
908 assert!(!from_someone_else(&own, ASKER.0));
909 assert!(from_someone_else(&own, "usr_reviewer"));
910 }
911
912 /// Checks on a head commit: `(context, state)` statuses, against the
913 /// required check names.
914 fn checks(statuses: &[(&str, &str)], required: &[&str]) -> WorkflowFacts {
915 let statuses: Vec<CommitStatus> = statuses
916 .iter()
917 .map(|(context, state)| CommitStatus {
918 context: (*context).to_owned(),
919 state: (*state).to_owned(),
920 description: None,
921 target_url: None,
922 updated_at: String::new(),
923 source: None,
924 check_run_id: None,
925 })
926 .collect();
927 let required: Vec<String> = required.iter().map(|name| (*name).to_owned()).collect();
928 WorkflowFacts::of(&statuses, &required)
929 }
930
931 /// A pull request that is ready for review, whose required check
932 /// passed, and nothing else yet.
933 fn facts() -> Facts {
934 Facts {
935 draft: false,
936 stalled: None,
937 working_on: None,
938 check_status: None,
939 revisions: 0,
940 behind: false,
941 conflicting: false,
942 auto_merge: false,
943 require_up_to_date: false,
944 approvals_missing: None,
945 person_request: None,
946 queued: None,
947 workflows: checks(&[("CI / pull_request", "success")], &["CI"]),
948 }
949 }
950
951 /// The stage, and a word for the step to take.
952 fn outcome(facts: Facts) -> (Stage, &'static str) {
953 let (lifecycle, next) = decide(facts);
954 let step = match next {
955 Next::Wait => "wait",
956 Next::Merge => "merge",
957 };
958 (lifecycle.stage, step)
959 }
960
961 #[test]
962 fn a_ready_change_lands_by_itself_only_where_the_repository_says_so() {
963 assert_eq!(outcome(facts()), (Stage::Ready, "wait"));
964 assert_eq!(outcome(Facts { auto_merge: true, ..facts() }), (Stage::Ready, "merge"));
965 // One that is behind is merged too: merging brings it up to date.
966 assert_eq!(outcome(Facts { auto_merge: true, behind: true, ..facts() }), (Stage::Ready, "merge"));
967 // Unless the repository wants it caught up first.
968 let strict = Facts { auto_merge: true, behind: true, require_up_to_date: true, ..facts() };
969 assert_eq!(outcome(strict), (Stage::NeedsYou, "wait"));
970 }
971
972 #[test]
973 fn approvals_the_repository_wants_are_waited_for() {
974 let short = Facts {
975 approvals_missing: Some("This repository requires 1 approving review.".to_owned()),
976 auto_merge: true,
977 ..facts()
978 };
979 let (lifecycle, next) = decide(short);
980 assert_eq!(lifecycle.stage, Stage::NeedsYou);
981 assert_eq!(lifecycle.detail, "This repository requires 1 approving review.");
982 assert!(matches!(next, Next::Wait));
983 }
984
985 #[test]
986 fn checks_are_waited_for_and_a_failing_required_one_needs_a_person() {
987 let running = Facts { workflows: checks(&[("CI / pull_request", "pending")], &["CI"]), auto_merge: true, ..facts() };
988 let (lifecycle, next) = decide(running);
989 assert_eq!(lifecycle.stage, Stage::Checking);
990 assert!(lifecycle.detail.contains("Waiting for CI / pull_request"));
991 assert!(matches!(next, Next::Wait));
992 let failed = Facts { workflows: checks(&[("CI / pull_request", "failure")], &["CI"]), ..facts() };
993 let (lifecycle, _) = decide(failed);
994 assert_eq!(lifecycle.stage, Stage::NeedsYou);
995 assert_eq!(lifecycle.detail, "The required check CI fails.");
996 // A check the branch does not require does not hold it.
997 let lint = checks(&[("CI / pull_request", "success"), ("Lint / pull_request", "failure")], &["CI"]);
998 assert_eq!(outcome(Facts { workflows: lint, ..facts() }), (Stage::Ready, "wait"));
999 let missing = Facts { workflows: checks(&[("CI / pull_request", "success")], &["CI", "Deploy"]), auto_merge: true, ..facts() };
1000 let (lifecycle, next) = decide(missing);
1001 assert_eq!(lifecycle.detail, "Waiting for the required check Deploy to report on its latest commit.");
1002 assert!(matches!(next, Next::Wait), "auto-merge must not land it");
1003 }
1004
1005 #[test]
1006 fn the_merge_queue_conflicts_and_people_hold_it() {
1007 let queued = Facts { queued: Some((QueueState::Testing, vec![12, 14])), auto_merge: true, ..facts() };
1008 let (lifecycle, _) = decide(queued);
1009 assert_eq!(lifecycle.stage, Stage::Queued);
1010 assert!(lifecycle.detail.contains("#12, #14"));
1011 assert_eq!(outcome(Facts { check_status: Some(CheckStatus::Failed), auto_merge: true, ..facts() }), (Stage::NeedsYou, "wait"));
1012 assert_eq!(outcome(Facts { check_status: Some(CheckStatus::Errored), ..facts() }), (Stage::NeedsYou, "wait"));
1013 assert_eq!(outcome(Facts { conflicting: true, auto_merge: true, ..facts() }), (Stage::NeedsYou, "wait"));
1014 let asked = Facts { person_request: Some("ana".to_owned()), auto_merge: true, ..facts() };
1015 let (lifecycle, next) = decide(asked);
1016 assert_eq!(lifecycle.detail, "ana asked for changes.");
1017 assert!(matches!(next, Next::Wait));
1018 }
1019
1020 #[test]
1021 fn a_merge_brought_up_to_date_lands_once_its_required_checks_pass_on_the_new_commit() {
1022 // The catch-up's merge commit, just pushed: nothing has reported yet.
1023 assert_eq!(landing_step(&checks(&[], &["CI"])), Landing::Wait);
1024 assert_eq!(landing_step(&checks(&[("CI / pull_request", "pending")], &["CI"])), Landing::Wait);
1025 assert_eq!(landing_step(&checks(&[("CI / pull_request", "success")], &["CI"])), Landing::Now);
1026 // Every required one, not just the first to report.
1027 assert_eq!(
1028 landing_step(&checks(&[("CI / pull_request", "success")], &["CI", "Deploy"])),
1029 Landing::Wait
1030 );
1031 // Nothing required: it lands at once, whatever else runs or failed.
1032 assert_eq!(landing_step(&checks(&[], &[])), Landing::Now);
1033 assert_eq!(landing_step(&checks(&[("Lint / pull_request", "failure")], &[])), Landing::Now);
1034 assert_eq!(
1035 landing_step(&checks(&[("CI / pull_request", "success"), ("Lint / pull_request", "failure")], &["CI"])),
1036 Landing::Now
1037 );
1038 // A required one that failed stops it, saying which.
1039 let Landing::Stall(reason) = landing_step(&checks(&[("CI / pull_request", "failure")], &["CI"])) else {
1040 panic!("a failing required check must stop it");
1041 };
1042 assert_eq!(
1043 reason,
1044 "It was brought up to date, but the required check CI failed on that commit, so it was not merged. Push a fix, then merge again."
1045 );
1046 let Landing::Stall(reason) =
1047 landing_step(&checks(&[("CI / push", "error"), ("Deploy", "failure")], &["CI", "Deploy"]))
1048 else {
1049 panic!("failing required checks must stop it");
1050 };
1051 assert!(reason.contains("required checks CI and Deploy failed"), "{reason}");
1052 }
1053
1054 #[test]
1055 fn a_merge_that_cannot_reach_the_repository_says_so() {
1056 let Outcome::Fail(failure) = repository_unreachable::<()>() else {
1057 panic!("it must fail");
1058 };
1059 assert_eq!(failure.code, FailureCode::Unavailable);
1060 assert!(failure.message.contains("couldn't reach the repository"));
1061 assert!(failure.message.contains("Try again"));
1062 // Well inside the site's own 60 s wait on a merge, catch-up and landing both.
1063 assert!(REPOS_WRITE_WITHIN.as_secs() * 2 < 60);
1064 }
1065
1066 #[test]
1067 fn a_landing_asked_for_before_it_carried_ignore_checks_still_reads() {
1068 let old: LandRequest = serde_json::from_value(serde_json::json!({
1069 "actor": serde_json::to_value(User::default()).unwrap(),
1070 "keepIssueOpen": true,
1071 }))
1072 .unwrap();
1073 assert!(old.keep_issue_open && !old.ignore_checks);
1074 }
1075
1076 #[test]
1077 fn drafts_steps_under_way_and_stops_wait() {
1078 assert_eq!(
1079 outcome(Facts { working_on: Some(WAITING_FOR_CHECKS.into()), auto_merge: true, ..facts() }),
1080 (Stage::Checking, "wait")
1081 );
1082 assert_eq!(outcome(Facts { draft: true, ..facts() }), (Stage::Working, "wait"));
1083 assert_eq!(outcome(Facts { working_on: Some("catch_up".into()), auto_merge: true, ..facts() }), (Stage::CatchingUp, "wait"));
1084 assert_eq!(outcome(Facts { working_on: Some("merge".into()), auto_merge: true, ..facts() }), (Stage::Ready, "wait"));
1085 let (lifecycle, next) = decide(Facts { stalled: Some("It could not be merged.".to_owned()), auto_merge: true, ..facts() });
1086 assert_eq!(lifecycle.stage, Stage::NeedsYou);
1087 assert_eq!(lifecycle.detail, "It could not be merged.");
1088 assert!(matches!(next, Next::Wait));
1089 }
1090}