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
786 linesCodeBlameRaw

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.

Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet1//! The cache of `actions/cache`: which entries each repository has, kept
2//! here, while the API keeps their bytes in R2 (the ACTIONS_CACHE bucket).
3//!
Merge branch 'worktree-agent-a3abfcce648e87dca'4//! - Entries are scoped by ref, as on GitHub (`scopes`): an entry belongs
5//! to the ref whose run saved it, and a run restores from its own ref,
6//! then its pull request's base branch, then the default branch. A pull
7//! request from outside the repository saves under `untrusted:<ref>`,
8//! which no other ref reads, so it can never plant an entry the default
9//! branch restores. g1t's own `actions/cache` and the toolkit's protocols
10//! follow the same rule (`Actions::cache_scope`).
11//! - An entry's version is the hash of its paths and compression, which
12//! the toolkit's client and g1t's runner both send: the same key saved
13//! for other paths is another entry.
14//! - A key is written once in its scope and version. In each scope in
15//! turn, a restore finds its key exactly, else the newest entry whose key
Merge branch 'worktree-agent-a63425cb2d4935c5f'16//! starts with its key, else with each of its restore keys in order
17//! (`prefixes`), as GitHub's cache does.
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet18//! - An entry is at most `CACHE_MAX_ENTRY_BYTES`. A repository's entries
19//! hold at most `CACHE_REPO_QUOTA_BYTES` together: saving past it evicts
20//! the entries restored longest ago.
21//! - An entry not restored for `CACHE_UNUSED_DAYS`, or saved more than
22//! `CACHE_MAX_AGE_DAYS` ago, is deleted by the hourly sweep, which also
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R223//! writes down what each workspace holds, its artifacts included (they
24//! are in the same bucket), and reports this month's storage to billing
25//! (`note_pending`, source `cache`) at R2's price.
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet26//!
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'27//! - People list a repository's entries and what they hold together
28//! (`caches`, `cache_usage`), and delete them (`delete_caches`), through
29//! the API's `/repos/{owner}/{repo}/actions/caches`, as on GitHub.
30//!
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R231//! The API calls these with the job's token, or its runtime token for the
32//! toolkit's protocols (runtime.rs), which is checked here. An entry the
33//! toolkit saves has the toolkit's version, and is found only by the same
34//! version; g1t's own `actions/cache` saves and finds entries without one.
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet35
36use g1t_contracts::FailureCode;
37use g1t_contracts::Outcome;
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'38use g1t_contracts::access::Capability;
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet39use g1t_contracts::actions::{
40 CACHE_MAX_AGE_DAYS, CACHE_MAX_ENTRY_BYTES, CACHE_MICROS_PER_GB_MONTH, CACHE_REPO_QUOTA_BYTES, CACHE_UNUSED_DAYS,
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'41 CacheAbortArgs, CacheCommitArgs, CacheCommitted, CacheEntry, CacheHit, CacheList, CacheLookupArgs, CacheReservation,
42 CacheReserveArgs, CacheUploadArgs, CacheUsage, CacheUsageArgs, CachesArgs, DeleteCachesArgs,
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet43};
44use g1t_contracts::billing::NotePendingArgs;
45use g1t_contracts::new_id;
46use g1t_contracts::time::rfc3339;
47use g1t_kit::now_ms;
48use serde::Deserialize;
49use serde_json::Value;
50use worker::Result;
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'51use worker::wasm_bindgen::JsValue;
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet52
53use crate::{Actions, check, fail};
54
55const DAY_MS: u64 = 24 * 60 * 60 * 1000;
56/// An upload not finished after this long is given up.
57const PENDING_MS: u64 = 6 * 60 * 60 * 1000;
58/// A GB, as storage is billed.
59const GB: f64 = 1_000_000_000.0;
60
61#[derive(Debug, Deserialize)]
62struct EntryRow {
63 id: String,
64 key: String,
65 object: String,
66 size: f64,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R267 created_at: String,
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet68}
69
70/// The longest key: GitHub's limit.
71const MAX_KEY_CHARS: usize = 512;
72
73/// Whether a key can be kept: 1 to 512 characters, no commas (GitHub's rule).
74pub(crate) fn valid_key(key: &str) -> bool {
75 !key.is_empty() && key.chars().count() <= MAX_KEY_CHARS && !key.contains(',')
76}
77
78/// Which of `entries` (id, size, newest use first) to evict so that the
79/// repository holds at most `quota` with `keep` (just saved) among them.
80/// The entry just saved is never evicted.
81pub(crate) fn to_evict(entries: &[(String, u64)], keep: &str, quota: u64) -> Vec<String> {
82 let mut total: u64 = entries.iter().map(|(_, size)| size).sum();
83 let mut out = Vec::new();
84 for (id, size) in entries.iter().rev() {
85 if total <= quota {
86 break;
87 }
88 if id == keep {
89 continue;
90 }
91 total = total.saturating_sub(*size);
92 out.push(id.clone());
93 }
94 out
95}
96
Merge remote-tracking branch 'origin/main' into workspace-chat97/// Whether a repository holding `total` bytes must evict to stay within
98/// `quota`: what `to_evict` would take something from.
99pub(crate) fn over_quota(total: u64, quota: u64) -> bool {
100 total > quota
101}
102
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet103/// A month's GB-months from the bytes held each day so far: each day's
104/// bytes over 30 days.
105pub(crate) fn gb_months(days: &[u64]) -> f64 {
106 days.iter().map(|bytes| *bytes as f64).sum::<f64>() / GB / 30.0
107}
108
109/// What `gb_months` of cache cost g1t, in millionths of a dollar.
110pub(crate) fn storage_cost(gb_months: f64) -> i64 {
111 (gb_months * CACHE_MICROS_PER_GB_MONTH as f64).ceil() as i64
112}
113
Merge branch 'worktree-agent-a63425cb2d4935c5f'114/// The prefixes a restore tries in each scope once its key has no exact
115/// match, in order: the key itself, then its restore keys. Blank ones and
116/// repeats are dropped.
117pub(crate) fn prefixes<'a>(key: &'a str, restore: &'a [String]) -> Vec<&'a str> {
118 let mut out: Vec<&str> = Vec::new();
119 for prefix in std::iter::once(key).chain(restore.iter().map(String::as_str)).map(str::trim) {
120 if !prefix.is_empty() && !out.contains(&prefix) {
121 out.push(prefix);
122 }
123 }
124 out
125}
126
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'127/// The `ORDER BY` of a list of entries, from GitHub's `sort`
128/// (`last_accessed_at` by default) and `direction` (`desc` by default);
129/// `None` for a name it does not have. Ties go by when each was saved.
130pub(crate) fn cache_order(sort: Option<&str>, direction: Option<&str>) -> Option<String> {
131 let column = match sort.map(str::trim).filter(|s| !s.is_empty()).unwrap_or("last_accessed_at") {
132 "last_accessed_at" => "last_used_at",
133 "created_at" => "created_at",
134 "size_in_bytes" => "size",
135 _ => return None,
136 };
137 let direction = match direction.map(str::trim).filter(|d| !d.is_empty()).unwrap_or("desc") {
138 "asc" => "ASC",
139 "desc" => "DESC",
140 _ => return None,
141 };
142 Some(format!("{column} {direction}, rowid {direction}"))
143}
144
145/// A saved entry, as `caches` lists it.
146#[derive(Debug, Deserialize)]
147struct ListedRow {
148 id: String,
149 number: f64,
150 key: String,
151 scope: String,
152 version: String,
153 object: String,
154 size: f64,
155 created_at: String,
156 last_used_at: String,
157}
158
159impl ListedRow {
160 fn view(&self) -> CacheEntry {
161 CacheEntry {
162 id: self.number as u64,
163 key: self.key.clone(),
164 git_ref: self.scope.clone(),
165 version: self.version.clone(),
166 size_in_bytes: self.size as u64,
167 created_at: self.created_at.clone(),
168 last_accessed_at: self.last_used_at.clone(),
169 }
170 }
171}
172
173const LISTED: &str = "SELECT id, rowid AS number, key, scope, version, object, size, created_at, last_used_at FROM cache_entries";
174
Merge branch 'worktree-agent-a3abfcce648e87dca'175/// Where a job's cache entries are found and saved: its repository, and in
176/// it the refs it restores from, in order, and the one it saves to.
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2177pub(crate) struct CacheScope {
178 pub repo_id: String,
Merge branch 'worktree-agent-a3abfcce648e87dca'179 pub restore: Vec<String>,
180 pub save: String,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2181}
182
Merge branch 'worktree-agent-a3abfcce648e87dca'183/// The scopes a run's jobs restore from, in order, and the one they save
184/// to: its own ref, then its pull request's base branch, then the default
185/// branch. A run that is not trusted (a pull request from outside) saves to
186/// a scope of its own that no other ref reads.
187pub(crate) fn scopes(git_ref: &str, base_ref: Option<&str>, default_branch: &str, trusted: bool) -> (Vec<String>, String) {
188 let own = if trusted { git_ref.to_owned() } else { format!("untrusted:{git_ref}") };
189 let mut restore = vec![own.clone()];
190 let base = base_ref.filter(|base| !base.is_empty()).map(|base| format!("refs/heads/{}", base.trim_start_matches("refs/heads/")));
191 for scope in base.into_iter().chain(std::iter::once(format!("refs/heads/{default_branch}"))) {
192 if !restore.contains(&scope) {
193 restore.push(scope);
194 }
195 }
196 (restore, own)
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2197}
198
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet199impl Actions {
Merge branch 'worktree-agent-a3abfcce648e87dca'200 /// Where a job's entries are found and saved (`scopes`), for g1t's own
201 /// `actions/cache` and the toolkit's protocols alike.
202 pub(crate) async fn cache_scope(&self, job: &crate::plan::JobRow) -> Result<CacheScope> {
203 let (restore, save) = match self.run_row(&job.run_id).await? {
204 Some(run) => {
205 let info = run.info();
206 scopes(&run.git_ref, info.base_ref.as_deref(), &info.default_branch, run.trusted != 0)
207 }
208 None => (Vec::new(), format!("untrusted:{}", job.run_id)),
209 };
210 Ok(CacheScope { repo_id: job.repo_id.clone(), restore, save })
211 }
212
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2213 /// A job by its own token or its runtime token (runtime.rs).
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet214 async fn cache_job(&self, job: &str, token: &str) -> Result<Outcome<crate::plan::JobRow>> {
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2215 self.job_for_credential(job, token).await
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet216 }
217
218 /// `cache_lookup`.
219 pub async fn cache_lookup(&self, a: CacheLookupArgs) -> Result<Outcome<Option<CacheHit>>> {
220 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'221 let scope = self.cache_scope(&job).await?;
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet222 let now = now_ms();
223 let fresh = rfc3339(now.saturating_sub(CACHE_MAX_AGE_DAYS * DAY_MS));
Merge branch 'worktree-agent-a3abfcce648e87dca'224 // Found only by the same version: runners from before it was sent
225 // send none, and find only entries saved without one.
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2226 let version = a.version.clone().unwrap_or_default();
Merge branch 'worktree-agent-a3abfcce648e87dca'227 let mut found = None;
228 'scopes: for ref_scope in &scope.restore {
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet229 found = self
230 .db
231 .prepare(
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2232 "SELECT id, key, object, size, created_at FROM cache_entries
Merge branch 'worktree-agent-a3abfcce648e87dca'233 WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'ready' AND created_at > ?",
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet234 )
Merge branch 'worktree-agent-a3abfcce648e87dca'235 .bind(&[
236 scope.repo_id.as_str().into(),
237 ref_scope.as_str().into(),
238 a.key.as_str().into(),
239 version.as_str().into(),
240 fresh.as_str().into(),
241 ])?
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet242 .first::<EntryRow>(None)
243 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'244 if found.is_some() {
245 break;
246 }
Merge branch 'worktree-agent-a63425cb2d4935c5f'247 for prefix in prefixes(&a.key, &a.restore) {
Merge branch 'worktree-agent-a3abfcce648e87dca'248 found = self
249 .db
250 .prepare(
251 "SELECT id, key, object, size, created_at FROM cache_entries
252 WHERE repo_id = ?1 AND scope = ?5 AND version = ?4 AND status = 'ready' AND created_at > ?3
253 AND substr(key, 1, length(?2)) = ?2
254 ORDER BY created_at DESC LIMIT 1",
255 )
256 .bind(&[
257 scope.repo_id.as_str().into(),
258 prefix.into(),
259 fresh.as_str().into(),
260 version.as_str().into(),
261 ref_scope.as_str().into(),
262 ])?
263 .first::<EntryRow>(None)
264 .await?;
265 if found.is_some() {
266 break 'scopes;
267 }
268 }
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet269 }
270 let Some(entry) = found else { return Ok(Outcome::Ok(None)) };
271 self.db
272 .prepare("UPDATE cache_entries SET last_used_at = ? WHERE id = ?")
273 .bind(&[rfc3339(now).into(), entry.id.as_str().into()])?
274 .run()
275 .await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2276 let blob = a.version.as_ref().and_then(|_| self.download_token("cache", &entry.id, &entry.object, crate::runtime::DOWNLOAD_SECONDS));
277 Ok(Outcome::Ok(Some(CacheHit { key: entry.key, object: entry.object, size: entry.size as u64, created_at: entry.created_at, blob })))
278 }
279
280 /// `cache_upload`.
281 pub async fn cache_upload(&self, a: CacheUploadArgs) -> Result<Outcome<CacheReservation>> {
282 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'283 let scope = self.cache_scope(&job).await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2284 #[derive(Deserialize)]
285 struct Pending {
286 id: String,
287 object: String,
288 number: f64,
289 upload: Option<String>,
290 }
291 let columns = "SELECT id, object, rowid AS number, upload FROM cache_entries";
292 let pending = match (a.number, a.key.as_deref()) {
293 (Some(number), _) => {
294 self.db
295 .prepare(format!("{columns} WHERE rowid = ? AND repo_id = ? AND status = 'pending'"))
296 .bind(&[(number as f64).into(), scope.repo_id.as_str().into()])?
297 .first::<Pending>(None)
298 .await?
299 }
300 (None, Some(key)) => {
301 self.db
Merge branch 'worktree-agent-a3abfcce648e87dca'302 .prepare(format!("{columns} WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'pending'"))
303 .bind(&[scope.repo_id.as_str().into(), scope.save.as_str().into(), key.into(), a.version.clone().unwrap_or_default().into()])?
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2304 .first::<Pending>(None)
305 .await?
306 }
307 (None, None) => None,
308 };
309 let Some(pending) = pending else {
310 return Ok(fail(FailureCode::NotFound, "No upload of that entry is in progress."));
311 };
312 let blob = match &pending.upload {
313 Some(upload) => match self.upload_token("cache", &pending.id, &pending.object, upload) {
314 Some(blob) => Some(blob),
315 None => return Ok(fail(FailureCode::Invalid, "The toolkit's storage is not set up here: the actions service has no ACTIONS_KEY.")),
316 },
317 None => None,
318 };
319 Ok(Outcome::Ok(CacheReservation { id: pending.id, object: pending.object, number: pending.number as u64, upload: pending.upload, blob }))
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet320 }
321
322 /// `cache_reserve`.
323 pub async fn cache_reserve(&self, a: CacheReserveArgs) -> Result<Outcome<CacheReservation>> {
324 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'325 let scope = self.cache_scope(&job).await?;
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet326 if !valid_key(&a.key) {
327 return Ok(fail(FailureCode::Invalid, format!("A cache key is 1 to {MAX_KEY_CHARS} characters, without commas.")));
328 }
329 if a.size > CACHE_MAX_ENTRY_BYTES {
330 return Ok(fail(
331 FailureCode::Invalid,
332 format!("It is {} MB; a cache entry is at most {} MB.", a.size / 1_048_576, CACHE_MAX_ENTRY_BYTES / 1_048_576),
333 ));
334 }
Merge main into rm-agents: mirroring removal, open signups, free-tier limits, the g1t CLI and runner fixes, with the agent removal kept. The CLI's table of operations is regenerated from the API without the agent, customize and AI credit operations; no CLI command used them.335 if let Some(why) = self.storage_refused(&job.namespace, a.size).await? {
336 return Ok(fail(FailureCode::Limit, why));
337 }
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet338 let now = now_ms();
339 let at = rfc3339(now);
Merge branch 'worktree-agent-a3abfcce648e87dca'340 let version = a.version.clone().unwrap_or_default();
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet341 // An upload left unfinished long ago no longer holds its key.
342 self.db
Merge branch 'worktree-agent-a3abfcce648e87dca'343 .prepare(
344 "UPDATE cache_entries SET status = 'expired'
345 WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'pending' AND created_at < ?",
346 )
347 .bind(&[
348 scope.repo_id.as_str().into(),
349 scope.save.as_str().into(),
350 a.key.as_str().into(),
351 version.as_str().into(),
352 rfc3339(now.saturating_sub(PENDING_MS)).into(),
353 ])?
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet354 .run()
355 .await?;
356 let id = new_id("cache", now);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2357 let object = format!("c/{}/{id}", scope.repo_id);
358 #[derive(Deserialize)]
359 struct Inserted {
360 number: f64,
361 }
Merge branch 'worktree-agent-a3abfcce648e87dca'362 // A key is written once in its scope and version, never into
363 // another ref's scope.
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet364 let inserted = self
365 .db
366 .prepare(
Merge branch 'worktree-agent-a3abfcce648e87dca'367 "INSERT INTO cache_entries (id, repo_id, namespace, key, object, size, status, created_at, last_used_at, version, scope)
368 VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'pending', ?7, ?7, ?8, ?9)
369 ON CONFLICT (repo_id, scope, key, version) DO UPDATE SET
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2370 id = ?1, object = ?5, size = ?6, status = 'pending', created_at = ?7, last_used_at = ?7, version = ?8, upload = NULL
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet371 WHERE cache_entries.status = 'expired'
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2372 RETURNING rowid AS number",
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet373 )
374 .bind(&[
375 id.as_str().into(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2376 scope.repo_id.as_str().into(),
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet377 job.namespace.as_str().into(),
378 a.key.as_str().into(),
379 object.as_str().into(),
380 (a.size as f64).into(),
381 at.as_str().into(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2382 version.as_str().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'383 scope.save.as_str().into(),
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet384 ])?
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2385 .first::<Inserted>(None)
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet386 .await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2387 let Some(inserted) = inserted else {
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet388 return Ok(fail(FailureCode::Conflict, "That key is already cached."));
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2389 };
390 Ok(Outcome::Ok(CacheReservation { id, object, number: inserted.number as u64, upload: None, blob: None }))
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet391 }
392
393 /// `cache_commit`: the entry is ready; entries past the quota are
394 /// evicted, restored longest ago first.
395 pub async fn cache_commit(&self, a: CacheCommitArgs) -> Result<Outcome<CacheCommitted>> {
396 let job = check!(self.cache_job(&a.job, &a.token).await?);
397 if a.size > CACHE_MAX_ENTRY_BYTES {
398 return Ok(fail(FailureCode::Invalid, "That entry is larger than a cache entry may be."));
399 }
400 let ready = self
401 .db
402 .prepare("UPDATE cache_entries SET status = 'ready', size = ?, last_used_at = ? WHERE id = ? AND repo_id = ? AND status = 'pending' RETURNING id")
403 .bind(&[(a.size as f64).into(), rfc3339(now_ms()).into(), a.id.as_str().into(), job.repo_id.as_str().into()])?
404 .first::<Value>(None)
405 .await?;
406 if ready.is_none() {
407 return Ok(fail(FailureCode::NotFound, "No upload of that entry is in progress."));
408 }
Merge remote-tracking branch 'origin/main' into workspace-chat409 // What the repository holds, summed: only past the quota are its
410 // entries listed to choose what to evict. A tool that saves an
411 // entry per compiled file (sccache) commits thousands of small
412 // entries a build, and listing them all on each was quadratic.
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet413 #[derive(Deserialize)]
Merge remote-tracking branch 'origin/main' into workspace-chat414 struct Total {
415 bytes: Option<f64>,
416 }
417 let total = self
418 .db
419 .prepare("SELECT SUM(size) AS bytes FROM cache_entries WHERE repo_id = ? AND status = 'ready'")
420 .bind(&[job.repo_id.as_str().into()])?
421 .first::<Total>(None)
422 .await?
423 .and_then(|t| t.bytes)
424 .unwrap_or(0.0);
425 if !over_quota(total as u64, CACHE_REPO_QUOTA_BYTES) {
426 return Ok(Outcome::Ok(CacheCommitted { evicted: Vec::new() }));
427 }
428 #[derive(Deserialize)]
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet429 struct Held {
430 id: String,
431 object: String,
432 size: f64,
433 }
434 let held = self
435 .db
436 .prepare("SELECT id, object, size FROM cache_entries WHERE repo_id = ? AND status = 'ready' ORDER BY last_used_at DESC")
437 .bind(&[job.repo_id.as_str().into()])?
438 .all()
439 .await?
440 .results::<Held>()?;
441 let sizes: Vec<(String, u64)> = held.iter().map(|h| (h.id.clone(), h.size as u64)).collect();
442 let evict = to_evict(&sizes, &a.id, CACHE_REPO_QUOTA_BYTES);
443 let mut evicted = Vec::new();
444 for id in &evict {
445 if let Some(entry) = held.iter().find(|h| &h.id == id) {
446 self.db.prepare("DELETE FROM cache_entries WHERE id = ?").bind(&[id.as_str().into()])?.run().await?;
447 evicted.push(entry.object.clone());
448 }
449 }
450 Ok(Outcome::Ok(CacheCommitted { evicted }))
451 }
452
453 /// `cache_abort`.
454 pub async fn cache_abort(&self, a: CacheAbortArgs) -> Result<Outcome<bool>> {
455 let job = check!(self.cache_job(&a.job, &a.token).await?);
456 self.db
457 .prepare("DELETE FROM cache_entries WHERE id = ? AND repo_id = ? AND status = 'pending'")
458 .bind(&[a.id.as_str().into(), job.repo_id.as_str().into()])?
459 .run()
460 .await?;
461 Ok(Outcome::Ok(true))
462 }
463
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'464 /// `caches`.
465 pub async fn caches(&self, a: CachesArgs) -> Result<Outcome<CacheList>> {
466 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
467 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
468 };
469 let Some(order) = cache_order(a.sort.as_deref(), a.direction.as_deref()) else {
470 return Ok(fail(FailureCode::Invalid, "sort is last_accessed_at, created_at or size_in_bytes, and direction asc or desc."));
471 };
472 let mut filter = "repo_id = ? AND status = 'ready'".to_owned();
473 let mut values: Vec<JsValue> = vec![repo.id.as_str().into()];
474 if let Some(key) = a.key.as_deref().filter(|k| !k.is_empty()) {
475 filter.push_str(" AND substr(key, 1, length(?)) = ?");
476 values.push(key.into());
477 values.push(key.into());
478 }
479 if let Some(git_ref) = a.git_ref.as_deref().map(str::trim).filter(|r| !r.is_empty()) {
480 filter.push_str(" AND scope = ?");
481 values.push(git_ref.into());
482 }
483 #[derive(Deserialize)]
484 struct Count {
485 n: f64,
486 }
487 let total = self
488 .db
489 .prepare(format!("SELECT COUNT(*) AS n FROM cache_entries WHERE {filter}"))
490 .bind(&values)?
491 .first::<Count>(None)
492 .await?
493 .map_or(0, |c| c.n as u64);
494 let per_page = a.per_page.unwrap_or(30).clamp(1, 100);
495 let page = a.page.unwrap_or(1).max(1);
496 let mut paged = values.clone();
497 paged.push(f64::from(per_page).into());
498 paged.push(((f64::from(page) - 1.0) * f64::from(per_page)).into());
499 let rows = self
500 .db
501 .prepare(format!("{LISTED} WHERE {filter} ORDER BY {order} LIMIT ? OFFSET ?"))
502 .bind(&paged)?
503 .all()
504 .await?
505 .results::<ListedRow>()?;
506 Ok(Outcome::Ok(CacheList { total_count: total, actions_caches: rows.iter().map(ListedRow::view).collect() }))
507 }
508
509 /// `cache_usage`: what the entries `cache_commit` counts against the
510 /// quota hold.
511 pub async fn cache_usage(&self, a: CacheUsageArgs) -> Result<Outcome<CacheUsage>> {
512 let Some(repo) = self.visible_repo(&a.repo, &a.viewer).await? else {
513 return Ok(fail(FailureCode::NotFound, "There is no such repository."));
514 };
515 #[derive(Deserialize)]
516 struct Held {
517 bytes: Option<f64>,
518 n: f64,
519 }
520 let held = self
521 .db
522 .prepare("SELECT SUM(size) AS bytes, COUNT(*) AS n FROM cache_entries WHERE repo_id = ? AND status = 'ready'")
523 .bind(&[repo.id.as_str().into()])?
524 .first::<Held>(None)
525 .await?;
526 let (bytes, count) = held.map_or((0, 0), |h| (h.bytes.unwrap_or(0.0) as u64, h.n as u64));
527 Ok(Outcome::Ok(CacheUsage { active_caches_size_in_bytes: bytes, active_caches_count: count, quota_bytes: CACHE_REPO_QUOTA_BYTES }))
528 }
529
530 /// `delete_caches`: the entries go from restores at once, and their
531 /// bytes with them (or at the next sweep, if the bucket refuses).
532 pub async fn delete_caches(&self, a: DeleteCachesArgs) -> Result<Outcome<CacheList>> {
533 let repo = check!(self.may(&a.actor, &a.repo, Capability::Run).await?);
534 let rows = match (a.id, a.key.as_deref().filter(|k| !k.is_empty())) {
535 (Some(id), _) => {
536 self.db
537 .prepare(format!("{LISTED} WHERE rowid = ? AND repo_id = ? AND status = 'ready'"))
538 .bind(&[(id as f64).into(), repo.id.as_str().into()])?
539 .all()
540 .await?
541 .results::<ListedRow>()?
542 }
543 (None, Some(key)) => {
544 let mut filter = "key = ? AND repo_id = ? AND status = 'ready'".to_owned();
545 let mut values: Vec<JsValue> = vec![key.into(), repo.id.as_str().into()];
546 if let Some(git_ref) = a.git_ref.as_deref().map(str::trim).filter(|r| !r.is_empty()) {
547 filter.push_str(" AND scope = ?");
548 values.push(git_ref.into());
549 }
550 self.db.prepare(format!("{LISTED} WHERE {filter} ORDER BY created_at DESC")).bind(&values)?.all().await?.results::<ListedRow>()?
551 }
552 (None, None) => return Ok(fail(FailureCode::Invalid, "Name the cache entry by its id, or the entries to delete by their key.")),
553 };
554 if rows.is_empty() {
555 return Ok(fail(FailureCode::NotFound, "No cache entry matches."));
556 }
557 for row in &rows {
558 self.db.prepare("UPDATE cache_entries SET status = 'expired' WHERE id = ?").bind(&[row.id.as_str().into()])?.run().await?;
559 }
560 let gone: Vec<(String, String)> = rows.iter().map(|r| (r.id.clone(), r.object.clone())).collect();
561 self.forget_cache(&gone).await?;
562 Ok(Outcome::Ok(CacheList { total_count: rows.len() as u64, actions_caches: rows.iter().map(ListedRow::view).collect() }))
563 }
564
565 /// Deletes expired entries' objects, then their rows: whether it could.
566 /// If the bucket refuses, the rows stay expired for the next sweep.
567 async fn forget_cache(&self, gone: &[(String, String)]) -> Result<bool> {
568 let Some(bucket) = &self.cache else { return Ok(false) };
569 for chunk in gone.chunks(100) {
570 if let Err(error) = bucket.delete_multiple(chunk.iter().map(|(_, object)| object.as_str()).collect()).await {
571 worker::console_error!("actions: cache objects not deleted: {error}");
572 return Ok(false);
573 }
574 for (id, _) in chunk {
575 self.db.prepare("DELETE FROM cache_entries WHERE id = ?").bind(&[id.as_str().into()])?.run().await?;
576 }
577 }
578 Ok(true)
579 }
580
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet581 /// Hourly: deletes entries unused for a week, saved too long ago, or
582 /// left unfinished, and their objects; then what each workspace's
583 /// cache holds today, and this month's storage, for billing.
584 pub async fn sweep_cache(&self, now: u64) -> Result<()> {
585 let at = |ms: u64| rfc3339(now.saturating_sub(ms));
586 self.db
587 .prepare(
588 "UPDATE cache_entries SET status = 'expired'
589 WHERE (status = 'ready' AND (last_used_at < ?1 OR created_at < ?2)) OR (status = 'pending' AND created_at < ?3)",
590 )
591 .bind(&[at(CACHE_UNUSED_DAYS * DAY_MS).into(), at(CACHE_MAX_AGE_DAYS * DAY_MS).into(), at(PENDING_MS).into()])?
592 .run()
593 .await?;
594 #[derive(Deserialize)]
595 struct Gone {
596 id: String,
597 object: String,
598 }
599 let gone = self
600 .db
601 .prepare("SELECT id, object FROM cache_entries WHERE status = 'expired' LIMIT 500")
602 .all()
603 .await?
604 .results::<Gone>()?;
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'605 let gone: Vec<(String, String)> = gone.into_iter().map(|g| (g.id, g.object)).collect();
606 if self.cache.is_some() && !self.forget_cache(&gone).await? {
607 return Ok(());
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet608 }
609 self.measure_cache(now).await
610 }
611
612 /// What each workspace's cache holds today, and this month's storage so
613 /// far reported to billing, which charges it to workspaces on the plan
614 /// once the month is over.
615 async fn measure_cache(&self, now: u64) -> Result<()> {
616 #[derive(Deserialize)]
617 struct Held {
618 namespace: String,
619 bytes: f64,
620 }
621 let today = rfc3339(now);
622 let (day, month) = (&today[..10], &today[..7]);
623 let held = self
624 .db
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2625 // Artifacts are kept in the same bucket, at the same price.
626 .prepare(
627 "SELECT namespace, SUM(size) AS bytes FROM (
628 SELECT namespace, size FROM cache_entries WHERE status = 'ready'
629 UNION ALL SELECT namespace, size FROM artifacts WHERE status = 'ready'
630 ) GROUP BY namespace",
631 )
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet632 .all()
633 .await?
634 .results::<Held>()?;
635 for workspace in &held {
636 self.db
637 .prepare(
638 "INSERT INTO cache_days (namespace, day, bytes) VALUES (?1, ?2, ?3)
639 ON CONFLICT (namespace, day) DO UPDATE SET bytes = max(bytes, ?3)",
640 )
641 .bind(&[workspace.namespace.as_str().into(), day.into(), workspace.bytes.into()])?
642 .run()
643 .await?;
644 }
645 #[derive(Deserialize)]
646 struct Day {
647 namespace: String,
648 bytes: f64,
649 }
650 let days = self
651 .db
652 .prepare("SELECT namespace, bytes FROM cache_days WHERE substr(day, 1, 7) = ?")
653 .bind(&[month.into()])?
654 .all()
655 .await?
656 .results::<Day>()?;
657 let mut by_workspace: std::collections::BTreeMap<String, Vec<u64>> = std::collections::BTreeMap::new();
658 for d in days {
659 by_workspace.entry(d.namespace).or_default().push(d.bytes as u64);
660 }
661 for (workspace, days) in by_workspace {
662 let months = gb_months(&days);
663 let cost = storage_cost(months);
664 if cost <= 0 {
665 continue;
666 }
667 let noted: Result<bool> = g1t_kit::call(
668 &self.billing,
669 "note_pending",
670 &NotePendingArgs { workspace: workspace.clone(), source: "cache".to_owned(), cost_micros: cost, detail: Some(format!("{months:.2} GB-months")) },
671 )
672 .await;
673 if let Err(error) = noted {
674 worker::console_error!("actions: cache storage not reported for {workspace}: {error}");
675 }
676 }
677 Ok(())
678 }
679}
680
681#[cfg(test)]
682mod tests {
683 use super::*;
684
685 fn entries(sizes: &[(&str, u64)]) -> Vec<(String, u64)> {
686 sizes.iter().map(|(id, size)| ((*id).to_owned(), *size)).collect()
687 }
688
689 #[test]
Merge branch 'worktree-agent-a3abfcce648e87dca'690 fn a_run_restores_from_its_ref_then_its_base_then_the_default_branch() {
691 let (restore, save) = scopes("refs/heads/feature", None, "main", true);
692 assert_eq!(restore, ["refs/heads/feature", "refs/heads/main"]);
693 assert_eq!(save, "refs/heads/feature");
694 // A pull request: its own merge ref, the branch it merges into, the default.
695 let (restore, save) = scopes("refs/pull/7/merge", Some("release/1.x"), "main", true);
696 assert_eq!(restore, ["refs/pull/7/merge", "refs/heads/release/1.x", "refs/heads/main"]);
697 assert_eq!(save, "refs/pull/7/merge");
698 // The default branch reads only its own.
699 let (restore, _) = scopes("refs/heads/main", Some("main"), "main", true);
700 assert_eq!(restore, ["refs/heads/main"]);
701 // From outside: it saves where nothing else reads.
702 let (restore, save) = scopes("refs/pull/9/merge", Some("main"), "main", false);
703 assert_eq!(save, "untrusted:refs/pull/9/merge");
704 assert_eq!(restore, ["untrusted:refs/pull/9/merge", "refs/heads/main"]);
705 for trusted_ref in ["refs/heads/main", "refs/pull/9/merge", "refs/heads/feature"] {
706 let (others, _) = scopes(trusted_ref, Some("main"), "main", true);
707 assert!(!others.contains(&save), "{trusted_ref} must never read an untrusted entry");
708 }
709 }
710
711 #[test]
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet712 fn eviction_takes_the_entries_restored_longest_ago() {
713 // Newest use first.
714 let held = entries(&[("new", 4), ("b", 3), ("c", 3), ("old", 2)]);
715 assert_eq!(to_evict(&held, "new", 12), Vec::<String>::new());
716 assert_eq!(to_evict(&held, "new", 10), ["old"]);
717 assert_eq!(to_evict(&held, "new", 7), ["old", "c"]);
718 // The entry just saved stays, even when it alone is past the quota.
719 assert_eq!(to_evict(&entries(&[("big", 20), ("a", 1)]), "big", 10), ["a"]);
Merge remote-tracking branch 'origin/main' into workspace-chat720 // Entries are listed only when the sum is over: at or under the
721 // quota, nothing would be evicted.
722 for (held, quota) in [(entries(&[("a", 4), ("b", 6)]), 10), (entries(&[("a", 1)]), 12)] {
723 let total = held.iter().map(|(_, size)| size).sum();
724 assert!(!over_quota(total, quota));
725 assert!(to_evict(&held, "a", quota).is_empty());
726 }
727 assert!(over_quota(11, 10));
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet728 }
729
730 #[test]
Merge branch 'worktree-agent-a63425cb2d4935c5f'731 fn a_restore_tries_its_key_as_a_prefix_before_its_restore_keys() {
732 let restore = vec!["cargo-Linux-1.99.0-".to_owned(), " ".to_owned(), "cargo-".to_owned(), "cargo-".to_owned()];
733 assert_eq!(prefixes("cargo-Linux-1.99.0-abc", &restore), ["cargo-Linux-1.99.0-abc", "cargo-Linux-1.99.0-", "cargo-"]);
734 assert_eq!(prefixes("k", &[]), ["k"]);
735 }
736
737 #[test]
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'738 fn entries_are_listed_in_github_s_orders() {
739 assert_eq!(cache_order(None, None).unwrap(), "last_used_at DESC, rowid DESC");
740 assert_eq!(cache_order(Some("created_at"), Some("asc")).unwrap(), "created_at ASC, rowid ASC");
741 assert_eq!(cache_order(Some("size_in_bytes"), Some(" ")).unwrap(), "size DESC, rowid DESC");
742 // Anything else is refused rather than put into the query.
743 assert_eq!(cache_order(Some("size; DROP TABLE cache_entries"), None), None);
744 assert_eq!(cache_order(Some("created_at"), Some("sideways")), None);
745 }
746
747 #[test]
748 fn a_listed_entry_is_shown_as_github_shows_one() {
749 let row = ListedRow {
750 id: "cache_1".into(),
751 number: 42.0,
752 key: "cargo-test-Linux-1.99.0-abc".into(),
753 scope: "refs/heads/main".into(),
754 version: String::new(),
755 object: "c/repo_1/cache_1".into(),
756 size: 1_687_000_000.0,
757 created_at: "2026-10-12T10:00:00.000Z".into(),
758 last_used_at: "2026-10-12T11:00:00.000Z".into(),
759 };
760 let shown = serde_json::to_value(row.view()).unwrap();
761 assert_eq!(shown["id"], 42);
762 assert_eq!(shown["ref"], "refs/heads/main");
763 assert_eq!(shown["size_in_bytes"], 1_687_000_000u64);
764 assert_eq!(shown["last_accessed_at"], "2026-10-12T11:00:00.000Z");
765 assert!(shown.get("object").is_none(), "where its bytes are is not shown");
766 }
767
768 #[test]
Code button: the SSH tab shows the address it will have, disabled, with why it is not on yet769 fn keys_are_checked() {
770 assert!(valid_key("cargo-Linux-abc123"));
771 assert!(!valid_key(""));
772 assert!(!valid_key("a,b"));
773 assert!(!valid_key(&"k".repeat(513)));
774 }
775
776 #[test]
777 fn storage_is_charged_by_the_gb_month_at_r2s_price() {
778 // 10 GB held for 30 days is 10 GB-months: $0.15.
779 let month = vec![10_000_000_000u64; 30];
780 assert!((gb_months(&month) - 10.0).abs() < 1e-9);
781 assert_eq!(storage_cost(gb_months(&month)), 150_000);
782 // A day of 1 GB: a thirtieth of a GB-month, rounded up.
783 assert_eq!(storage_cost(gb_months(&[1_000_000_000])), 500);
784 assert_eq!(storage_cost(0.0), 0);
785 }
786}

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