Skip to content
Sign up

flagon-io/g1t

public
0
923 linesCodeBlameRaw
1//! The merge queue: pull requests are tested together with the ones ahead
2//! of them, and only a combination that passed reaches the default branch.
3//!
4//! A batch of up to the merge queue rule's `max_entries_to_build` entries (4 unless
5//! the default branch's rules say otherwise) is tested at once, speculatively: one
6//! sandbox per entry builds the default branch with that entry and every
7//! entry ahead of it merged in, and pushes the result to
8//! `g1t-queue/<entry>`. The repository's `merge_group` workflows then run
9//! on that commit, and the default branch's required checks must pass
10//! there. Entries land in order, each by moving the default branch to its
11//! tested state, once everything ahead has landed. One that fails leaves
12//! the queue with a failed check run; the entries behind it are tested
13//! again without it. Which entry answers for a failure, which entries
14//! ahead it breaks with, who is told and how it rejoins: integration.rs.
15
16use futures_util::future::try_join_all;
17use g1t_contracts::events::{ChecksEvent, QueueChanged};
18use g1t_contracts::repos::{DeleteBranchArgs, GetByIdArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath};
19use g1t_contracts::time::rfc3339;
20use g1t_contracts::work::*;
21use g1t_contracts::{FailureCode, Outcome, User, new_id};
22use g1t_kit::now_ms;
23use serde::Deserialize;
24use worker::Result;
25use worker::wasm_bindgen::JsValue;
26
27use crate::Work;
28use crate::checks::{hash, new_token};
29
30/// How many entries that left the queue are shown.
31const RECENT: u32 = 20;
32/// Where tested states are pushed, in the repository itself.
33pub(crate) const BRANCH_PREFIX: &str = "g1t-queue/";
34
35#[derive(Clone, Deserialize)]
36pub(crate) struct EntryRow {
37 pub(crate) id: String,
38 pub(crate) repo_id: String,
39 pub(crate) pull_id: String,
40 pub(crate) number: u32,
41 state: String,
42 enqueued_by: String,
43 pub(crate) keep_issue_open: u32,
44 pub(crate) head_commit: Option<String>,
45 pub(crate) base_commit: Option<String>,
46 ahead: Option<String>,
47 combined_commit: Option<String>,
48 results: Option<String>,
49 error: Option<String>,
50 pub(crate) tested_at: Option<String>,
51 created_at: String,
52 finished_at: Option<String>,
53 /// What failed its combined state while entries ahead were still
54 /// being tested (integration.rs), as JSON.
55 #[serde(default)]
56 pub(crate) held: Option<String>,
57 /// What the queue worked out about its failure (integration.rs).
58 #[serde(default)]
59 pub(crate) diagnosis: Option<String>,
60}
61
62impl EntryRow {
63 pub(crate) fn state(&self) -> QueueState {
64 match self.state.as_str() {
65 "testing" => QueueState::Testing,
66 "passed" => QueueState::Passed,
67 "failed" => QueueState::Failed,
68 "landed" => QueueState::Landed,
69 "removed" => QueueState::Removed,
70 _ => QueueState::Waiting,
71 }
72 }
73
74 pub(crate) fn actor(&self) -> Option<User> {
75 serde_json::from_str(&self.enqueued_by).ok()
76 }
77
78 pub(crate) fn ahead(&self) -> Vec<u32> {
79 self.ahead
80 .as_deref()
81 .and_then(|ahead| serde_json::from_str(ahead).ok())
82 .unwrap_or_default()
83 }
84
85 fn branch(&self) -> String {
86 format!("{BRANCH_PREFIX}{}", self.id)
87 }
88}
89
90fn minutes_ago(minutes: u64) -> String {
91 rfc3339(now_ms().saturating_sub(minutes * 60 * 1000))
92}
93
94impl Work {
95 pub(crate) async fn entries(&self, repo_id: &str, active: bool) -> Result<Vec<EntryRow>> {
96 let sql = if active {
97 "SELECT * FROM queue_entries
98 WHERE repo_id = ? AND state IN ('waiting', 'testing', 'passed')
99 ORDER BY created_at, id"
100 } else {
101 "SELECT * FROM queue_entries
102 WHERE repo_id = ? AND state IN ('failed', 'landed', 'removed')
103 ORDER BY finished_at DESC LIMIT ?"
104 };
105 let statement = self.db.prepare(sql);
106 let statement = if active {
107 statement.bind(&[repo_id.into()])?
108 } else {
109 statement.bind(&[repo_id.into(), RECENT.into()])?
110 };
111 statement.all().await?.results::<EntryRow>()
112 }
113
114 /// The entry a pull request has in the queue now, if any.
115 pub(crate) async fn queued_entry(&self, pull_id: &str) -> Result<Option<(QueueState, Vec<u32>)>> {
116 let row = match self.prefetched_pull(pull_id) {
117 Some(found) => found.first::<EntryRow>(crate::prefetch::Slot::Queued)?,
118 None => self
119 .db
120 .prepare(
121 "SELECT * FROM queue_entries
122 WHERE pull_id = ? AND state IN ('waiting', 'testing', 'passed') LIMIT 1",
123 )
124 .bind(&[pull_id.into()])?
125 .first::<EntryRow>(None)
126 .await?,
127 };
128 Ok(row.map(|row| (row.state(), row.ahead())))
129 }
130
131 pub(crate) async fn changed(&self, repo_id: &str) -> Result<()> {
132 self.publish_as(
133 "queue.changed",
134 repo_id,
135 None,
136 QueueChanged {
137 repo_id: repo_id.to_owned(),
138 },
139 )
140 .await
141 }
142
143 /// Puts a pull request that may merge into the queue instead. Merging
144 /// it again while it is queued changes nothing.
145 pub(crate) async fn enqueue(
146 &self,
147 repo: &Repo,
148 pull: &Pull,
149 actor: &User,
150 keep_issue_open: bool,
151 ) -> Result<Outcome<Pull>> {
152 if self.queued_entry(&pull.id).await?.is_some() {
153 return Ok(Outcome::Ok(pull.clone()));
154 }
155 let now = now_ms();
156 self.db
157 .prepare(
158 "INSERT INTO queue_entries
159 (id, repo_id, pull_id, number, state, enqueued_by, keep_issue_open, created_at)
160 VALUES (?, ?, ?, ?, 'waiting', ?, ?, ?)",
161 )
162 .bind(&[
163 new_id("qen", now).into(),
164 repo.id.as_str().into(),
165 pull.id.as_str().into(),
166 pull.number.into(),
167 serde_json::to_string(actor)?.into(),
168 u32::from(keep_issue_open).into(),
169 rfc3339(now).into(),
170 ])?
171 .run()
172 .await?;
173 let who = (actor.id.as_str(), actor.username.as_str());
174 self.note(&repo.id, pull.number, who, "added this to the merge queue")
175 .await?;
176 self.changed(&repo.id).await?;
177 Ok(Outcome::Ok(pull.clone()))
178 }
179
180 pub(crate) async fn queue(&self, a: QueueArgs) -> Result<Outcome<QueueView>> {
181 let repo = match self.repo(&a.repo, &a.viewer).await? {
182 Outcome::Ok(repo) => repo,
183 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
184 };
185 let enabled = self.default_branch_settings(&repo).await?.merge_queue;
186 let (active, recent) = (self.entries(&repo.id, true).await?, self.entries(&repo.id, false).await?);
187 let numbers: Vec<u32> = active.iter().chain(&recent).map(|row| row.number).collect();
188 let pulls = try_join_all(numbers.iter().map(|number| self.pull(&repo.id, *number))).await?;
189 let view = |rows: Vec<EntryRow>, pulls: &[Option<Pull>]| -> Vec<QueueEntry> {
190 rows.into_iter()
191 .zip(pulls)
192 .map(|(row, pull)| QueueEntry {
193 state: row.state(),
194 ahead: row.ahead(),
195 enqueued_by: row.actor().map_or_else(|| "g1t".to_owned(), |actor| actor.username),
196 title: pull.as_ref().map(|p| p.title.clone()).unwrap_or_default(),
197 agent: pull.as_ref().map(|p| p.agent.clone()).unwrap_or_default(),
198 results: row
199 .results
200 .as_deref()
201 .and_then(|results| serde_json::from_str(results).ok())
202 .unwrap_or_default(),
203 id: row.id,
204 number: row.number,
205 base_commit: row.base_commit,
206 combined_commit: row.combined_commit,
207 error: row.error,
208 created_at: row.created_at,
209 finished_at: row.finished_at,
210 })
211 .collect()
212 };
213 let split = active.len();
214 Ok(Outcome::Ok(QueueView {
215 enabled,
216 active: view(active, &pulls[..split]),
217 recent: view(recent, &pulls[split..]),
218 }))
219 }
220
221 /// Takes a pull request out of the queue, by the hand of someone who
222 /// may merge.
223 pub(crate) async fn remove_from_queue(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
224 let viewer = Some(a.actor.clone());
225 let (repo, pull) = match self.pull_at(&a.repo, a.number, &viewer).await? {
226 Outcome::Ok(found) => found,
227 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
228 };
229 if let Outcome::Fail(failure) = crate::retired::writable(&repo) {
230 return Ok(Outcome::Fail(failure));
231 }
232 if !a.actor.verified {
233 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
234 }
235 if let Outcome::Fail(failure) =
236 crate::allowed(Some(&a.actor), &repo, g1t_contracts::access::Capability::Merge)
237 {
238 return Ok(Outcome::Fail(failure));
239 }
240 if self.leave(&repo.id, &pull, QueueState::Removed, None).await? {
241 let who = (a.actor.id.as_str(), a.actor.username.as_str());
242 self.note(&repo.id, pull.number, who, "removed this from the merge queue")
243 .await?;
244 self.changed(&repo.id).await?;
245 }
246 Ok(Outcome::Ok(pull))
247 }
248
249 /// Takes a pull request's entry out of the queue, and sends every entry
250 /// whose tested state included it back to waiting. Whether it had one.
251 pub(crate) async fn leave(
252 &self,
253 repo_id: &str,
254 pull: &Pull,
255 state: QueueState,
256 error: Option<&str>,
257 ) -> Result<bool> {
258 let active = self.entries(repo_id, true).await?;
259 let Some(entry) = active.iter().find(|row| row.pull_id == pull.id) else {
260 return Ok(false);
261 };
262 let now = rfc3339(now_ms());
263 self.db
264 .prepare(
265 "UPDATE queue_entries SET state = ?, error = COALESCE(?, error), finished_at = ?
266 WHERE id = ?",
267 )
268 .bind(&[
269 state.as_str().into(),
270 error.map_or(JsValue::NULL, JsValue::from),
271 now.as_str().into(),
272 entry.id.as_str().into(),
273 ])?
274 .run()
275 .await?;
276 self.drop_branch(entry).await;
277 let behind: Vec<&EntryRow> = active
278 .iter()
279 .filter(|row| row.state() != QueueState::Waiting && row.ahead().contains(&pull.number))
280 .collect();
281 self.retest(&behind).await?;
282 Ok(true)
283 }
284
285 /// Removes an entry's tested state from the repository once it has
286 /// left the queue, as GitHub does with its queue's branches. A branch
287 /// left behind is untidy, not wrong, so a failure only logs.
288 async fn drop_branch(&self, row: &EntryRow) {
289 self.drop_branch_named(&row.repo_id, &row.branch()).await;
290 }
291
292 /// Removes one of the queue's branches.
293 pub(crate) async fn drop_branch_named(&self, repo_id: &str, branch: &str) {
294 let deleted: Result<Outcome<bool>> = g1t_kit::call(
295 &self.repos,
296 "delete_branch",
297 &DeleteBranchArgs {
298 repo_id: repo_id.to_owned(),
299 branch: branch.to_owned(),
300 head: None,
301 },
302 )
303 .await;
304 match deleted {
305 Ok(Outcome::Ok(_)) => {}
306 Ok(Outcome::Fail(failure)) => worker::console_warn!("{branch}: {}", failure.message),
307 Err(error) => worker::console_warn!("{branch}: {error}"),
308 }
309 }
310
311 /// Sends entries back to waiting, to be tested again.
312 async fn retest(&self, rows: &[&EntryRow]) -> Result<()> {
313 for row in rows {
314 self.db
315 .prepare(
316 "UPDATE queue_entries
317 SET state = 'waiting', token_hash = NULL, combined_commit = NULL,
318 results = NULL, error = NULL, ahead = NULL, held = NULL
319 WHERE id = ? AND state IN ('testing', 'passed')",
320 )
321 .bind(&[row.id.as_str().into()])?
322 .run()
323 .await?;
324 }
325 Ok(())
326 }
327
328 /// The next batch to test, if nothing is being tested now: one job per
329 /// entry, each building the default branch with that entry and every
330 /// entry ahead of it.
331 ///
332 /// Smaller combinations a failure is being narrowed down with
333 /// (integration.rs) are built alongside, whatever the queue is doing.
334 pub(crate) async fn queue_build(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
335 let mut jobs = self.probe_jobs(&a.repo_id).await?;
336 jobs.extend(self.queue_batch(a).await?);
337 Ok(jobs)
338 }
339
340 async fn queue_batch(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
341 let active = self.entries(&a.repo_id, true).await?;
342 // How the default branch's merge queue rule batches (rulesets.rs).
343 let rule = self.queue_rule(&a.repo_id).await?;
344 // A batch that has taken too long is tested again.
345 let stale = minutes_ago(u64::from(rule.check_response_timeout_minutes).max(1));
346 let stuck: Vec<&EntryRow> = active
347 .iter()
348 .filter(|row| {
349 row.state() == QueueState::Testing
350 && row.tested_at.as_deref().is_none_or(|at| at < stale.as_str())
351 })
352 .collect();
353 if !stuck.is_empty() {
354 self.retest(&stuck).await?;
355 return Box::pin(self.queue_batch(a)).await;
356 }
357 if active.iter().any(|row| row.state() != QueueState::Waiting) {
358 return Ok(Vec::new());
359 }
360 let batch: Vec<EntryRow> = active.into_iter().take(rule.max_entries_to_build.max(1) as usize).collect();
361 let Some(first) = batch.first() else {
362 return Ok(Vec::new());
363 };
364 // Too few to start yet, and the oldest has not waited long enough.
365 if batch.len() < rule.min_entries_to_merge as usize
366 && first.created_at.as_str() > minutes_ago(u64::from(rule.min_entries_wait_minutes)).as_str()
367 {
368 return Ok(Vec::new());
369 }
370 let Some(actor) = first.actor() else {
371 return Ok(Vec::new());
372 };
373 let repo: Outcome<Repo> = g1t_kit::call(
374 &self.repos,
375 "get_by_id",
376 &GetByIdArgs {
377 id: a.repo_id.clone(),
378 viewer: Some(actor),
379 },
380 )
381 .await?;
382 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
383 return Ok(Vec::new());
384 };
385 let base: Option<String> = g1t_kit::call(
386 &self.repos,
387 "head",
388 &HeadArgs {
389 repo_id: repo.id.clone(),
390 branch: repo.default_branch.clone(),
391 },
392 )
393 .await?;
394 let Some(base) = base else {
395 return Ok(Vec::new());
396 };
397
398 // Each entry's change, as it is now.
399 let mut items: Vec<(EntryRow, Pull, QueueStackItem)> = Vec::new();
400 for row in batch {
401 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
402 continue;
403 };
404 if pull.status != PullStatus::Open {
405 self.leave(&repo.id, &pull, QueueState::Removed, Some("It was closed."))
406 .await?;
407 continue;
408 }
409 let head: Option<String> = g1t_kit::call(
410 &self.repos,
411 "head",
412 &HeadArgs {
413 repo_id: pull.fork_repo_id.clone().unwrap_or_else(|| repo.id.clone()),
414 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
415 },
416 )
417 .await?;
418 let Some(commit) = head else {
419 continue;
420 };
421 let source = pull.fork.clone().unwrap_or_else(|| RepoPath {
422 namespace: repo.namespace.clone(),
423 name: repo.name.clone(),
424 });
425 let item = QueueStackItem {
426 number: pull.number,
427 title: pull.title.clone(),
428 source,
429 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
430 commit,
431 };
432 items.push((row, pull, item));
433 }
434
435 let now = rfc3339(now_ms());
436 let mut jobs = Vec::new();
437 for index in 0..items.len() {
438 let (row, pull, _) = &items[index];
439 let stack: Vec<QueueStackItem> = items[..=index].iter().map(|(_, _, item)| item.clone()).collect();
440 let ahead: Vec<u32> = stack[..index].iter().map(|item| item.number).collect();
441 let token = new_token();
442 self.db
443 .prepare(
444 "UPDATE queue_entries
445 SET state = 'testing', token_hash = ?, head_commit = ?, base_commit = ?,
446 ahead = ?, combined_commit = NULL, results = NULL, error = NULL,
447 tested_at = ?
448 WHERE id = ? AND state = 'waiting'",
449 )
450 .bind(&[
451 hash(&token).into(),
452 stack[index].commit.as_str().into(),
453 base.as_str().into(),
454 serde_json::to_string(&ahead)?.into(),
455 now.as_str().into(),
456 row.id.as_str().into(),
457 ])?
458 .run()
459 .await?;
460 // The sandbox reads the change and pushes the tested state as
461 // the pull request's owner (whoever asked g1t for it, or its
462 // author): a real account, whose token carries its memberships.
463 // (Whoever queued it may be g1t itself.)
464 let actor = pull.owner().clone();
465 jobs.push(QueueJob {
466 entry_id: row.id.clone(),
467 token,
468 repo: RepoPath {
469 namespace: repo.namespace.clone(),
470 name: repo.name.clone(),
471 },
472 default_branch: repo.default_branch.clone(),
473 base_commit: base.clone(),
474 branch: row.branch(),
475 // The state is checked by its merge_group workflows.
476 contract_checks: Vec::new(),
477 stack,
478 checks: Vec::new(),
479 actor,
480 });
481 }
482 Ok(jobs)
483 }
484
485 /// A sandbox's result for one combined state.
486 pub(crate) async fn report_queue(&self, a: ReportQueueArgs) -> Result<Outcome<QueueState>> {
487 // A smaller combination built to narrow down a failure.
488 if a.entry_id.starts_with("qpr_") {
489 return self.report_probe(a).await;
490 }
491 let row = self
492 .db
493 .prepare("SELECT * FROM queue_entries WHERE id = ? AND token_hash = ?")
494 .bind(&[a.entry_id.as_str().into(), hash(&a.token).into()])?
495 .first::<EntryRow>(None)
496 .await?;
497 let Some(row) = row else {
498 return Ok(Outcome::fail(FailureCode::NotFound, "No such queue entry."));
499 };
500 if row.state() != QueueState::Testing {
501 return Ok(Outcome::fail(
502 FailureCode::Conflict,
503 "This state is no longer being tested.",
504 ));
505 }
506 let built = a.error.is_none() && a.results.iter().all(|result| result.passed);
507 // A state that was built runs the repository's `merge_group`
508 // workflows; it stays in testing until they finish (see `statuses`),
509 // and the branch's required checks must pass on it.
510 let workflows = match (&a.combined_commit, built) {
511 (Some(commit), true) => self
512 .run_merge_group(
513 &row.repo_id,
514 &row.id,
515 &row.branch(),
516 row.base_commit.as_deref(),
517 row.number,
518 &row.ahead(),
519 commit,
520 )
521 .await
522 .unwrap_or(0),
523 _ => 0,
524 };
525 let required = self.settings_by_id(&row.repo_id).await?.required_checks;
526 // Nothing runs on it, so the required checks never would report.
527 let unchecked = (built && workflows == 0 && !required.is_empty()).then(|| {
528 format!(
529 "the required {} {} cannot report on it: no workflow runs on merge_group events. Add merge_group to the on: of the workflows the branch requires",
530 if required.len() == 1 { "check" } else { "checks" },
531 crate::statuses::list(&required)
532 )
533 });
534 let passed = built && unchecked.is_none();
535 let error = a.error.clone().or(unchecked.clone());
536 let report = ReportQueueArgs { error: error.clone(), ..a };
537 let failure = (!passed).then(|| {
538 crate::integration::Failure::of_report(&report, unchecked.is_some().then_some(&required[..]))
539 });
540 // Failed while an entry ahead of it is still being tested: held
541 // until those ahead are settled, so that it is not blamed for
542 // theirs (integration.rs).
543 let held = match &failure {
544 Some(failure) if self.ahead_unsettled(&row).await? => Some(serde_json::to_string(failure)?),
545 _ => None,
546 };
547 let state = if passed && workflows == 0 {
548 QueueState::Passed
549 } else if passed || held.is_some() {
550 QueueState::Testing
551 } else {
552 QueueState::Failed
553 };
554 self.db
555 .prepare(
556 "UPDATE queue_entries
557 SET state = ?, combined_commit = ?, results = ?, error = ?, token_hash = NULL, held = ?,
558 finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
559 WHERE id = ?",
560 )
561 .bind(&[
562 state.as_str().into(),
563 report.combined_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
564 serde_json::to_string(&report.results)?.into(),
565 error.as_deref().map_or(JsValue::NULL, JsValue::from),
566 held.as_deref().map_or(JsValue::NULL, JsValue::from),
567 state.as_str().into(),
568 rfc3339(now_ms()).into(),
569 row.id.as_str().into(),
570 ])?
571 .run()
572 .await?;
573 if let Some(failure) = failure.filter(|_| held.is_none()) {
574 self.eject(&row, &report).await?;
575 self.diagnose(&row, failure).await?;
576 }
577 if state == QueueState::Passed {
578 self.integration_passed(&row, report.combined_commit.as_deref()).await?;
579 self.resolve_held(&row.repo_id).await?;
580 }
581 self.settle(&row.repo_id).await?;
582 self.changed(&row.repo_id).await?;
583 Ok(Outcome::Ok(state))
584 }
585
586 /// Asks the actions service to run the repository's `merge_group`
587 /// workflows on a combined state: `entry`'s, pushed to `branch`, with
588 /// `ahead` merged in before `number`. Returns how many runs started.
589 #[allow(clippy::too_many_arguments)]
590 pub(crate) async fn run_merge_group(
591 &self,
592 repo_id: &str,
593 entry: &str,
594 branch: &str,
595 base_commit: Option<&str>,
596 number: u32,
597 ahead: &[u32],
598 commit: &str,
599 ) -> Result<u32> {
600 #[derive(serde::Deserialize)]
601 struct Started {
602 runs: u32,
603 }
604 let started: Outcome<Started> = g1t_kit::call(
605 &self.actions,
606 "merge_group",
607 &serde_json::json!({
608 "repoId": repo_id,
609 "entry": entry,
610 "sha": commit,
611 "headRef": format!("refs/heads/{branch}"),
612 "baseSha": base_commit,
613 "number": number,
614 "ahead": ahead,
615 }),
616 )
617 .await?;
618 Ok(match started {
619 Outcome::Ok(started) => started.runs,
620 Outcome::Fail(_) => 0,
621 })
622 }
623
624 /// Workflows on a combined state finished: it passes and lands in turn
625 /// when they all passed and so did every required check, or fails and
626 /// leaves the queue.
627 pub(crate) async fn merge_group_finished(
628 &self,
629 repo_id: &str,
630 commit: &str,
631 facts: &crate::statuses::WorkflowFacts,
632 ) -> Result<()> {
633 let row = self
634 .db
635 .prepare(
636 "SELECT * FROM queue_entries
637 WHERE repo_id = ? AND combined_commit = ? AND state = 'testing' AND token_hash IS NULL AND held IS NULL",
638 )
639 .bind(&[repo_id.into(), commit.into()])?
640 .first::<EntryRow>(None)
641 .await?;
642 let missing = facts.expected();
643 let passed = facts.failed.is_empty() && facts.required_failed().is_empty() && missing.is_empty();
644 let Some(row) = row else {
645 // Or a smaller combination built to narrow down a failure.
646 if self.probe_finished(repo_id, commit, passed).await? {
647 self.changed(repo_id).await?;
648 }
649 return Ok(());
650 };
651 let failure = if passed {
652 None
653 } else {
654 let error = if facts.failed.is_empty() && facts.required_failed().is_empty() {
655 format!(
656 "the required {} {} did not report on it. Add merge_group to the on: of the workflows the branch requires",
657 if missing.len() == 1 { "check" } else { "checks" },
658 crate::statuses::list(&missing)
659 )
660 } else if facts.failed.is_empty() {
661 format!("the required {} failed on it", crate::statuses::list(&facts.required_failed()))
662 } else {
663 format!("the workflow {} failed on it", crate::statuses::list(&facts.failed))
664 };
665 // Where to see the check that failed: its run's page.
666 let statuses = self.statuses(repo_id, Some(commit)).await?;
667 let url = |name: &str| {
668 statuses
669 .iter()
670 .find(|status| status.context == name || check_name(&status.context).0 == name)
671 .and_then(|status| status.target_url.clone())
672 };
673 Some(crate::integration::Failure::of_workflows(
674 commit,
675 &facts.failed,
676 &facts.required_failed(),
677 &missing,
678 error,
679 url,
680 ))
681 };
682 let held = match &failure {
683 Some(failure) if self.ahead_unsettled(&row).await? => Some(serde_json::to_string(failure)?),
684 _ => None,
685 };
686 let state = if passed {
687 "passed"
688 } else if held.is_some() {
689 "testing"
690 } else {
691 "failed"
692 };
693 let moved = self
694 .db
695 .prepare(
696 "UPDATE queue_entries SET state = ?, held = ?, finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
697 WHERE id = ? AND state = 'testing' AND held IS NULL RETURNING id AS value",
698 )
699 .bind(&[
700 state.into(),
701 held.as_deref().map_or(JsValue::NULL, JsValue::from),
702 state.into(),
703 rfc3339(now_ms()).into(),
704 row.id.as_str().into(),
705 ])?
706 .first::<crate::rows::ValueRow>(None)
707 .await?;
708 if moved.is_none() {
709 return Ok(());
710 }
711 if let Some(failure) = failure.filter(|_| held.is_none()) {
712 self.eject(&row, &failure.report(&row.id)).await?;
713 self.diagnose(&row, failure).await?;
714 }
715 if passed {
716 self.integration_passed(&row, Some(commit)).await?;
717 self.resolve_held(repo_id).await?;
718 }
719 self.settle(repo_id).await?;
720 self.changed(repo_id).await?;
721 Ok(())
722 }
723
724 /// An entry whose combined state failed: the entries tested on top of
725 /// it are tested again without it, and the failure is recorded as a
726 /// failed check run of its pull request.
727 pub(crate) async fn eject(&self, row: &EntryRow, report: &ReportQueueArgs) -> Result<()> {
728 self.drop_branch(row).await;
729 let active = self.entries(&row.repo_id, true).await?;
730 let behind: Vec<&EntryRow> = active
731 .iter()
732 .filter(|other| other.state() != QueueState::Waiting && other.ahead().contains(&row.number))
733 .collect();
734 self.retest(&behind).await?;
735
736 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
737 return Ok(());
738 };
739 let ahead = row.ahead();
740 let state = if ahead.is_empty() {
741 "the default branch as it is now".to_owned()
742 } else {
743 format!(
744 "the default branch with {} merged in first",
745 ahead.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", ")
746 )
747 };
748 // Named in backticks, so the conversation can link each to the diff.
749 let files = crate::mergeability::tidy(report.conflicts.clone())
750 .iter()
751 .map(|path| format!("`{path}`"))
752 .collect::<Vec<_>>()
753 .join(", ");
754 let why = match (&report.error, report.conflict_with) {
755 (_, Some(other)) if other != row.number && !files.is_empty() => format!(
756 "Its change conflicts with #{other}, which is ahead of it in the merge queue, in {files}. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
757 ),
758 (_, Some(other)) if other != row.number => format!(
759 "Its change conflicts with #{other}, which is ahead of it in the merge queue. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
760 ),
761 (Some(_), None) if !files.is_empty() => format!(
762 "Its change conflicts with the default branch in {files}. Bring it up to date with the default branch and push: it rejoins the merge queue once its checks pass."
763 ),
764 (Some(error), _) if error.starts_with("the workflow ") => {
765 format!("{} when it was combined with {state}.", error.replacen("the workflow", "The workflow", 1).trim_end_matches(" on it"))
766 }
767 (Some(error), _) if error.starts_with("the required ") => {
768 format!("Combined with {state}, {error}.")
769 }
770 (Some(error), _) => format!("Its combined state could not be built or checked: {error}"),
771 (None, _) => format!(
772 "It failed when it was combined with {state}, though it may pass on its own."
773 ),
774 };
775 let now = now_ms();
776 let run_id = new_id("chk", now);
777 let mut results = report.results.clone();
778 for result in &mut results {
779 result.command = format!("{} (merge queue, on {state})", result.command);
780 }
781 self.db
782 .batch(vec![
783 self.db
784 .prepare(
785 "INSERT INTO check_runs
786 (id, pull_id, head_commit, status, results, error, token_hash,
787 created_at, finished_at)
788 VALUES (?, ?, ?, 'failed', ?, ?, ?, ?, ?)",
789 )
790 .bind(&[
791 run_id.as_str().into(),
792 pull.id.as_str().into(),
793 row.head_commit.as_deref().unwrap_or_default().into(),
794 serde_json::to_string(&results)?.into(),
795 why.as_str().into(),
796 hash(&new_token()).into(),
797 rfc3339(now).into(),
798 rfc3339(now).into(),
799 ])?,
800 self.db
801 .prepare("UPDATE pulls SET check_status = 'failed', check_run_id = ? WHERE id = ?")
802 .bind(&[run_id.as_str().into(), pull.id.as_str().into()])?,
803 // Its merge stays asked for: it rejoins once a fix's checks
804 // pass (integration.rs `requeue_if_fixed`).
805 self.db
806 .prepare("UPDATE queue_entries SET requeue = 1 WHERE id = ?")
807 .bind(&[row.id.as_str().into()])?,
808 ])
809 .await?;
810 self.note(
811 &row.repo_id,
812 pull.number,
813 ("g1t", "g1t"),
814 &format!("was taken out of the merge queue. {why}"),
815 )
816 .await?;
817 self.publish_as(
818 "checks.completed",
819 &row.repo_id,
820 None,
821 ChecksEvent {
822 pull_id: pull.id.clone(),
823 repo_id: row.repo_id.clone(),
824 number: pull.number,
825 status: "failed",
826 commit: row.head_commit.clone().unwrap_or_default(),
827 },
828 )
829 .await?;
830 Ok(())
831 }
832
833 /// Lands every entry at the front of the queue whose tested state
834 /// passed, in order.
835 async fn settle(&self, repo_id: &str) -> Result<()> {
836 let active = self.entries(repo_id, true).await?;
837 let Some(viewer) = active.first().and_then(EntryRow::actor) else {
838 return Ok(());
839 };
840 let repo: Outcome<Repo> = g1t_kit::call(
841 &self.repos,
842 "get_by_id",
843 &GetByIdArgs {
844 id: repo_id.to_owned(),
845 viewer: Some(viewer),
846 },
847 )
848 .await?;
849 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
850 return Ok(());
851 };
852 for row in active {
853 if row.state() != QueueState::Passed {
854 // The front has not passed yet: nothing behind it may land.
855 return Ok(());
856 }
857 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
858 continue;
859 };
860 let Some(actor) = row.actor() else {
861 continue;
862 };
863 // Pushed to since it was tested: test it again as it is now.
864 if pull.status != PullStatus::Open {
865 self.leave(repo_id, &pull, QueueState::Removed, Some("It was closed."))
866 .await?;
867 continue;
868 }
869 let head: Option<String> = g1t_kit::call(
870 &self.repos,
871 "head",
872 &HeadArgs {
873 repo_id: pull.fork_repo_id.clone().unwrap_or_else(|| repo_id.to_owned()),
874 branch: pull.branch.clone().unwrap_or_else(|| repo.default_branch.clone()),
875 },
876 )
877 .await?;
878 if head.is_some() && head != row.head_commit {
879 let active = self.entries(repo_id, true).await?;
880 let again: Vec<&EntryRow> = active
881 .iter()
882 .filter(|other| other.id == row.id || other.ahead().contains(&row.number))
883 .collect();
884 self.retest(&again).await?;
885 return Ok(());
886 }
887 let landed: Outcome<Landed> = g1t_kit::call(
888 &self.repos,
889 "land",
890 &LandArgs {
891 source_id: repo_id.to_owned(),
892 branch: Some(row.branch()),
893 actor: actor.clone(),
894 // The queue lands on the default branch.
895 target_branch: None,
896 },
897 )
898 .await?;
899 let landed = match landed {
900 Outcome::Ok(landed) => landed,
901 Outcome::Fail(_) => {
902 // The default branch moved outside the queue: every
903 // tested state is built on something that is gone.
904 let active = self.entries(repo_id, true).await?;
905 let all: Vec<&EntryRow> = active.iter().collect();
906 self.retest(&all).await?;
907 return Ok(());
908 }
909 };
910 self.db
911 .prepare(
912 "UPDATE queue_entries SET state = 'landed', finished_at = ? WHERE id = ?",
913 )
914 .bind(&[rfc3339(now_ms()).into(), row.id.as_str().into()])?
915 .run()
916 .await?;
917 self.drop_branch(&row).await;
918 self.record_merge(&repo, pull, &actor, row.keep_issue_open != 0, landed)
919 .await?;
920 }
921 Ok(())
922 }
923}