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.

gitagentscloudflarerustforgemcp

Commit

Merge branch 'worktree-agent-a31e55ea24332287a' into ship-batch

syntaqxcommitted Parents4da0a625be0ae5Browse files
7 files+73−280/7 viewed
+2−1
478478 Each Worker keeps a share of its invocations' logs, set by
479479 `head_sampling_rate` in its `wrangler.jsonc`: every one for billing,
480480 identity, runner, actions, deployments, status and sudo, and a tenth for
481−the rest. Metrics are not sampled. To chase a rare error on a sampled
481+the rest. Repos and work keep every one for now, while rare ~60 s stalls
482+are traced, and go back to a tenth once they are found. Metrics are not sampled. To chase a rare error on a sampled
482483 Worker, raise its rate to 1 while you look.
+3−1
501501 .results::<Id>()?;
502502 for Id { repo_id } in ids {
503503 match self.repo_by_id(&repo_id).await? {
504− Some((repo, ws)) => self.sync(&repo, &ws).await?,
504+ Some((repo, ws)) => {
505+ self.sync(&repo, &ws).await?;
506+ }
505507 // Gone: nothing to sync, so it is not looked for again.
506508 None => {
507509 self.db
+33−18
1111 use serde::Deserialize;
1212 use worker::Result;
1313
14+use futures_util::{StreamExt, stream};
15+
1416 use crate::{Actions, Count, MAX_WORKFLOWS, optional};
1517
18+/// Workflow files read from repos at once: under the six connections a
19+/// Worker keeps open.
20+const BLOBS_AT_ONCE: usize = 6;
21+
1622 /// One workflow file as read at a commit.
23+#[derive(Clone)]
1724 pub struct WorkflowFile {
1825 pub path: String,
1926 pub source: String,
2330 }
2431
2532 /// The files at a commit, and the commit the ref resolved to.
33+#[derive(Clone)]
2634 pub struct Read {
2735 pub files: Vec<WorkflowFile>,
2836 pub head: Option<String>,
100108 };
101109 let at = head.clone().unwrap_or(resolved);
102110 found_head = found_head.or(head);
103− for entry in entries
111+ let file_paths: Vec<String> = entries
104112 .into_iter()
105113 .filter(|entry| matches!(entry.kind, EntryKind::Blob | EntryKind::Exec))
106114 .filter(|entry| entry.name.ends_with(".yml") || entry.name.ends_with(".yaml"))
107115 .take(MAX_WORKFLOWS.saturating_sub(files.len()))
108− {
109− let file_path = format!("{folder}/{}", entry.name);
110− let blob: Outcome<BlobView> = g1t_kit::call(
111− &self.repos,
112− "blob",
113− &BlobArgs {
114− path: path.clone(),
115− viewer: viewer.clone(),
116− git_ref: at.clone(),
117− file_path: file_path.clone(),
118− },
119− )
120− .await?;
121− if let Outcome::Ok(BlobView { text: Some(source), .. }) = blob {
116+ .map(|entry| format!("{folder}/{}", entry.name))
117+ .collect();
118+ // A few at a time, in the tree's order: one after another, a
119+ // repository's dozen workflows took 4 s of every push.
120+ let repos = &self.repos;
121+ let blobs: Vec<Result<Outcome<BlobView>>> = stream::iter(file_paths.iter().map(|file_path| {
122+ let args = BlobArgs {
123+ path: path.clone(),
124+ viewer: viewer.clone(),
125+ git_ref: at.clone(),
126+ file_path: file_path.clone(),
127+ };
128+ async move { g1t_kit::call(repos, "blob", &args).await }
129+ }))
130+ .buffered(BLOBS_AT_ONCE)
131+ .collect()
132+ .await;
133+ for (file_path, blob) in file_paths.into_iter().zip(blobs) {
134+ if let Outcome::Ok(BlobView { text: Some(source), .. }) = blob? {
122135 files.push(WorkflowFile { path: file_path, source, required: false });
123136 }
124137 }
126139 Ok(Read { files, head: found_head })
127140 }
128141
129− /// Keeps the `workflows` table in step with the default branch.
130− pub async fn sync(&self, repo: &Repo, actor: &User) -> Result<()> {
142+ /// Keeps the `workflows` table in step with the default branch, and
143+ /// gives back the files it read there (`on_event` starts a default
144+ /// branch push's runs from them rather than reading them again).
145+ pub async fn sync(&self, repo: &Repo, actor: &User) -> Result<Read> {
131146 let read = self.read_for(repo, actor, None).await?;
132147 let full_name = format!("{}/{}", repo.namespace, repo.name);
133148 let now = rfc3339(now_ms());
200215 .bind(&[repo.id.as_str().into(), now.into()])?,
201216 );
202217 self.db.batch(statements).await?;
203− Ok(())
218+ Ok(read)
204219 }
205220
206221 /// Whether any of the repository's default-branch workflows could
+9−4
776776 return Ok(());
777777 }
778778 let Some((repo, ws)) = self.repo_by_id(repo_id).await? else { return Ok(()) };
779− if pushed_default {
780− self.sync(&repo, &ws).await?;
781− }
779+ // What the sync read, which a push to the default branch starts
780+ // from while the branch is still at the pushed commit.
781+ let synced = if pushed_default { Some(self.sync(&repo, &ws).await?) } else { None };
782782 // Nothing starts while the repository's Actions are off (a fork's
783783 // are, until someone turns them on). Its workflows are still synced
784784 // above, so turning them on finds them listed. `create_run` holds
822822 }
823823 // A pull request from a working copy reads the copy's files.
824824 let read = if subject.source == Self::repo_path(&repo) {
825− self.read_for(&repo, &ws, subject.source_ref.as_deref()).await?
825+ match synced.as_ref().filter(|read| read.head.as_deref() == Some(subject.sha.as_str())) {
826+ // The same files at the same commit: reading them again
827+ // took seconds of every push to the default branch.
828+ Some(read) => read.clone(),
829+ None => self.read_for(&repo, &ws, subject.source_ref.as_deref()).await?,
830+ }
826831 } else if !subject.source.namespace.eq_ignore_ascii_case(WORKING_COPIES) {
827832 // A branch of someone's fork, built as its head (there was
828833 // no merge commit): the workflows are this repository's, at
+18−0
27352735 flush_later(&env, &ctx);
27362736 return answered;
27372737 };
2738+ let started = now_ms();
27382739 // A replica near the caller when it asks for one (crates/kit/src/d1.rs).
27392740 // Git over HTTPS above always reads the primary.
27402741 let (db, served) = g1t_kit::d1::open(&env, "DB", &request)?;
29982999 };
29993000 repos.deferred.hand_over(&ctx);
30003001 flush_later(&env, &ctx);
3002+ slow_answer(&method, now_ms().saturating_sub(started));
30013003 served.finish(answered)
30023004 }
30033005
3006+/// RPC answers this slow are logged with how long the answer took.
3007+const SLOW_RPC_MS: u64 = 1_000;
3008+
3009+/// Logs an RPC answer that took [`SLOW_RPC_MS`] or more, as a JSON line
3010+/// Workers Logs keeps as fields (`rpc`, `answered_ms`). An invocation's
3011+/// wall time there also counts what `ctx.wait_until` does after the
3012+/// answer, which here includes waiting up to `meters::FLUSH_EVERY_MS` to
3013+/// write the meters (`flush_later`), so it reads as seconds for answers
3014+/// that took milliseconds; `answered_ms` is the time the caller waited,
3015+/// as the `svc` entry of the answer's `server-timing` says.
3016+fn slow_answer(method: &str, took: u64) {
3017+ if took >= SLOW_RPC_MS {
3018+ worker::console_log!("{}", serde_json::json!({ "rpc": method, "answered_ms": took }));
3019+ }
3020+}
3021+
30043022 /// The nightly cron in wrangler.jsonc: tonight's backups are queued.
30053023 const BACKUP_CRON: &str = "53 2 * * *";
30063024
+4−2
136136 { "name": "PACK_FILL_LIMIT", "namespace_id": "4301", "simple": { "limit": 30, "period": 60 } },
137137 { "name": "ANONYMOUS_FETCH_LIMIT", "namespace_id": "4302", "simple": { "limit": 120, "period": 60 } }
138138 ],
139− // Logs of a tenth of requests: every git request and page is one.
140− "observability": { "enabled": true, "head_sampling_rate": 0.1 }
139+ // Logs of every request for now, to trace the rare ~60 s stalls a tenth
140+ // of requests kept missing. Put it back to 0.1 once they are found:
141+ // every git request and page is one.
142+ "observability": { "enabled": true, "head_sampling_rate": 1 }
141143 }
+4−2
2626 "queues": {
2727 "consumers": [{ "queue": "g1t-events-work", "max_batch_size": 100, "max_batch_timeout": 1, "max_retries": 3, "dead_letter_queue": "g1t-events-dlq" }]
2828 },
29− // Logs of a tenth of requests: every issue and pull request page asks it.
30− "observability": { "enabled": true, "head_sampling_rate": 0.1 }
29+ // Logs of every request for now, to trace the rare ~60 s stalls a tenth
30+ // of requests kept missing. Put it back to 0.1 once they are found:
31+ // every issue and pull request page asks it.
32+ "observability": { "enabled": true, "head_sampling_rate": 1 }
3133 }

Comments

Sign in to comment.