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