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,358 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, 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 a sandbox 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 sandbox 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 /// One job for each smaller combination waiting to be built, and the
655 /// end of any that took too long.
656 pub(crate) async fn probe_jobs(&self, repo_id: &str) -> Result<Vec<QueueJob>> {
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(Vec::new());
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 let mut jobs = Vec::new();
670 for probe in probes {
671 if probe.state == "testing" {
672 if probe.tested_at.as_deref().is_none_or(|at| at < stale.as_str()) {
673 self.probe_result(&probe, ProbeResult::Stopped, Some("It took too long.")).await?;
674 }
675 continue;
676 }
677 if let Some(job) = self.probe_job(&probe).await? {
678 jobs.push(job);
679 }
680 }
681 Ok(jobs)
682 }
683
684 async fn probe_job(&self, probe: &ProbeRow) -> Result<Option<QueueJob>> {
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, pull, actor)) = found else {
702 self.probe_result(probe, ProbeResult::Stopped, Some("Its failure was already reported.")).await?;
703 return Ok(None);
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),
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(None);
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(None);
753 }
754 Ok(Some(QueueJob {
755 entry_id: probe.id.clone(),
756 token,
757 repo: RepoPath {
758 namespace: repo.namespace.clone(),
759 name: repo.name.clone(),
760 },
761 default_branch: repo.default_branch.clone(),
762 base_commit: diagnosis.base_commit.clone(),
763 branch: probe.branch(),
764 stack,
765 checks: Vec::new(),
766 contract_checks: Vec::new(),
767 actor: pull.owner().clone(),
768 }))
769 }
770
771 /// A sandbox's result for a smaller combination: built, it waits for
772 /// its `merge_group` workflows as a queue state does.
773 pub(crate) async fn report_probe(&self, a: ReportQueueArgs) -> Result<Outcome<QueueState>> {
774 let probe = self
775 .db
776 .prepare("SELECT * FROM queue_probes WHERE id = ? AND token_hash = ?")
777 .bind(&[a.entry_id.as_str().into(), hash(&a.token).into()])?
778 .first::<ProbeRow>(None)
779 .await?;
780 let Some(mut probe) = probe else {
781 return Ok(Outcome::fail(FailureCode::NotFound, "No such queue entry."));
782 };
783 if probe.state != "testing" {
784 return Ok(Outcome::fail(FailureCode::Conflict, "This state is no longer being tested."));
785 }
786 if a.conflict_with.is_some() || !a.conflicts.is_empty() {
787 self.probe_result(&probe, ProbeResult::Broke, None).await?;
788 return Ok(Outcome::Ok(QueueState::Failed));
789 }
790 let built = a.error.is_none() && a.results.iter().all(|result| result.passed);
791 let Some(commit) = a.combined_commit.as_deref().filter(|_| built) else {
792 self.probe_result(&probe, ProbeResult::Stopped, a.error.as_deref()).await?;
793 return Ok(Outcome::Ok(QueueState::Failed));
794 };
795 let Some(row) = self.entry_by_id(&probe.entry_id).await? else {
796 return Ok(Outcome::Ok(QueueState::Removed));
797 };
798 let base = row
799 .diagnosis
800 .as_deref()
801 .and_then(|text| serde_json::from_str::<Diagnosis>(text).ok())
802 .map(|diagnosis| diagnosis.base_commit);
803 let accepted = self.db
804 .prepare(REPORT_PROBE)
805 .bind(&[commit.into(), probe.id.as_str().into(), hash(&a.token).into()])?
806 .first::<ValueRow>(None)
807 .await?;
808 if accepted.is_none() {
809 return Ok(Outcome::fail(FailureCode::Conflict, "This probe attempt is no longer being tested."));
810 }
811 probe.token_hash = None;
812 probe.combined_commit = Some(commit.to_owned());
813 let runs = self
814 .run_merge_group(&probe.repo_id, &probe.id, &probe.branch(), base.as_deref(), probe.number, &probe.subset(), commit)
815 .await?;
816 if runs == 0 {
817 self.probe_result(&probe, ProbeResult::Stopped, Some("No workflow runs on merge_group events.")).await?;
818 return Ok(Outcome::Ok(QueueState::Failed));
819 }
820 self.db.prepare(OPEN_PROBE)
821 .bind(&[runs.into(), probe.id.as_str().into(), commit.into()])?.run().await?;
822 // A run may have finished while dispatching was fenced.
823 self.probe_finished(&probe.repo_id, commit).await?;
824 Ok(Outcome::Ok(QueueState::Testing))
825 }
826
827 /// The `merge_group` workflows on a smaller combination finished.
828 /// Whether `commit` was one.
829 pub(crate) async fn probe_finished(&self, repo_id: &str, commit: &str) -> Result<bool> {
830 let probe = self
831 .db
832 .prepare(
833 "SELECT * FROM queue_probes
834 WHERE repo_id = ? AND combined_commit = ? AND state = 'testing' AND token_hash IS NULL AND dispatching = 0",
835 )
836 .bind(&[repo_id.into(), commit.into()])?
837 .first::<ProbeRow>(None)
838 .await?;
839 let Some(probe) = probe else { return Ok(false) };
840 // Judged on its own runs only: the same commit may have been built
841 // before, by the queue or another probe.
842 let attempt = crate::statuses::Attempt {
843 sha: commit,
844 head_ref: format!("refs/heads/{}", probe.branch()),
845 since: probe.tested_at.as_deref(),
846 runs: probe.workflow_runs,
847 };
848 let facts = self.attempt_facts(repo_id, &attempt).await?;
849 if !facts.pending.is_empty() {
850 return Ok(false);
851 }
852 let result = if facts.passed() { ProbeResult::Passed } else { ProbeResult::Broke };
853 self.probe_result(&probe, result, None).await?;
854 Ok(true)
855 }
856
857 /// Narrows the suspects by how a smaller combination came out, and
858 /// builds the next one or reports.
859 async fn probe_result(&self, probe: &ProbeRow, result: ProbeResult, error: Option<&str>) -> Result<()> {
860 let state = match result {
861 ProbeResult::Broke => "failed",
862 ProbeResult::Passed => "passed",
863 ProbeResult::Stopped => "stopped",
864 };
865 let claimed = self
866 .db
867 .prepare(FINISH_PROBE)
868 .bind(&[
869 state.into(),
870 error.map_or(JsValue::NULL, JsValue::from),
871 rfc3339(now_ms()).into(),
872 probe.id.as_str().into(),
873 probe.state.as_str().into(),
874 probe.token_hash.as_deref().map_or(JsValue::NULL, JsValue::from),
875 probe.combined_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
876 ])?
877 .first::<ValueRow>(None)
878 .await?;
879 if claimed.is_none() {
880 return Ok(());
881 }
882 self.drop_branch_named(&probe.repo_id, &probe.branch()).await;
883 let Some(row) = self.entry_by_id(&probe.entry_id).await? else {
884 return Ok(());
885 };
886 let Some(before) = row.diagnosis.clone() else { return Ok(()) };
887 let Ok(mut diagnosis) = serde_json::from_str::<Diagnosis>(&before) else {
888 return Ok(());
889 };
890 if diagnosis.done {
891 return Ok(());
892 }
893 match result {
894 ProbeResult::Stopped => {
895 // Nothing more can be learned: report what is known.
896 diagnosis.done = true;
897 if !self.keep_diagnosis(&row.id, Some(&before), &diagnosis).await? {
898 return Ok(());
899 }
900 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
901 return Ok(());
902 };
903 self.integration_failed(&row, &pull, &diagnosis.failure, narrowed(&diagnosis), &diagnosis.suspects, &[])
904 .await
905 }
906 ProbeResult::Broke | ProbeResult::Passed => {
907 diagnosis.observed(&probe.subset(), result == ProbeResult::Broke);
908 self.go_on(&row, Some(&before), diagnosis).await
909 }
910 }
911 }
912
913 // --- Telling the work -------------------------------------------------------
914
915 /// Tells the pull request, its issue, its session and anyone following
916 /// that its combined state failed, and who it failed with. `landed`:
917 /// with nothing ahead, the pull requests merged into the default branch
918 /// since it branched.
919 pub(crate) async fn integration_failed(
920 &self,
921 row: &EntryRow,
922 pull: &Pull,
923 failure: &Failure,
924 bisected: bool,
925 suspects: &[u32],
926 landed: &[u32],
927 ) -> Result<()> {
928 let ahead = row.ahead();
929 let mut combination = ahead.clone();
930 combination.push(pull.number);
931 let g1t = ("g1t", "g1t");
932 let note = pull_note(&ahead, suspects, bisected, landed, &failure.check);
933 self.note(&row.repo_id, pull.number, g1t, &note).await?;
934 let summary = summary(pull.number, suspects, landed, &failure.check);
935 if let Some(issue) = pull.issue {
936 self.note(&row.repo_id, issue, g1t, &format!("noted: {summary}")).await?;
937 }
938 self.session_note(pull, &format!("g1t {note}")).await?;
939 self.keep_decision_quietly(
940 &row.repo_id,
941 NewDecision {
942 issue: pull.issue,
943 pull: Some(pull.number),
944 session_id: None,
945 actor: None,
946 kind: DecisionKind::IntegrationFailure,
947 confidence: DecisionConfidence::Tested,
948 summary,
949 evidence: json!({
950 "commit": failure.combined_commit,
951 "check": failure.check,
952 "combination": combination,
953 "other_pulls": ahead,
954 "suspects": suspects,
955 "bisected": bisected,
956 "landed": landed,
957 }),
958 },
959 )
960 .await;
961 // Named by the attempt and the commit it tested, so a result told
962 // twice reaches each session once.
963 let key = format!("{}:{}:{}", row.id, row.attempt_id.as_deref().unwrap_or(""), failure.combined_commit.as_deref().unwrap_or(""));
964 let data = serde_json::to_value(IntegrationFailed {
965 repo_id: row.repo_id.clone(),
966 pull_id: pull.id.clone(),
967 number: pull.number,
968 pull: pull.number,
969 issue: pull.issue,
970 combination,
971 check: failure.check.clone(),
972 bisected,
973 suspects: suspects.to_vec(),
974 other_pulls: ahead,
975 landed: landed.to_vec(),
976 commit: failure.combined_commit.clone(),
977 })?;
978 self.publish_keyed("integration.failed", &row.repo_id, None, data, &key).await
979 }
980
981 /// Tells anyone following that an entry's combined state passed.
982 pub(crate) async fn integration_passed(&self, row: &EntryRow, commit: Option<&str>) -> Result<()> {
983 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
984 return Ok(());
985 };
986 let ahead = row.ahead();
987 let mut combination = ahead.clone();
988 combination.push(pull.number);
989 let summary = if ahead.is_empty() {
990 format!("#{} passed in the merge queue on the default branch as it is.", pull.number)
991 } else {
992 format!("#{} passed in the merge queue combined with {}.", pull.number, numbers(&ahead))
993 };
994 self.keep_decision_quietly(
995 &row.repo_id,
996 NewDecision {
997 issue: pull.issue,
998 pull: Some(pull.number),
999 session_id: None,
1000 actor: None,
1001 kind: DecisionKind::IntegrationPassed,
1002 confidence: DecisionConfidence::Tested,
1003 summary,
1004 evidence: json!({ "commit": commit, "combination": combination, "other_pulls": ahead }),
1005 },
1006 )
1007 .await;
1008 let key = format!("{}:{}:{}", row.id, row.attempt_id.as_deref().unwrap_or(""), commit.unwrap_or(""));
1009 let data = serde_json::to_value(IntegrationPassed {
1010 repo_id: row.repo_id.clone(),
1011 pull_id: pull.id.clone(),
1012 number: pull.number,
1013 pull: pull.number,
1014 issue: pull.issue,
1015 combination,
1016 commit: commit.map(str::to_owned),
1017 })?;
1018 self.publish_keyed("integration.passed", &row.repo_id, None, data, &key).await
1019 }
1020
1021 /// Appends a note by g1t to a pull request's session, and to the
1022 /// harness sessions working on it or its issue (coordination/).
1023 pub(crate) async fn session_note(&self, pull: &Pull, text: &str) -> Result<()> {
1024 self.note_sessions(pull, text).await;
1025 let now = rfc3339(now_ms());
1026 self.db
1027 .prepare(
1028 "INSERT INTO session_entries (pull_id, seq, kind, text, tool, \"commit\", at)
1029 SELECT ?, COALESCE(MAX(seq), 0) + 1, 'note', ?, NULL, ?, ?
1030 FROM session_entries WHERE pull_id = ?",
1031 )
1032 .bind(&[
1033 pull.id.as_str().into(),
1034 text.into(),
1035 pull.head_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
1036 now.as_str().into(),
1037 pull.id.as_str().into(),
1038 ])?
1039 .run()
1040 .await?;
1041 self.publish_as(
1042 "session.appended",
1043 &pull.repo_id,
1044 None,
1045 SessionAppended {
1046 pull_id: pull.id.clone(),
1047 repo_id: pull.repo_id.clone(),
1048 number: pull.number,
1049 count: 1,
1050 },
1051 )
1052 .await
1053 }
1054
1055 // --- Rejoining ---------------------------------------------------------------
1056
1057 /// Puts a pull request the queue took out back in, once a new commit's
1058 /// checks pass: merged again as whoever queued it, so the branch's
1059 /// rules are checked as they are now. A refusal is noted.
1060 pub(crate) async fn requeue_if_fixed(&self, pull_id: &str, sha: &str) -> Result<()> {
1061 let row = self
1062 .db
1063 .prepare(
1064 "SELECT * FROM queue_entries WHERE pull_id = ? AND requeue = 1 AND state = 'failed'
1065 ORDER BY created_at DESC LIMIT 1",
1066 )
1067 .bind(&[pull_id.into()])?
1068 .first::<EntryRow>(None)
1069 .await?;
1070 let Some(row) = row else { return Ok(()) };
1071 // The commit that failed: nothing has been fixed yet.
1072 if row.head_commit.as_deref() == Some(sha) {
1073 return Ok(());
1074 }
1075 let claimed = self
1076 .db
1077 .prepare("UPDATE queue_entries SET requeue = 0 WHERE id = ? AND requeue = 1 RETURNING id AS value")
1078 .bind(&[row.id.as_str().into()])?
1079 .first::<ValueRow>(None)
1080 .await?;
1081 let (Some(_), Some(actor), Some(pull)) = (claimed, row.actor(), self.pull_by_id(pull_id).await?) else {
1082 return Ok(());
1083 };
1084 if pull.status != PullStatus::Open || pull.head_commit.as_deref() != Some(sha) {
1085 return Ok(());
1086 }
1087 let path: Option<RepoPath> = g1t_kit::call(
1088 &self.repos,
1089 "path_by_id",
1090 &PathByIdArgs { id: pull.repo_id.clone() },
1091 )
1092 .await?;
1093 let Some(repo) = path else { return Ok(()) };
1094 let merged = self
1095 .merge_pull(PullActionArgs {
1096 actor,
1097 repo,
1098 number: pull.number,
1099 summary: String::new(),
1100 keep_issue_open: row.keep_issue_open != 0,
1101 ignore_checks: false,
1102 bypass_rules: false,
1103 merge_method: None,
1104 commit_title: None,
1105 commit_message: None,
1106 expected_head: None,
1107 delete_branch: None,
1108 update_method: None,
1109 })
1110 .await?;
1111 let short = &sha[..sha.len().min(12)];
1112 let text = match merged {
1113 Outcome::Ok(_) => format!("put this back in the merge queue: its checks pass on {short}, pushed after it was taken out"),
1114 Outcome::Fail(failure) => format!(
1115 "could not put this back in the merge queue though its checks pass on {short}: {} Merge it again when that is dealt with.",
1116 failure.message
1117 ),
1118 };
1119 self.note(&pull.repo_id, pull.number, ("g1t", "g1t"), &text).await
1120 }
1121
1122 async fn entry_by_id(&self, id: &str) -> Result<Option<EntryRow>> {
1123 self.db
1124 .prepare("SELECT * FROM queue_entries WHERE id = ?")
1125 .bind(&[id.into()])?
1126 .first::<EntryRow>(None)
1127 .await
1128 }
1129}
1130
1131/// Whether the suspects were narrowed by any build.
1132fn narrowed(diagnosis: &Diagnosis) -> bool {
1133 let ahead = diagnosis.stack.len().saturating_sub(1);
1134 diagnosis.suspects.len() < ahead
1135}
1136
1137#[cfg(test)]
1138mod tests {
1139 use super::*;
1140
1141 fn failure(kind: FailureKind) -> Failure {
1142 Failure {
1143 kind,
1144 check: check("CI / merge_group", "failure", Some("https://g1t.sh/a/b/actions/runs/1".into())),
1145 error: None,
1146 conflict_with: None,
1147 conflicts: Vec::new(),
1148 results: Vec::new(),
1149 combined_commit: Some("c0ffee".into()),
1150 }
1151 }
1152
1153 #[test]
1154 fn a_failure_waits_for_the_entries_ahead_of_it() {
1155 use QueueState::*;
1156 let active = [(11, Passed), (12, Testing), (14, Testing)];
1157 assert!(!ahead_settled(&[11, 12], &active), "#12 is still being tested");
1158 assert!(ahead_settled(&[11], &active));
1159 // #10 landed and left the queue.
1160 assert!(ahead_settled(&[10, 11], &active));
1161 assert!(ahead_settled(&[], &active));
1162 }
1163
1164 #[test]
1165 fn who_answers_for_a_failure() {
1166 let checks = failure(FailureKind::Checks);
1167 // Tested alone on the default branch, failing there though it
1168 // passes on its own: the default branch moved under it.
1169 assert_eq!(plan(&[], &checks, true), Plan::BaseMoved);
1170 // Alone, and failing on its own too: it is the cause.
1171 assert_eq!(plan(&[], &checks, false), Plan::Blame(vec![]));
1172 // Alone, but not a check failure: nothing to look up.
1173 for kind in [FailureKind::Conflict, FailureKind::Missing, FailureKind::Build] {
1174 assert_eq!(plan(&[], &failure(kind), true), Plan::Blame(vec![]), "{kind:?}");
1175 }
1176 // Its own checks fail too: still the cause.
1177 assert_eq!(plan(&[11, 12], &checks, false), Plan::Blame(vec![]));
1178 // One entry ahead: that one, without building anything.
1179 assert_eq!(plan(&[11], &checks, true), Plan::Blame(vec![11]));
1180 // More: narrow them down.
1181 assert_eq!(plan(&[11, 12, 13], &checks, true), Plan::Bisect);
1182 // A conflict names the entry it collided with.
1183 let conflict = Failure { conflict_with: Some(12), ..failure(FailureKind::Conflict) };
1184 assert_eq!(plan(&[11, 12], &conflict, true), Plan::Blame(vec![12]));
1185 // A required check that never reports is not about combining.
1186 assert_eq!(plan(&[11, 12], &failure(FailureKind::Missing), true), Plan::Blame(vec![]));
1187 assert_eq!(plan(&[11, 12], &failure(FailureKind::Build), true), Plan::Blame(vec![]));
1188 }
1189
1190 /// Plays a bisection through against `breaks`, the suspects the entry
1191 /// really fails with: how many builds it took and what was left.
1192 fn bisect(ahead: &[u32], breaks: u32) -> (u32, Vec<u32>) {
1193 let mut diagnosis = diagnosis(ahead);
1194 while let Some(tested) = diagnosis.next_subset() {
1195 diagnosis.observed(&tested, tested.contains(&breaks));
1196 }
1197 (diagnosis.builds, diagnosis.suspects)
1198 }
1199
1200 fn diagnosis(ahead: &[u32]) -> Diagnosis {
1201 Diagnosis { base_commit: "base".into(), stack: vec![], suspects: ahead.to_vec(), builds: 0,
1202 failure: failure(FailureKind::Checks), done: false, complement: None }
1203 }
1204
1205 #[test]
1206 fn passing_halves_do_not_falsely_identify_a_single_cause() {
1207 let mut diagnosis = diagnosis(&[11, 12]);
1208 assert_eq!(diagnosis.next_subset(), Some(vec![11]));
1209 diagnosis.observed(&[11], false);
1210 assert_eq!(diagnosis.suspects, [11, 12]);
1211 assert_eq!(diagnosis.next_subset(), Some(vec![12]));
1212 diagnosis.observed(&[12], false);
1213 assert_eq!(diagnosis.next_subset(), None);
1214 assert_eq!(diagnosis.suspects, [11, 12]);
1215 }
1216
1217 #[test]
1218 fn exhausted_budget_keeps_the_last_demonstrably_failing_combination() {
1219 let mut diagnosis = diagnosis(&[11, 12]);
1220 diagnosis.builds = PROBE_BUDGET - 1;
1221 diagnosis.observed(&[11], false);
1222 assert_eq!(diagnosis.next_subset(), None);
1223 assert_eq!(diagnosis.suspects, [11, 12]);
1224 }
1225
1226 #[test]
1227 fn bisection_halves_the_suspects_within_the_budget() {
1228 assert_eq!(next_probe(&[11, 12, 13], 0), Some(vec![11, 12]));
1229 assert_eq!(next_probe(&[11, 12], 0), Some(vec![11]));
1230 assert_eq!(next_probe(&[11], 0), None);
1231 assert_eq!(next_probe(&[11, 12], PROBE_BUDGET), None);
1232 assert_eq!(narrow(&[11, 12, 13], &[11, 12], true), vec![11, 12]);
1233 assert_eq!(narrow(&[11, 12, 13], &[11, 12], false), vec![13]);
1234 // Complementary probes also consume the bounded budget.
1235 assert_eq!(bisect(&[11, 12, 13], 11), (2, vec![11]));
1236 assert_eq!(bisect(&[11, 12, 13], 12), (3, vec![12]));
1237 assert_eq!(bisect(&[11, 12, 13], 13), (2, vec![13]));
1238 // Never exclude the cause merely to fit the budget.
1239 let eight: Vec<u32> = (1..=8).collect();
1240 for breaks in 1..=8 {
1241 let (builds, suspects) = bisect(&eight, breaks);
1242 assert!(builds <= PROBE_BUDGET);
1243 assert!(suspects.contains(&breaks));
1244 }
1245 // More than the budget can narrow to one: what is left is reported.
1246 let twelve: Vec<u32> = (1..=12).collect();
1247 let (builds, left) = bisect(&twelve, 1);
1248 assert_eq!(builds, PROBE_BUDGET);
1249 assert_eq!(left, vec![1, 2]);
1250 }
1251
1252 #[test]
1253 fn the_notes_name_the_other_changes_and_the_check() {
1254 let check = failure(FailureKind::Checks).check;
1255 let note = pull_note(&[11, 12, 13], &[12], true, &[], &check);
1256 assert!(note.starts_with("traced its merge queue failure to how it combines with #12 (narrowed down from #11, #12 and #13"), "{note}");
1257 assert!(note.contains("`CI / merge_group` failed"));
1258 assert!(note.contains("once #12 has landed"));
1259 let own = pull_note(&[], &[], false, &[], &check);
1260 assert!(own.contains("on the default branch as it is now. No other change has been isolated as the cause"), "{own}");
1261 let conflict = pull_note(&[11], &[11], false, &[], &FailedCheck { name: "merge".into(), conclusion: "conflict".into(), url: None });
1262 assert!(conflict.contains("it did not merge cleanly"), "{conflict}");
1263 assert!(!conflict.contains("narrowed"));
1264 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.");
1265 assert_eq!(numbers(&[]), "");
1266 }
1267
1268 #[test]
1269 fn the_notes_name_what_landed_since_it_branched() {
1270 let check = failure(FailureKind::Checks).check;
1271 // One landed: named as the likely cause.
1272 let one = pull_note(&[], &[], false, &[1], &check);
1273 assert_eq!(
1274 one,
1275 "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."
1276 );
1277 assert_eq!(
1278 summary(14, &[], &[1], &check),
1279 "#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."
1280 );
1281 // Several: listed, newest first, as possible causes.
1282 let several = pull_note(&[], &[], false, &[7, 5, 1], &check);
1283 assert!(several.contains("Since this branched, #7, #5 and #1 landed and may be what it conflicts with."), "{several}");
1284 assert!(!several.contains("the change is the cause"));
1285 assert!(summary(14, &[], &[7, 5], &check).ends_with("since it branched, #7 and #5 landed and may be what it conflicts with."));
1286 // No attribution evidence is not proof that the change itself caused it.
1287 assert_eq!(summary(14, &[], &[], &check), "#14 failed in the merge queue: `CI / merge_group` failed. No other change has been isolated as the cause.");
1288 }
1289
1290 fn commit(hash: &str, at: &str) -> Commit {
1291 Commit {
1292 hash: hash.into(),
1293 tree_hash: format!("t{hash}"),
1294 message: String::new(),
1295 author: g1t_contracts::repos::Signature { name: "a".into(), email: "a@example.com".into() },
1296 parents: Vec::new(),
1297 authored_at: at.into(),
1298 }
1299 }
1300
1301 fn merged(number: u32, commit: &str, at: &str) -> MergedRow {
1302 MergedRow { number, head_commit: Some(commit.into()), merged_at: Some(at.into()) }
1303 }
1304
1305 #[test]
1306 fn what_landed_is_what_its_history_does_not_hold() {
1307 // #3 and #2 landed after it branched from #1's commit.
1308 let pulls = [
1309 merged(3, "c3", "2026-10-12T12:00:00.000Z"),
1310 merged(2, "c2", "2026-10-12T11:00:00.000Z"),
1311 merged(1, "c1", "2026-10-12T10:00:00.000Z"),
1312 ];
1313 let history = [commit("mine", "2026-10-12T10:30:00Z"), commit("c1", "2026-10-12T10:00:00Z"), commit("c0", "2026-10-11T00:00:00Z")];
1314 assert_eq!(landed_since(&pulls, &history, None), vec![3, 2]);
1315 // Brought up to date with #2 since: only #3 is new to it.
1316 let updated = [commit("merge", "2026-10-12T11:30:00Z"), commit("c2", "2026-10-12T11:00:00Z"), commit("c1", "2026-10-12T10:00:00Z")];
1317 assert_eq!(landed_since(&pulls, &updated, None), vec![3]);
1318 // A history cut short does not count what is older than it.
1319 let short = [commit("mine", "2026-10-12T11:30:00Z"), commit("x", "2026-10-12T10:30:00Z")];
1320 assert_eq!(landed_since(&pulls, &short, Some("2026-10-12T10:30:00Z")), vec![3, 2]);
1321 // A merged pull request with no recorded commit is not counted.
1322 let unknown = [MergedRow { number: 9, head_commit: None, merged_at: None }];
1323 assert!(landed_since(&unknown, &history, None).is_empty());
1324 // At most MAX_LANDED, newest first.
1325 let many: Vec<MergedRow> = (1..=15).rev().map(|n| merged(n, &format!("m{n}"), "2026-10-12T12:00:00.000Z")).collect();
1326 let found = landed_since(&many, &history, None);
1327 assert_eq!(found.len(), MAX_LANDED);
1328 assert_eq!(found.first(), Some(&15));
1329 }
1330
1331 #[test]
1332 fn failures_are_read_from_what_was_reported() {
1333 let conflict = ReportQueueArgs {
1334 entry_id: "qen_1".into(),
1335 token: String::new(),
1336 combined_commit: None,
1337 results: Vec::new(),
1338 error: Some("conflict".into()),
1339 conflict_with: Some(12),
1340 conflicts: vec!["src/a.rs".into()],
1341 };
1342 let read = Failure::of_report(&conflict, None);
1343 assert_eq!(read.kind, FailureKind::Conflict);
1344 assert_eq!(read.check.name, "merge");
1345 let back = read.report("qen_1");
1346 assert_eq!((back.conflict_with, back.conflicts.len()), (Some(12), 1));
1347 let missing = Failure::of_report(&ReportQueueArgs { conflict_with: None, conflicts: Vec::new(), ..conflict }, Some(&["CI".to_owned()]));
1348 assert_eq!((missing.kind, missing.check.conclusion.as_str()), (FailureKind::Missing, "missing"));
1349 let workflows = Failure::of_workflows("abc", &["CI / merge_group".into()], &[], &[], "the workflow failed".into(), |name| Some(format!("https://x/{name}")));
1350 assert_eq!(workflows.kind, FailureKind::Checks);
1351 assert_eq!(workflows.check.url.as_deref(), Some("https://x/CI / merge_group"));
1352 let never = Failure::of_workflows("abc", &[], &[], &["CI".into()], "missing".into(), |_| None);
1353 assert_eq!(never.kind, FailureKind::Missing);
1354 // A held failure survives being kept as JSON.
1355 let kept: Failure = serde_json::from_str(&serde_json::to_string(&workflows).unwrap()).unwrap();
1356 assert_eq!(kept.check, workflows.check);
1357 }
1358}