Skip to content
Sign up

flagon-io/g1t

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

Merge branch 'worktree-agent-a6a121745e81f639f'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//!
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.
Merge branch 'worktree-agent-a6a121745e81f639f'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
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.
Merge branch 'worktree-agent-a6a121745e81f639f'26//!
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R227//! The API calls these with the job's token, or its runtime token for the
28//! toolkit's protocols (runtime.rs), which is checked here. An entry the
29//! toolkit saves has the toolkit's version, and is found only by the same
30//! version; g1t's own `actions/cache` saves and finds entries without one.
Merge branch 'worktree-agent-a6a121745e81f639f'31
32use g1t_contracts::FailureCode;
33use g1t_contracts::Outcome;
34use g1t_contracts::actions::{
35 CACHE_MAX_AGE_DAYS, CACHE_MAX_ENTRY_BYTES, CACHE_MICROS_PER_GB_MONTH, CACHE_REPO_QUOTA_BYTES, CACHE_UNUSED_DAYS,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R236 CacheAbortArgs, CacheCommitArgs, CacheCommitted, CacheHit, CacheLookupArgs, CacheReservation, CacheReserveArgs, CacheUploadArgs,
Merge branch 'worktree-agent-a6a121745e81f639f'37};
38use g1t_contracts::billing::NotePendingArgs;
39use g1t_contracts::new_id;
40use g1t_contracts::time::rfc3339;
41use g1t_kit::now_ms;
42use serde::Deserialize;
43use serde_json::Value;
44use worker::Result;
45
46use crate::{Actions, check, fail};
47
48const DAY_MS: u64 = 24 * 60 * 60 * 1000;
49/// An upload not finished after this long is given up.
50const PENDING_MS: u64 = 6 * 60 * 60 * 1000;
51/// A GB, as storage is billed.
52const GB: f64 = 1_000_000_000.0;
53
54#[derive(Debug, Deserialize)]
55struct EntryRow {
56 id: String,
57 key: String,
58 object: String,
59 size: f64,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R260 created_at: String,
Merge branch 'worktree-agent-a6a121745e81f639f'61}
62
63/// The longest key: GitHub's limit.
64const MAX_KEY_CHARS: usize = 512;
65
66/// Whether a key can be kept: 1 to 512 characters, no commas (GitHub's rule).
67pub(crate) fn valid_key(key: &str) -> bool {
68 !key.is_empty() && key.chars().count() <= MAX_KEY_CHARS && !key.contains(',')
69}
70
71/// Which of `entries` (id, size, newest use first) to evict so that the
72/// repository holds at most `quota` with `keep` (just saved) among them.
73/// The entry just saved is never evicted.
74pub(crate) fn to_evict(entries: &[(String, u64)], keep: &str, quota: u64) -> Vec<String> {
75 let mut total: u64 = entries.iter().map(|(_, size)| size).sum();
76 let mut out = Vec::new();
77 for (id, size) in entries.iter().rev() {
78 if total <= quota {
79 break;
80 }
81 if id == keep {
82 continue;
83 }
84 total = total.saturating_sub(*size);
85 out.push(id.clone());
86 }
87 out
88}
89
Merge remote-tracking branch 'origin/main' into workspace-chat90/// Whether a repository holding `total` bytes must evict to stay within
91/// `quota`: what `to_evict` would take something from.
92pub(crate) fn over_quota(total: u64, quota: u64) -> bool {
93 total > quota
94}
95
Merge branch 'worktree-agent-a6a121745e81f639f'96/// A month's GB-months from the bytes held each day so far: each day's
97/// bytes over 30 days.
98pub(crate) fn gb_months(days: &[u64]) -> f64 {
99 days.iter().map(|bytes| *bytes as f64).sum::<f64>() / GB / 30.0
100}
101
102/// What `gb_months` of cache cost g1t, in millionths of a dollar.
103pub(crate) fn storage_cost(gb_months: f64) -> i64 {
104 (gb_months * CACHE_MICROS_PER_GB_MONTH as f64).ceil() as i64
105}
106
Merge branch 'worktree-agent-a63425cb2d4935c5f'107/// The prefixes a restore tries in each scope once its key has no exact
108/// match, in order: the key itself, then its restore keys. Blank ones and
109/// repeats are dropped.
110pub(crate) fn prefixes<'a>(key: &'a str, restore: &'a [String]) -> Vec<&'a str> {
111 let mut out: Vec<&str> = Vec::new();
112 for prefix in std::iter::once(key).chain(restore.iter().map(String::as_str)).map(str::trim) {
113 if !prefix.is_empty() && !out.contains(&prefix) {
114 out.push(prefix);
115 }
116 }
117 out
118}
119
Merge branch 'worktree-agent-a3abfcce648e87dca'120/// Where a job's cache entries are found and saved: its repository, and in
121/// 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 R2122pub(crate) struct CacheScope {
123 pub repo_id: String,
Merge branch 'worktree-agent-a3abfcce648e87dca'124 pub restore: Vec<String>,
125 pub save: String,
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2126}
127
Merge branch 'worktree-agent-a3abfcce648e87dca'128/// The scopes a run's jobs restore from, in order, and the one they save
129/// to: its own ref, then its pull request's base branch, then the default
130/// branch. A run that is not trusted (a pull request from outside) saves to
131/// a scope of its own that no other ref reads.
132pub(crate) fn scopes(git_ref: &str, base_ref: Option<&str>, default_branch: &str, trusted: bool) -> (Vec<String>, String) {
133 let own = if trusted { git_ref.to_owned() } else { format!("untrusted:{git_ref}") };
134 let mut restore = vec![own.clone()];
135 let base = base_ref.filter(|base| !base.is_empty()).map(|base| format!("refs/heads/{}", base.trim_start_matches("refs/heads/")));
136 for scope in base.into_iter().chain(std::iter::once(format!("refs/heads/{default_branch}"))) {
137 if !restore.contains(&scope) {
138 restore.push(scope);
139 }
140 }
141 (restore, own)
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2142}
143
Merge branch 'worktree-agent-a6a121745e81f639f'144impl Actions {
Merge branch 'worktree-agent-a3abfcce648e87dca'145 /// Where a job's entries are found and saved (`scopes`), for g1t's own
146 /// `actions/cache` and the toolkit's protocols alike.
147 pub(crate) async fn cache_scope(&self, job: &crate::plan::JobRow) -> Result<CacheScope> {
148 let (restore, save) = match self.run_row(&job.run_id).await? {
149 Some(run) => {
150 let info = run.info();
151 scopes(&run.git_ref, info.base_ref.as_deref(), &info.default_branch, run.trusted != 0)
152 }
153 None => (Vec::new(), format!("untrusted:{}", job.run_id)),
154 };
155 Ok(CacheScope { repo_id: job.repo_id.clone(), restore, save })
156 }
157
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2158 /// A job by its own token or its runtime token (runtime.rs).
Merge branch 'worktree-agent-a6a121745e81f639f'159 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 R2160 self.job_for_credential(job, token).await
Merge branch 'worktree-agent-a6a121745e81f639f'161 }
162
163 /// `cache_lookup`.
164 pub async fn cache_lookup(&self, a: CacheLookupArgs) -> Result<Outcome<Option<CacheHit>>> {
165 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'166 let scope = self.cache_scope(&job).await?;
Merge branch 'worktree-agent-a6a121745e81f639f'167 let now = now_ms();
168 let fresh = rfc3339(now.saturating_sub(CACHE_MAX_AGE_DAYS * DAY_MS));
Merge branch 'worktree-agent-a3abfcce648e87dca'169 // Found only by the same version: runners from before it was sent
170 // send none, and find only entries saved without one.
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2171 let version = a.version.clone().unwrap_or_default();
Merge branch 'worktree-agent-a3abfcce648e87dca'172 let mut found = None;
173 'scopes: for ref_scope in &scope.restore {
Merge branch 'worktree-agent-a6a121745e81f639f'174 found = self
175 .db
176 .prepare(
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2177 "SELECT id, key, object, size, created_at FROM cache_entries
Merge branch 'worktree-agent-a3abfcce648e87dca'178 WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'ready' AND created_at > ?",
Merge branch 'worktree-agent-a6a121745e81f639f'179 )
Merge branch 'worktree-agent-a3abfcce648e87dca'180 .bind(&[
181 scope.repo_id.as_str().into(),
182 ref_scope.as_str().into(),
183 a.key.as_str().into(),
184 version.as_str().into(),
185 fresh.as_str().into(),
186 ])?
Merge branch 'worktree-agent-a6a121745e81f639f'187 .first::<EntryRow>(None)
188 .await?;
Merge branch 'worktree-agent-a3abfcce648e87dca'189 if found.is_some() {
190 break;
191 }
Merge branch 'worktree-agent-a63425cb2d4935c5f'192 for prefix in prefixes(&a.key, &a.restore) {
Merge branch 'worktree-agent-a3abfcce648e87dca'193 found = self
194 .db
195 .prepare(
196 "SELECT id, key, object, size, created_at FROM cache_entries
197 WHERE repo_id = ?1 AND scope = ?5 AND version = ?4 AND status = 'ready' AND created_at > ?3
198 AND substr(key, 1, length(?2)) = ?2
199 ORDER BY created_at DESC LIMIT 1",
200 )
201 .bind(&[
202 scope.repo_id.as_str().into(),
203 prefix.into(),
204 fresh.as_str().into(),
205 version.as_str().into(),
206 ref_scope.as_str().into(),
207 ])?
208 .first::<EntryRow>(None)
209 .await?;
210 if found.is_some() {
211 break 'scopes;
212 }
213 }
Merge branch 'worktree-agent-a6a121745e81f639f'214 }
215 let Some(entry) = found else { return Ok(Outcome::Ok(None)) };
216 self.db
217 .prepare("UPDATE cache_entries SET last_used_at = ? WHERE id = ?")
218 .bind(&[rfc3339(now).into(), entry.id.as_str().into()])?
219 .run()
220 .await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2221 let blob = a.version.as_ref().and_then(|_| self.download_token("cache", &entry.id, &entry.object, crate::runtime::DOWNLOAD_SECONDS));
222 Ok(Outcome::Ok(Some(CacheHit { key: entry.key, object: entry.object, size: entry.size as u64, created_at: entry.created_at, blob })))
223 }
224
225 /// `cache_upload`.
226 pub async fn cache_upload(&self, a: CacheUploadArgs) -> Result<Outcome<CacheReservation>> {
227 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'228 let scope = self.cache_scope(&job).await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2229 #[derive(Deserialize)]
230 struct Pending {
231 id: String,
232 object: String,
233 number: f64,
234 upload: Option<String>,
235 }
236 let columns = "SELECT id, object, rowid AS number, upload FROM cache_entries";
237 let pending = match (a.number, a.key.as_deref()) {
238 (Some(number), _) => {
239 self.db
240 .prepare(format!("{columns} WHERE rowid = ? AND repo_id = ? AND status = 'pending'"))
241 .bind(&[(number as f64).into(), scope.repo_id.as_str().into()])?
242 .first::<Pending>(None)
243 .await?
244 }
245 (None, Some(key)) => {
246 self.db
Merge branch 'worktree-agent-a3abfcce648e87dca'247 .prepare(format!("{columns} WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'pending'"))
248 .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 R2249 .first::<Pending>(None)
250 .await?
251 }
252 (None, None) => None,
253 };
254 let Some(pending) = pending else {
255 return Ok(fail(FailureCode::NotFound, "No upload of that entry is in progress."));
256 };
257 let blob = match &pending.upload {
258 Some(upload) => match self.upload_token("cache", &pending.id, &pending.object, upload) {
259 Some(blob) => Some(blob),
260 None => return Ok(fail(FailureCode::Invalid, "The toolkit's storage is not set up here: the actions service has no ACTIONS_KEY.")),
261 },
262 None => None,
263 };
264 Ok(Outcome::Ok(CacheReservation { id: pending.id, object: pending.object, number: pending.number as u64, upload: pending.upload, blob }))
Merge branch 'worktree-agent-a6a121745e81f639f'265 }
266
267 /// `cache_reserve`.
268 pub async fn cache_reserve(&self, a: CacheReserveArgs) -> Result<Outcome<CacheReservation>> {
269 let job = check!(self.cache_job(&a.job, &a.token).await?);
Merge branch 'worktree-agent-a3abfcce648e87dca'270 let scope = self.cache_scope(&job).await?;
Merge branch 'worktree-agent-a6a121745e81f639f'271 if !valid_key(&a.key) {
272 return Ok(fail(FailureCode::Invalid, format!("A cache key is 1 to {MAX_KEY_CHARS} characters, without commas.")));
273 }
274 if a.size > CACHE_MAX_ENTRY_BYTES {
275 return Ok(fail(
276 FailureCode::Invalid,
277 format!("It is {} MB; a cache entry is at most {} MB.", a.size / 1_048_576, CACHE_MAX_ENTRY_BYTES / 1_048_576),
278 ));
279 }
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.280 if let Some(why) = self.storage_refused(&job.namespace, a.size).await? {
281 return Ok(fail(FailureCode::Limit, why));
282 }
Merge branch 'worktree-agent-a6a121745e81f639f'283 let now = now_ms();
284 let at = rfc3339(now);
Merge branch 'worktree-agent-a3abfcce648e87dca'285 let version = a.version.clone().unwrap_or_default();
Merge branch 'worktree-agent-a6a121745e81f639f'286 // An upload left unfinished long ago no longer holds its key.
287 self.db
Merge branch 'worktree-agent-a3abfcce648e87dca'288 .prepare(
289 "UPDATE cache_entries SET status = 'expired'
290 WHERE repo_id = ? AND scope = ? AND key = ? AND version = ? AND status = 'pending' AND created_at < ?",
291 )
292 .bind(&[
293 scope.repo_id.as_str().into(),
294 scope.save.as_str().into(),
295 a.key.as_str().into(),
296 version.as_str().into(),
297 rfc3339(now.saturating_sub(PENDING_MS)).into(),
298 ])?
Merge branch 'worktree-agent-a6a121745e81f639f'299 .run()
300 .await?;
301 let id = new_id("cache", now);
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2302 let object = format!("c/{}/{id}", scope.repo_id);
303 #[derive(Deserialize)]
304 struct Inserted {
305 number: f64,
306 }
Merge branch 'worktree-agent-a3abfcce648e87dca'307 // A key is written once in its scope and version, never into
308 // another ref's scope.
Merge branch 'worktree-agent-a6a121745e81f639f'309 let inserted = self
310 .db
311 .prepare(
Merge branch 'worktree-agent-a3abfcce648e87dca'312 "INSERT INTO cache_entries (id, repo_id, namespace, key, object, size, status, created_at, last_used_at, version, scope)
313 VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'pending', ?7, ?7, ?8, ?9)
314 ON CONFLICT (repo_id, scope, key, version) DO UPDATE SET
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2315 id = ?1, object = ?5, size = ?6, status = 'pending', created_at = ?7, last_used_at = ?7, version = ?8, upload = NULL
Merge branch 'worktree-agent-a6a121745e81f639f'316 WHERE cache_entries.status = 'expired'
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2317 RETURNING rowid AS number",
Merge branch 'worktree-agent-a6a121745e81f639f'318 )
319 .bind(&[
320 id.as_str().into(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2321 scope.repo_id.as_str().into(),
Merge branch 'worktree-agent-a6a121745e81f639f'322 job.namespace.as_str().into(),
323 a.key.as_str().into(),
324 object.as_str().into(),
325 (a.size as f64).into(),
326 at.as_str().into(),
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2327 version.as_str().into(),
Merge branch 'worktree-agent-a3abfcce648e87dca'328 scope.save.as_str().into(),
Merge branch 'worktree-agent-a6a121745e81f639f'329 ])?
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2330 .first::<Inserted>(None)
Merge branch 'worktree-agent-a6a121745e81f639f'331 .await?;
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2332 let Some(inserted) = inserted else {
Merge branch 'worktree-agent-a6a121745e81f639f'333 return Ok(fail(FailureCode::Conflict, "That key is already cached."));
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2334 };
335 Ok(Outcome::Ok(CacheReservation { id, object, number: inserted.number as u64, upload: None, blob: None }))
Merge branch 'worktree-agent-a6a121745e81f639f'336 }
337
338 /// `cache_commit`: the entry is ready; entries past the quota are
339 /// evicted, restored longest ago first.
340 pub async fn cache_commit(&self, a: CacheCommitArgs) -> Result<Outcome<CacheCommitted>> {
341 let job = check!(self.cache_job(&a.job, &a.token).await?);
342 if a.size > CACHE_MAX_ENTRY_BYTES {
343 return Ok(fail(FailureCode::Invalid, "That entry is larger than a cache entry may be."));
344 }
345 let ready = self
346 .db
347 .prepare("UPDATE cache_entries SET status = 'ready', size = ?, last_used_at = ? WHERE id = ? AND repo_id = ? AND status = 'pending' RETURNING id")
348 .bind(&[(a.size as f64).into(), rfc3339(now_ms()).into(), a.id.as_str().into(), job.repo_id.as_str().into()])?
349 .first::<Value>(None)
350 .await?;
351 if ready.is_none() {
352 return Ok(fail(FailureCode::NotFound, "No upload of that entry is in progress."));
353 }
Merge remote-tracking branch 'origin/main' into workspace-chat354 // What the repository holds, summed: only past the quota are its
355 // entries listed to choose what to evict. A tool that saves an
356 // entry per compiled file (sccache) commits thousands of small
357 // entries a build, and listing them all on each was quadratic.
Merge branch 'worktree-agent-a6a121745e81f639f'358 #[derive(Deserialize)]
Merge remote-tracking branch 'origin/main' into workspace-chat359 struct Total {
360 bytes: Option<f64>,
361 }
362 let total = self
363 .db
364 .prepare("SELECT SUM(size) AS bytes FROM cache_entries WHERE repo_id = ? AND status = 'ready'")
365 .bind(&[job.repo_id.as_str().into()])?
366 .first::<Total>(None)
367 .await?
368 .and_then(|t| t.bytes)
369 .unwrap_or(0.0);
370 if !over_quota(total as u64, CACHE_REPO_QUOTA_BYTES) {
371 return Ok(Outcome::Ok(CacheCommitted { evicted: Vec::new() }));
372 }
373 #[derive(Deserialize)]
Merge branch 'worktree-agent-a6a121745e81f639f'374 struct Held {
375 id: String,
376 object: String,
377 size: f64,
378 }
379 let held = self
380 .db
381 .prepare("SELECT id, object, size FROM cache_entries WHERE repo_id = ? AND status = 'ready' ORDER BY last_used_at DESC")
382 .bind(&[job.repo_id.as_str().into()])?
383 .all()
384 .await?
385 .results::<Held>()?;
386 let sizes: Vec<(String, u64)> = held.iter().map(|h| (h.id.clone(), h.size as u64)).collect();
387 let evict = to_evict(&sizes, &a.id, CACHE_REPO_QUOTA_BYTES);
388 let mut evicted = Vec::new();
389 for id in &evict {
390 if let Some(entry) = held.iter().find(|h| &h.id == id) {
391 self.db.prepare("DELETE FROM cache_entries WHERE id = ?").bind(&[id.as_str().into()])?.run().await?;
392 evicted.push(entry.object.clone());
393 }
394 }
395 Ok(Outcome::Ok(CacheCommitted { evicted }))
396 }
397
398 /// `cache_abort`.
399 pub async fn cache_abort(&self, a: CacheAbortArgs) -> Result<Outcome<bool>> {
400 let job = check!(self.cache_job(&a.job, &a.token).await?);
401 self.db
402 .prepare("DELETE FROM cache_entries WHERE id = ? AND repo_id = ? AND status = 'pending'")
403 .bind(&[a.id.as_str().into(), job.repo_id.as_str().into()])?
404 .run()
405 .await?;
406 Ok(Outcome::Ok(true))
407 }
408
409 /// Hourly: deletes entries unused for a week, saved too long ago, or
410 /// left unfinished, and their objects; then what each workspace's
411 /// cache holds today, and this month's storage, for billing.
412 pub async fn sweep_cache(&self, now: u64) -> Result<()> {
413 let at = |ms: u64| rfc3339(now.saturating_sub(ms));
414 self.db
415 .prepare(
416 "UPDATE cache_entries SET status = 'expired'
417 WHERE (status = 'ready' AND (last_used_at < ?1 OR created_at < ?2)) OR (status = 'pending' AND created_at < ?3)",
418 )
419 .bind(&[at(CACHE_UNUSED_DAYS * DAY_MS).into(), at(CACHE_MAX_AGE_DAYS * DAY_MS).into(), at(PENDING_MS).into()])?
420 .run()
421 .await?;
422 #[derive(Deserialize)]
423 struct Gone {
424 id: String,
425 object: String,
426 }
427 let gone = self
428 .db
429 .prepare("SELECT id, object FROM cache_entries WHERE status = 'expired' LIMIT 500")
430 .all()
431 .await?
432 .results::<Gone>()?;
433 if let Some(bucket) = &self.cache {
434 for chunk in gone.chunks(100) {
435 if let Err(error) = bucket.delete_multiple(chunk.iter().map(|g| g.object.as_str()).collect()).await {
436 worker::console_error!("actions: cache objects not deleted: {error}");
437 return Ok(());
438 }
439 for entry in chunk {
440 self.db.prepare("DELETE FROM cache_entries WHERE id = ?").bind(&[entry.id.as_str().into()])?.run().await?;
441 }
442 }
443 }
444 self.measure_cache(now).await
445 }
446
447 /// What each workspace's cache holds today, and this month's storage so
448 /// far reported to billing, which charges it to workspaces on the plan
449 /// once the month is over.
450 async fn measure_cache(&self, now: u64) -> Result<()> {
451 #[derive(Deserialize)]
452 struct Held {
453 namespace: String,
454 bytes: f64,
455 }
456 let today = rfc3339(now);
457 let (day, month) = (&today[..10], &today[..7]);
458 let held = self
459 .db
Actions: OIDC tokens, the toolkit's cache and artifact services, and artifacts in R2460 // Artifacts are kept in the same bucket, at the same price.
461 .prepare(
462 "SELECT namespace, SUM(size) AS bytes FROM (
463 SELECT namespace, size FROM cache_entries WHERE status = 'ready'
464 UNION ALL SELECT namespace, size FROM artifacts WHERE status = 'ready'
465 ) GROUP BY namespace",
466 )
Merge branch 'worktree-agent-a6a121745e81f639f'467 .all()
468 .await?
469 .results::<Held>()?;
470 for workspace in &held {
471 self.db
472 .prepare(
473 "INSERT INTO cache_days (namespace, day, bytes) VALUES (?1, ?2, ?3)
474 ON CONFLICT (namespace, day) DO UPDATE SET bytes = max(bytes, ?3)",
475 )
476 .bind(&[workspace.namespace.as_str().into(), day.into(), workspace.bytes.into()])?
477 .run()
478 .await?;
479 }
480 #[derive(Deserialize)]
481 struct Day {
482 namespace: String,
483 bytes: f64,
484 }
485 let days = self
486 .db
487 .prepare("SELECT namespace, bytes FROM cache_days WHERE substr(day, 1, 7) = ?")
488 .bind(&[month.into()])?
489 .all()
490 .await?
491 .results::<Day>()?;
492 let mut by_workspace: std::collections::BTreeMap<String, Vec<u64>> = std::collections::BTreeMap::new();
493 for d in days {
494 by_workspace.entry(d.namespace).or_default().push(d.bytes as u64);
495 }
496 for (workspace, days) in by_workspace {
497 let months = gb_months(&days);
498 let cost = storage_cost(months);
499 if cost <= 0 {
500 continue;
501 }
502 let noted: Result<bool> = g1t_kit::call(
503 &self.billing,
504 "note_pending",
505 &NotePendingArgs { workspace: workspace.clone(), source: "cache".to_owned(), cost_micros: cost, detail: Some(format!("{months:.2} GB-months")) },
506 )
507 .await;
508 if let Err(error) = noted {
509 worker::console_error!("actions: cache storage not reported for {workspace}: {error}");
510 }
511 }
512 Ok(())
513 }
514}
515
516#[cfg(test)]
517mod tests {
518 use super::*;
519
520 fn entries(sizes: &[(&str, u64)]) -> Vec<(String, u64)> {
521 sizes.iter().map(|(id, size)| ((*id).to_owned(), *size)).collect()
522 }
523
524 #[test]
Merge branch 'worktree-agent-a3abfcce648e87dca'525 fn a_run_restores_from_its_ref_then_its_base_then_the_default_branch() {
526 let (restore, save) = scopes("refs/heads/feature", None, "main", true);
527 assert_eq!(restore, ["refs/heads/feature", "refs/heads/main"]);
528 assert_eq!(save, "refs/heads/feature");
529 // A pull request: its own merge ref, the branch it merges into, the default.
530 let (restore, save) = scopes("refs/pull/7/merge", Some("release/1.x"), "main", true);
531 assert_eq!(restore, ["refs/pull/7/merge", "refs/heads/release/1.x", "refs/heads/main"]);
532 assert_eq!(save, "refs/pull/7/merge");
533 // The default branch reads only its own.
534 let (restore, _) = scopes("refs/heads/main", Some("main"), "main", true);
535 assert_eq!(restore, ["refs/heads/main"]);
536 // From outside: it saves where nothing else reads.
537 let (restore, save) = scopes("refs/pull/9/merge", Some("main"), "main", false);
538 assert_eq!(save, "untrusted:refs/pull/9/merge");
539 assert_eq!(restore, ["untrusted:refs/pull/9/merge", "refs/heads/main"]);
540 for trusted_ref in ["refs/heads/main", "refs/pull/9/merge", "refs/heads/feature"] {
541 let (others, _) = scopes(trusted_ref, Some("main"), "main", true);
542 assert!(!others.contains(&save), "{trusted_ref} must never read an untrusted entry");
543 }
544 }
545
546 #[test]
Merge branch 'worktree-agent-a6a121745e81f639f'547 fn eviction_takes_the_entries_restored_longest_ago() {
548 // Newest use first.
549 let held = entries(&[("new", 4), ("b", 3), ("c", 3), ("old", 2)]);
550 assert_eq!(to_evict(&held, "new", 12), Vec::<String>::new());
551 assert_eq!(to_evict(&held, "new", 10), ["old"]);
552 assert_eq!(to_evict(&held, "new", 7), ["old", "c"]);
553 // The entry just saved stays, even when it alone is past the quota.
554 assert_eq!(to_evict(&entries(&[("big", 20), ("a", 1)]), "big", 10), ["a"]);
Merge remote-tracking branch 'origin/main' into workspace-chat555 // Entries are listed only when the sum is over: at or under the
556 // quota, nothing would be evicted.
557 for (held, quota) in [(entries(&[("a", 4), ("b", 6)]), 10), (entries(&[("a", 1)]), 12)] {
558 let total = held.iter().map(|(_, size)| size).sum();
559 assert!(!over_quota(total, quota));
560 assert!(to_evict(&held, "a", quota).is_empty());
561 }
562 assert!(over_quota(11, 10));
Merge branch 'worktree-agent-a6a121745e81f639f'563 }
564
565 #[test]
Merge branch 'worktree-agent-a63425cb2d4935c5f'566 fn a_restore_tries_its_key_as_a_prefix_before_its_restore_keys() {
567 let restore = vec!["cargo-Linux-1.99.0-".to_owned(), " ".to_owned(), "cargo-".to_owned(), "cargo-".to_owned()];
568 assert_eq!(prefixes("cargo-Linux-1.99.0-abc", &restore), ["cargo-Linux-1.99.0-abc", "cargo-Linux-1.99.0-", "cargo-"]);
569 assert_eq!(prefixes("k", &[]), ["k"]);
570 }
571
572 #[test]
Merge branch 'worktree-agent-a6a121745e81f639f'573 fn keys_are_checked() {
574 assert!(valid_key("cargo-Linux-abc123"));
575 assert!(!valid_key(""));
576 assert!(!valid_key("a,b"));
577 assert!(!valid_key(&"k".repeat(513)));
578 }
579
580 #[test]
581 fn storage_is_charged_by_the_gb_month_at_r2s_price() {
582 // 10 GB held for 30 days is 10 GB-months: $0.15.
583 let month = vec![10_000_000_000u64; 30];
584 assert!((gb_months(&month) - 10.0).abs() < 1e-9);
585 assert_eq!(storage_cost(gb_months(&month)), 150_000);
586 // A day of 1 GB: a thirtieth of a GB-month, rounded up.
587 assert_eq!(storage_cost(gb_months(&[1_000_000_000])), 500);
588 assert_eq!(storage_cost(0.0), 0);
589 }
590}

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