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
569 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.

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

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