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,078 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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

This file's history is long; its oldest lines are credited to the oldest commit read.