Skip to content
Sign up

flagon-io/g1t

public
0
1,286 linesCodeBlameRaw
1//! What the merge queue learns when it combines pull requests, and who it
2//! tells.
3//!
4//! **Blame.** Entries are tested speculatively: entry *i* on the default
5//! branch with every entry ahead of it merged in. A combined state that
6//! fails while an entry ahead of it is still being tested is *held*: if
7//! that entry fails too, the held one is tested again without it and is
8//! not blamed; once every entry ahead has passed, the held one answers for
9//! the failure. So a broken change is blamed on itself, never on those
10//! queued behind it.
11//!
12//! **Bisection.** An entry whose own checks pass but whose combined state
13//! fails breaks *with* some of the entries ahead of it. With more than one
14//! ahead, g1t builds smaller combinations, the default branch with half of
15//! the suspects and the entry, halving the suspects each time, within
16//! [`PROBE_BUDGET`] extra builds (built and checked as the queue's states
17//! are, by the `merge_group` workflows, and never landed). What is left is
18//! reported as `suspects`.
19//!
20//! **A default branch that moved.** An entry that fails with nothing ahead
21//! of it, though its own checks pass on its head, fails on the default
22//! branch as it is now: something landed there since it branched. g1t looks
23//! up the pull requests merged into the default branch whose commits the
24//! entry's history does not hold, newest first and at most [`MAX_LANDED`],
25//! and reports them as `landed`. The entry still answers for the failure
26//! (it is the one that must adapt); one landed pull request is named as the
27//! likely cause.
28//!
29//! **Telling the work.** The answer goes on the pull request's and its
30//! issue's timelines, into the pull request's session, as a decision
31//! record (`integration_failure`, tested) and as `integration.failed`. A
32//! passing combination is `integration.passed`, and a decision too.
33//!
34//! **Rejoining.** An entry the queue took out keeps its merge request: once
35//! a new commit is pushed and its checks pass, it is merged again (into the
36//! queue) as whoever queued it, by the branch's rules as they are then.
37
38use g1t_contracts::decisions::{DecisionConfidence, DecisionKind, FailedCheck, IntegrationFailed, IntegrationPassed};
39use g1t_contracts::events::SessionAppended;
40use g1t_contracts::repos::{Commit, GetByIdArgs, LogArgs, PathByIdArgs, Repo, RepoPath};
41use g1t_contracts::time::rfc3339;
42use g1t_contracts::work::*;
43use g1t_contracts::{FailureCode, Outcome, new_id};
44use g1t_kit::now_ms;
45use serde::{Deserialize, Serialize};
46use serde_json::json;
47use worker::Result;
48use worker::wasm_bindgen::JsValue;
49
50use crate::Work;
51use crate::checks::{hash, new_token};
52use crate::decisions::NewDecision;
53use crate::queue::{BRANCH_PREFIX, EntryRow};
54use crate::rows::ValueRow;
55
56/// The most extra combinations built to narrow down one failure.
57pub(crate) const PROBE_BUDGET: u32 = 3;
58
59/// The most pull requests reported as landed since an entry branched.
60pub(crate) const MAX_LANDED: usize = 10;
61
62/// How far back the entry's own history is read to tell what it holds.
63const LANDED_HISTORY: u32 = 1000;
64
65/// How many of the latest merged pull requests are looked at.
66const LANDED_CANDIDATES: u32 = 50;
67
68/// What kind of failure a combined state had.
69#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
70#[serde(rename_all = "snake_case")]
71pub(crate) enum FailureKind {
72 /// A check failed on it.
73 Checks,
74 /// The changes did not merge cleanly.
75 Conflict,
76 /// A required check never reported on it.
77 Missing,
78 /// It could not be built or checked.
79 Build,
80}
81
82/// What failed a combined state, as kept while it is held and while the
83/// failure is narrowed down.
84#[derive(Clone, Debug, Serialize, Deserialize)]
85pub(crate) struct Failure {
86 pub(crate) kind: FailureKind,
87 pub(crate) check: FailedCheck,
88 #[serde(default)]
89 pub(crate) error: Option<String>,
90 #[serde(default)]
91 pub(crate) conflict_with: Option<u32>,
92 #[serde(default)]
93 pub(crate) conflicts: Vec<String>,
94 #[serde(default)]
95 pub(crate) results: Vec<CheckResult>,
96 #[serde(default)]
97 pub(crate) combined_commit: Option<String>,
98}
99
100impl Failure {
101 /// What a sandbox reported for a state that failed. `unchecked` is set
102 /// when the state was built but nothing could check it.
103 pub(crate) fn of_report(report: &ReportQueueArgs, unchecked: Option<&[String]>) -> Failure {
104 let (kind, check) = if report.conflict_with.is_some() || !report.conflicts.is_empty() {
105 (FailureKind::Conflict, check("merge", "conflict", None))
106 } else if let Some(missing) = unchecked {
107 (FailureKind::Missing, check(&crate::statuses::list(missing), "missing", None))
108 } else if report.error.is_some() {
109 (FailureKind::Build, check("build", "error", None))
110 } else {
111 let name = report
112 .results
113 .iter()
114 .find(|result| !result.passed)
115 .map_or("checks", |result| result.command.as_str());
116 (FailureKind::Checks, check(name, "failure", None))
117 };
118 Failure {
119 kind,
120 check,
121 error: report.error.clone(),
122 conflict_with: report.conflict_with,
123 conflicts: report.conflicts.clone(),
124 results: report.results.clone(),
125 combined_commit: report.combined_commit.clone(),
126 }
127 }
128
129 /// What its `merge_group` workflows said: the first that failed, or the
130 /// required checks that failed or never reported.
131 pub(crate) fn of_workflows(
132 commit: &str,
133 failed: &[String],
134 required_failed: &[String],
135 missing: &[String],
136 error: String,
137 url: impl Fn(&str) -> Option<String>,
138 ) -> Failure {
139 let (kind, check) = match (failed.first(), required_failed.first()) {
140 (Some(name), _) | (None, Some(name)) => (FailureKind::Checks, check(name, "failure", url(name))),
141 (None, None) => (FailureKind::Missing, check(&crate::statuses::list(missing), "missing", None)),
142 };
143 Failure {
144 kind,
145 check,
146 error: Some(error),
147 conflict_with: None,
148 conflicts: Vec::new(),
149 results: Vec::new(),
150 combined_commit: Some(commit.to_owned()),
151 }
152 }
153
154 /// The report the queue ejects an entry with.
155 pub(crate) fn report(&self, entry_id: &str) -> ReportQueueArgs {
156 ReportQueueArgs {
157 entry_id: entry_id.to_owned(),
158 token: String::new(),
159 combined_commit: self.combined_commit.clone(),
160 results: self.results.clone(),
161 error: self.error.clone(),
162 conflict_with: self.conflict_with,
163 conflicts: self.conflicts.clone(),
164 }
165 }
166}
167
168fn check(name: &str, conclusion: &str, url: Option<String>) -> FailedCheck {
169 FailedCheck {
170 name: name.to_owned(),
171 conclusion: conclusion.to_owned(),
172 url,
173 }
174}
175
176/// Whether every entry ahead has passed (or landed, and so left the
177/// queue). `active` is the queue's entries now, as (number, state).
178pub(crate) fn ahead_settled(ahead: &[u32], active: &[(u32, QueueState)]) -> bool {
179 ahead.iter().all(|number| {
180 active
181 .iter()
182 .find(|(n, _)| n == number)
183 .is_none_or(|(_, state)| *state == QueueState::Passed)
184 })
185}
186
187/// Who answers for a failed combined state, before any extra builds.
188#[derive(Debug, PartialEq, Eq)]
189pub(crate) enum Plan {
190 /// These entries ahead are what it fails with; none when it fails by
191 /// itself.
192 Blame(Vec<u32>),
193 /// It fails with some of the entries ahead: narrow them down.
194 Bisect,
195 /// Nothing is ahead and it passes its checks on its own, yet it fails
196 /// on the default branch as it is now: it answers for it, and what
197 /// landed there since it branched is looked up.
198 BaseMoved,
199}
200
201/// Decides who answers for a failure of the entry tested with `ahead`
202/// merged in first. `passed_alone`: its own checks pass on its head.
203pub(crate) fn plan(ahead: &[u32], failure: &Failure, passed_alone: bool) -> Plan {
204 if ahead.is_empty() {
205 return if failure.kind == FailureKind::Checks && passed_alone {
206 Plan::BaseMoved
207 } else {
208 Plan::Blame(Vec::new())
209 };
210 }
211 match failure.kind {
212 // The sandbox names the entry ahead that changed the same files.
213 FailureKind::Conflict => Plan::Blame(failure.conflict_with.filter(|n| ahead.contains(n)).into_iter().collect()),
214 // Neither is about how the changes combine.
215 FailureKind::Missing | FailureKind::Build => Plan::Blame(Vec::new()),
216 FailureKind::Checks if !passed_alone => Plan::Blame(Vec::new()),
217 FailureKind::Checks if ahead.len() == 1 => Plan::Blame(ahead.to_vec()),
218 FailureKind::Checks => Plan::Bisect,
219 }
220}
221
222/// A pull request merged into the default branch, as looked at for what
223/// landed since an entry branched.
224#[derive(Clone, Debug, Deserialize)]
225pub(crate) struct MergedRow {
226 pub(crate) number: u32,
227 /// The commit it landed as.
228 pub(crate) head_commit: Option<String>,
229 pub(crate) merged_at: Option<String>,
230}
231
232/// The pull requests among `merged` (newest first) that landed since the
233/// entry branched: those whose commit `history`, the entry's own, does not
234/// hold. When that history was cut short, `oldest` is the time of its
235/// oldest commit, and a pull request merged before it is not counted, as
236/// its commit may be further back. Newest first, at most [`MAX_LANDED`].
237pub(crate) fn landed_since(merged: &[MergedRow], history: &[Commit], oldest: Option<&str>) -> Vec<u32> {
238 merged
239 .iter()
240 .filter(|pull| {
241 pull.head_commit
242 .as_deref()
243 .is_some_and(|commit| !history.iter().any(|known| known.hash == commit))
244 })
245 .filter(|pull| match (oldest, pull.merged_at.as_deref()) {
246 (Some(oldest), Some(at)) => at >= oldest,
247 (Some(_), None) => false,
248 (None, _) => true,
249 })
250 .map(|pull| pull.number)
251 .take(MAX_LANDED)
252 .collect()
253}
254
255/// The suspects to build with the entry next, the first half of them, or
256/// none when one is left or the budget is spent.
257pub(crate) fn next_probe(suspects: &[u32], builds: u32) -> Option<Vec<u32>> {
258 if suspects.len() < 2 || builds >= PROBE_BUDGET {
259 return None;
260 }
261 Some(suspects[..suspects.len().div_ceil(2)].to_vec())
262}
263
264/// The suspects left once a build with `tested` and the entry `broke` or
265/// passed: those tested when it broke, the others when it passed.
266pub(crate) fn narrow(suspects: &[u32], tested: &[u32], broke: bool) -> Vec<u32> {
267 suspects
268 .iter()
269 .filter(|n| tested.contains(n) == broke)
270 .copied()
271 .collect()
272}
273
274/// The pull requests in a combination, `#11, #12 and #14`.
275pub(crate) fn numbers(list: &[u32]) -> String {
276 let named: Vec<String> = list.iter().map(|n| format!("#{n}")).collect();
277 match named.as_slice() {
278 [] => String::new(),
279 [one] => one.clone(),
280 [rest @ .., last] => format!("{} and {last}", rest.join(", ")),
281 }
282}
283
284/// What failed, as the start of a sentence.
285fn what_failed(check: &FailedCheck) -> String {
286 match check.conclusion.as_str() {
287 "conflict" => "it did not merge cleanly".to_owned(),
288 "missing" => format!("the required {} never reported", check.name),
289 "error" if check.name == "build" => "the combined state could not be built".to_owned(),
290 _ => format!("`{}` failed", check.name),
291 }
292}
293
294/// The note on the pull request, as the rest of a sentence starting with
295/// `g1t`.
296pub(crate) fn pull_note(ahead: &[u32], suspects: &[u32], bisected: bool, landed: &[u32], check: &FailedCheck) -> String {
297 let what = what_failed(check);
298 let rejoin = "Push a fix: once its checks pass on the new commit, it rejoins the merge queue by itself.";
299 if suspects.is_empty() && ahead.is_empty() && !landed.is_empty() {
300 return format!(
301 "traced its merge queue failure to the default branch as it is now: {what} with this change merged into it, though this passes its checks on its own. Since this branched, {}. Bring it up to date with the default branch, fix what breaks, and push. {rejoin}",
302 landed_clause(landed),
303 );
304 }
305 if suspects.is_empty() {
306 let on = if ahead.is_empty() {
307 "on the default branch as it is now".to_owned()
308 } else {
309 format!("with {} merged in ahead of it", numbers(ahead))
310 };
311 return format!("traced its merge queue failure to this change: {what} {on}, and the change is the cause. {rejoin}");
312 }
313 let narrowed = if bisected && suspects.len() < ahead.len() {
314 format!(" (narrowed down from {} by building smaller combinations)", numbers(ahead))
315 } else {
316 String::new()
317 };
318 format!(
319 "traced its merge queue failure to how it combines with {}{narrowed}: {what} when they were merged together, though this passes its checks on its own. Bring it up to date with the default branch once {} {} landed, fix what breaks, and push. {rejoin}",
320 numbers(suspects),
321 numbers(suspects),
322 if suspects.len() == 1 { "has" } else { "have" },
323 )
324}
325
326/// What landed on the default branch since the entry branched, as a
327/// clause: one is named as the likely cause.
328fn landed_clause(landed: &[u32]) -> String {
329 match landed {
330 [one] => format!("#{one} landed and is the likely cause"),
331 _ => format!("{} landed and may be what it conflicts with", numbers(landed)),
332 }
333}
334
335/// The note on its issue, and the one-line summary of the decision.
336pub(crate) fn summary(number: u32, suspects: &[u32], landed: &[u32], check: &FailedCheck) -> String {
337 let what = what_failed(check);
338 if suspects.is_empty() && !landed.is_empty() {
339 format!(
340 "#{number} failed in the merge queue on the default branch as it is now: {what}, though #{number} passes its checks on its own; since it branched, {}.",
341 landed_clause(landed),
342 )
343 } else if suspects.is_empty() {
344 format!("#{number} failed in the merge queue: {what}, and #{number} itself is the cause.")
345 } else {
346 format!("#{number} failed in the merge queue together with {}: {what} when they were combined.", numbers(suspects))
347 }
348}
349
350/// What the queue is working out about a failed entry, kept on it as
351/// JSON while smaller combinations are built.
352#[derive(Clone, Debug, Serialize, Deserialize)]
353pub(crate) struct Diagnosis {
354 pub(crate) base_commit: String,
355 /// The entries ahead, in queue order, then the entry: each as it was
356 /// tested.
357 pub(crate) stack: Vec<QueueStackItem>,
358 pub(crate) suspects: Vec<u32>,
359 pub(crate) builds: u32,
360 pub(crate) failure: Failure,
361 #[serde(default)]
362 pub(crate) done: bool,
363}
364
365/// How one smaller combination came out.
366#[derive(Clone, Copy, Debug, PartialEq, Eq)]
367pub(crate) enum ProbeResult {
368 /// The entry broke with these suspects.
369 Broke,
370 /// It passed with them.
371 Passed,
372 /// It could not be built or checked: narrowing stops here.
373 Stopped,
374}
375
376#[derive(Clone, Deserialize)]
377pub(crate) struct ProbeRow {
378 id: String,
379 repo_id: String,
380 entry_id: String,
381 number: u32,
382 subset: String,
383 state: String,
384 #[serde(default)]
385 tested_at: Option<String>,
386}
387
388impl ProbeRow {
389 fn subset(&self) -> Vec<u32> {
390 serde_json::from_str(&self.subset).unwrap_or_default()
391 }
392
393 fn branch(&self) -> String {
394 format!("{BRANCH_PREFIX}{}", self.id)
395 }
396}
397
398impl Work {
399 // --- Blame ---------------------------------------------------------------
400
401 /// Whether an entry ahead of `row` has not passed yet.
402 pub(crate) async fn ahead_unsettled(&self, row: &EntryRow) -> Result<bool> {
403 let ahead = row.ahead();
404 if ahead.is_empty() {
405 return Ok(false);
406 }
407 let active = self.entries(&row.repo_id, true).await?;
408 let states: Vec<(u32, QueueState)> = active.iter().map(|other| (other.number, other.state())).collect();
409 Ok(!ahead_settled(&ahead, &states))
410 }
411
412 /// Blames each held entry whose entries ahead have all passed by now:
413 /// it leaves the queue, and the failure is worked out.
414 pub(crate) async fn resolve_held(&self, repo_id: &str) -> Result<()> {
415 for _ in 0..16 {
416 let active = self.entries(repo_id, true).await?;
417 let states: Vec<(u32, QueueState)> = active.iter().map(|row| (row.number, row.state())).collect();
418 let Some(row) = active
419 .iter()
420 .find(|row| row.held.is_some() && ahead_settled(&row.ahead(), &states))
421 else {
422 return Ok(());
423 };
424 let claimed = self
425 .db
426 .prepare(
427 "UPDATE queue_entries SET state = 'failed', held = NULL, finished_at = ?
428 WHERE id = ? AND state = 'testing' AND held IS NOT NULL RETURNING id AS value",
429 )
430 .bind(&[rfc3339(now_ms()).into(), row.id.as_str().into()])?
431 .first::<ValueRow>(None)
432 .await?;
433 if claimed.is_none() {
434 continue;
435 }
436 let Some(failure) = row.held.as_deref().and_then(|held| serde_json::from_str::<Failure>(held).ok()) else {
437 continue;
438 };
439 self.eject(row, &failure.report(&row.id)).await?;
440 self.diagnose(row, failure).await?;
441 }
442 Ok(())
443 }
444
445 /// Whether the entry's own checks pass on the commit it was queued at.
446 async fn passed_alone(&self, row: &EntryRow) -> Result<bool> {
447 let facts = self.facts(&row.repo_id, row.head_commit.as_deref()).await?;
448 Ok(facts.failed.is_empty() && facts.required_failed().is_empty())
449 }
450
451 /// Works out who answers for an entry's failure, after it left the
452 /// queue: at once, or by building smaller combinations first.
453 pub(crate) async fn diagnose(&self, row: &EntryRow, failure: Failure) -> Result<()> {
454 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
455 return Ok(());
456 };
457 let ahead = row.ahead();
458 // Only a check failure asks whether it fails alone.
459 let passed_alone = failure.kind != FailureKind::Checks || self.passed_alone(row).await?;
460 match plan(&ahead, &failure, passed_alone) {
461 Plan::Blame(suspects) => self.integration_failed(row, &pull, &failure, false, &suspects, &[]).await,
462 Plan::BaseMoved => {
463 // What landed is context: without it, the failure is still told.
464 let landed = self.landed_since_branched(row, &pull).await.unwrap_or_default();
465 self.integration_failed(row, &pull, &failure, false, &[], &landed).await
466 }
467 Plan::Bisect => {
468 let Some(stack) = self.tested_stack(row, &pull, &ahead).await? else {
469 // What was tested can no longer be found: say what is known.
470 return self.integration_failed(row, &pull, &failure, false, &ahead, &[]).await;
471 };
472 let diagnosis = Diagnosis {
473 base_commit: row.base_commit.clone().unwrap_or_default(),
474 stack,
475 suspects: ahead,
476 builds: 0,
477 failure,
478 done: false,
479 };
480 self.go_on(row, None, diagnosis).await
481 }
482 }
483 }
484
485 /// The pull requests merged into the default branch since `pull`'s
486 /// entry branched from it and before it was tested, newest first: the
487 /// latest merged ones whose commits the entry's history does not hold.
488 async fn landed_since_branched(&self, row: &EntryRow, pull: &Pull) -> Result<Vec<u32>> {
489 let (Some(head), Some(tested_at)) = (row.head_commit.clone(), row.tested_at.clone()) else {
490 return Ok(Vec::new());
491 };
492 let viewer = Some(pull.owner().clone());
493 let repo: Outcome<Repo> = g1t_kit::call(
494 &self.repos,
495 "get_by_id",
496 &GetByIdArgs { id: row.repo_id.clone(), viewer: viewer.clone() },
497 )
498 .await?;
499 let Outcome::Ok(repo) = repo else { return Ok(Vec::new()) };
500 let path = pull.fork.clone().unwrap_or_else(|| RepoPath {
501 namespace: repo.namespace.clone(),
502 name: repo.name.clone(),
503 });
504 let merged = async {
505 self.db
506 .prepare(
507 "SELECT number, head_commit, merged_at FROM pulls
508 WHERE repo_id = ? AND status = 'merged' AND id != ? AND merged_at <= ?
509 AND (base_branch IS NULL OR base_branch = '' OR base_branch = ?)
510 ORDER BY merged_at DESC LIMIT ?",
511 )
512 .bind(&[
513 row.repo_id.as_str().into(),
514 pull.id.as_str().into(),
515 tested_at.as_str().into(),
516 repo.default_branch.as_str().into(),
517 LANDED_CANDIDATES.into(),
518 ])?
519 .all()
520 .await?
521 .results::<MergedRow>()
522 };
523 let log = LogArgs { path, viewer, git_ref: Some(head), limit: LANDED_HISTORY };
524 let history = g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &log);
525 let (merged, history) = futures_util::future::try_join(merged, history).await?;
526 let Outcome::Ok(history) = history else { return Ok(Vec::new()) };
527 // A history cut short holds only what is newer than its oldest commit.
528 let oldest = (history.len() >= LANDED_HISTORY as usize)
529 .then(|| history.iter().map(|commit| commit.authored_at.as_str()).min())
530 .flatten();
531 Ok(landed_since(&merged, &history, oldest))
532 }
533
534 /// Each entry ahead and then `row`'s, as they were tested together.
535 async fn tested_stack(&self, row: &EntryRow, pull: &Pull, ahead: &[u32]) -> Result<Option<Vec<QueueStackItem>>> {
536 let path: Option<RepoPath> = g1t_kit::call(
537 &self.repos,
538 "path_by_id",
539 &PathByIdArgs { id: row.repo_id.clone() },
540 )
541 .await?;
542 let Some(path) = path else { return Ok(None) };
543 let item = |pull: &Pull, commit: String| QueueStackItem {
544 number: pull.number,
545 title: pull.title.clone(),
546 source: pull.fork.clone().unwrap_or_else(|| path.clone()),
547 branch: pull.branch.clone().unwrap_or_default(),
548 commit,
549 };
550 let mut stack = Vec::new();
551 for number in ahead {
552 let tested = self
553 .db
554 .prepare(
555 "SELECT * FROM queue_entries WHERE repo_id = ? AND number = ? AND head_commit IS NOT NULL
556 ORDER BY created_at DESC LIMIT 1",
557 )
558 .bind(&[row.repo_id.as_str().into(), (*number).into()])?
559 .first::<EntryRow>(None)
560 .await?;
561 let (Some(tested), Some(other)) = (tested, self.pull(&row.repo_id, *number).await?) else {
562 return Ok(None);
563 };
564 let Some(commit) = tested.head_commit else { return Ok(None) };
565 stack.push(item(&other, commit));
566 }
567 let Some(commit) = row.head_commit.clone() else { return Ok(None) };
568 stack.push(item(pull, commit));
569 Ok(Some(stack))
570 }
571
572 /// Builds the next smaller combination, or reports what is known when
573 /// there is none to build. `before` is the diagnosis as last kept, so
574 /// that only one result moves it on.
575 async fn go_on(&self, row: &EntryRow, before: Option<&str>, mut diagnosis: Diagnosis) -> Result<()> {
576 let subset = next_probe(&diagnosis.suspects, diagnosis.builds);
577 diagnosis.done = subset.is_none();
578 if !self.keep_diagnosis(&row.id, before, &diagnosis).await? {
579 return Ok(());
580 }
581 let Some(subset) = subset else {
582 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
583 return Ok(());
584 };
585 return self
586 .integration_failed(row, &pull, &diagnosis.failure, narrowed(&diagnosis), &diagnosis.suspects, &[])
587 .await;
588 };
589 let now = now_ms();
590 self.db
591 .prepare(
592 "INSERT INTO queue_probes (id, repo_id, entry_id, number, subset, state, created_at)
593 VALUES (?, ?, ?, ?, ?, 'waiting', ?)",
594 )
595 .bind(&[
596 new_id("qpr", now).into(),
597 row.repo_id.as_str().into(),
598 row.id.as_str().into(),
599 row.number.into(),
600 serde_json::to_string(&subset)?.into(),
601 rfc3339(now).into(),
602 ])?
603 .run()
604 .await?;
605 self.changed(&row.repo_id).await
606 }
607
608 /// Keeps the diagnosis on its entry if it is still as `before` was.
609 /// Whether it was kept.
610 async fn keep_diagnosis(&self, entry_id: &str, before: Option<&str>, diagnosis: &Diagnosis) -> Result<bool> {
611 let kept = self
612 .db
613 .prepare("UPDATE queue_entries SET diagnosis = ? WHERE id = ? AND diagnosis IS ? RETURNING id AS value")
614 .bind(&[
615 serde_json::to_string(diagnosis)?.into(),
616 entry_id.into(),
617 before.map_or(JsValue::NULL, JsValue::from),
618 ])?
619 .first::<ValueRow>(None)
620 .await?;
621 Ok(kept.is_some())
622 }
623
624 // --- Smaller combinations -------------------------------------------------
625
626 /// One job for each smaller combination waiting to be built, and the
627 /// end of any that took too long.
628 pub(crate) async fn probe_jobs(&self, repo_id: &str) -> Result<Vec<QueueJob>> {
629 let probes = self
630 .db
631 .prepare("SELECT * FROM queue_probes WHERE repo_id = ? AND state IN ('waiting', 'testing')")
632 .bind(&[repo_id.into()])?
633 .all()
634 .await?
635 .results::<ProbeRow>()?;
636 if probes.is_empty() {
637 return Ok(Vec::new());
638 }
639 let rule = self.queue_rule(repo_id).await?;
640 let stale = rfc3339(now_ms().saturating_sub(u64::from(rule.check_response_timeout_minutes.max(1)) * 60 * 1000));
641 let mut jobs = Vec::new();
642 for probe in probes {
643 if probe.state == "testing" {
644 if probe.tested_at.as_deref().is_none_or(|at| at < stale.as_str()) {
645 self.probe_result(&probe, ProbeResult::Stopped, Some("It took too long.")).await?;
646 }
647 continue;
648 }
649 if let Some(job) = self.probe_job(&probe).await? {
650 jobs.push(job);
651 }
652 }
653 Ok(jobs)
654 }
655
656 async fn probe_job(&self, probe: &ProbeRow) -> Result<Option<QueueJob>> {
657 let found = match self.entry_by_id(&probe.entry_id).await? {
658 Some(row) => {
659 let diagnosis = row
660 .diagnosis
661 .as_deref()
662 .and_then(|text| serde_json::from_str::<Diagnosis>(text).ok())
663 .filter(|diagnosis| !diagnosis.done);
664 let pull = self.pull_by_id(&row.pull_id).await?;
665 match (diagnosis, pull, row.actor()) {
666 (Some(diagnosis), Some(pull), Some(actor)) => Some((diagnosis, pull, actor)),
667 _ => None,
668 }
669 }
670 None => None,
671 };
672 // Nothing is waiting on it any more.
673 let Some((diagnosis, pull, actor)) = found else {
674 self.probe_result(probe, ProbeResult::Stopped, Some("Its failure was already reported.")).await?;
675 return Ok(None);
676 };
677 let repo: Outcome<Repo> = g1t_kit::call(
678 &self.repos,
679 "get_by_id",
680 &GetByIdArgs {
681 id: probe.repo_id.clone(),
682 viewer: Some(actor),
683 },
684 )
685 .await?;
686 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
687 self.probe_result(probe, ProbeResult::Stopped, Some("The repository is archived.")).await?;
688 return Ok(None);
689 };
690 let subset = probe.subset();
691 let mut stack = Vec::new();
692 for item in &diagnosis.stack {
693 if item.number != probe.number && !subset.contains(&item.number) {
694 continue;
695 }
696 let mut item = item.clone();
697 // A pull request that landed since may have lost its branch;
698 // its commit is in the default branch's history now.
699 if let Some(other) = self.pull(&repo.id, item.number).await?
700 && other.status == PullStatus::Merged
701 {
702 item.source = RepoPath {
703 namespace: repo.namespace.clone(),
704 name: repo.name.clone(),
705 };
706 item.branch = repo.default_branch.clone();
707 }
708 if item.branch.is_empty() {
709 item.branch = repo.default_branch.clone();
710 }
711 stack.push(item);
712 }
713 let token = new_token();
714 let claimed = self
715 .db
716 .prepare(
717 "UPDATE queue_probes SET state = 'testing', token_hash = ?, tested_at = ?
718 WHERE id = ? AND state = 'waiting' RETURNING id AS value",
719 )
720 .bind(&[hash(&token).into(), rfc3339(now_ms()).into(), probe.id.as_str().into()])?
721 .first::<ValueRow>(None)
722 .await?;
723 if claimed.is_none() {
724 return Ok(None);
725 }
726 Ok(Some(QueueJob {
727 entry_id: probe.id.clone(),
728 token,
729 repo: RepoPath {
730 namespace: repo.namespace.clone(),
731 name: repo.name.clone(),
732 },
733 default_branch: repo.default_branch.clone(),
734 base_commit: diagnosis.base_commit.clone(),
735 branch: probe.branch(),
736 stack,
737 checks: Vec::new(),
738 contract_checks: Vec::new(),
739 actor: pull.owner().clone(),
740 }))
741 }
742
743 /// A sandbox's result for a smaller combination: built, it waits for
744 /// its `merge_group` workflows as a queue state does.
745 pub(crate) async fn report_probe(&self, a: ReportQueueArgs) -> Result<Outcome<QueueState>> {
746 let probe = self
747 .db
748 .prepare("SELECT * FROM queue_probes WHERE id = ? AND token_hash = ?")
749 .bind(&[a.entry_id.as_str().into(), hash(&a.token).into()])?
750 .first::<ProbeRow>(None)
751 .await?;
752 let Some(probe) = probe else {
753 return Ok(Outcome::fail(FailureCode::NotFound, "No such queue entry."));
754 };
755 if probe.state != "testing" {
756 return Ok(Outcome::fail(FailureCode::Conflict, "This state is no longer being tested."));
757 }
758 if a.conflict_with.is_some() || !a.conflicts.is_empty() {
759 self.probe_result(&probe, ProbeResult::Broke, None).await?;
760 return Ok(Outcome::Ok(QueueState::Failed));
761 }
762 let built = a.error.is_none() && a.results.iter().all(|result| result.passed);
763 let Some(commit) = a.combined_commit.as_deref().filter(|_| built) else {
764 self.probe_result(&probe, ProbeResult::Stopped, a.error.as_deref()).await?;
765 return Ok(Outcome::Ok(QueueState::Failed));
766 };
767 let Some(row) = self.entry_by_id(&probe.entry_id).await? else {
768 return Ok(Outcome::Ok(QueueState::Removed));
769 };
770 let base = row
771 .diagnosis
772 .as_deref()
773 .and_then(|text| serde_json::from_str::<Diagnosis>(text).ok())
774 .map(|diagnosis| diagnosis.base_commit);
775 self.db
776 .prepare(
777 "UPDATE queue_probes SET combined_commit = ?, token_hash = NULL WHERE id = ? AND state = 'testing'",
778 )
779 .bind(&[commit.into(), probe.id.as_str().into()])?
780 .run()
781 .await?;
782 let runs = self
783 .run_merge_group(&probe.repo_id, &probe.id, &probe.branch(), base.as_deref(), probe.number, &probe.subset(), commit)
784 .await
785 .unwrap_or(0);
786 if runs == 0 {
787 self.probe_result(&probe, ProbeResult::Stopped, Some("No workflow runs on merge_group events.")).await?;
788 return Ok(Outcome::Ok(QueueState::Failed));
789 }
790 Ok(Outcome::Ok(QueueState::Testing))
791 }
792
793 /// The `merge_group` workflows on a smaller combination finished.
794 /// Whether `commit` was one.
795 pub(crate) async fn probe_finished(&self, repo_id: &str, commit: &str, passed: bool) -> Result<bool> {
796 let probe = self
797 .db
798 .prepare(
799 "SELECT * FROM queue_probes
800 WHERE repo_id = ? AND combined_commit = ? AND state = 'testing' AND token_hash IS NULL",
801 )
802 .bind(&[repo_id.into(), commit.into()])?
803 .first::<ProbeRow>(None)
804 .await?;
805 let Some(probe) = probe else { return Ok(false) };
806 let result = if passed { ProbeResult::Passed } else { ProbeResult::Broke };
807 self.probe_result(&probe, result, None).await?;
808 Ok(true)
809 }
810
811 /// Narrows the suspects by how a smaller combination came out, and
812 /// builds the next one or reports.
813 async fn probe_result(&self, probe: &ProbeRow, result: ProbeResult, error: Option<&str>) -> Result<()> {
814 let state = match result {
815 ProbeResult::Broke => "failed",
816 ProbeResult::Passed => "passed",
817 ProbeResult::Stopped => "stopped",
818 };
819 let claimed = self
820 .db
821 .prepare(
822 "UPDATE queue_probes SET state = ?, token_hash = NULL, error = ?, finished_at = ?
823 WHERE id = ? AND state IN ('waiting', 'testing') RETURNING id AS value",
824 )
825 .bind(&[
826 state.into(),
827 error.map_or(JsValue::NULL, JsValue::from),
828 rfc3339(now_ms()).into(),
829 probe.id.as_str().into(),
830 ])?
831 .first::<ValueRow>(None)
832 .await?;
833 if claimed.is_none() {
834 return Ok(());
835 }
836 self.drop_branch_named(&probe.repo_id, &probe.branch()).await;
837 let Some(row) = self.entry_by_id(&probe.entry_id).await? else {
838 return Ok(());
839 };
840 let Some(before) = row.diagnosis.clone() else { return Ok(()) };
841 let Ok(mut diagnosis) = serde_json::from_str::<Diagnosis>(&before) else {
842 return Ok(());
843 };
844 if diagnosis.done {
845 return Ok(());
846 }
847 match result {
848 ProbeResult::Stopped => {
849 // Nothing more can be learned: report what is known.
850 diagnosis.done = true;
851 if !self.keep_diagnosis(&row.id, Some(&before), &diagnosis).await? {
852 return Ok(());
853 }
854 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
855 return Ok(());
856 };
857 self.integration_failed(&row, &pull, &diagnosis.failure, narrowed(&diagnosis), &diagnosis.suspects, &[])
858 .await
859 }
860 ProbeResult::Broke | ProbeResult::Passed => {
861 diagnosis.suspects = narrow(&diagnosis.suspects, &probe.subset(), result == ProbeResult::Broke);
862 diagnosis.builds += 1;
863 self.go_on(&row, Some(&before), diagnosis).await
864 }
865 }
866 }
867
868 // --- Telling the work -------------------------------------------------------
869
870 /// Tells the pull request, its issue, its session and anyone following
871 /// that its combined state failed, and who it failed with. `landed`:
872 /// with nothing ahead, the pull requests merged into the default branch
873 /// since it branched.
874 pub(crate) async fn integration_failed(
875 &self,
876 row: &EntryRow,
877 pull: &Pull,
878 failure: &Failure,
879 bisected: bool,
880 suspects: &[u32],
881 landed: &[u32],
882 ) -> Result<()> {
883 let ahead = row.ahead();
884 let mut combination = ahead.clone();
885 combination.push(pull.number);
886 let g1t = ("g1t", "g1t");
887 let note = pull_note(&ahead, suspects, bisected, landed, &failure.check);
888 self.note(&row.repo_id, pull.number, g1t, &note).await?;
889 let summary = summary(pull.number, suspects, landed, &failure.check);
890 if let Some(issue) = pull.issue {
891 self.note(&row.repo_id, issue, g1t, &format!("noted: {summary}")).await?;
892 }
893 self.session_note(pull, &format!("g1t {note}")).await?;
894 self.keep_decision_quietly(
895 &row.repo_id,
896 NewDecision {
897 issue: pull.issue,
898 pull: Some(pull.number),
899 session_id: None,
900 actor: None,
901 kind: DecisionKind::IntegrationFailure,
902 confidence: DecisionConfidence::Tested,
903 summary,
904 evidence: json!({
905 "commit": failure.combined_commit,
906 "check": failure.check,
907 "combination": combination,
908 "other_pulls": ahead,
909 "suspects": suspects,
910 "bisected": bisected,
911 "landed": landed,
912 }),
913 },
914 )
915 .await;
916 self.publish_as(
917 "integration.failed",
918 &row.repo_id,
919 None,
920 IntegrationFailed {
921 repo_id: row.repo_id.clone(),
922 pull_id: pull.id.clone(),
923 number: pull.number,
924 pull: pull.number,
925 issue: pull.issue,
926 combination,
927 check: failure.check.clone(),
928 bisected,
929 suspects: suspects.to_vec(),
930 other_pulls: ahead,
931 landed: landed.to_vec(),
932 commit: failure.combined_commit.clone(),
933 },
934 )
935 .await
936 }
937
938 /// Tells anyone following that an entry's combined state passed.
939 pub(crate) async fn integration_passed(&self, row: &EntryRow, commit: Option<&str>) -> Result<()> {
940 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
941 return Ok(());
942 };
943 let ahead = row.ahead();
944 let mut combination = ahead.clone();
945 combination.push(pull.number);
946 let summary = if ahead.is_empty() {
947 format!("#{} passed in the merge queue on the default branch as it is.", pull.number)
948 } else {
949 format!("#{} passed in the merge queue combined with {}.", pull.number, numbers(&ahead))
950 };
951 self.keep_decision_quietly(
952 &row.repo_id,
953 NewDecision {
954 issue: pull.issue,
955 pull: Some(pull.number),
956 session_id: None,
957 actor: None,
958 kind: DecisionKind::IntegrationPassed,
959 confidence: DecisionConfidence::Tested,
960 summary,
961 evidence: json!({ "commit": commit, "combination": combination, "other_pulls": ahead }),
962 },
963 )
964 .await;
965 self.publish_as(
966 "integration.passed",
967 &row.repo_id,
968 None,
969 IntegrationPassed {
970 repo_id: row.repo_id.clone(),
971 pull_id: pull.id.clone(),
972 number: pull.number,
973 pull: pull.number,
974 issue: pull.issue,
975 combination,
976 commit: commit.map(str::to_owned),
977 },
978 )
979 .await
980 }
981
982 /// Appends a note by g1t to a pull request's session, and to the
983 /// harness sessions working on it or its issue (coordination/).
984 pub(crate) async fn session_note(&self, pull: &Pull, text: &str) -> Result<()> {
985 self.note_sessions(pull, text).await;
986 let now = rfc3339(now_ms());
987 self.db
988 .prepare(
989 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
990 SELECT ?, COALESCE(MAX(seq), 0) + 1, 'note', ?, NULL, ?, ?
991 FROM session_entries WHERE pull_id = ?",
992 )
993 .bind(&[
994 pull.id.as_str().into(),
995 text.into(),
996 pull.head_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
997 now.as_str().into(),
998 pull.id.as_str().into(),
999 ])?
1000 .run()
1001 .await?;
1002 self.publish_as(
1003 "session.appended",
1004 &pull.repo_id,
1005 None,
1006 SessionAppended {
1007 pull_id: pull.id.clone(),
1008 repo_id: pull.repo_id.clone(),
1009 number: pull.number,
1010 count: 1,
1011 },
1012 )
1013 .await
1014 }
1015
1016 // --- Rejoining ---------------------------------------------------------------
1017
1018 /// Puts a pull request the queue took out back in, once a new commit's
1019 /// checks pass: merged again as whoever queued it, so the branch's
1020 /// rules are checked as they are now. A refusal is noted.
1021 pub(crate) async fn requeue_if_fixed(&self, pull_id: &str, sha: &str) -> Result<()> {
1022 let row = self
1023 .db
1024 .prepare(
1025 "SELECT * FROM queue_entries WHERE pull_id = ? AND requeue = 1 AND state = 'failed'
1026 ORDER BY created_at DESC LIMIT 1",
1027 )
1028 .bind(&[pull_id.into()])?
1029 .first::<EntryRow>(None)
1030 .await?;
1031 let Some(row) = row else { return Ok(()) };
1032 // The commit that failed: nothing has been fixed yet.
1033 if row.head_commit.as_deref() == Some(sha) {
1034 return Ok(());
1035 }
1036 let claimed = self
1037 .db
1038 .prepare("UPDATE queue_entries SET requeue = 0 WHERE id = ? AND requeue = 1 RETURNING id AS value")
1039 .bind(&[row.id.as_str().into()])?
1040 .first::<ValueRow>(None)
1041 .await?;
1042 let (Some(_), Some(actor), Some(pull)) = (claimed, row.actor(), self.pull_by_id(pull_id).await?) else {
1043 return Ok(());
1044 };
1045 if pull.status != PullStatus::Open || pull.head_commit.as_deref() != Some(sha) {
1046 return Ok(());
1047 }
1048 let path: Option<RepoPath> = g1t_kit::call(
1049 &self.repos,
1050 "path_by_id",
1051 &PathByIdArgs { id: pull.repo_id.clone() },
1052 )
1053 .await?;
1054 let Some(repo) = path else { return Ok(()) };
1055 let merged = self
1056 .merge_pull(PullActionArgs {
1057 actor,
1058 repo,
1059 number: pull.number,
1060 summary: String::new(),
1061 keep_issue_open: row.keep_issue_open != 0,
1062 ignore_checks: false,
1063 bypass_rules: false,
1064 })
1065 .await?;
1066 let short = &sha[..sha.len().min(12)];
1067 let text = match merged {
1068 Outcome::Ok(_) => format!("put this back in the merge queue: its checks pass on {short}, pushed after it was taken out"),
1069 Outcome::Fail(failure) => format!(
1070 "could not put this back in the merge queue though its checks pass on {short}: {} Merge it again when that is dealt with.",
1071 failure.message
1072 ),
1073 };
1074 self.note(&pull.repo_id, pull.number, ("g1t", "g1t"), &text).await
1075 }
1076
1077 async fn entry_by_id(&self, id: &str) -> Result<Option<EntryRow>> {
1078 self.db
1079 .prepare("SELECT * FROM queue_entries WHERE id = ?")
1080 .bind(&[id.into()])?
1081 .first::<EntryRow>(None)
1082 .await
1083 }
1084}
1085
1086/// Whether the suspects were narrowed by any build.
1087fn narrowed(diagnosis: &Diagnosis) -> bool {
1088 let ahead = diagnosis.stack.len().saturating_sub(1);
1089 diagnosis.suspects.len() < ahead
1090}
1091
1092#[cfg(test)]
1093mod tests {
1094 use super::*;
1095
1096 fn failure(kind: FailureKind) -> Failure {
1097 Failure {
1098 kind,
1099 check: check("CI / merge_group", "failure", Some("https://g1t.sh/a/b/actions/runs/1".into())),
1100 error: None,
1101 conflict_with: None,
1102 conflicts: Vec::new(),
1103 results: Vec::new(),
1104 combined_commit: Some("c0ffee".into()),
1105 }
1106 }
1107
1108 #[test]
1109 fn a_failure_waits_for_the_entries_ahead_of_it() {
1110 use QueueState::*;
1111 let active = [(11, Passed), (12, Testing), (14, Testing)];
1112 assert!(!ahead_settled(&[11, 12], &active), "#12 is still being tested");
1113 assert!(ahead_settled(&[11], &active));
1114 // #10 landed and left the queue.
1115 assert!(ahead_settled(&[10, 11], &active));
1116 assert!(ahead_settled(&[], &active));
1117 }
1118
1119 #[test]
1120 fn who_answers_for_a_failure() {
1121 let checks = failure(FailureKind::Checks);
1122 // Tested alone on the default branch, failing there though it
1123 // passes on its own: the default branch moved under it.
1124 assert_eq!(plan(&[], &checks, true), Plan::BaseMoved);
1125 // Alone, and failing on its own too: it is the cause.
1126 assert_eq!(plan(&[], &checks, false), Plan::Blame(vec![]));
1127 // Alone, but not a check failure: nothing to look up.
1128 for kind in [FailureKind::Conflict, FailureKind::Missing, FailureKind::Build] {
1129 assert_eq!(plan(&[], &failure(kind), true), Plan::Blame(vec![]), "{kind:?}");
1130 }
1131 // Its own checks fail too: still the cause.
1132 assert_eq!(plan(&[11, 12], &checks, false), Plan::Blame(vec![]));
1133 // One entry ahead: that one, without building anything.
1134 assert_eq!(plan(&[11], &checks, true), Plan::Blame(vec![11]));
1135 // More: narrow them down.
1136 assert_eq!(plan(&[11, 12, 13], &checks, true), Plan::Bisect);
1137 // A conflict names the entry it collided with.
1138 let conflict = Failure { conflict_with: Some(12), ..failure(FailureKind::Conflict) };
1139 assert_eq!(plan(&[11, 12], &conflict, true), Plan::Blame(vec![12]));
1140 // A required check that never reports is not about combining.
1141 assert_eq!(plan(&[11, 12], &failure(FailureKind::Missing), true), Plan::Blame(vec![]));
1142 assert_eq!(plan(&[11, 12], &failure(FailureKind::Build), true), Plan::Blame(vec![]));
1143 }
1144
1145 /// Plays a bisection through against `breaks`, the suspects the entry
1146 /// really fails with: how many builds it took and what was left.
1147 fn bisect(ahead: &[u32], breaks: u32) -> (u32, Vec<u32>) {
1148 let (mut suspects, mut builds) = (ahead.to_vec(), 0);
1149 while let Some(tested) = next_probe(&suspects, builds) {
1150 suspects = narrow(&suspects, &tested, tested.contains(&breaks));
1151 builds += 1;
1152 }
1153 (builds, suspects)
1154 }
1155
1156 #[test]
1157 fn bisection_halves_the_suspects_within_the_budget() {
1158 assert_eq!(next_probe(&[11, 12, 13], 0), Some(vec![11, 12]));
1159 assert_eq!(next_probe(&[11, 12], 0), Some(vec![11]));
1160 assert_eq!(next_probe(&[11], 0), None);
1161 assert_eq!(next_probe(&[11, 12], PROBE_BUDGET), None);
1162 assert_eq!(narrow(&[11, 12, 13], &[11, 12], true), vec![11, 12]);
1163 assert_eq!(narrow(&[11, 12, 13], &[11, 12], false), vec![13]);
1164 // A full default batch (three ahead) needs two builds at most.
1165 assert_eq!(bisect(&[11, 12, 13], 11), (2, vec![11]));
1166 assert_eq!(bisect(&[11, 12, 13], 12), (2, vec![12]));
1167 assert_eq!(bisect(&[11, 12, 13], 13), (1, vec![13]));
1168 // Eight ahead: found in three.
1169 let eight: Vec<u32> = (1..=8).collect();
1170 for breaks in 1..=8 {
1171 assert_eq!(bisect(&eight, breaks), (3, vec![breaks]));
1172 }
1173 // More than the budget can narrow to one: what is left is reported.
1174 let twelve: Vec<u32> = (1..=12).collect();
1175 let (builds, left) = bisect(&twelve, 1);
1176 assert_eq!(builds, PROBE_BUDGET);
1177 assert_eq!(left, vec![1, 2]);
1178 }
1179
1180 #[test]
1181 fn the_notes_name_the_other_changes_and_the_check() {
1182 let check = failure(FailureKind::Checks).check;
1183 let note = pull_note(&[11, 12, 13], &[12], true, &[], &check);
1184 assert!(note.starts_with("traced its merge queue failure to how it combines with #12 (narrowed down from #11, #12 and #13"), "{note}");
1185 assert!(note.contains("`CI / merge_group` failed"));
1186 assert!(note.contains("once #12 has landed"));
1187 let own = pull_note(&[], &[], false, &[], &check);
1188 assert!(own.contains("on the default branch as it is now, and the change is the cause"), "{own}");
1189 let conflict = pull_note(&[11], &[11], false, &[], &FailedCheck { name: "merge".into(), conclusion: "conflict".into(), url: None });
1190 assert!(conflict.contains("it did not merge cleanly"), "{conflict}");
1191 assert!(!conflict.contains("narrowed"));
1192 assert_eq!(summary(14, &[11, 12], &[], &check), "#14 failed in the merge queue together with #11 and #12: `CI / merge_group` failed when they were combined.");
1193 assert_eq!(numbers(&[]), "");
1194 }
1195
1196 #[test]
1197 fn the_notes_name_what_landed_since_it_branched() {
1198 let check = failure(FailureKind::Checks).check;
1199 // One landed: named as the likely cause.
1200 let one = pull_note(&[], &[], false, &[1], &check);
1201 assert_eq!(
1202 one,
1203 "traced its merge queue failure to the default branch as it is now: `CI / merge_group` failed with this change merged into it, though this passes its checks on its own. Since this branched, #1 landed and is the likely cause. Bring it up to date with the default branch, fix what breaks, and push. Push a fix: once its checks pass on the new commit, it rejoins the merge queue by itself."
1204 );
1205 assert_eq!(
1206 summary(14, &[], &[1], &check),
1207 "#14 failed in the merge queue on the default branch as it is now: `CI / merge_group` failed, though #14 passes its checks on its own; since it branched, #1 landed and is the likely cause."
1208 );
1209 // Several: listed, newest first, as possible causes.
1210 let several = pull_note(&[], &[], false, &[7, 5, 1], &check);
1211 assert!(several.contains("Since this branched, #7, #5 and #1 landed and may be what it conflicts with."), "{several}");
1212 assert!(!several.contains("the change is the cause"));
1213 assert!(summary(14, &[], &[7, 5], &check).ends_with("since it branched, #7 and #5 landed and may be what it conflicts with."));
1214 // None found: the change is the cause, as before.
1215 assert_eq!(summary(14, &[], &[], &check), "#14 failed in the merge queue: `CI / merge_group` failed, and #14 itself is the cause.");
1216 }
1217
1218 fn commit(hash: &str, at: &str) -> Commit {
1219 Commit {
1220 hash: hash.into(),
1221 tree_hash: format!("t{hash}"),
1222 message: String::new(),
1223 author: g1t_contracts::repos::Signature { name: "a".into(), email: "a@example.com".into() },
1224 parents: Vec::new(),
1225 authored_at: at.into(),
1226 }
1227 }
1228
1229 fn merged(number: u32, commit: &str, at: &str) -> MergedRow {
1230 MergedRow { number, head_commit: Some(commit.into()), merged_at: Some(at.into()) }
1231 }
1232
1233 #[test]
1234 fn what_landed_is_what_its_history_does_not_hold() {
1235 // #3 and #2 landed after it branched from #1's commit.
1236 let pulls = [
1237 merged(3, "c3", "2026-10-12T12:00:00.000Z"),
1238 merged(2, "c2", "2026-10-12T11:00:00.000Z"),
1239 merged(1, "c1", "2026-10-12T10:00:00.000Z"),
1240 ];
1241 let history = [commit("mine", "2026-10-12T10:30:00Z"), commit("c1", "2026-10-12T10:00:00Z"), commit("c0", "2026-10-11T00:00:00Z")];
1242 assert_eq!(landed_since(&pulls, &history, None), vec![3, 2]);
1243 // Brought up to date with #2 since: only #3 is new to it.
1244 let updated = [commit("merge", "2026-10-12T11:30:00Z"), commit("c2", "2026-10-12T11:00:00Z"), commit("c1", "2026-10-12T10:00:00Z")];
1245 assert_eq!(landed_since(&pulls, &updated, None), vec![3]);
1246 // A history cut short does not count what is older than it.
1247 let short = [commit("mine", "2026-10-12T11:30:00Z"), commit("x", "2026-10-12T10:30:00Z")];
1248 assert_eq!(landed_since(&pulls, &short, Some("2026-10-12T10:30:00Z")), vec![3, 2]);
1249 // A merged pull request with no recorded commit is not counted.
1250 let unknown = [MergedRow { number: 9, head_commit: None, merged_at: None }];
1251 assert!(landed_since(&unknown, &history, None).is_empty());
1252 // At most MAX_LANDED, newest first.
1253 let many: Vec<MergedRow> = (1..=15).rev().map(|n| merged(n, &format!("m{n}"), "2026-10-12T12:00:00.000Z")).collect();
1254 let found = landed_since(&many, &history, None);
1255 assert_eq!(found.len(), MAX_LANDED);
1256 assert_eq!(found.first(), Some(&15));
1257 }
1258
1259 #[test]
1260 fn failures_are_read_from_what_was_reported() {
1261 let conflict = ReportQueueArgs {
1262 entry_id: "qen_1".into(),
1263 token: String::new(),
1264 combined_commit: None,
1265 results: Vec::new(),
1266 error: Some("conflict".into()),
1267 conflict_with: Some(12),
1268 conflicts: vec!["src/a.rs".into()],
1269 };
1270 let read = Failure::of_report(&conflict, None);
1271 assert_eq!(read.kind, FailureKind::Conflict);
1272 assert_eq!(read.check.name, "merge");
1273 let back = read.report("qen_1");
1274 assert_eq!((back.conflict_with, back.conflicts.len()), (Some(12), 1));
1275 let missing = Failure::of_report(&ReportQueueArgs { conflict_with: None, conflicts: Vec::new(), ..conflict }, Some(&["CI".to_owned()]));
1276 assert_eq!((missing.kind, missing.check.conclusion.as_str()), (FailureKind::Missing, "missing"));
1277 let workflows = Failure::of_workflows("abc", &["CI / merge_group".into()], &[], &[], "the workflow failed".into(), |name| Some(format!("https://x/{name}")));
1278 assert_eq!(workflows.kind, FailureKind::Checks);
1279 assert_eq!(workflows.check.url.as_deref(), Some("https://x/CI / merge_group"));
1280 let never = Failure::of_workflows("abc", &[], &[], &["CI".into()], "missing".into(), |_| None);
1281 assert_eq!(never.kind, FailureKind::Missing);
1282 // A held failure survives being kept as JSON.
1283 let kept: Failure = serde_json::from_str(&serde_json::to_string(&workflows).unwrap()).unwrap();
1284 assert_eq!(kept.check, workflows.check);
1285 }
1286}