Sign in

flagon-io/g1t

public

g1t/services/work/src/checks.rs

244 lines8,610 bytesCodeBlame
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
7use g1t_contracts::events::ChecksEvent;
8use g1t_contracts::time::rfc3339;
9use g1t_contracts::work::*;
10use g1t_contracts::{FailureCode, Outcome};
11use g1t_kit::now_ms;
12use serde::Deserialize;
13use sha2::{Digest, Sha256};
14use worker::Result;
15use worker::wasm_bindgen::JsValue;
16
17use crate::prefetch::Slot;
18use crate::rows::{PULL_COLUMNS, PullRow};
19use crate::{Work, optional};
20
21const MAX_OUTPUT_CHARS: usize = 16_000;
22const MAX_RESULTS: usize = 20;
23/// How many earlier runs a pull request shows.
24const EARLIER_RUNS: u32 = 10;
25
26#[derive(Deserialize)]
27struct 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
40impl 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
54pub(crate) fn hash(token: &str) -> String {
55 hex::encode(Sha256::digest(token.as_bytes()))
56}
57
58pub(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
64fn refused<T>(message: &str) -> Outcome<T> {
65 Outcome::fail(FailureCode::Conflict, message)
66}
67
68impl 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 /// Commands written on issues are no longer run: a pull request's
116 /// checks are the workflows run on it, and the default branch's
117 /// protection says which must pass. Refused, for a runner from before.
118 pub(crate) async fn start_checks(&self, _: StartChecksArgs) -> Result<Outcome<CheckJob>> {
119 Ok(refused(
120 "Checks are the workflows run on a pull request; there are no commands to run.",
121 ))
122 }
123
124 /// Records what a sandbox reports for its run: that it has started, its
125 /// results, that it could not run, or that the run should be forgotten.
126 /// The run's token is the only credential, and a finished run accepts
127 /// nothing more.
128 pub(crate) async fn report_checks(&self, a: ReportChecksArgs) -> Result<Outcome<CheckRun>> {
129 let run = self
130 .db
131 .prepare("SELECT * FROM check_runs WHERE id = ?")
132 .bind(&[a.run_id.as_str().into()])?
133 .first::<RunRow>(None)
134 .await?;
135 let Some(run) = run.filter(|run| run.token_hash == hash(&a.token)) else {
136 return Ok(Outcome::fail(FailureCode::NotFound, "Check run not found."));
137 };
138 if run.finished_at.is_some() {
139 return Ok(refused("This check run has already finished."));
140 }
141 let latest = "id = ? AND check_run_id = ?";
142 let pull_keys =
143 || -> [JsValue; 2] { [run.pull_id.as_str().into(), run.id.as_str().into()] };
144
145 if a.skip {
146 self.db
147 .batch(vec![
148 self.db
149 .prepare("DELETE FROM check_runs WHERE id = ?")
150 .bind(&[run.id.as_str().into()])?,
151 self.db
152 .prepare(format!(
153 "UPDATE pulls SET check_status = NULL, check_run_id = NULL WHERE {latest}"
154 ))
155 .bind(&pull_keys())?,
156 ])
157 .await?;
158 return Ok(Outcome::Ok(run.into()));
159 }
160
161 let results: Vec<CheckResult> = a
162 .results
163 .into_iter()
164 .take(MAX_RESULTS)
165 .map(|mut result| {
166 // Keep the end of long output: that is where failures are.
167 let length = result.output.chars().count();
168 if length > MAX_OUTPUT_CHARS {
169 result.output = result
170 .output
171 .chars()
172 .skip(length - MAX_OUTPUT_CHARS)
173 .collect();
174 }
175 result
176 })
177 .collect();
178 let status = if a.error.is_some() {
179 CheckStatus::Errored
180 } else if results.is_empty() {
181 CheckStatus::Running
182 } else if results.iter().all(|result| result.passed) {
183 CheckStatus::Passed
184 } else {
185 CheckStatus::Failed
186 };
187 let finished = (status != CheckStatus::Running).then(|| rfc3339(now_ms()));
188 self.db
189 .batch(vec![
190 self.db
191 .prepare(
192 "UPDATE check_runs SET status = ?, results = ?, error = ?, finished_at = ?
193 WHERE id = ?",
194 )
195 .bind(&[
196 status.as_str().into(),
197 serde_json::to_string(&results)?.into(),
198 optional(&a.error),
199 optional(&finished),
200 run.id.as_str().into(),
201 ])?,
202 self.db
203 .prepare(format!("UPDATE pulls SET check_status = ? WHERE {latest}"))
204 .bind(&[
205 status.as_str().into(),
206 run.pull_id.as_str().into(),
207 run.id.as_str().into(),
208 ])?,
209 ])
210 .await?;
211
212 if finished.is_some() {
213 let pull = self
214 .db
215 .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE id = ?"))
216 .bind(&[run.pull_id.as_str().into()])?
217 .first::<PullRow>(None)
218 .await?
219 .map(Pull::from);
220 if let Some(pull) = pull {
221 self.publish_as(
222 "checks.completed",
223 &pull.repo_id,
224 None,
225 ChecksEvent {
226 pull_id: pull.id.clone(),
227 repo_id: pull.repo_id.clone(),
228 number: pull.number,
229 status: status.as_str(),
230 commit: run.head_commit.clone(),
231 },
232 )
233 .await?;
234 }
235 }
236 Ok(Outcome::Ok(CheckRun {
237 status,
238 results,
239 error: a.error,
240 finished_at: finished,
241 ..run.into()
242 }))
243 }
244}