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