Skip to content
Sign in

flagon-io/g1t

public
0

g1t/services/work/src/statuses.rs

323 lines12,771 bytesCodeBlame

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

Fast pages, required checks on the branch, self-hosted runners, honest incidents1//! Statuses on commits: what workflow runs (and other tools, such as
2//! deployments) say about a pull request's head. These are its checks.
3//!
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge4//! The rules of the branch a pull request merges into name the checks that
5//! must pass (rulesets.rs; `RepoSettings::required_checks` as they stack):
6//! a required check that failed, is still running or has not reported
7//! refuses the merge, for everyone and for the merge queue (`g1t_rules`
8//! says so). Where g1t sees an agent's pull request through,
Fast pages, required checks on the branch, self-hosted runners, honest incidents9//! any check that failed sends the agent back to fix it, with what the
10//! failing jobs printed; once it is out of revisions, only a required
11//! check holds the pull request for a person.
GitHub Actions on g1t, part two: running workflows12
13use g1t_contracts::events::ChecksEvent;
14use g1t_contracts::time::rfc3339;
Fast pages, required checks on the branch, self-hosted runners, honest incidents15use g1t_contracts::work::{
16 CommitStatus, RequiredCheck, RequiredState, SeenCheck, SeenChecksArgs, SetCommitStatusArgs, check_name,
17 required_checks,
18};
GitHub Actions on g1t, part two: running workflows19use g1t_contracts::{FailureCode, Outcome};
20use g1t_kit::now_ms;
21use serde::Deserialize;
22use worker::Result;
23
24use crate::Work;
25
26#[derive(Deserialize)]
27struct StatusRow {
28 context: String,
29 state: String,
30 description: Option<String>,
31 target_url: Option<String>,
32 updated_at: String,
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge33 #[serde(default)]
34 source: Option<String>,
GitHub Actions on g1t, part two: running workflows35}
36
37impl From<StatusRow> for CommitStatus {
38 fn from(row: StatusRow) -> Self {
39 CommitStatus {
40 context: row.context,
41 state: row.state,
42 description: row.description,
43 target_url: row.target_url,
44 updated_at: row.updated_at,
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge45 source: row.source,
GitHub Actions on g1t, part two: running workflows46 }
47 }
48}
49
50#[derive(Deserialize)]
51struct HeadRow {
52 id: String,
53 number: u32,
54}
55
Fast pages, required checks on the branch, self-hosted runners, honest incidents56/// What a commit's checks say: every status still running and every one
57/// that failed, by context, and where each required check stands.
58#[derive(Clone, Debug, Default, PartialEq)]
GitHub Actions on g1t, part two: running workflows59pub(crate) struct WorkflowFacts {
60 pub(crate) pending: Vec<String>,
61 pub(crate) failed: Vec<String>,
Fast pages, required checks on the branch, self-hosted runners, honest incidents62 pub(crate) required: Vec<RequiredCheck>,
GitHub Actions on g1t, part two: running workflows63}
64
65impl WorkflowFacts {
Fast pages, required checks on the branch, self-hosted runners, honest incidents66 /// `required` names the checks the default branch's protection requires.
67 pub(crate) fn of(statuses: &[CommitStatus], required: &[String]) -> WorkflowFacts {
GitHub Actions on g1t, part two: running workflows68 WorkflowFacts {
69 pending: statuses.iter().filter(|s| s.state == "pending").map(|s| s.context.clone()).collect(),
70 failed: statuses.iter().filter(|s| s.state == "failure" || s.state == "error").map(|s| s.context.clone()).collect(),
Fast pages, required checks on the branch, self-hosted runners, honest incidents71 required: required_checks(required, statuses),
GitHub Actions on g1t, part two: running workflows72 }
73 }
74
Fast pages, required checks on the branch, self-hosted runners, honest incidents75 fn required_in(&self, state: RequiredState) -> Vec<String> {
76 self.required.iter().filter(|check| check.state == state).map(|check| check.name.clone()).collect()
77 }
78
79 /// The required checks that failed, by name.
80 pub(crate) fn required_failed(&self) -> Vec<String> {
81 self.required_in(RequiredState::Failure)
82 }
83
84 /// The required checks nothing has reported on the commit yet.
85 pub(crate) fn expected(&self) -> Vec<String> {
86 self.required_in(RequiredState::Expected)
87 }
88}
89
90/// The check names in `(context, last reported)` rows, most recent first:
91/// each name once, with the events it was reported for.
92pub(crate) fn seen(rows: Vec<(String, String)>) -> Vec<SeenCheck> {
93 let mut rows = rows;
94 rows.sort_by(|a, b| b.1.cmp(&a.1));
95 let mut out: Vec<SeenCheck> = Vec::new();
96 for (context, at) in rows {
97 let (name, event) = check_name(&context);
98 match out.iter_mut().find(|seen| seen.name.eq_ignore_ascii_case(name)) {
99 Some(seen) => {
100 if let Some(event) = event
101 && !seen.events.iter().any(|known| known == event)
102 {
103 seen.events.push(event.to_owned());
104 }
105 }
106 None => out.push(SeenCheck {
107 name: name.to_owned(),
108 events: event.map(|event| vec![event.to_owned()]).unwrap_or_default(),
109 last_seen: at,
110 }),
111 }
112 }
113 out
114}
115
116/// How far back `seen_checks` looks, and the most contexts it reads.
117const SEEN_DAYS: u64 = 30;
118const SEEN_LIMIT: u32 = 200;
119
120#[derive(Deserialize)]
121struct SeenRow {
122 context: String,
123 at: String,
124}
125
GitHub Actions on g1t, part two: running workflows126pub(crate) fn list(names: &[String]) -> String {
127 match names {
128 [] => String::new(),
129 [one] => one.clone(),
130 [rest @ .., last] => format!("{} and {last}", rest.join(", ")),
131 }
132}
133
134impl Work {
Fast pages, required checks on the branch, self-hosted runners, honest incidents135 /// Where a commit's checks stand, against the repository's required ones.
136 pub(crate) async fn facts(&self, repo_id: &str, sha: Option<&str>) -> Result<WorkflowFacts> {
137 let (statuses, settings) =
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge138 futures_util::future::try_join(self.statuses(repo_id, sha), self.settings_by_id(repo_id)).await?;
Fast pages, required checks on the branch, self-hosted runners, honest incidents139 Ok(WorkflowFacts::of(&statuses, &settings.required_checks))
140 }
141
142 /// The check names reported on a repository's commits lately, for
143 /// choosing which to require.
144 pub(crate) async fn seen_checks(&self, a: SeenChecksArgs) -> Result<Outcome<Vec<SeenCheck>>> {
145 let repo = match self.repo(&a.repo, &a.viewer).await? {
146 Outcome::Ok(repo) => repo,
147 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
148 };
149 let since = rfc3339(now_ms().saturating_sub(SEEN_DAYS * 24 * 60 * 60 * 1000));
150 let rows = self
151 .db
152 .prepare(
153 "SELECT context, MAX(updated_at) AS at FROM commit_statuses
154 WHERE repo_id = ? AND updated_at >= ? GROUP BY context ORDER BY at DESC LIMIT ?",
155 )
156 .bind(&[repo.id.as_str().into(), since.into(), SEEN_LIMIT.into()])?
157 .all()
158 .await?
159 .results::<SeenRow>()?;
160 Ok(Outcome::Ok(seen(rows.into_iter().map(|row| (row.context, row.at)).collect())))
161 }
162
GitHub Actions on g1t, part two: running workflows163 pub(crate) async fn statuses(&self, repo_id: &str, sha: Option<&str>) -> Result<Vec<CommitStatus>> {
164 let Some(sha) = sha else { return Ok(Vec::new()) };
Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily165 if let Some(found) = self.prefetched_repo(repo_id).filter(|found| found.head.as_deref() == Some(sha)) {
166 return Ok(found
167 .rows::<StatusRow>(crate::prefetch::Slot::Statuses)?
168 .into_iter()
169 .map(CommitStatus::from)
170 .collect());
171 }
GitHub Actions on g1t, part two: running workflows172 Ok(self
173 .db
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge174 .prepare("SELECT context, state, description, target_url, updated_at, source FROM commit_statuses WHERE repo_id = ? AND sha = ? ORDER BY context")
GitHub Actions on g1t, part two: running workflows175 .bind(&[repo_id.into(), sha.into()])?
176 .all()
177 .await?
178 .results::<StatusRow>()?
179 .into_iter()
180 .map(CommitStatus::from)
181 .collect())
182 }
183
184 pub(crate) async fn set_commit_status(&self, a: SetCommitStatusArgs) -> Result<Outcome<bool>> {
185 if !matches!(a.state.as_str(), "pending" | "success" | "failure" | "error") {
186 return Ok(Outcome::fail(FailureCode::Invalid, "`state` is pending, success, failure or error."));
187 }
188 self.db
189 .prepare(
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge190 "INSERT INTO commit_statuses (repo_id, sha, context, state, description, target_url, updated_at, source)
191 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
GitHub Actions on g1t, part two: running workflows192 ON CONFLICT (repo_id, sha, context) DO UPDATE SET
193 state = excluded.state, description = excluded.description,
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge194 target_url = excluded.target_url, updated_at = excluded.updated_at,
195 source = excluded.source",
GitHub Actions on g1t, part two: running workflows196 )
197 .bind(&[
198 a.repo_id.as_str().into(),
199 a.sha.as_str().into(),
200 a.context.as_str().into(),
201 a.state.as_str().into(),
202 a.description.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
203 a.target_url.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
204 rfc3339(now_ms()).into(),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge205 a.source.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
GitHub Actions on g1t, part two: running workflows206 ])?
207 .run()
208 .await?;
209 if a.state == "pending" {
210 return Ok(Outcome::Ok(true));
211 }
212 // Once every workflow on a pull request's head has finished, its
213 // lifecycle moves on, as it does when its checks finish.
Fast pages, required checks on the branch, self-hosted runners, honest incidents214 let facts = self.facts(&a.repo_id, Some(&a.sha)).await?;
GitHub Actions on g1t, part two: running workflows215 if !facts.pending.is_empty() {
216 return Ok(Outcome::Ok(true));
217 }
Sidebar: the panels really slide218 // A merge queue state waiting on its merge_group workflows.
Fast pages, required checks on the branch, self-hosted runners, honest incidents219 self.merge_group_finished(&a.repo_id, &a.sha, &facts).await?;
GitHub Actions on g1t, part two: running workflows220 let heads = self
221 .db
222 .prepare("SELECT id, number FROM pulls WHERE repo_id = ? AND head_commit = ? AND status IN ('draft', 'open')")
223 .bind(&[a.repo_id.as_str().into(), a.sha.as_str().into()])?
224 .all()
225 .await?
226 .results::<HeadRow>()?;
227 for head in heads {
A stalled pull request picks back up when its workflows pass228 // A pull request g1t stopped on picks back up once what stopped
229 // it passes: its workflows, and its checks if it has any. The
230 // lifecycle then decides again, within its usual limits.
231 if facts.failed.is_empty() {
232 let resumed = self
233 .db
234 .prepare(
235 "UPDATE pulls SET stalled = NULL WHERE id = ? AND managed = 1 AND stalled IS NOT NULL
236 AND (check_status IS NULL OR check_status = 'passed') RETURNING id AS value",
237 )
238 .bind(&[head.id.as_str().into()])?
239 .first::<crate::rows::ValueRow>(None)
240 .await?;
241 if resumed.is_some() {
Events: review requests, assignments, stops and deployments are published242 self.announce_resumed(&head.id, None).await?;
A stalled pull request picks back up when its workflows pass243 self.note(
244 &a.repo_id,
245 head.number,
246 (crate::lifecycle::POLICY_ACTOR_ID, crate::lifecycle::POLICY_ACTOR_NAME),
247 "picked this back up: its workflows pass now",
248 )
249 .await?;
250 }
251 }
GitHub Actions on g1t, part two: running workflows252 self.publish_as(
253 "checks.completed",
254 &a.repo_id,
255 None,
256 ChecksEvent {
257 pull_id: head.id,
258 repo_id: a.repo_id.clone(),
259 number: head.number,
260 status: if facts.failed.is_empty() { "passed" } else { "failed" },
261 commit: a.sha.clone(),
262 },
263 )
264 .await?;
265 }
266 Ok(Outcome::Ok(true))
267 }
268}
269
270#[cfg(test)]
271mod tests {
272 use super::*;
273
274 fn status(context: &str, state: &str) -> CommitStatus {
275 CommitStatus {
276 context: context.into(),
277 state: state.into(),
278 description: None,
279 target_url: None,
280 updated_at: String::new(),
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge281 source: None,
GitHub Actions on g1t, part two: running workflows282 }
283 }
284
Fast pages, required checks on the branch, self-hosted runners, honest incidents285 fn names(list: &[&str]) -> Vec<String> {
286 list.iter().map(|name| (*name).to_owned()).collect()
287 }
288
289 #[test]
290 fn only_required_checks_hold_a_merge() {
291 let statuses = [status("CI / push", "pending"), status("Lint / pull_request", "failure"), status("Docs", "success")];
292 // Nothing required: nothing holds it, whatever failed.
293 let free = WorkflowFacts::of(&statuses, &[]);
294 assert_eq!(free.pending, ["CI / push"]);
295 assert_eq!(free.failed, ["Lint / pull_request"]);
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge296 assert!(free.required_failed().is_empty() && free.expected().is_empty());
Fast pages, required checks on the branch, self-hosted runners, honest incidents297 let both = WorkflowFacts::of(&statuses, &names(&["CI", "Lint"]));
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge298 assert_eq!(both.required_failed(), ["Lint"]);
299 assert!(WorkflowFacts::of(&[status("Docs", "success")], &names(&["Docs"])).required_failed().is_empty());
Fast pages, required checks on the branch, self-hosted runners, honest incidents300 }
301
302 #[test]
Merge rulesets: branch and tag rules, agent-first, enforced on push and merge303 fn a_required_check_nothing_reported_is_expected() {
Fast pages, required checks on the branch, self-hosted runners, honest incidents304 let facts = WorkflowFacts::of(&[status("CI / pull_request", "success")], &names(&["CI", "Deploy"]));
305 assert_eq!(facts.expected(), ["Deploy"]);
306 }
307
GitHub Actions on g1t, part two: running workflows308 #[test]
Fast pages, required checks on the branch, self-hosted runners, honest incidents309 fn seen_checks_are_named_once_with_their_events() {
310 let rows = vec![
311 ("CI / push".to_owned(), "2026-10-01T00:00:00Z".to_owned()),
312 ("CI / pull_request".to_owned(), "2026-10-03T00:00:00Z".to_owned()),
313 ("g1t / deploy".to_owned(), "2026-10-02T00:00:00Z".to_owned()),
314 ];
315 let seen = seen(rows);
316 assert_eq!(seen.len(), 2);
317 assert_eq!(seen[0].name, "CI");
318 assert_eq!(seen[0].events, ["pull_request", "push"]);
319 assert_eq!(seen[0].last_seen, "2026-10-03T00:00:00Z");
320 assert_eq!(seen[1].name, "g1t / deploy");
321 assert!(seen[1].events.is_empty());
GitHub Actions on g1t, part two: running workflows322 }
323}

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