| 1 | //! Check runs: the record of the merge queue taking a pull request out, |
| 2 | //! and of commands written on issues, which g1t ran before a pull |
| 3 | //! request's checks were the workflows run on it. Those runs are kept so |
| 4 | //! their history still reads; a sandbox still finishing one reports |
| 5 | //! through the API with the job's one-time token. |
| 6 | |
| 7 | use g1t_contracts::events::ChecksEvent; |
| 8 | use g1t_contracts::time::rfc3339; |
| 9 | use g1t_contracts::work::*; |
| 10 | use g1t_contracts::{FailureCode, Outcome}; |
| 11 | use g1t_kit::now_ms; |
| 12 | use serde::Deserialize; |
| 13 | use sha2::{Digest, Sha256}; |
| 14 | use worker::Result; |
| 15 | use worker::wasm_bindgen::JsValue; |
| 16 | |
| 17 | use crate::prefetch::Slot; |
| 18 | use crate::rows::{PULL_COLUMNS, PullRow}; |
| 19 | use crate::{Work, optional}; |
| 20 | |
| 21 | const MAX_OUTPUT_CHARS: usize = 16_000; |
| 22 | const MAX_RESULTS: usize = 20; |
| 23 | /// How many earlier runs a pull request shows. |
| 24 | const EARLIER_RUNS: u32 = 10; |
| 25 | |
| 26 | #[derive(Deserialize)] |
| 27 | struct RunRow { |
| 28 | id: String, |
| 29 | pull_id: String, |
| 30 | head_commit: String, |
| 31 | status: CheckStatus, |
| 32 | /// JSON array of results. |
| 33 | results: String, |
| 34 | error: Option<String>, |
| 35 | token_hash: String, |
| 36 | created_at: String, |
| 37 | finished_at: Option<String>, |
| 38 | } |
| 39 | |
| 40 | impl From<RunRow> for CheckRun { |
| 41 | fn from(row: RunRow) -> Self { |
| 42 | CheckRun { |
| 43 | id: row.id, |
| 44 | head_commit: row.head_commit, |
| 45 | status: row.status, |
| 46 | results: serde_json::from_str(&row.results).unwrap_or_default(), |
| 47 | error: row.error, |
| 48 | created_at: row.created_at, |
| 49 | finished_at: row.finished_at, |
| 50 | } |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | pub(crate) fn hash(token: &str) -> String { |
| 55 | hex::encode(Sha256::digest(token.as_bytes())) |
| 56 | } |
| 57 | |
| 58 | pub(crate) fn new_token() -> String { |
| 59 | let mut bytes = [0u8; 32]; |
| 60 | getrandom::getrandom(&mut bytes).expect("no source of randomness"); |
| 61 | hex::encode(bytes) |
| 62 | } |
| 63 | |
| 64 | fn refused<T>(message: &str) -> Outcome<T> { |
| 65 | Outcome::fail(FailureCode::Conflict, message) |
| 66 | } |
| 67 | |
| 68 | impl Work { |
| 69 | /// The most recent check run of a pull request. |
| 70 | pub(crate) async fn latest_checks(&self, pull_id: &str) -> Result<Option<CheckRun>> { |
| 71 | if let Some(found) = self.prefetched_pull(pull_id) { |
| 72 | return Ok(found.first::<RunRow>(Slot::Runs)?.map(CheckRun::from)); |
| 73 | } |
| 74 | Ok(self |
| 75 | .db |
| 76 | .prepare("SELECT * FROM check_runs WHERE pull_id = ? ORDER BY id DESC LIMIT 1") |
| 77 | .bind(&[pull_id.into()])? |
| 78 | .first::<RunRow>(None) |
| 79 | .await? |
| 80 | .map(CheckRun::from)) |
| 81 | } |
| 82 | |
| 83 | /// The runs before the latest, newest first, without what each command |
| 84 | /// printed: enough to see how the checks went over time. |
| 85 | pub(crate) async fn earlier_checks(&self, pull_id: &str) -> Result<Vec<CheckRun>> { |
| 86 | let rows = match self.prefetched_pull(pull_id) { |
| 87 | Some(found) => found |
| 88 | .rows::<RunRow>(Slot::Runs)? |
| 89 | .into_iter() |
| 90 | .skip(1) |
| 91 | .take(EARLIER_RUNS as usize) |
| 92 | .collect(), |
| 93 | None => self |
| 94 | .db |
| 95 | .prepare( |
| 96 | "SELECT * FROM check_runs WHERE pull_id = ? ORDER BY id DESC LIMIT ? OFFSET 1", |
| 97 | ) |
| 98 | .bind(&[pull_id.into(), EARLIER_RUNS.into()])? |
| 99 | .all() |
| 100 | .await? |
| 101 | .results::<RunRow>()?, |
| 102 | }; |
| 103 | Ok(rows |
| 104 | .into_iter() |
| 105 | .map(|row| { |
| 106 | let mut run = CheckRun::from(row); |
| 107 | for result in &mut run.results { |
| 108 | result.output = String::new(); |
| 109 | } |
| 110 | run |
| 111 | }) |
| 112 | .collect()) |
| 113 | } |
| 114 | |
| 115 | /// Records what a sandbox reports for its run: that it has started, its |
| 116 | /// results, that it could not run, or that the run should be forgotten. |
| 117 | /// The run's token is the only credential, and a finished run accepts |
| 118 | /// nothing more. |
| 119 | pub(crate) async fn report_checks(&self, a: ReportChecksArgs) -> Result<Outcome<CheckRun>> { |
| 120 | let run = self |
| 121 | .db |
| 122 | .prepare("SELECT * FROM check_runs WHERE id = ?") |
| 123 | .bind(&[a.run_id.as_str().into()])? |
| 124 | .first::<RunRow>(None) |
| 125 | .await?; |
| 126 | let Some(run) = run.filter(|run| run.token_hash == hash(&a.token)) else { |
| 127 | return Ok(Outcome::fail(FailureCode::NotFound, "Check run not found.")); |
| 128 | }; |
| 129 | if run.finished_at.is_some() { |
| 130 | return Ok(refused("This check run has already finished.")); |
| 131 | } |
| 132 | let latest = "id = ? AND check_run_id = ?"; |
| 133 | let pull_keys = |
| 134 | || -> [JsValue; 2] { [run.pull_id.as_str().into(), run.id.as_str().into()] }; |
| 135 | |
| 136 | if a.skip { |
| 137 | self.db |
| 138 | .batch(vec![ |
| 139 | self.db |
| 140 | .prepare("DELETE FROM check_runs WHERE id = ?") |
| 141 | .bind(&[run.id.as_str().into()])?, |
| 142 | self.db |
| 143 | .prepare(format!( |
| 144 | "UPDATE pulls SET check_status = NULL, check_run_id = NULL WHERE {latest}" |
| 145 | )) |
| 146 | .bind(&pull_keys())?, |
| 147 | ]) |
| 148 | .await?; |
| 149 | return Ok(Outcome::Ok(run.into())); |
| 150 | } |
| 151 | |
| 152 | let results: Vec<CheckResult> = a |
| 153 | .results |
| 154 | .into_iter() |
| 155 | .take(MAX_RESULTS) |
| 156 | .map(|mut result| { |
| 157 | // Keep the end of long output: that is where failures are. |
| 158 | let length = result.output.chars().count(); |
| 159 | if length > MAX_OUTPUT_CHARS { |
| 160 | result.output = result |
| 161 | .output |
| 162 | .chars() |
| 163 | .skip(length - MAX_OUTPUT_CHARS) |
| 164 | .collect(); |
| 165 | } |
| 166 | result |
| 167 | }) |
| 168 | .collect(); |
| 169 | let status = if a.error.is_some() { |
| 170 | CheckStatus::Errored |
| 171 | } else if results.is_empty() { |
| 172 | CheckStatus::Running |
| 173 | } else if results.iter().all(|result| result.passed) { |
| 174 | CheckStatus::Passed |
| 175 | } else { |
| 176 | CheckStatus::Failed |
| 177 | }; |
| 178 | let finished = (status != CheckStatus::Running).then(|| rfc3339(now_ms())); |
| 179 | self.db |
| 180 | .batch(vec![ |
| 181 | self.db |
| 182 | .prepare( |
| 183 | "UPDATE check_runs SET status = ?, results = ?, error = ?, finished_at = ? |
| 184 | WHERE id = ?", |
| 185 | ) |
| 186 | .bind(&[ |
| 187 | status.as_str().into(), |
| 188 | serde_json::to_string(&results)?.into(), |
| 189 | optional(&a.error), |
| 190 | optional(&finished), |
| 191 | run.id.as_str().into(), |
| 192 | ])?, |
| 193 | self.db |
| 194 | .prepare(format!("UPDATE pulls SET check_status = ? WHERE {latest}")) |
| 195 | .bind(&[ |
| 196 | status.as_str().into(), |
| 197 | run.pull_id.as_str().into(), |
| 198 | run.id.as_str().into(), |
| 199 | ])?, |
| 200 | ]) |
| 201 | .await?; |
| 202 | |
| 203 | if finished.is_some() { |
| 204 | let pull = self |
| 205 | .db |
| 206 | .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE id = ?")) |
| 207 | .bind(&[run.pull_id.as_str().into()])? |
| 208 | .first::<PullRow>(None) |
| 209 | .await? |
| 210 | .map(Pull::from); |
| 211 | if let Some(pull) = pull { |
| 212 | self.publish_as( |
| 213 | "checks.completed", |
| 214 | &pull.repo_id, |
| 215 | None, |
| 216 | ChecksEvent { |
| 217 | pull_id: pull.id.clone(), |
| 218 | repo_id: pull.repo_id.clone(), |
| 219 | number: pull.number, |
| 220 | status: status.as_str(), |
| 221 | commit: run.head_commit.clone(), |
| 222 | }, |
| 223 | ) |
| 224 | .await?; |
| 225 | } |
| 226 | } |
| 227 | Ok(Outcome::Ok(CheckRun { |
| 228 | status, |
| 229 | results, |
| 230 | error: a.error, |
| 231 | finished_at: finished, |
| 232 | ..run.into() |
| 233 | })) |
| 234 | } |
| 235 | } |