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,372 linesCodeBlameRaw
1//! The merge queue: pull requests are tested together with the ones ahead
2//! of them, and only a combination that passed reaches the default branch.
3//!
4//! A batch of up to the merge queue rule's `max_entries_to_build` entries (4 unless
5//! the default branch's rules say otherwise) is tested at once, speculatively: one
6//! sandbox per entry builds the default branch with that entry and every
7//! entry ahead of it merged in, and pushes the result to
8//! `g1t-queue/<entry>-<attempt>`. The repository's `merge_group` workflows then run
9//! on that commit, and the default branch's required checks must pass
10//! there. Entries land in order, each by moving the default branch to its
11//! tested state, once everything ahead has landed. One that fails leaves
12//! the queue with a failed check run; the entries behind it are tested
13//! again without it. Which entry answers for a failure, which entries
14//! ahead it breaks with, who is told and how it rejoins: integration.rs.
15
16use futures_util::future::try_join_all;
17use g1t_contracts::events::{ChecksEvent, QueueChanged};
18use g1t_contracts::repos::{
19 DeleteBranchArgs, GetByIdArgs, HeadArgs, LandArgs, Landed, Repo, RepoPath,
20};
21use g1t_contracts::time::rfc3339;
22use g1t_contracts::work::*;
23use g1t_contracts::{FailureCode, Outcome, User, new_id};
24use g1t_kit::now_ms;
25use serde::Deserialize;
26use worker::Result;
27use worker::wasm_bindgen::JsValue;
28
29use crate::Work;
30use crate::checks::{hash, new_token};
31
32/// How many entries that left the queue are shown.
33const RECENT: u32 = 20;
34/// Where tested states are pushed, in the repository itself.
35pub(crate) const BRANCH_PREFIX: &str = "g1t-queue/";
36
37// Each transition must still own the attempt when it writes: service
38// calls between reading and writing allow removal, retries and callbacks
39// from another request to interleave.
40const CLAIM_BUILD: &str = "UPDATE queue_entries
41 SET state = 'testing', attempt_id = ?1, token_hash = ?1, head_commit = ?2, base_commit = ?3,
42 ahead = ?4, combined_commit = NULL, results = NULL, error = NULL,
43 tested_at = ?5, dispatching = 0, tested_stack = ?7, workflow_runs = 0
44 WHERE id = ?6 AND state = 'waiting' RETURNING id AS value";
45
46const ENQUEUE: &str = "INSERT INTO queue_entries
47 (id, repo_id, pull_id, number, state, enqueued_by, keep_issue_open, created_at, bypass_rules)
48 SELECT ?1, ?2, ?3, ?4, 'waiting', ?5, ?6, ?7, ?8
49 WHERE NOT EXISTS (SELECT 1 FROM queue_entries WHERE pull_id = ?3 AND state IN ('waiting', 'testing', 'passed'))
50 RETURNING id AS value";
51
52const RETEST: &str = "UPDATE queue_entries
53 SET state = 'waiting', token_hash = NULL, combined_commit = NULL,
54 results = NULL, error = NULL, ahead = NULL, held = NULL, dispatching = 0, workflow_runs = 0
55 WHERE id = ?1 AND state IN ('testing', 'passed')
56 AND attempt_id IS ?2 AND landing_commit IS NULL";
57
58const BEGIN_LANDING: &str = "UPDATE queue_entries SET landing_commit = combined_commit
59 WHERE id = ?1 AND state = 'passed' AND combined_commit = ?2 AND attempt_id IS ?3
60 AND (landing_commit = combined_commit OR EXISTS
61 (SELECT 1 FROM pulls WHERE pulls.id = queue_entries.pull_id AND pulls.status = 'open'))
62 AND NOT EXISTS (SELECT 1 FROM json_each(COALESCE(queue_entries.ahead, '[]')) dependency
63 WHERE NOT EXISTS (SELECT 1 FROM queue_entries landed
64 WHERE landed.repo_id = queue_entries.repo_id AND landed.number = dependency.value AND landed.state = 'landed'))
65 AND NOT EXISTS (SELECT 1 FROM queue_entries earlier
66 WHERE earlier.repo_id = queue_entries.repo_id AND earlier.state IN ('waiting', 'testing', 'passed')
67 AND (earlier.created_at < queue_entries.created_at
68 OR (earlier.created_at = queue_entries.created_at AND earlier.id < queue_entries.id)))
69 RETURNING id AS value";
70
71// The attempt was reset: no later claim or publication can use its branch.
72const RETESTED: &str = "SELECT id AS value FROM queue_entries
73 WHERE id = ?1 AND state = 'waiting' AND attempt_id IS ?2 AND landing_commit IS NULL";
74
75const PUBLISHING: &str = "SELECT id AS value FROM queue_entries
76 WHERE pull_id = ?1 AND state = 'passed' AND landing_commit IS NOT NULL LIMIT 1";
77
78const STILL_LANDING: &str = "SELECT id AS value FROM queue_entries
79 WHERE id = ?1 AND state = 'passed' AND landing_commit = ?2";
80
81// Only the request that still holds the intent may give it up: one that
82// lost Git's race to a concurrent publisher must not undo its intent.
83const ABANDON_LANDING: &str = "UPDATE queue_entries SET landing_commit = NULL
84 WHERE id = ?1 AND state = 'passed' AND landing_commit = ?2 RETURNING id AS value";
85
86const FINISH_LANDING: &str = "UPDATE queue_entries SET state = 'landed', finished_at = ?1
87 WHERE id = ?2 AND state = 'passed' AND landing_commit = combined_commit";
88
89const LEAVE: &str =
90 "UPDATE queue_entries SET state = ?1, error = COALESCE(?2, error), finished_at = ?3
91 WHERE id = ?4 AND state IN ('waiting', 'testing', 'passed') AND landing_commit IS NULL
92 RETURNING id AS value";
93
94const ACCEPT_BUILD: &str =
95 "UPDATE queue_entries SET combined_commit = ?1, token_hash = NULL, dispatching = 1
96 WHERE id = ?2 AND state = 'testing' AND token_hash = ?3 RETURNING id AS value";
97
98const REPORT_BUILD: &str = "UPDATE queue_entries
99 SET state = ?, combined_commit = ?, results = ?, error = ?, token_hash = NULL, held = ?, workflow_runs = ?, dispatching = 0,
100 finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
101 WHERE id = ? AND state = 'testing' AND token_hash IS NULL AND held IS NULL
102 AND attempt_id IS ? AND combined_commit IS ? RETURNING id AS value";
103
104const FINISH_CHECKS: &str = "UPDATE queue_entries
105 SET state = ?, held = ?, finished_at = CASE WHEN ? = 'failed' THEN ? ELSE NULL END
106 WHERE id = ? AND state = 'testing' AND held IS NULL
107 AND combined_commit = ? AND token_hash IS NULL AND dispatching = 0 AND attempt_id IS ?
108 RETURNING id AS value";
109
110#[cfg(test)]
111#[path = "queue_tests.rs"]
112mod tests;
113
114#[derive(Clone, Deserialize)]
115pub(crate) struct EntryRow {
116 pub(crate) id: String,
117 pub(crate) repo_id: String,
118 pub(crate) pull_id: String,
119 pub(crate) number: u32,
120 state: String,
121 enqueued_by: String,
122 pub(crate) keep_issue_open: u32,
123 pub(crate) head_commit: Option<String>,
124 pub(crate) base_commit: Option<String>,
125 ahead: Option<String>,
126 combined_commit: Option<String>,
127 results: Option<String>,
128 error: Option<String>,
129 pub(crate) tested_at: Option<String>,
130 #[serde(default)]
131 pub(crate) attempt_id: Option<String>,
132 #[serde(default)]
133 landing_commit: Option<String>,
134 /// Whoever queued it asked to bypass the rules they may bypass.
135 #[serde(default)]
136 bypass_rules: u32,
137 created_at: String,
138 finished_at: Option<String>,
139 /// What failed its combined state while entries ahead were still
140 /// being tested (integration.rs), as JSON.
141 #[serde(default)]
142 pub(crate) held: Option<String>,
143 /// What the queue worked out about its failure (integration.rs).
144 #[serde(default)]
145 pub(crate) diagnosis: Option<String>,
146 /// The stack this attempt built, as JSON `[QueueStackItem]`.
147 #[serde(default)]
148 tested_stack: Option<String>,
149 /// How many merge_group runs this attempt's dispatch registered.
150 #[serde(default)]
151 workflow_runs: u32,
152}
153
154impl EntryRow {
155 pub(crate) fn state(&self) -> QueueState {
156 match self.state.as_str() {
157 "testing" => QueueState::Testing,
158 "passed" => QueueState::Passed,
159 "failed" => QueueState::Failed,
160 "landed" => QueueState::Landed,
161 "removed" => QueueState::Removed,
162 _ => QueueState::Waiting,
163 }
164 }
165
166 pub(crate) fn actor(&self) -> Option<User> {
167 serde_json::from_str(&self.enqueued_by).ok()
168 }
169
170 pub(crate) fn ahead(&self) -> Vec<u32> {
171 self.ahead
172 .as_deref()
173 .and_then(|ahead| serde_json::from_str(ahead).ok())
174 .unwrap_or_default()
175 }
176
177 fn branch(&self) -> String {
178 attempt_branch(&self.id, self.attempt_id.as_deref())
179 }
180
181 /// Exactly what this attempt built, as recorded when it was claimed.
182 pub(crate) fn tested_stack(&self) -> Option<Vec<QueueStackItem>> {
183 self.tested_stack.as_deref().and_then(|stack| serde_json::from_str(stack).ok())
184 }
185
186 /// The attempt whose evidence decides this entry, once it has a commit.
187 fn attempt(&self) -> Option<crate::statuses::Attempt<'_>> {
188 Some(crate::statuses::Attempt {
189 sha: self.combined_commit.as_deref()?,
190 head_ref: format!("refs/heads/{}", self.branch()),
191 since: self.tested_at.as_deref(),
192 runs: self.workflow_runs,
193 })
194 }
195}
196
197fn attempt_branch(id: &str, attempt: Option<&str>) -> String {
198 match attempt {
199 Some(attempt) => format!("{BRANCH_PREFIX}{id}-{attempt}"),
200 None => format!("{BRANCH_PREFIX}{id}"),
201 }
202}
203
204/// Why an entry left because what it would build is gone.
205const BRANCH_GONE: &str = "Its branch no longer exists.";
206const PULL_GONE: &str = "Its pull request no longer exists.";
207
208fn gone_note(reason: &str) -> String {
209 let what = reason.trim_end_matches('.');
210 let what = what.strip_prefix("Its ").map_or_else(|| what.to_owned(), |rest| format!("its {rest}"));
211 format!("{what}, so it cannot be tested or landed. Push the branch again and merge to rejoin.")
212}
213
214fn minutes_ago(minutes: u64) -> String {
215 rfc3339(now_ms().saturating_sub(minutes * 60 * 1000))
216}
217
218impl Work {
219 pub(crate) async fn entries(&self, repo_id: &str, active: bool) -> Result<Vec<EntryRow>> {
220 let sql = if active {
221 "SELECT * FROM queue_entries
222 WHERE repo_id = ? AND state IN ('waiting', 'testing', 'passed')
223 ORDER BY created_at, id"
224 } else {
225 "SELECT * FROM queue_entries
226 WHERE repo_id = ? AND state IN ('failed', 'landed', 'removed')
227 ORDER BY finished_at DESC LIMIT ?"
228 };
229 let statement = self.db.prepare(sql);
230 let statement = if active {
231 statement.bind(&[repo_id.into()])?
232 } else {
233 statement.bind(&[repo_id.into(), RECENT.into()])?
234 };
235 statement.all().await?.results::<EntryRow>()
236 }
237
238 /// The entry a pull request has in the queue now, if any.
239 pub(crate) async fn queued_entry(
240 &self,
241 pull_id: &str,
242 ) -> Result<Option<(QueueState, Vec<u32>)>> {
243 let row = match self.prefetched_pull(pull_id) {
244 Some(found) => found.first::<EntryRow>(crate::prefetch::Slot::Queued)?,
245 None => {
246 self.db
247 .prepare(
248 "SELECT * FROM queue_entries
249 WHERE pull_id = ? AND state IN ('waiting', 'testing', 'passed') LIMIT 1",
250 )
251 .bind(&[pull_id.into()])?
252 .first::<EntryRow>(None)
253 .await?
254 }
255 };
256 Ok(row.map(|row| (row.state(), row.ahead())))
257 }
258
259 pub(crate) async fn changed(&self, repo_id: &str) -> Result<()> {
260 self.publish_as(
261 "queue.changed",
262 repo_id,
263 None,
264 QueueChanged {
265 repo_id: repo_id.to_owned(),
266 },
267 )
268 .await
269 }
270
271 /// Puts a pull request that may merge into the queue instead. Merging
272 /// it again while it is queued changes nothing.
273 pub(crate) async fn enqueue(
274 &self,
275 repo: &Repo,
276 pull: &Pull,
277 actor: &User,
278 keep_issue_open: bool,
279 bypass_rules: bool,
280 ) -> Result<Outcome<Pull>> {
281 if self.queued_entry(&pull.id).await?.is_some() {
282 return Ok(Outcome::Ok(pull.clone()));
283 }
284 let now = now_ms();
285 let inserted = self
286 .db
287 .prepare(ENQUEUE)
288 .bind(&[
289 new_id("qen", now).into(),
290 repo.id.as_str().into(),
291 pull.id.as_str().into(),
292 pull.number.into(),
293 serde_json::to_string(actor)?.into(),
294 u32::from(keep_issue_open).into(),
295 rfc3339(now).into(),
296 u32::from(bypass_rules).into(),
297 ])?
298 .first::<crate::rows::ValueRow>(None)
299 .await?;
300 if inserted.is_none() {
301 return Ok(Outcome::Ok(pull.clone()));
302 }
303 let who = (actor.id.as_str(), actor.username.as_str());
304 self.note(&repo.id, pull.number, who, "added this to the merge queue")
305 .await?;
306 self.changed(&repo.id).await?;
307 Ok(Outcome::Ok(pull.clone()))
308 }
309
310 pub(crate) async fn queue(&self, a: QueueArgs) -> Result<Outcome<QueueView>> {
311 let repo = match self.repo(&a.repo, &a.viewer).await? {
312 Outcome::Ok(repo) => repo,
313 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
314 };
315 let enabled = self.default_branch_settings(&repo).await?.merge_queue;
316 let (active, recent) = (
317 self.entries(&repo.id, true).await?,
318 self.entries(&repo.id, false).await?,
319 );
320 let numbers: Vec<u32> = active.iter().chain(&recent).map(|row| row.number).collect();
321 let pulls = try_join_all(numbers.iter().map(|number| self.pull(&repo.id, *number))).await?;
322 let view = |rows: Vec<EntryRow>, pulls: &[Option<Pull>]| -> Vec<QueueEntry> {
323 rows.into_iter()
324 .zip(pulls)
325 .map(|(row, pull)| QueueEntry {
326 state: row.state(),
327 ahead: row.ahead(),
328 enqueued_by: row
329 .actor()
330 .map_or_else(|| "g1t".to_owned(), |actor| actor.username),
331 title: pull.as_ref().map(|p| p.title.clone()).unwrap_or_default(),
332 agent: pull.as_ref().map(|p| p.agent.clone()).unwrap_or_default(),
333 results: row
334 .results
335 .as_deref()
336 .and_then(|results| serde_json::from_str(results).ok())
337 .unwrap_or_default(),
338 id: row.id,
339 number: row.number,
340 base_commit: row.base_commit,
341 combined_commit: row.combined_commit,
342 error: row.error,
343 created_at: row.created_at,
344 finished_at: row.finished_at,
345 })
346 .collect()
347 };
348 let split = active.len();
349 Ok(Outcome::Ok(QueueView {
350 enabled,
351 active: view(active, &pulls[..split]),
352 recent: view(recent, &pulls[split..]),
353 }))
354 }
355
356 /// Takes a pull request out of the queue, by the hand of someone who
357 /// may merge.
358 pub(crate) async fn remove_from_queue(&self, a: PullActionArgs) -> Result<Outcome<Pull>> {
359 let viewer = Some(a.actor.clone());
360 let (repo, pull) = match self.pull_at(&a.repo, a.number, &viewer).await? {
361 Outcome::Ok(found) => found,
362 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
363 };
364 if let Outcome::Fail(failure) = crate::retired::writable(&repo) {
365 return Ok(Outcome::Fail(failure));
366 }
367 if !a.actor.verified {
368 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
369 }
370 if let Outcome::Fail(failure) = crate::allowed(
371 Some(&a.actor),
372 &repo,
373 g1t_contracts::access::Capability::Merge,
374 ) {
375 return Ok(Outcome::Fail(failure));
376 }
377 if self
378 .leave(&repo.id, &pull, QueueState::Removed, None)
379 .await?
380 {
381 let who = (a.actor.id.as_str(), a.actor.username.as_str());
382 self.note(
383 &repo.id,
384 pull.number,
385 who,
386 "removed this from the merge queue",
387 )
388 .await?;
389 self.changed(&repo.id).await?;
390 } else if self.publishing(&pull.id).await? {
391 return Ok(Outcome::fail(
392 FailureCode::Conflict,
393 "This tested commit is already being published. Wait for landing to finish.",
394 ));
395 }
396 Ok(Outcome::Ok(pull))
397 }
398
399 /// Whether a pull request's tested commit is being published: its
400 /// entry holds publication intent, which removal cannot revoke.
401 pub(crate) async fn publishing(&self, pull_id: &str) -> Result<bool> {
402 Ok(self
403 .db
404 .prepare(PUBLISHING)
405 .bind(&[pull_id.into()])?
406 .first::<crate::rows::ValueRow>(None)
407 .await?
408 .is_some())
409 }
410
411 /// Takes a pull request's entry out of the queue, and sends every entry
412 /// whose tested state included it back to waiting. Whether it had one.
413 pub(crate) async fn leave(
414 &self,
415 repo_id: &str,
416 pull: &Pull,
417 state: QueueState,
418 error: Option<&str>,
419 ) -> Result<bool> {
420 let active = self.entries(repo_id, true).await?;
421 let Some(entry) = active.iter().find(|row| row.pull_id == pull.id) else {
422 return Ok(false);
423 };
424 self.leave_entry(&active, entry, state, error).await
425 }
426
427 /// Takes `entry`, one of `active`, out of the queue, as `leave` does.
428 async fn leave_entry(
429 &self,
430 active: &[EntryRow],
431 entry: &EntryRow,
432 state: QueueState,
433 error: Option<&str>,
434 ) -> Result<bool> {
435 let now = rfc3339(now_ms());
436 let removed = self
437 .db
438 .prepare(LEAVE)
439 .bind(&[
440 state.as_str().into(),
441 error.map_or(JsValue::NULL, JsValue::from),
442 now.as_str().into(),
443 entry.id.as_str().into(),
444 ])?
445 .first::<crate::rows::ValueRow>(None)
446 .await?;
447 if removed.is_none() {
448 return Ok(false);
449 }
450 self.drop_branch(entry).await;
451 let behind: Vec<&EntryRow> = active
452 .iter()
453 .filter(|row| row.state() != QueueState::Waiting && row.ahead().contains(&entry.number))
454 .collect();
455 self.retest(&behind).await?;
456 Ok(true)
457 }
458
459 /// An entry whose pull request or branch no longer exists cannot be
460 /// built or landed: it leaves the queue, saying why, rather than wait
461 /// at the front for ever and hold up every entry behind it.
462 async fn leave_gone(&self, entry: &EntryRow, reason: &str) -> Result<()> {
463 let active = self.entries(&entry.repo_id, true).await?;
464 let Some(entry) = active.iter().find(|row| row.id == entry.id) else {
465 return Ok(());
466 };
467 if !self.leave_entry(&active, entry, QueueState::Removed, Some(reason)).await? {
468 return Ok(());
469 }
470 // A pull request that is gone has no conversation to say so in.
471 if reason != PULL_GONE {
472 self.note(
473 &entry.repo_id,
474 entry.number,
475 ("g1t", "g1t"),
476 &format!("removed this from the merge queue: {}", gone_note(reason)),
477 )
478 .await?;
479 }
480 self.changed(&entry.repo_id).await
481 }
482
483 /// Removes an entry's tested state from the repository once it has
484 /// left the queue, as GitHub does with its queue's branches. A branch
485 /// left behind is untidy, not wrong, so a failure only logs.
486 async fn drop_branch(&self, row: &EntryRow) {
487 self.drop_branch_named(&row.repo_id, &row.branch()).await;
488 }
489
490 /// Removes one of the queue's branches.
491 pub(crate) async fn drop_branch_named(&self, repo_id: &str, branch: &str) {
492 let deleted: Result<Outcome<bool>> = g1t_kit::call(
493 &self.repos,
494 "delete_branch",
495 &DeleteBranchArgs {
496 repo_id: repo_id.to_owned(),
497 branch: branch.to_owned(),
498 head: None,
499 },
500 )
501 .await;
502 match deleted {
503 Ok(Outcome::Ok(_)) => {}
504 Ok(Outcome::Fail(failure)) => worker::console_warn!("{branch}: {}", failure.message),
505 Err(error) => worker::console_warn!("{branch}: {error}"),
506 }
507 }
508
509 /// Sends entries back to waiting, to be tested again.
510 async fn retest(&self, rows: &[&EntryRow]) -> Result<()> {
511 for row in rows {
512 let attempt = row
513 .attempt_id
514 .as_deref()
515 .map_or(JsValue::NULL, JsValue::from);
516 self.db
517 .prepare(RETEST)
518 .bind(&[row.id.as_str().into(), attempt.clone()])?
519 .run()
520 .await?;
521 // Each attempt pushes its own branch; one that was given up is
522 // never landed from, so remove it rather than leave one behind
523 // per retry. Only once the reset took: an attempt that is being
524 // published keeps its branch.
525 if row.attempt_id.is_some()
526 && self
527 .db
528 .prepare(RETESTED)
529 .bind(&[row.id.as_str().into(), attempt])?
530 .first::<crate::rows::ValueRow>(None)
531 .await?
532 .is_some()
533 {
534 self.drop_branch(row).await;
535 }
536 }
537 Ok(())
538 }
539
540 /// The next batch to test, if nothing is being tested now: one job per
541 /// entry, each building the default branch with that entry and every
542 /// entry ahead of it.
543 ///
544 /// Smaller combinations a failure is being narrowed down with
545 /// (integration.rs) are built alongside, whatever the queue is doing.
546 pub(crate) async fn queue_build(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
547 // Also resumes a publication interrupted after the Git update.
548 self.settle(&a.repo_id).await?;
549 let mut jobs = self.probe_jobs(&a.repo_id).await?;
550 jobs.extend(self.queue_batch(a).await?);
551 Ok(jobs)
552 }
553
554 async fn queue_batch(&self, a: QueueBuildArgs) -> Result<Vec<QueueJob>> {
555 let active = self.entries(&a.repo_id, true).await?;
556 // How the default branch's merge queue rule batches (rulesets.rs).
557 let rule = self.queue_rule(&a.repo_id).await?;
558 // A batch that has taken too long is tested again.
559 let stale = minutes_ago(u64::from(rule.check_response_timeout_minutes).max(1));
560 let stuck: Vec<&EntryRow> = active
561 .iter()
562 .filter(|row| {
563 row.state() == QueueState::Testing
564 && row
565 .tested_at
566 .as_deref()
567 .is_none_or(|at| at < stale.as_str())
568 })
569 .collect();
570 if !stuck.is_empty() {
571 self.retest(&stuck).await?;
572 return Box::pin(self.queue_batch(a)).await;
573 }
574 if active.iter().any(|row| row.state() != QueueState::Waiting) {
575 return Ok(Vec::new());
576 }
577 let batch: Vec<EntryRow> = active
578 .into_iter()
579 .take(rule.max_entries_to_build.max(1) as usize)
580 .collect();
581 let Some(first) = batch.first() else {
582 return Ok(Vec::new());
583 };
584 // Too few to start yet, and the oldest has not waited long enough.
585 if batch.len() < rule.min_entries_to_merge as usize
586 && first.created_at.as_str()
587 > minutes_ago(u64::from(rule.min_entries_wait_minutes)).as_str()
588 {
589 return Ok(Vec::new());
590 }
591 let Some(actor) = first.actor() else {
592 return Ok(Vec::new());
593 };
594 let repo: Outcome<Repo> = g1t_kit::call(
595 &self.repos,
596 "get_by_id",
597 &GetByIdArgs {
598 id: a.repo_id.clone(),
599 viewer: Some(actor),
600 },
601 )
602 .await?;
603 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
604 return Ok(Vec::new());
605 };
606 let base: Option<String> = g1t_kit::call(
607 &self.repos,
608 "head",
609 &HeadArgs {
610 repo_id: repo.id.clone(),
611 branch: repo.default_branch.clone(),
612 },
613 )
614 .await?;
615 let Some(base) = base else {
616 return Ok(Vec::new());
617 };
618
619 // Each entry's change, as it is now.
620 let mut items: Vec<(EntryRow, Pull, QueueStackItem)> = Vec::new();
621 for row in batch {
622 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
623 self.leave_gone(&row, PULL_GONE).await?;
624 continue;
625 };
626 if pull.status != PullStatus::Open {
627 self.leave(&repo.id, &pull, QueueState::Removed, Some("It was closed."))
628 .await?;
629 continue;
630 }
631 let head: Option<String> = g1t_kit::call(
632 &self.repos,
633 "head",
634 &HeadArgs {
635 repo_id: pull.source_id().to_owned(),
636 branch: pull
637 .branch
638 .clone()
639 .unwrap_or_else(|| repo.default_branch.clone()),
640 },
641 )
642 .await?;
643 let Some(commit) = head else {
644 self.leave_gone(&row, BRANCH_GONE).await?;
645 continue;
646 };
647 let source = pull.source_path(&RepoPath {
648 namespace: repo.namespace.clone(),
649 name: repo.name.clone(),
650 });
651 let item = QueueStackItem {
652 number: pull.number,
653 title: pull.title.clone(),
654 source,
655 branch: pull
656 .branch
657 .clone()
658 .unwrap_or_else(|| repo.default_branch.clone()),
659 commit,
660 };
661 items.push((row, pull, item));
662 }
663
664 let now = rfc3339(now_ms());
665 let mut jobs = Vec::new();
666 for index in 0..items.len() {
667 let (row, pull, _) = &items[index];
668 let stack: Vec<QueueStackItem> = items[..=index]
669 .iter()
670 .map(|(_, _, item)| item.clone())
671 .collect();
672 let ahead: Vec<u32> = stack[..index].iter().map(|item| item.number).collect();
673 let token = new_token();
674 let claimed = self
675 .db
676 .prepare(CLAIM_BUILD)
677 .bind(&[
678 hash(&token).into(),
679 stack[index].commit.as_str().into(),
680 base.as_str().into(),
681 serde_json::to_string(&ahead)?.into(),
682 now.as_str().into(),
683 row.id.as_str().into(),
684 // Diagnosis rebuilds exactly these commits (integration.rs).
685 serde_json::to_string(&stack)?.into(),
686 ])?
687 .first::<crate::rows::ValueRow>(None)
688 .await?;
689 if claimed.is_none() {
690 continue;
691 }
692 // The sandbox reads the change and pushes the tested state as
693 // the pull request's owner (whoever asked g1t for it, or its
694 // author): a real account, whose token carries its memberships.
695 // (Whoever queued it may be g1t itself.)
696 let actor = pull.owner().clone();
697 let branch = attempt_branch(&row.id, Some(&hash(&token)));
698 jobs.push(QueueJob {
699 entry_id: row.id.clone(),
700 token,
701 repo: RepoPath {
702 namespace: repo.namespace.clone(),
703 name: repo.name.clone(),
704 },
705 default_branch: repo.default_branch.clone(),
706 base_commit: base.clone(),
707 branch,
708 // The state is checked by its merge_group workflows.
709 contract_checks: Vec::new(),
710 stack,
711 checks: Vec::new(),
712 actor,
713 });
714 }
715 Ok(jobs)
716 }
717
718 /// A sandbox's result for one combined state.
719 pub(crate) async fn report_queue(&self, a: ReportQueueArgs) -> Result<Outcome<QueueState>> {
720 // A smaller combination built to narrow down a failure.
721 if a.entry_id.starts_with("qpr_") {
722 return self.report_probe(a).await;
723 }
724 let row = self
725 .db
726 .prepare("SELECT * FROM queue_entries WHERE id = ? AND token_hash = ?")
727 .bind(&[a.entry_id.as_str().into(), hash(&a.token).into()])?
728 .first::<EntryRow>(None)
729 .await?;
730 let Some(row) = row else {
731 return Ok(Outcome::fail(FailureCode::NotFound, "No such queue entry."));
732 };
733 if row.state() != QueueState::Testing {
734 return Ok(Outcome::fail(
735 FailureCode::Conflict,
736 "This state is no longer being tested.",
737 ));
738 }
739 if a.combined_commit.is_none() && a.error.is_none() {
740 return Ok(Outcome::fail(
741 FailureCode::Invalid,
742 "A successful build must name the combined commit it tested.",
743 ));
744 }
745 // Publish the commit before dispatching workflows: a very fast callback
746 // must be able to find it. Only the winning report dispatches work.
747 let accepted = self
748 .db
749 .prepare(ACCEPT_BUILD)
750 .bind(&[
751 a.combined_commit
752 .as_deref()
753 .map_or(JsValue::NULL, JsValue::from),
754 row.id.as_str().into(),
755 hash(&a.token).into(),
756 ])?
757 .first::<crate::rows::ValueRow>(None)
758 .await?;
759 if accepted.is_none() {
760 return Ok(Outcome::fail(
761 FailureCode::Conflict,
762 "This build attempt is no longer being tested.",
763 ));
764 }
765 let built = a.error.is_none() && a.results.iter().all(|result| result.passed);
766 // A state that was built runs the repository's `merge_group`
767 // workflows; it stays in testing until they finish (see `statuses`),
768 // and the branch's required checks must pass on it.
769 let workflows = match (&a.combined_commit, built) {
770 (Some(commit), true) => {
771 let dispatched = self
772 .run_merge_group(
773 &row.repo_id,
774 &row.id,
775 &row.branch(),
776 row.base_commit.as_deref(),
777 row.number,
778 &row.ahead(),
779 commit,
780 )
781 .await;
782 match dispatched {
783 Ok(runs) => runs,
784 Err(error) => {
785 // The token is spent and the fence stays up, so no
786 // report or callback could ever move this attempt
787 // on: build it again now rather than after the
788 // check response timeout.
789 self.retest(&[&row]).await?;
790 return Err(error);
791 }
792 }
793 }
794 _ => 0,
795 };
796 let required = self.settings_by_id(&row.repo_id).await?.required_checks;
797 // Nothing runs on it, so the required checks never would report.
798 let unchecked = (built && workflows == 0 && !required.is_empty()).then(|| {
799 format!(
800 "the required {} {} cannot report on it: no workflow runs on merge_group events. Add merge_group to the on: of the workflows the branch requires",
801 if required.len() == 1 { "check" } else { "checks" },
802 crate::statuses::list(&required)
803 )
804 });
805 let passed = built && unchecked.is_none();
806 let error = a.error.clone().or(unchecked.clone());
807 let report = ReportQueueArgs {
808 error: error.clone(),
809 ..a
810 };
811 let failure = (!passed).then(|| {
812 crate::integration::Failure::of_report(
813 &report,
814 unchecked.is_some().then_some(&required[..]),
815 )
816 });
817 // Failed while an entry ahead of it is still being tested: held
818 // until those ahead are settled, so that it is not blamed for
819 // theirs (integration.rs).
820 let held = match &failure {
821 Some(failure) if self.ahead_unsettled(&row).await? => {
822 Some(serde_json::to_string(failure)?)
823 }
824 _ => None,
825 };
826 let state = if passed && workflows == 0 {
827 QueueState::Passed
828 } else if passed || held.is_some() {
829 QueueState::Testing
830 } else {
831 QueueState::Failed
832 };
833 let moved = self
834 .db
835 .prepare(REPORT_BUILD)
836 .bind(&[
837 state.as_str().into(),
838 report
839 .combined_commit
840 .as_deref()
841 .map_or(JsValue::NULL, JsValue::from),
842 serde_json::to_string(&report.results)?.into(),
843 error.as_deref().map_or(JsValue::NULL, JsValue::from),
844 held.as_deref().map_or(JsValue::NULL, JsValue::from),
845 workflows.into(),
846 state.as_str().into(),
847 rfc3339(now_ms()).into(),
848 row.id.as_str().into(),
849 row.attempt_id
850 .as_deref()
851 .map_or(JsValue::NULL, JsValue::from),
852 report
853 .combined_commit
854 .as_deref()
855 .map_or(JsValue::NULL, JsValue::from),
856 ])?
857 .first::<crate::rows::ValueRow>(None)
858 .await?;
859 if moved.is_none() {
860 return Ok(Outcome::fail(
861 FailureCode::Conflict,
862 "This build attempt is no longer being tested.",
863 ));
864 }
865 if passed && workflows > 0 {
866 // A workflow may have finished while dispatching was fenced. Read
867 // the whole registered group now so that callback is not lost.
868 self.merge_group_finished(&row.repo_id, report.combined_commit.as_deref().unwrap())
869 .await?;
870 }
871 if let Some(failure) = failure.filter(|_| held.is_none()) {
872 self.eject(&row, &report).await?;
873 self.diagnose(&row, failure).await?;
874 }
875 if state == QueueState::Passed {
876 self.integration_passed(&row, report.combined_commit.as_deref())
877 .await?;
878 self.resolve_held(&row.repo_id).await?;
879 }
880 self.settle(&row.repo_id).await?;
881 self.changed(&row.repo_id).await?;
882 Ok(Outcome::Ok(state))
883 }
884
885 /// Asks the actions service to run the repository's `merge_group`
886 /// workflows on a combined state: `entry`'s, pushed to `branch`, with
887 /// `ahead` merged in before `number`. Returns how many runs started.
888 #[allow(clippy::too_many_arguments)]
889 pub(crate) async fn run_merge_group(
890 &self,
891 repo_id: &str,
892 entry: &str,
893 branch: &str,
894 base_commit: Option<&str>,
895 number: u32,
896 ahead: &[u32],
897 commit: &str,
898 ) -> Result<u32> {
899 #[derive(serde::Deserialize)]
900 struct Started {
901 runs: u32,
902 }
903 let started: Outcome<Started> = g1t_kit::call(
904 &self.actions,
905 "merge_group",
906 &serde_json::json!({
907 "repoId": repo_id,
908 "entry": entry,
909 "sha": commit,
910 "headRef": format!("refs/heads/{branch}"),
911 "baseSha": base_commit,
912 "number": number,
913 "ahead": ahead,
914 }),
915 )
916 .await?;
917 Ok(match started {
918 Outcome::Ok(started) => started.runs,
919 Outcome::Fail(failure) => {
920 return Err(worker::Error::RustError(format!(
921 "Merge-group dispatch failed: {}",
922 failure.message
923 )));
924 }
925 })
926 }
927
928 /// Workflows on a combined state finished: it passes and lands in turn
929 /// when they all passed and so did every required check, or fails and
930 /// leaves the queue.
931 pub(crate) async fn merge_group_finished(
932 &self,
933 repo_id: &str,
934 commit: &str,
935 ) -> Result<()> {
936 let row = self
937 .db
938 .prepare(
939 "SELECT * FROM queue_entries
940 WHERE repo_id = ? AND combined_commit = ? AND state = 'testing' AND token_hash IS NULL AND held IS NULL AND dispatching = 0",
941 )
942 .bind(&[repo_id.into(), commit.into()])?
943 .first::<EntryRow>(None)
944 .await?;
945 let Some(row) = row else {
946 // Or a smaller combination built to narrow down a failure.
947 if self.probe_finished(repo_id, commit).await? {
948 self.changed(repo_id).await?;
949 }
950 return Ok(());
951 };
952 // Only this attempt's own runs, and what reported since it was
953 // claimed, decide it: the same commit may have been built before.
954 let Some(attempt) = row.attempt() else { return Ok(()) };
955 let facts = self.attempt_facts(repo_id, &attempt).await?;
956 if !facts.pending.is_empty() {
957 return Ok(());
958 }
959 let missing = facts.expected();
960 let passed = facts.passed();
961 let failure = if passed {
962 None
963 } else {
964 let error = if facts.failed.is_empty() && facts.required_failed().is_empty() {
965 format!(
966 "the required {} {} did not report on it. Add merge_group to the on: of the workflows the branch requires",
967 if missing.len() == 1 {
968 "check"
969 } else {
970 "checks"
971 },
972 crate::statuses::list(&missing)
973 )
974 } else if facts.failed.is_empty() {
975 format!(
976 "the required {} failed on it",
977 crate::statuses::list(&facts.required_failed())
978 )
979 } else {
980 format!(
981 "the workflow {} failed on it",
982 crate::statuses::list(&facts.failed)
983 )
984 };
985 // Where to see the check that failed: its run's page.
986 let statuses = self.statuses(repo_id, Some(commit)).await?;
987 let url = |name: &str| {
988 statuses
989 .iter()
990 .find(|status| status.context == name || check_name(&status.context).0 == name)
991 .and_then(|status| status.target_url.clone())
992 };
993 Some(crate::integration::Failure::of_workflows(
994 commit,
995 &facts.failed,
996 &facts.required_failed(),
997 &missing,
998 error,
999 url,
1000 ))
1001 };
1002 let held = match &failure {
1003 Some(failure) if self.ahead_unsettled(&row).await? => {
1004 Some(serde_json::to_string(failure)?)
1005 }
1006 _ => None,
1007 };
1008 let state = if passed {
1009 "passed"
1010 } else if held.is_some() {
1011 "testing"
1012 } else {
1013 "failed"
1014 };
1015 let moved = self
1016 .db
1017 .prepare(FINISH_CHECKS)
1018 .bind(&[
1019 state.into(),
1020 held.as_deref().map_or(JsValue::NULL, JsValue::from),
1021 state.into(),
1022 rfc3339(now_ms()).into(),
1023 row.id.as_str().into(),
1024 commit.into(),
1025 row.attempt_id.as_deref().map_or(JsValue::NULL, JsValue::from),
1026 ])?
1027 .first::<crate::rows::ValueRow>(None)
1028 .await?;
1029 if moved.is_none() {
1030 return Ok(());
1031 }
1032 if let Some(failure) = failure.filter(|_| held.is_none()) {
1033 self.eject(&row, &failure.report(&row.id)).await?;
1034 self.diagnose(&row, failure).await?;
1035 }
1036 if passed {
1037 self.integration_passed(&row, Some(commit)).await?;
1038 self.resolve_held(repo_id).await?;
1039 }
1040 self.settle(repo_id).await?;
1041 self.changed(repo_id).await?;
1042 Ok(())
1043 }
1044
1045 /// An entry whose combined state failed: the entries tested on top of
1046 /// it are tested again without it, and the failure is recorded as a
1047 /// failed check run of its pull request.
1048 pub(crate) async fn eject(&self, row: &EntryRow, report: &ReportQueueArgs) -> Result<()> {
1049 self.drop_branch(row).await;
1050 let active = self.entries(&row.repo_id, true).await?;
1051 let behind: Vec<&EntryRow> = active
1052 .iter()
1053 .filter(|other| {
1054 other.state() != QueueState::Waiting && other.ahead().contains(&row.number)
1055 })
1056 .collect();
1057 self.retest(&behind).await?;
1058
1059 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
1060 return Ok(());
1061 };
1062 let ahead = row.ahead();
1063 let state = if ahead.is_empty() {
1064 "the default branch as it is now".to_owned()
1065 } else {
1066 format!(
1067 "the default branch with {} merged in first",
1068 ahead
1069 .iter()
1070 .map(|n| format!("#{n}"))
1071 .collect::<Vec<_>>()
1072 .join(", ")
1073 )
1074 };
1075 // Named in backticks, so the conversation can link each to the diff.
1076 let files = crate::mergeability::tidy(report.conflicts.clone())
1077 .iter()
1078 .map(|path| format!("`{path}`"))
1079 .collect::<Vec<_>>()
1080 .join(", ");
1081 let why = match (&report.error, report.conflict_with) {
1082 (_, Some(other)) if other != row.number && !files.is_empty() => format!(
1083 "Its change conflicts with #{other}, which is ahead of it in the merge queue, in {files}. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
1084 ),
1085 (_, Some(other)) if other != row.number => format!(
1086 "Its change conflicts with #{other}, which is ahead of it in the merge queue. Bring it up to date with the default branch once #{other} lands and push: it rejoins the merge queue once its checks pass."
1087 ),
1088 (Some(_), None) if !files.is_empty() => format!(
1089 "Its change conflicts with the default branch in {files}. Bring it up to date with the default branch and push: it rejoins the merge queue once its checks pass."
1090 ),
1091 (Some(error), _) if error.starts_with("the workflow ") => {
1092 format!(
1093 "{} when it was combined with {state}.",
1094 error
1095 .replacen("the workflow", "The workflow", 1)
1096 .trim_end_matches(" on it")
1097 )
1098 }
1099 (Some(error), _) if error.starts_with("the required ") => {
1100 format!("Combined with {state}, {error}.")
1101 }
1102 (Some(error), _) => {
1103 format!("Its combined state could not be built or checked: {error}")
1104 }
1105 (None, _) => format!(
1106 "It failed when it was combined with {state}, though it may pass on its own."
1107 ),
1108 };
1109 let now = now_ms();
1110 let run_id = new_id("chk", now);
1111 let mut results = report.results.clone();
1112 for result in &mut results {
1113 result.command = format!("{} (merge queue, on {state})", result.command);
1114 }
1115 self.db
1116 .batch(vec![
1117 self.db
1118 .prepare(
1119 "INSERT INTO check_runs
1120 (id, pull_id, head_commit, status, results, error, token_hash,
1121 created_at, finished_at)
1122 VALUES (?, ?, ?, 'failed', ?, ?, ?, ?, ?)",
1123 )
1124 .bind(&[
1125 run_id.as_str().into(),
1126 pull.id.as_str().into(),
1127 row.head_commit.as_deref().unwrap_or_default().into(),
1128 serde_json::to_string(&results)?.into(),
1129 why.as_str().into(),
1130 hash(&new_token()).into(),
1131 rfc3339(now).into(),
1132 rfc3339(now).into(),
1133 ])?,
1134 self.db
1135 .prepare(
1136 "UPDATE pulls SET check_status = 'failed', check_run_id = ? WHERE id = ?",
1137 )
1138 .bind(&[run_id.as_str().into(), pull.id.as_str().into()])?,
1139 // Its merge stays asked for: it rejoins once a fix's checks
1140 // pass (integration.rs `requeue_if_fixed`).
1141 self.db
1142 .prepare("UPDATE queue_entries SET requeue = 1 WHERE id = ?")
1143 .bind(&[row.id.as_str().into()])?,
1144 ])
1145 .await?;
1146 self.note(
1147 &row.repo_id,
1148 pull.number,
1149 ("g1t", "g1t"),
1150 &format!("was taken out of the merge queue. {why}"),
1151 )
1152 .await?;
1153 self.publish_as(
1154 "checks.completed",
1155 &row.repo_id,
1156 None,
1157 ChecksEvent {
1158 pull_id: pull.id.clone(),
1159 repo_id: row.repo_id.clone(),
1160 number: pull.number,
1161 status: "failed",
1162 commit: row.head_commit.clone().unwrap_or_default(),
1163 },
1164 )
1165 .await?;
1166 Ok(())
1167 }
1168
1169 /// Lands every entry at the front of the queue whose tested state
1170 /// passed, in order.
1171 async fn settle(&self, repo_id: &str) -> Result<()> {
1172 let active = self.entries(repo_id, true).await?;
1173 let Some(viewer) = active.first().and_then(EntryRow::actor) else {
1174 return Ok(());
1175 };
1176 let repo: Outcome<Repo> = g1t_kit::call(
1177 &self.repos,
1178 "get_by_id",
1179 &GetByIdArgs {
1180 id: repo_id.to_owned(),
1181 viewer: Some(viewer),
1182 },
1183 )
1184 .await?;
1185 let Outcome::Ok(repo) = crate::retired::unless_archived(repo) else {
1186 return Ok(());
1187 };
1188 for row in active {
1189 if row.state() != QueueState::Passed {
1190 // The front has not passed yet: nothing behind it may land.
1191 return Ok(());
1192 }
1193 let Some(pull) = self.pull_by_id(&row.pull_id).await? else {
1194 self.leave_gone(&row, PULL_GONE).await?;
1195 return Ok(());
1196 };
1197 let Some(actor) = row.actor() else {
1198 continue;
1199 };
1200 // Git may have succeeded before a previous request stopped. Once
1201 // the merge record exists, only queue bookkeeping remains.
1202 if row.landing_commit.is_some()
1203 && pull.status == PullStatus::Merged
1204 && pull.head_commit == row.landing_commit
1205 {
1206 self.finish_landing(&row).await?;
1207 continue;
1208 }
1209 if row.landing_commit.is_none() {
1210 // Pushed to since it was tested: test it again as it is now.
1211 if pull.status != PullStatus::Open {
1212 self.leave(repo_id, &pull, QueueState::Removed, Some("It was closed."))
1213 .await?;
1214 continue;
1215 }
1216 let head: Option<String> = g1t_kit::call(
1217 &self.repos,
1218 "head",
1219 &HeadArgs {
1220 repo_id: pull.source_id().to_owned(),
1221 branch: pull
1222 .branch
1223 .clone()
1224 .unwrap_or_else(|| repo.default_branch.clone()),
1225 },
1226 )
1227 .await?;
1228 if head.is_none() {
1229 self.leave_gone(&row, BRANCH_GONE).await?;
1230 return Ok(());
1231 }
1232 if head != row.head_commit || row.combined_commit.is_none() {
1233 let active = self.entries(repo_id, true).await?;
1234 let again: Vec<&EntryRow> = active
1235 .iter()
1236 .filter(|other| other.id == row.id || other.ahead().contains(&row.number))
1237 .collect();
1238 self.retest(&again).await?;
1239 return Ok(());
1240 }
1241 // Judged on this attempt's own evidence on the tested commit,
1242 // not on an earlier attempt's or the pull request's own.
1243 let Some(attempt) = row.attempt() else { return Ok(()) };
1244 let evidence = self.attempt_statuses(repo_id, &attempt).await?;
1245 if evidence.runs_missing {
1246 // Its runs cannot be read just now: try again later.
1247 return Ok(());
1248 }
1249 let gate = self
1250 .queue_landing_gate(
1251 &repo,
1252 &pull,
1253 &actor,
1254 row.combined_commit.as_deref().unwrap(),
1255 row.bypass_rules != 0,
1256 evidence.all(),
1257 )
1258 .await?;
1259 if let Some(refusal) = gate.refusal() {
1260 self.record_merge_evaluations(&repo, &pull, &gate).await;
1261 self.leave(repo_id, &pull, QueueState::Removed, Some(&refusal))
1262 .await?;
1263 self.changed(repo_id).await?;
1264 return Ok(());
1265 }
1266 }
1267 let acquired = self
1268 .db
1269 .prepare(BEGIN_LANDING)
1270 .bind(&[
1271 row.id.as_str().into(),
1272 row.combined_commit
1273 .as_deref()
1274 .map_or(JsValue::NULL, JsValue::from),
1275 row.attempt_id
1276 .as_deref()
1277 .map_or(JsValue::NULL, JsValue::from),
1278 ])?
1279 .first::<crate::rows::ValueRow>(None)
1280 .await?;
1281 if acquired.is_none() {
1282 return Ok(());
1283 }
1284 let land = LandArgs {
1285 source_id: repo_id.to_owned(),
1286 branch: Some(row.branch()),
1287 actor: actor.clone(),
1288 // The queue lands on the default branch.
1289 target_branch: None,
1290 expected_commit: row.combined_commit.clone(),
1291 target_id: None,
1292 // The state the queue tested lands as it is: the queue
1293 // rule's method is not applied to it.
1294 method: Some(g1t_contracts::rules::MergeMethod::FastForward),
1295 message: None,
1296 author: None,
1297 sign_off: false,
1298 };
1299 let mut landed: Outcome<Landed> = g1t_kit::call(&self.repos, "land", &land).await?;
1300 if matches!(landed, Outcome::Fail(_)) {
1301 // BEGIN_LANDING lets a crashed publication resume, so another
1302 // request may be publishing this same commit: its push makes
1303 // ours lose Git's compare-and-swap, and its bookkeeping may
1304 // already have dropped the tested branch. Ask Git again before
1305 // giving up on a commit that may already be published.
1306 if !self.still_landing(&row).await? {
1307 return Ok(());
1308 }
1309 landed = g1t_kit::call(&self.repos, "land", &land).await?;
1310 }
1311 let landed = match landed {
1312 Outcome::Ok(landed) => landed,
1313 Outcome::Fail(_) => {
1314 let abandoned = self
1315 .db
1316 .prepare(ABANDON_LANDING)
1317 .bind(&[
1318 row.id.as_str().into(),
1319 row.combined_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
1320 ])?
1321 .first::<crate::rows::ValueRow>(None)
1322 .await?;
1323 if abandoned.is_none() {
1324 // Another request finished (or abandoned) it.
1325 return Ok(());
1326 }
1327 // The default branch moved outside the queue: every
1328 // tested state is built on something that is gone.
1329 let active = self.entries(repo_id, true).await?;
1330 let all: Vec<&EntryRow> = active.iter().collect();
1331 self.retest(&all).await?;
1332 return Ok(());
1333 }
1334 };
1335 // Record the merge before removing its recoverable queue intent.
1336 let head = row.head_commit.clone();
1337 let merged = self.record_merge(&repo, pull, &actor, row.keep_issue_open != 0, landed)
1338 .await?;
1339 self.finish_landing(&row).await?;
1340 // Its branch, where the repository deletes branches on merge,
1341 // if it is still where it was tested.
1342 if merged.status == PullStatus::Merged {
1343 self.delete_branch_after_merge(&repo, &merged, &actor, None, head).await?;
1344 }
1345 }
1346 Ok(())
1347 }
1348
1349 async fn finish_landing(&self, row: &EntryRow) -> Result<()> {
1350 self.db
1351 .prepare(FINISH_LANDING)
1352 .bind(&[rfc3339(now_ms()).into(), row.id.as_str().into()])?
1353 .run()
1354 .await?;
1355 self.drop_branch(row).await;
1356 Ok(())
1357 }
1358
1359 /// Whether `row` still holds the publication intent it was read with.
1360 async fn still_landing(&self, row: &EntryRow) -> Result<bool> {
1361 Ok(self
1362 .db
1363 .prepare(STILL_LANDING)
1364 .bind(&[
1365 row.id.as_str().into(),
1366 row.combined_commit.as_deref().map_or(JsValue::NULL, JsValue::from),
1367 ])?
1368 .first::<crate::rows::ValueRow>(None)
1369 .await?
1370 .is_some())
1371 }
1372}