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

A push to the default branch starts its runs about 8 seconds sooner, repos and work log every request while the ~60 s stalls are traced, and repos logs its slow answers with the time the caller waited. Measured on flagon-io/g1t over its last 35 pushes (the timeline's git.push times against each run's created_at and started_at): a push's first run is made 12 to 46 s after its event (usually 16 to 20), each further workflow 4 to 6 s after the one before, and a run's first job starts 1 to 3 s after it is made whenever the workspace has room. For 6e24edf2 the event was at 20:37:25.7, the actions consumer started at about 20:37:32.5 (two queue hops), Caches was made at 20:37:43.7, CI at 20:37:47.4 and Deploy at 20:37:52.8, each starting 1.4 to 2 s later. Of the 11 s inside the consumer before the first run, most was reading the twelve workflow files from repos one after another (blob at 300 to 400 ms each), twice: once to sync the workflows table for the default branch and once more, at the same commit, to start the runs. Now the sync hands back what it read, and on_event starts a default-branch push's runs from it while the branch is still at the pushed commit (read.head equals the push's after; otherwise it reads as before), and read_folders reads the files six at a time, in the tree's order. The waits of minutes some runs show (29aba25d: 7 to 8.5 min, 58def404: 5 min, Deploy runs: up to 28 min) are RUNNING_PER_WORKSPACE (four of a workspace's jobs at once) and Deploy's concurrency group, by design, not dispatch; nothing here changes them. Repos and work keep every invocation's logs (head_sampling_rate 1), with a comment to put them back to 0.1 once the stalls are found, and the deploy guide says so. Repos' p99 wall time in Workers Logs counts ctx.wait_until, which waits up to meters::FLUSH_EVERY_MS to write the meters; an RPC answer that takes a second or more now logs {"rpc", "answered_ms"} as JSON, which Workers Logs keeps as fields, so slow answers can be told from slow flushes; the svc entry of every answer's server-timing already gave callers the same time. Metering is unchanged. Checked: cargo test for g1t-actions-service (113), g1t-repos (341) and g1t-work (189), clippy for wasm32 on both services with no new warnings, node scripts/deploy.mjs manifest --check, and npm run test:deploy (92).

G1t-Session:
ses_01m4y12gdde2r9q832m7xk87kd
syntaqxcommitted Parent6e24edfBrowse 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.