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

Commit

Merge branch 'worktree-agent-a3548f77b68bcc35c' into queue-methods

syntaqxcommitted Parents928e7ed0c9ad8cBrowse files
4 files+114−120/4 viewed
+3−0
128128 blob_key: Option<String>,
129129 /// Log retention and free workspaces' caps, from wrangler vars (limits.rs).
130130 limits: limits::Limits,
131+ /// Whether repositories' Actions are on, read once per request.
132+ actions_on: protection::ActionsOn,
131133 }
132134
133135 impl Actions {
147149 sealer: env.secret("ACTIONS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())),
148150 blob_key: env.secret("ACTIONS_KEY").ok().map(|key| runtime::blob_key(&key.to_string())),
149151 limits: limits::Limits::from_env(env),
152+ actions_on: protection::ActionsOn::default(),
150153 })
151154 }
152155
+8−2
635635 return Ok(None);
636636 }
637637 // Nor while its Actions are off (Settings, Actions; a fork's are
638− // until someone turns them on), whatever started it.
638+ // until someone turns them on), whatever started it. This is the
639+ // gate; read once per request, so the runs of one push share it.
639640 if !self.actions_on(&new.repo).await? {
640641 return Ok(None);
641642 }
808809 self.advance(&run.id).await
809810 }
810811
811− /// A run that could not start, such as for a workflow file that does not read.
812+ /// A run that could not start, such as for a workflow file that does
813+ /// not read. None while `repo`'s Actions are off, as for `create_run`.
812814 #[allow(clippy::too_many_arguments)]
813815 pub async fn record_failed_run(
814816 &self,
817+ repo: &Repo,
815818 row: &WorkflowRow,
816819 git_ref: &str,
817820 sha: &str,
820823 actor: &str,
821824 problem: &str,
822825 ) -> Result<()> {
826+ if !self.actions_on(repo).await? {
827+ return Ok(());
828+ }
823829 let numbered = self
824830 .db
825831 .prepare("UPDATE workflows SET run_count = run_count + 1 WHERE id = ? RETURNING run_count AS n")
+97−8
2929 use g1t_kit::now_ms;
3030 use serde::Deserialize;
3131 use serde_json::{Value, json};
32+use std::cell::RefCell;
33+use std::collections::HashMap;
3234 use worker::Result;
3335
3436 use crate::plan::{JobRow, RunRow};
322324 }
323325
324326 /// Whether `repo`'s workflows run at all: as chosen, else on, and off
325− /// for a fork. Nothing starts a run while they are off.
327+ /// for a fork. Nothing starts a run while they are off. Read once per
328+ /// request ([`ActionsOn`]), however many runs it starts.
326329 pub(crate) async fn actions_on(&self, repo: &Repo) -> Result<bool> {
327330 #[derive(Deserialize)]
328331 struct Enabled {
329332 enabled: Option<u32>,
330333 }
331− let row = self
332− .db
333− .prepare("SELECT enabled FROM repo_settings WHERE repo_id = ?")
334− .bind(&[repo.id.as_str().into()])?
335− .first::<Enabled>(None)
336− .await?;
337− Ok(actions_enabled(row.and_then(|row| row.enabled).map(|on| on == 1), repo.forked_from.is_some()))
334+ self.actions_on
335+ .get(&repo.id, || async {
336+ let row = self
337+ .db
338+ .prepare("SELECT enabled FROM repo_settings WHERE repo_id = ?")
339+ .bind(&[repo.id.as_str().into()])?
340+ .first::<Enabled>(None)
341+ .await?;
342+ Ok(actions_enabled(row.and_then(|row| row.enabled).map(|on| on == 1), repo.forked_from.is_some()))
343+ })
344+ .await
338345 }
339346
340347 /// What a workflow without `permissions:` gets in the repository, and
433440 ])?
434441 .run()
435442 .await?;
443+ self.actions_on.forget(&repo.id);
436444 Ok(Outcome::Ok(self.repo_settings(&repo).await?))
437445 }
438446
11781186 }
11791187 }
11801188
1189+/// Whether repositories' Actions are on, as this request read them: a push
1190+/// or a minute's schedules that start several runs read each repository's
1191+/// setting once. The service is made afresh for each request, and
1192+/// `Actions::on_event` clears it for each queued event, so a change is
1193+/// seen by the next one.
1194+#[derive(Default)]
1195+pub(crate) struct ActionsOn(RefCell<HashMap<String, bool>>);
1196+
1197+impl ActionsOn {
1198+ /// `repo_id`'s, from `read` the first time it is asked for.
1199+ pub(crate) async fn get<F, Fut>(&self, repo_id: &str, read: F) -> Result<bool>
1200+ where
1201+ F: FnOnce() -> Fut,
1202+ Fut: Future<Output = Result<bool>>,
1203+ {
1204+ if let Some(on) = self.0.borrow().get(repo_id).copied() {
1205+ return Ok(on);
1206+ }
1207+ let on = read().await?;
1208+ self.0.borrow_mut().insert(repo_id.to_owned(), on);
1209+ Ok(on)
1210+ }
1211+
1212+ /// Reads `repo_id`'s again next time, after it was changed.
1213+ pub(crate) fn forget(&self, repo_id: &str) {
1214+ self.0.borrow_mut().remove(repo_id);
1215+ }
1216+
1217+ /// Reads every repository's again.
1218+ pub(crate) fn clear(&self) {
1219+ self.0.borrow_mut().clear();
1220+ }
1221+}
1222+
11811223 /// A workflow job's own token may not approve or change what keeps runs
11821224 /// safe: a workflow could otherwise let itself through.
11831225 fn refuse_job_token<T>(actor: &User) -> Option<Outcome<T>> {
12081250 );
12091251 }
12101252
1253+ /// Runs a future that never waits, as `ActionsOn`'s do here.
1254+ fn ready<T>(future: impl Future<Output = T>) -> T {
1255+ let mut future = std::pin::pin!(future);
1256+ let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
1257+ match future.as_mut().poll(&mut cx) {
1258+ std::task::Poll::Ready(value) => value,
1259+ std::task::Poll::Pending => panic!("the future waited"),
1260+ }
1261+ }
1262+
1263+ #[test]
1264+ fn a_request_reads_each_repositorys_setting_once() {
1265+ let on = ActionsOn::default();
1266+ let reads = std::cell::Cell::new(0);
1267+ let read = |enabled: bool| {
1268+ reads.set(reads.get() + 1);
1269+ async move { Ok(enabled) }
1270+ };
1271+ // A fork with Actions off: every run of one push is refused, from
1272+ // one read.
1273+ for _ in 0..5 {
1274+ assert!(!ready(on.get("rep_fork", || read(actions_enabled(None, true)))).unwrap());
1275+ }
1276+ assert_eq!(reads.get(), 1);
1277+ // Another repository is its own read.
1278+ for _ in 0..3 {
1279+ assert!(ready(on.get("rep_main", || read(actions_enabled(None, false)))).unwrap());
1280+ }
1281+ assert_eq!(reads.get(), 2);
1282+ // Turned on in Settings: read again, and runs start.
1283+ on.forget("rep_fork");
1284+ assert!(ready(on.get("rep_fork", || read(actions_enabled(Some(true), true)))).unwrap());
1285+ assert_eq!(reads.get(), 3);
1286+ // The next queued event reads both afresh.
1287+ on.clear();
1288+ assert!(!ready(on.get("rep_fork", || read(false))).unwrap());
1289+ assert!(ready(on.get("rep_main", || read(true))).unwrap());
1290+ assert_eq!(reads.get(), 5);
1291+ }
1292+
1293+ #[test]
1294+ fn a_failed_read_is_not_kept() {
1295+ let on = ActionsOn::default();
1296+ assert!(ready(on.get("rep_a", || async { Err(worker::Error::RustError("D1 is down".into())) })).is_err());
1297+ assert!(ready(on.get("rep_a", || async { Ok(true) })).unwrap());
1298+ }
1299+
12111300 fn env(reviewers: &[(&str, &str)]) -> Environment {
12121301 Environment {
12131302 reviewers: reviewers.iter().map(|(kind, name)| EnvironmentReviewer { kind: (*kind).into(), name: (*name).into() }).collect(),
+6−2
760760 }
761761
762762 pub async fn on_event(&self, event: &Event) -> Result<()> {
763+ // Each queued event reads repositories' settings afresh.
764+ self.actions_on.clear();
763765 let Some(repo_id) = event.repo_id.as_deref() else { return Ok(()) };
764766 let mut mapped = github_events(&event.kind);
765767 // A new branch or tag is also `create`.
779781 }
780782 // Nothing starts while the repository's Actions are off (a fork's
781783 // are, until someone turns them on). Its workflows are still synced
782− // above, so turning them on finds them listed.
784+ // above, so turning them on finds them listed. `create_run` holds
785+ // the gate; asking here (the same read) skips reading workflows
786+ // from git for nothing.
783787 if !self.actions_on(&repo).await? {
784788 return Ok(());
785789 }
10121016 problem: &str,
10131017 ) -> Result<()> {
10141018 let row = self.workflow_row(repo, path, path, source).await?;
1015− self.record_failed_run(&row, subject.git_ref.as_str(), &subject.sha, event_key, actor_id, sender, problem).await
1019+ self.record_failed_run(repo, &row, subject.git_ref.as_str(), &subject.sha, event_key, actor_id, sender, problem).await
10161020 }
10171021
10181022 /// Scheduled workflows that run this minute, on the default branch: at

Comments

Sign in to comment.