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
794 linesCodeBlameRaw
1//! 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//!
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
16//! starts with its key, else with each of its restore keys in order
17//! (`prefixes`), as GitHub's cache does.
18//! - 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
23//! 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.
26//!
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//!
31//! 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.
35
36use g1t_contracts::FailureCode;
37use g1t_contracts::Outcome;
38use g1t_contracts::access::Capability;
39use g1t_contracts::actions::{
40 CACHE_MAX_AGE_DAYS, CACHE_MAX_ENTRY_BYTES, CACHE_MICROS_PER_GB_MONTH, CACHE_REPO_QUOTA_BYTES, CACHE_UNUSED_DAYS,
41 CacheAbortArgs, CacheCommitArgs, CacheCommitted, CacheEntry, CacheHit, CacheList, CacheLookupArgs, CacheReservation,
42 CacheReserveArgs, CacheUploadArgs, CacheUsage, CacheUsageArgs, CachesArgs, DeleteCachesArgs,
43};
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;
51use worker::wasm_bindgen::JsValue;
52
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,
67 created_at: String,
68}
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
97/// 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
103/// 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
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
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
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.
177pub(crate) struct CacheScope {
178 pub repo_id: String,
179 pub restore: Vec<String>,
180 pub save: String,
181}
182
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)
197}
198
199impl Actions {
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
213 /// A job by its own token or its runtime token (runtime.rs).
214 async fn cache_job(&self, job: &str, token: &str) -> Result<Outcome<crate::plan::JobRow>> {
215 self.job_for_credential(job, token).await
216 }
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?);
221 let scope = self.cache_scope(&job).await?;
222 let now = now_ms();
223 let fresh = rfc3339(now.saturating_sub(CACHE_MAX_AGE_DAYS * DAY_MS));
224 // Found only by the same version: runners from before it was sent
225 // send none, and find only entries saved without one.
226 let version = a.version.clone().unwrap_or_default();
227 let mut found = None;
228 'scopes: for ref_scope in &scope.restore {
229 found = self
230 .db
231 .prepare(
232 "SELECT id, key, object, size, created_at FROM cache_entries
233 WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'ready' AND created_at > ?",
234 )
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 ])?
242 .first::<EntryRow>(None)
243 .await?;
244 if found.is_some() {
245 break;
246 }
247 for prefix in prefixes(&a.key, &a.restore) {
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 }
269 }
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?;
276 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?);
283 let scope = self.cache_scope(&job).await?;
284 #[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
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()])?
304 .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 }))
320 }
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?);
325 let scope = self.cache_scope(&job).await?;
326 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 }
335 if let Some(why) = self.storage_refused(&job.namespace, a.size).await? {
336 return Ok(fail(FailureCode::Limit, why));
337 }
338 let now = now_ms();
339 let at = rfc3339(now);
340 let version = a.version.clone().unwrap_or_default();
341 // An upload left unfinished long ago no longer holds its key.
342 self.db
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 ])?
354 .run()
355 .await?;
356 let id = new_id("cache", now);
357 let object = format!("c/{}/{id}", scope.repo_id);
358 #[derive(Deserialize)]
359 struct Inserted {
360 number: f64,
361 }
362 // A key is written once in its scope and version, never into
363 // another ref's scope.
364 let inserted = self
365 .db
366 .prepare(
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
370 id = ?1, object = ?5, size = ?6, status = 'pending', created_at = ?7, last_used_at = ?7, version = ?8, upload = NULL
371 WHERE cache_entries.status = 'expired'
372 RETURNING rowid AS number",
373 )
374 .bind(&[
375 id.as_str().into(),
376 scope.repo_id.as_str().into(),
377 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(),
382 version.as_str().into(),
383 scope.save.as_str().into(),
384 ])?
385 .first::<Inserted>(None)
386 .await?;
387 let Some(inserted) = inserted else {
388 return Ok(fail(FailureCode::Conflict, "That key is already cached."));
389 };
390 Ok(Outcome::Ok(CacheReservation { id, object, number: inserted.number as u64, upload: None, blob: None }))
391 }
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 }
409 // 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.
413 #[derive(Deserialize)]
414 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)]
429 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
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
581 /// 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>()?;
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(());
608 }
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
625 // 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 )
632 .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 {
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 },
679 )
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]
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]
720 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"]);
728 // 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));
736 }
737
738 #[test]
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]
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]
777 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}