Skip to content
Sign up

flagon-io/g1t

public
0
794 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.

Mission control: a place to ask g1t, shown and not on yet, with the actions under it1//! 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.
Mission control: a place to ask g1t, shown and not on yet, with the actions under it18//! - 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.
Mission control: a place to ask g1t, shown and not on yet, with the actions under it26//!
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.
Mission control: a place to ask g1t, shown and not on yet, with the actions under it35
36use g1t_contracts::FailureCode;
37use g1t_contracts::Outcome;
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'38use g1t_contracts::access::Capability;
Mission control: a place to ask g1t, shown and not on yet, with the actions under it39use 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,
Mission control: a place to ask g1t, shown and not on yet, with the actions under it43};
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;
Mission control: a place to ask g1t, shown and not on yet, with the actions under it52
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,
Mission control: a place to ask g1t, shown and not on yet, with the actions under it68}
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
Mission control: a place to ask g1t, shown and not on yet, with the actions under it103/// 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
Mission control: a place to ask g1t, shown and not on yet, with the actions under it199impl 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).
Mission control: a place to ask g1t, shown and not on yet, with the actions under it214 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
Mission control: a place to ask g1t, shown and not on yet, with the actions under it216 }
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?;
Mission control: a place to ask g1t, shown and not on yet, with the actions under it222 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 {
Mission control: a place to ask g1t, shown and not on yet, with the actions under it229 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 > ?",
Mission control: a place to ask g1t, shown and not on yet, with the actions under it234 )
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 ])?
Mission control: a place to ask g1t, shown and not on yet, with the actions under it242 .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 }
Mission control: a place to ask g1t, shown and not on yet, with the actions under it269 }
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 }))
Mission control: a place to ask g1t, shown and not on yet, with the actions under it320 }
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?;
Mission control: a place to ask g1t, shown and not on yet, with the actions under it326 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 }
Mission control: a place to ask g1t, shown and not on yet, with the actions under it338 let now = now_ms();
339 let at = rfc3339(now);
Merge branch 'worktree-agent-a3abfcce648e87dca'340 let version = a.version.clone().unwrap_or_default();
Mission control: a place to ask g1t, shown and not on yet, with the actions under it341 // 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 ])?
Mission control: a place to ask g1t, shown and not on yet, with the actions under it354 .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.
Mission control: a place to ask g1t, shown and not on yet, with the actions under it364 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
Mission control: a place to ask g1t, shown and not on yet, with the actions under it371 WHERE cache_entries.status = 'expired'
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2372 RETURNING rowid AS number",
Mission control: a place to ask g1t, shown and not on yet, with the actions under it373 )
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(),
Mission control: a place to ask g1t, shown and not on yet, with the actions under it377 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(),
Mission control: a place to ask g1t, shown and not on yet, with the actions under it384 ])?
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2385 .first::<Inserted>(None)
Mission control: a place to ask g1t, shown and not on yet, with the actions under it386 .await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2387 let Some(inserted) = inserted else {
Mission control: a place to ask g1t, shown and not on yet, with the actions under it388 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 }))
Mission control: a place to ask g1t, shown and not on yet, with the actions under it391 }
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.
Mission control: a place to ask g1t, shown and not on yet, with the actions under it413 #[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)]
Mission control: a place to ask g1t, shown and not on yet, with the actions under it429 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
Mission control: a place to ask g1t, shown and not on yet, with the actions under it581 /// 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(());
Mission control: a place to ask g1t, shown and not on yet, with the actions under it608 }
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 )
Mission control: a place to ask g1t, shown and not on yet, with the actions under it632 .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",
Billing is ready to take real money, and stays on Stripe's test key until the switch. The g1t plan is $21 a month, all-in, with $10 of usage included: the card processing fee is part of the price, and no payment carries a fee line any more (old fee rows still read as they were). Card, billing address, tax IDs, invoices and cancelling are on Stripe's billing page (Manage billing), whose configuration billing keeps in step; Checkout asks for the billing address and a tax ID; g1t's Billing page keeps usage, the spend limit, budget alerts and prepaying. Members other than owners and billing managers no longer see billing details or invoices, from the page or the API.670 &NotePendingArgs {
671 workspace: workspace.clone(),
672 source: "cache".to_owned(),
673 cost_micros: cost,
674 detail: Some(format!("{months:.2} GB-months")),
675 // The month these days are in, not the one the report
676 // arrives in.
677 month: Some(month.to_owned()),
678 },
Mission control: a place to ask g1t, shown and not on yet, with the actions under it679 )
680 .await;
681 if let Err(error) = noted {
682 worker::console_error!("actions: cache storage not reported for {workspace}: {error}");
683 }
684 }
685 Ok(())
686 }
687}
688
689#[cfg(test)]
690mod tests {
691 use super::*;
692
693 fn entries(sizes: &[(&str, u64)]) -> Vec<(String, u64)> {
694 sizes.iter().map(|(id, size)| ((*id).to_owned(), *size)).collect()
695 }
696
697 #[test]
Merge branch 'worktree-agent-a3abfcce648e87dca'698 fn a_run_restores_from_its_ref_then_its_base_then_the_default_branch() {
699 let (restore, save) = scopes("refs/heads/feature", None, "main", true);
700 assert_eq!(restore, ["refs/heads/feature", "refs/heads/main"]);
701 assert_eq!(save, "refs/heads/feature");
702 // A pull request: its own merge ref, the branch it merges into, the default.
703 let (restore, save) = scopes("refs/pull/7/merge", Some("release/1.x"), "main", true);
704 assert_eq!(restore, ["refs/pull/7/merge", "refs/heads/release/1.x", "refs/heads/main"]);
705 assert_eq!(save, "refs/pull/7/merge");
706 // The default branch reads only its own.
707 let (restore, _) = scopes("refs/heads/main", Some("main"), "main", true);
708 assert_eq!(restore, ["refs/heads/main"]);
709 // From outside: it saves where nothing else reads.
710 let (restore, save) = scopes("refs/pull/9/merge", Some("main"), "main", false);
711 assert_eq!(save, "untrusted:refs/pull/9/merge");
712 assert_eq!(restore, ["untrusted:refs/pull/9/merge", "refs/heads/main"]);
713 for trusted_ref in ["refs/heads/main", "refs/pull/9/merge", "refs/heads/feature"] {
714 let (others, _) = scopes(trusted_ref, Some("main"), "main", true);
715 assert!(!others.contains(&save), "{trusted_ref} must never read an untrusted entry");
716 }
717 }
718
719 #[test]
Mission control: a place to ask g1t, shown and not on yet, with the actions under it720 fn eviction_takes_the_entries_restored_longest_ago() {
721 // Newest use first.
722 let held = entries(&[("new", 4), ("b", 3), ("c", 3), ("old", 2)]);
723 assert_eq!(to_evict(&held, "new", 12), Vec::<String>::new());
724 assert_eq!(to_evict(&held, "new", 10), ["old"]);
725 assert_eq!(to_evict(&held, "new", 7), ["old", "c"]);
726 // The entry just saved stays, even when it alone is past the quota.
727 assert_eq!(to_evict(&entries(&[("big", 20), ("a", 1)]), "big", 10), ["a"]);
Merge remote-tracking branch 'origin/main' into workspace-chat728 // Entries are listed only when the sum is over: at or under the
729 // quota, nothing would be evicted.
730 for (held, quota) in [(entries(&[("a", 4), ("b", 6)]), 10), (entries(&[("a", 1)]), 12)] {
731 let total = held.iter().map(|(_, size)| size).sum();
732 assert!(!over_quota(total, quota));
733 assert!(to_evict(&held, "a", quota).is_empty());
734 }
735 assert!(over_quota(11, 10));
Mission control: a place to ask g1t, shown and not on yet, with the actions under it736 }
737
738 #[test]
Merge branch 'worktree-agent-a63425cb2d4935c5f'739 fn a_restore_tries_its_key_as_a_prefix_before_its_restore_keys() {
740 let restore = vec!["cargo-Linux-1.99.0-".to_owned(), " ".to_owned(), "cargo-".to_owned(), "cargo-".to_owned()];
741 assert_eq!(prefixes("cargo-Linux-1.99.0-abc", &restore), ["cargo-Linux-1.99.0-abc", "cargo-Linux-1.99.0-", "cargo-"]);
742 assert_eq!(prefixes("k", &[]), ["k"]);
743 }
744
745 #[test]
Merge branch 'worktree-agent-a3b9b1f730f55b5bf'746 fn entries_are_listed_in_github_s_orders() {
747 assert_eq!(cache_order(None, None).unwrap(), "last_used_at DESC, rowid DESC");
748 assert_eq!(cache_order(Some("created_at"), Some("asc")).unwrap(), "created_at ASC, rowid ASC");
749 assert_eq!(cache_order(Some("size_in_bytes"), Some(" ")).unwrap(), "size DESC, rowid DESC");
750 // Anything else is refused rather than put into the query.
751 assert_eq!(cache_order(Some("size; DROP TABLE cache_entries"), None), None);
752 assert_eq!(cache_order(Some("created_at"), Some("sideways")), None);
753 }
754
755 #[test]
756 fn a_listed_entry_is_shown_as_github_shows_one() {
757 let row = ListedRow {
758 id: "cache_1".into(),
759 number: 42.0,
760 key: "cargo-test-Linux-1.99.0-abc".into(),
761 scope: "refs/heads/main".into(),
762 version: String::new(),
763 object: "c/repo_1/cache_1".into(),
764 size: 1_687_000_000.0,
765 created_at: "2026-10-12T10:00:00.000Z".into(),
766 last_used_at: "2026-10-12T11:00:00.000Z".into(),
767 };
768 let shown = serde_json::to_value(row.view()).unwrap();
769 assert_eq!(shown["id"], 42);
770 assert_eq!(shown["ref"], "refs/heads/main");
771 assert_eq!(shown["size_in_bytes"], 1_687_000_000u64);
772 assert_eq!(shown["last_accessed_at"], "2026-10-12T11:00:00.000Z");
773 assert!(shown.get("object").is_none(), "where its bytes are is not shown");
774 }
775
776 #[test]
Mission control: a place to ask g1t, shown and not on yet, with the actions under it777 fn keys_are_checked() {
778 assert!(valid_key("cargo-Linux-abc123"));
779 assert!(!valid_key(""));
780 assert!(!valid_key("a,b"));
781 assert!(!valid_key(&"k".repeat(513)));
782 }
783
784 #[test]
785 fn storage_is_charged_by_the_gb_month_at_r2s_price() {
786 // 10 GB held for 30 days is 10 GB-months: $0.15.
787 let month = vec![10_000_000_000u64; 30];
788 assert!((gb_months(&month) - 10.0).abs() < 1e-9);
789 assert_eq!(storage_cost(gb_months(&month)), 150_000);
790 // A day of 1 GB: a thirtieth of a GB-month, rounded up.
791 assert_eq!(storage_cost(gb_months(&[1_000_000_000])), 500);
792 assert_eq!(storage_cost(0.0), 0);
793 }
794}

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