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.

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