Skip to content
Sign up

flagon-io/g1t

public
00

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