Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/repos/src/store.rs

1,021 lines42,762 bytesCodeBlame
1//! The storage that actually holds git repositories.
2//!
3//! The service depends on the [`GitStore`] and [`GitRepo`] ports;
4//! [`ArtifactsStore`] is the adapter for Cloudflare Artifacts.
5//!
6//! Every call on the binding goes through [`invoke`]: it is counted
7//! (meters.rs), timed for the store's health, refused at once while its
8//! namespace's breaker is open, and tried again after a failure that may
9//! pass when it only reads (resilience.rs). Repositories may live in
10//! several namespaces (shards.rs).
11
12use g1t_contracts::repos::{Branch, Commit, EntryKind, GitAccess, Signature, TreeEntry};
13use g1t_contracts::time::rfc3339;
14use g1t_kit::js::{self, Thrown};
15use serde::{Deserialize, Serialize};
16use std::cell::RefCell;
17use std::collections::HashMap;
18use std::rc::Rc;
19use worker::js_sys::{Reflect, Uint8Array};
20use worker::wasm_bindgen::{JsCast, JsValue};
21use worker::{Env, Result};
22
23use crate::meters::{self, Outcome};
24use crate::resilience::{self, Admit, Busy, Failure};
25use crate::shards;
26
27/// Credentials that never leave this service: g1t has already decided who
28/// may do what before one is used. They live an hour and are used for 50
29/// minutes, so each one used has at least ten minutes left.
30const INTERNAL_TTL_SECONDS: u32 = 3_600;
31const INTERNAL_REUSE_MS: u64 = 50 * 60 * 1000;
32/// Credentials handed out (`git_access`, to sandboxes): five minutes, used
33/// for three, so whoever gets one has at least two.
34const HANDOUT_TTL_SECONDS: u32 = 300;
35const HANDOUT_REUSE_MS: u64 = 180_000;
36/// How long a handed-out credential stays valid, in milliseconds.
37pub const CREDENTIAL_LIFE_MS: u64 = HANDOUT_TTL_SECONDS as u64 * 1000;
38
39#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
40pub enum Scope {
41 Read,
42 Write,
43}
44
45impl Scope {
46 fn as_str(self) -> &'static str {
47 match self {
48 Scope::Read => "read",
49 Scope::Write => "write",
50 }
51 }
52}
53
54/// Who a credential is for.
55#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
56pub enum Use {
57 /// This service, talking to the store itself.
58 Internal,
59 /// Someone outside it, through `git_access`.
60 Handout,
61}
62
63impl Use {
64 pub fn ttl_seconds(self) -> u32 {
65 match self {
66 Use::Internal => INTERNAL_TTL_SECONDS,
67 Use::Handout => HANDOUT_TTL_SECONDS,
68 }
69 }
70
71 pub fn reuse_ms(self) -> u64 {
72 match self {
73 Use::Internal => INTERNAL_REUSE_MS,
74 Use::Handout => HANDOUT_REUSE_MS,
75 }
76 }
77
78 fn as_str(self) -> &'static str {
79 match self {
80 Use::Internal => "internal",
81 Use::Handout => "handout",
82 }
83 }
84}
85
86/// Where a credential handed out came from, for `Server-Timing`.
87#[derive(Clone, Copy, PartialEq, Eq, Debug)]
88pub enum Kept {
89 /// This isolate made it, or had it from another, a moment ago.
90 Isolate,
91 /// Another isolate made it and shared it.
92 Shared,
93}
94
95impl Kept {
96 pub fn as_str(self) -> &'static str {
97 match self {
98 Kept::Isolate => "isolate",
99 Kept::Shared => "shared",
100 }
101 }
102}
103
104/// A place repositories live. `key` is the store's own name for a repo.
105#[allow(async_fn_in_trait)]
106pub trait GitStore {
107 type Repo: GitRepo;
108
109 /// Creates an empty repository. Succeeds if it already exists.
110 async fn create(&self, key: &str, description: Option<&str>, default_branch: &str) -> Result<()>;
111 async fn open(&self, key: &str) -> Result<Self::Repo>;
112 /// A credential for `key` made a moment ago, if the store keeps one.
113 async fn kept_access(&self, _key: &str, _scope: Scope) -> Option<(GitAccess, Kept)> {
114 None
115 }
116 /// A new credential for `key`, which the store may keep for next time.
117 async fn mint_access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
118 self.open(key).await?.access(scope).await
119 }
120 /// A remote URL and credential for this service's own use. A store may
121 /// hand out one it made a moment ago.
122 async fn access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
123 match self.kept_access(key, scope).await {
124 Some((access, _)) => Ok(access),
125 None => self.mint_access(key, scope).await,
126 }
127 }
128 /// A short-lived credential for someone outside this service.
129 async fn handout(&self, key: &str, scope: Scope) -> Result<GitAccess> {
130 self.access(key, scope).await
131 }
132 /// Stops handing out the credentials it keeps for `key`: the store
133 /// turned one down, or the repository is gone.
134 async fn forget_access(&self, _key: &str) {}
135 /// Removes a repository and everything in it, for good. Succeeds if it
136 /// is already gone.
137 async fn delete(&self, key: &str) -> Result<()>;
138 /// The namespaces new repositories may be placed in (shards.rs).
139 fn namespaces(&self) -> Vec<String> {
140 vec![shards::DEFAULT_NAMESPACE.to_owned()]
141 }
142 /// The namespace bound as the default.
143 fn default_namespace(&self) -> String {
144 shards::DEFAULT_NAMESPACE.to_owned()
145 }
146}
147
148/// One open repository.
149#[allow(async_fn_in_trait)]
150pub trait GitRepo {
151 /// A remote URL and credential for git itself, for this service.
152 async fn access(&self, scope: Scope) -> Result<GitAccess>;
153 /// Every branch and the commit it points to.
154 async fn branches(&self) -> Result<Vec<Branch>>;
155 /// Newest first along the first-parent chain; empty for an unknown ref.
156 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>>;
157 /// The parents of a commit, or `None` if the commit does not exist.
158 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>>;
159 async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>>;
160 async fn read_blob(&self, blob_hash: &str) -> Result<Option<Vec<u8>>>;
161 /// `None` when the ref or path does not resolve to a file.
162 async fn read_file(&self, git_ref: &str, path: &str) -> Result<Option<Vec<u8>>>;
163 /// Makes a copy of this repository under `target_key`, in the same
164 /// namespace.
165 async fn fork(&self, target_key: &str) -> Result<()>;
166 /// The version of the repository's refs (registry.rs `RefsState`),
167 /// when answers about branches may be kept under it (R9). Reads by
168 /// branch name are then kept until the version moves.
169 fn at_refs_version(&mut self, _version: Option<u64>) {}
170}
171
172thread_local! {
173 /// The namespace bound to `ARTIFACTS`, for keys that name none.
174 static DEFAULT_NS: RefCell<String> = RefCell::new(shards::DEFAULT_NAMESPACE.to_owned());
175 /// Where each namespace's remotes start: `https://<account>.artifacts.cloudflare.net/git/<namespace>/`.
176 static REMOTE_PREFIX: RefCell<HashMap<String, String>> = RefCell::new(HashMap::new());
177}
178
179/// The namespace and name a store key stands for.
180pub fn locate(key: &str) -> (String, String) {
181 let (namespace, name) = shards::split(key);
182 let namespace = namespace.map_or_else(|| DEFAULT_NS.with(|ns| ns.borrow().clone()), str::to_owned);
183 (namespace, name.to_owned())
184}
185
186/// The store key a git remote is for, from its last two path segments.
187pub fn key_from_remote(remote: &str) -> Option<String> {
188 let path = remote.trim_end_matches('/');
189 let path = path.strip_suffix(".git").unwrap_or(path);
190 let (rest, name) = path.rsplit_once('/')?;
191 let namespace = rest.rsplit('/').next()?;
192 let default = DEFAULT_NS.with(|ns| ns.borrow().clone());
193 Some(shards::compose(Some(namespace), name, &default))
194}
195
196/// Where a namespace's remotes start, learned from one remote the store
197/// gave for `name`.
198fn learn_prefix(remote: &str, name: &str) -> Option<String> {
199 remote.strip_suffix(&format!("{name}.git")).filter(|prefix| prefix.ends_with('/')).map(str::to_owned)
200}
201
202/// A repository's remote, from where its namespace's remotes start.
203fn remote_from(prefix: &str, name: &str) -> String {
204 format!("{prefix}{name}.git")
205}
206
207struct Namespace {
208 name: String,
209 binding: JsValue,
210}
211
212pub struct ArtifactsStore {
213 namespaces: Rc<Vec<Namespace>>,
214 /// Where isolates share the credentials they make; see shared.rs.
215 shared: Option<Rc<crate::shared::Shared>>,
216}
217
218impl ArtifactsStore {
219 pub fn new(env: &Env, shared: Option<Rc<crate::shared::Shared>>) -> Result<Self> {
220 let config = env.var("ARTIFACTS_NAMESPACES").ok().map(|value| value.to_string());
221 let mut namespaces = Vec::new();
222 for (binding, name) in shards::bindings(config.as_deref()) {
223 match js::binding(env, &binding) {
224 Ok(value) => namespaces.push(Namespace { name, binding: value }),
225 // The default binding is required; the others are optional.
226 Err(error) if binding == shards::DEFAULT_BINDING => return Err(error),
227 Err(_) => worker::console_error!("ARTIFACTS_NAMESPACES names {binding}, which is not bound"),
228 }
229 }
230 if let Some(default) = namespaces.first() {
231 DEFAULT_NS.with(|ns| ns.borrow_mut().clone_from(&default.name));
232 }
233 // Optional: where remotes start, `https://<account>.artifacts.cloudflare.net/git`,
234 // so the first credential an isolate makes needs no `info()` either.
235 if let Ok(base) = env.var("ARTIFACTS_REMOTE_BASE") {
236 let base = base.to_string().trim_end_matches('/').to_owned();
237 if base.starts_with("https://") {
238 REMOTE_PREFIX.with(|prefixes| {
239 let mut prefixes = prefixes.borrow_mut();
240 for namespace in &namespaces {
241 prefixes.entry(namespace.name.clone()).or_insert_with(|| format!("{base}/{}/", namespace.name));
242 }
243 });
244 }
245 }
246 Ok(Self { namespaces: Rc::new(namespaces), shared })
247 }
248
249 fn binding(&self, namespace: &str) -> Result<&JsValue> {
250 self.namespaces
251 .iter()
252 .find(|candidate| candidate.name == namespace)
253 .map(|found| &found.binding)
254 .ok_or_else(|| worker::Error::RustError(format!("git store namespace {namespace} is not bound")))
255 }
256}
257
258/// A failed call on the binding.
259pub struct StoreError {
260 pub thrown: Thrown,
261 pub busy: Option<Busy>,
262}
263
264impl StoreError {
265 pub fn is(&self, code: &str) -> bool {
266 self.thrown.is(code)
267 }
268}
269
270impl From<StoreError> for worker::Error {
271 fn from(error: StoreError) -> Self {
272 match error.busy {
273 Some(busy) => busy.error(&error.thrown.to_string()),
274 None => error.thrown.into(),
275 }
276 }
277}
278
279/// The meter for a binding method: `binding.create_token`.
280fn meter_of(method: &str) -> String {
281 let mut out = String::from("binding.");
282 for c in method.chars() {
283 if c.is_ascii_uppercase() {
284 out.push('_');
285 out.push(c.to_ascii_lowercase());
286 } else {
287 out.push(c);
288 }
289 }
290 out
291}
292
293/// Seconds a caller is told to wait when the store is busy.
294const BUSY_RETRY_AFTER: u64 = 5;
295
296/// Calls `target[method](...args)` on namespace `namespace`'s binding, for
297/// the repository at `key`. `retry`: whether a failure that may pass is
298/// tried again (reads and credentials only).
299async fn invoke(
300 namespace: &str,
301 key: &str,
302 target: &JsValue,
303 method: &str,
304 args: &[JsValue],
305 retry: bool,
306) -> std::result::Result<JsValue, StoreError> {
307 let meter = meter_of(method);
308 let mut attempt = 0;
309 loop {
310 let now = g1t_kit::now_ms();
311 let admit = resilience::with_breaker(namespace, |breaker| breaker.admit(now));
312 if let Admit::Wait(ms) = admit {
313 meters::record_health(namespace, Outcome::Rejected, 0);
314 return Err(StoreError {
315 thrown: Thrown { code: None, message: format!("{method} not asked: the git store has been failing") },
316 busy: Some(Busy { rate_limited: false, retry_after: resilience::seconds(ms).max(BUSY_RETRY_AFTER) }),
317 });
318 }
319 meters::record(&meter, key, 0, 0);
320 let called = js::call(target, method, args).await;
321 let ms = g1t_kit::now_ms().saturating_sub(now);
322 match called {
323 Ok(value) => {
324 resilience::with_breaker(namespace, |breaker| breaker.succeeded());
325 meters::record_health(namespace, Outcome::Ok, ms);
326 return Ok(value);
327 }
328 Err(thrown) => {
329 let failure = resilience::classify(thrown.code.as_deref(), &thrown.message);
330 if failure == Failure::Permanent {
331 // An answer (NOT_FOUND, ALREADY_EXISTS): the store is up.
332 resilience::with_breaker(namespace, |breaker| breaker.succeeded());
333 meters::record_health(namespace, Outcome::Ok, ms);
334 return Err(StoreError { thrown, busy: None });
335 }
336 if retry && resilience::retry(failure, attempt) {
337 let wait = resilience::backoff_ms(failure, attempt, worker::js_sys::Math::random());
338 worker::Delay::from(std::time::Duration::from_millis(wait)).await;
339 attempt += 1;
340 continue;
341 }
342 resilience::with_breaker(namespace, |breaker| breaker.failed(failure, g1t_kit::now_ms()));
343 let outcome = if failure == Failure::RateLimited { Outcome::RateLimited } else { Outcome::Failed };
344 meters::record_health(namespace, outcome, ms);
345 worker::console_error!("git store {namespace}: {method} for {key} failed: {thrown}");
346 return Err(StoreError {
347 thrown,
348 busy: Some(Busy { rate_limited: failure == Failure::RateLimited, retry_after: BUSY_RETRY_AFTER }),
349 });
350 }
351 }
352 }
353}
354
355/// A credential as isolates share it, sealed (see shared.rs): with when it
356/// was made, so that one shared is reused no longer than one kept here.
357#[derive(Serialize, Deserialize)]
358struct SharedCredential {
359 remote: String,
360 token: String,
361 made: u64,
362}
363
364/// The shared cache's key for a credential: the store's key for the
365/// repository, the scope, and who it is for. (`cred2`: credentials kept
366/// before their lives differed by use are not read.)
367fn shared_key(key: &str, scope: Scope, using: Use) -> String {
368 format!("cred2:{key}:{}:{}", scope.as_str(), using.as_str())
369}
370
371/// A shared credential, if it was made less than its reuse window before
372/// `now`; with when it was made.
373fn shared_credential(bytes: &[u8], now: u64, using: Use) -> Option<(GitAccess, u64)> {
374 let kept: SharedCredential = serde_json::from_slice(bytes).ok()?;
375 (now.saturating_sub(kept.made) < using.reuse_ms()).then_some((
376 GitAccess {
377 remote: kept.remote,
378 token: kept.token,
379 },
380 kept.made,
381 ))
382}
383
384/// Credentials made a while ago, by repository, scope and use. Making one
385/// is a round trip to the store; reusing it saves that, and the store's
386/// lookup of the repository with it.
387#[derive(Default)]
388pub struct Credentials {
389 kept: HashMap<(String, Scope, Use), (GitAccess, u64)>,
390}
391
392impl Credentials {
393 /// One made for `key`, `scope` and `using` within its reuse window of `now`.
394 pub fn get(&self, key: &str, scope: Scope, using: Use, now: u64) -> Option<GitAccess> {
395 self.kept
396 .get(&(key.to_owned(), scope, using))
397 .filter(|(_, made)| now.saturating_sub(*made) < using.reuse_ms())
398 .map(|(access, _)| access.clone())
399 }
400
401 pub fn keep(&mut self, key: &str, scope: Scope, using: Use, access: GitAccess, made: u64, now: u64) {
402 // Expired ones go first, so the map stays as small as the isolate's
403 // recent repositories.
404 self.kept
405 .retain(|(_, _, kept_use), (_, at)| now.saturating_sub(*at) < kept_use.reuse_ms());
406 self.kept.insert((key.to_owned(), scope, using), (access, made));
407 }
408
409 pub fn forget(&mut self, key: &str) {
410 self.kept.retain(|(kept, _, _), _| kept != key);
411 }
412}
413
414thread_local! {
415 static CREDENTIALS: RefCell<Credentials> = RefCell::new(Credentials::default());
416}
417
418/// A credential kept in this isolate, else one another isolate shared.
419async fn kept(shared: Option<&crate::shared::Shared>, key: &str, scope: Scope, using: Use) -> Option<(GitAccess, Kept)> {
420 let now = g1t_kit::now_ms();
421 if let Some(access) = CREDENTIALS.with(|kept| kept.borrow().get(key, scope, using, now)) {
422 return Some((access, Kept::Isolate));
423 }
424 let bytes = shared?.get(&shared_key(key, scope, using)).await?;
425 let (access, made) = shared_credential(&bytes, now, using)?;
426 CREDENTIALS.with(|kept| kept.borrow_mut().keep(key, scope, using, access.clone(), made, now));
427 Some((access, Kept::Shared))
428}
429
430/// Keeps a credential just made here, and shares it.
431async fn keep(shared: Option<&crate::shared::Shared>, key: &str, scope: Scope, using: Use, access: &GitAccess) {
432 let now = g1t_kit::now_ms();
433 CREDENTIALS.with(|kept| kept.borrow_mut().keep(key, scope, using, access.clone(), now, now));
434 if let Some(shared) = shared {
435 let value = SharedCredential {
436 remote: access.remote.clone(),
437 token: access.token.clone(),
438 made: now,
439 };
440 if let Ok(bytes) = serde_json::to_vec(&value) {
441 shared.put(&shared_key(key, scope, using), &bytes, using.reuse_ms() / 1000).await;
442 }
443 }
444}
445
446impl GitStore for ArtifactsStore {
447 type Repo = ArtifactsRepo;
448
449 async fn kept_access(&self, key: &str, scope: Scope) -> Option<(GitAccess, Kept)> {
450 kept(self.shared.as_deref(), key, scope, Use::Internal).await
451 }
452
453 async fn mint_access(&self, key: &str, scope: Scope) -> Result<GitAccess> {
454 let repo = self.open(key).await?;
455 let access = repo.mint(scope, Use::Internal).await?;
456 keep(self.shared.as_deref(), key, scope, Use::Internal, &access).await;
457 Ok(access)
458 }
459
460 async fn handout(&self, key: &str, scope: Scope) -> Result<GitAccess> {
461 if let Some((access, _)) = kept(self.shared.as_deref(), key, scope, Use::Handout).await {
462 return Ok(access);
463 }
464 let access = self.open(key).await?.mint(scope, Use::Handout).await?;
465 keep(self.shared.as_deref(), key, scope, Use::Handout, &access).await;
466 Ok(access)
467 }
468
469 async fn forget_access(&self, key: &str) {
470 CREDENTIALS.with(|kept| kept.borrow_mut().forget(key));
471 if let Some(shared) = &self.shared {
472 let keys: Vec<String> = [Scope::Read, Scope::Write]
473 .into_iter()
474 .flat_map(|scope| [Use::Internal, Use::Handout].map(|using| shared_key(key, scope, using)))
475 .collect();
476 futures_util::future::join_all(keys.iter().map(|shared_key| shared.delete(shared_key))).await;
477 }
478 }
479
480 async fn create(&self, key: &str, description: Option<&str>, default_branch: &str) -> Result<()> {
481 let (namespace, name) = locate(key);
482 let options = js::to_js(&serde_json::json!({
483 "description": description,
484 "setDefaultBranch": default_branch,
485 }))?;
486 match invoke(&namespace, key, self.binding(&namespace)?, "create", &[name.as_str().into(), options], true).await {
487 // Left behind by an earlier failed attempt; adopt it.
488 Err(failed) if !failed.is("ALREADY_EXISTS") => Err(failed.into()),
489 _ => Ok(()),
490 }
491 }
492
493 async fn delete(&self, key: &str) -> Result<()> {
494 self.forget_access(key).await;
495 let (namespace, name) = locate(key);
496 match invoke(&namespace, key, self.binding(&namespace)?, "delete", &[name.as_str().into()], true).await {
497 // Gone already: an earlier purge got this far.
498 Err(failed) if !failed.is("NOT_FOUND") => Err(failed.into()),
499 _ => Ok(()),
500 }
501 }
502
503 async fn open(&self, key: &str) -> Result<ArtifactsRepo> {
504 let (namespace, name) = locate(key);
505 let handle = invoke(&namespace, key, self.binding(&namespace)?, "get", &[name.as_str().into()], true).await?;
506 Ok(ArtifactsRepo {
507 handle,
508 key: key.to_owned(),
509 name,
510 namespace,
511 shared: self.shared.clone(),
512 refs_version: None,
513 })
514 }
515
516 fn namespaces(&self) -> Vec<String> {
517 self.namespaces.iter().map(|namespace| namespace.name.clone()).collect()
518 }
519
520 fn default_namespace(&self) -> String {
521 DEFAULT_NS.with(|ns| ns.borrow().clone())
522 }
523}
524
525/// A handle to one Artifacts repository. It is an RPC stub, so it is
526/// released when dropped.
527pub struct ArtifactsRepo {
528 handle: JsValue,
529 /// The repository's store key, which scopes its cached objects.
530 key: String,
531 /// Its name in its namespace.
532 name: String,
533 namespace: String,
534 shared: Option<Rc<crate::shared::Shared>>,
535 /// See [`GitRepo::at_refs_version`].
536 refs_version: Option<u64>,
537}
538
539/// Where cached git objects live. Trees and blobs are named by their
540/// content, so a cached one is never stale; each is kept under its own
541/// repository's key, so a repository only ever finds its own objects.
542const OBJECT_CACHE: &str = "https://objects.g1t.internal/";
543/// Blobs and files larger than this are not cached.
544const MAX_CACHED_BLOB: usize = 1024 * 1024;
545const OBJECT_MAX_AGE: &str = "public, max-age=31536000, immutable";
546/// Answers kept under a refs version: until the version moves, and no
547/// longer than this, which bounds how stale one can be should a change
548/// ever fail to move it.
549const VERSIONED_MAX_AGE: &str = "public, max-age=300";
550
551/// Whether a ref is a full commit hash (SHA-1 or SHA-256), whose history
552/// can be kept for good.
553pub fn is_commit_hash(git_ref: &str) -> bool {
554 (git_ref.len() == 40 || git_ref.len() == 64) && git_ref.bytes().all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
555}
556
557/// Where a read is kept: under the object it names for good, under the
558/// refs version for a name that can move, or nowhere.
559#[derive(Debug, PartialEq, Eq)]
560pub enum CacheKey {
561 Forever(String),
562 Versioned(String),
563}
564
565/// The cache key for `log(git_ref, limit)`.
566pub fn log_key(git_ref: &str, limit: u32, version: Option<u64>) -> Option<CacheKey> {
567 if is_commit_hash(git_ref) {
568 return Some(CacheKey::Forever(format!("log/{git_ref}-{limit}")));
569 }
570 version.map(|version| CacheKey::Versioned(format!("vlog/{version}/{}-{limit}", g1t_secrets::sha256_hex(git_ref))))
571}
572
573/// The cache key for `read_file(git_ref, path)`.
574pub fn file_key(git_ref: &str, path: &str, version: Option<u64>) -> Option<CacheKey> {
575 let path = g1t_secrets::sha256_hex(path);
576 if is_commit_hash(git_ref) {
577 return Some(CacheKey::Forever(format!("file/{git_ref}/{path}")));
578 }
579 version.map(|version| CacheKey::Versioned(format!("vfile/{version}/{}/{path}", g1t_secrets::sha256_hex(git_ref))))
580}
581
582/// The cache key for the branch list.
583pub fn branches_key(version: Option<u64>) -> Option<CacheKey> {
584 version.map(|version| CacheKey::Versioned(format!("branches/{version}")))
585}
586
587impl ArtifactsRepo {
588 fn cache_url(&self, path: &str) -> String {
589 format!("{OBJECT_CACHE}{}/{path}", self.key)
590 }
591
592 async fn cached_at(&self, path: &str) -> Option<Vec<u8>> {
593 let mut response = worker::Cache::default().get(self.cache_url(path), false).await.ok()??;
594 response.bytes().await.ok()
595 }
596
597 async fn cached(&self, kind: &str, hash: &str) -> Option<Vec<u8>> {
598 self.cached_at(&format!("{kind}/{hash}")).await
599 }
600
601 async fn keep_at(&self, path: &str, bytes: Vec<u8>, max_age: &str) {
602 let Ok(mut response) = worker::Response::from_bytes(bytes) else {
603 return;
604 };
605 let _ = response.headers_mut().set("cache-control", max_age);
606 let _ = worker::Cache::default().put(self.cache_url(path), response).await;
607 }
608
609 /// Keeps an object for next time. A failure only costs a later read.
610 async fn keep(&self, kind: &str, hash: &str, bytes: Vec<u8>) {
611 self.keep_at(&format!("{kind}/{hash}"), bytes, OBJECT_MAX_AGE).await;
612 }
613
614 async fn get_key(&self, key: &CacheKey) -> Option<Vec<u8>> {
615 match key {
616 CacheKey::Forever(path) | CacheKey::Versioned(path) => self.cached_at(path).await,
617 }
618 }
619
620 async fn put_key(&self, key: &CacheKey, bytes: Vec<u8>) {
621 match key {
622 CacheKey::Forever(path) => self.keep_at(path, bytes, OBJECT_MAX_AGE).await,
623 CacheKey::Versioned(path) => self.keep_at(path, bytes, VERSIONED_MAX_AGE).await,
624 }
625 }
626
627 async fn call(&self, method: &str, args: &[JsValue], retry: bool) -> std::result::Result<JsValue, StoreError> {
628 invoke(&self.namespace, &self.key, &self.handle, method, args, retry).await
629 }
630
631 /// A new credential from the store. Its remote is worked out from the
632 /// key once this isolate knows where the namespace's remotes start;
633 /// until then the store is asked (`info()`) alongside the token.
634 pub async fn mint(&self, scope: Scope, using: Use) -> Result<GitAccess> {
635 let args = [scope.as_str().into(), using.ttl_seconds().into()];
636 let prefix = REMOTE_PREFIX.with(|prefixes| prefixes.borrow().get(&self.namespace).cloned());
637 if let Some(prefix) = prefix {
638 let token: RawToken = js::from_js(&self.call("createToken", &args, true).await?)?;
639 return Ok(GitAccess { remote: remote_from(&prefix, &self.name), token: token.plaintext });
640 }
641 let (info, token) = futures_util::future::join(self.call("info", &[], true), self.call("createToken", &args, true)).await;
642 let info: RawInfo = js::from_js(&info?)?;
643 let token: RawToken = js::from_js(&token?)?;
644 match learn_prefix(&info.remote, &self.name) {
645 Some(prefix) => REMOTE_PREFIX.with(|prefixes| {
646 prefixes.borrow_mut().insert(self.namespace.clone(), prefix);
647 }),
648 None => worker::console_error!("the git store's remote {} does not end in {}.git", info.remote, self.name),
649 }
650 Ok(GitAccess { remote: info.remote, token: token.plaintext })
651 }
652}
653
654impl Drop for ArtifactsRepo {
655 fn drop(&mut self) {
656 let symbol = js::get(&worker::js_sys::global(), "Symbol");
657 let dispose = js::get(&symbol, "dispose");
658 if let Ok(function) = Reflect::get(&self.handle, &dispose)
659 .and_then(|value| value.dyn_into::<worker::js_sys::Function>())
660 {
661 let _ = function.call0(&self.handle);
662 }
663 }
664}
665
666#[derive(Deserialize)]
667#[serde(rename_all = "camelCase")]
668struct RawCommit {
669 hash: String,
670 tree_hash: String,
671 message: String,
672 author: Signature,
673 parents: Vec<String>,
674 /// Seconds since the epoch.
675 authored_at: u64,
676}
677
678#[derive(Deserialize)]
679struct RawEntry {
680 name: String,
681 hash: String,
682 #[serde(rename = "type")]
683 kind: EntryKind,
684}
685
686#[derive(Deserialize)]
687struct RawInfo {
688 remote: String,
689}
690
691#[derive(Deserialize)]
692struct RawToken {
693 plaintext: String,
694}
695
696/// The bytes of a `Blob`, or `None` for null.
697async fn blob_bytes(blob: JsValue) -> Result<Option<Vec<u8>>> {
698 if blob.is_null() || blob.is_undefined() {
699 return Ok(None);
700 }
701 let buffer = js::call(&blob, "arrayBuffer", &[]).await?;
702 Ok(Some(Uint8Array::new(&buffer).to_vec()))
703}
704
705impl GitRepo for ArtifactsRepo {
706 /// One made a while ago, here or in another isolate, else a new one.
707 async fn access(&self, scope: Scope) -> Result<GitAccess> {
708 if let Some((access, _)) = kept(self.shared.as_deref(), &self.key, scope, Use::Internal).await {
709 return Ok(access);
710 }
711 let access = self.mint(scope, Use::Internal).await?;
712 keep(self.shared.as_deref(), &self.key, scope, Use::Internal, &access).await;
713 Ok(access)
714 }
715
716 async fn branches(&self) -> Result<Vec<Branch>> {
717 let key = branches_key(self.refs_version);
718 if let Some(key) = &key
719 && let Some(bytes) = self.get_key(key).await
720 && let Ok(branches) = serde_json::from_slice::<Vec<Branch>>(&bytes)
721 {
722 return Ok(branches);
723 }
724 let branches = crate::refs::branches(&self.access(Scope::Read).await?).await?;
725 if let (Some(key), Ok(bytes)) = (&key, serde_json::to_vec(&branches)) {
726 self.put_key(key, bytes).await;
727 }
728 Ok(branches)
729 }
730
731 async fn log(&self, git_ref: &str, limit: u32) -> Result<Vec<Commit>> {
732 // History from a commit never changes, so a log asked for by hash is
733 // kept like an object: walking it is a read per commit. One asked
734 // for by a branch's name is kept until the branch can have moved.
735 let key = log_key(git_ref, limit, self.refs_version);
736 if let Some(key) = &key
737 && let Some(bytes) = self.get_key(key).await
738 && let Ok(commits) = serde_json::from_slice::<Vec<Commit>>(&bytes)
739 {
740 return Ok(commits);
741 }
742 let options = js::to_js(&serde_json::json!({ "ref": git_ref, "limit": limit }))?;
743 let raw: Vec<RawCommit> = js::from_js(&self.call("log", &[options], true).await?)?;
744 let commits: Vec<Commit> = raw
745 .into_iter()
746 .map(|commit| Commit {
747 hash: commit.hash,
748 tree_hash: commit.tree_hash,
749 message: commit.message,
750 author: commit.author,
751 parents: commit.parents,
752 authored_at: rfc3339(commit.authored_at * 1000),
753 })
754 .collect();
755 // An unknown ref logs nothing; that is not kept, in case it arrives.
756 if !commits.is_empty()
757 && let Ok(bytes) = serde_json::to_vec(&commits)
758 {
759 if let Some(key) = &key {
760 self.put_key(key, bytes.clone()).await;
761 }
762 // The same history, by the commit the name led to.
763 if !is_commit_hash(git_ref)
764 && let Some(CacheKey::Forever(path)) = log_key(&commits[0].hash, limit, None)
765 {
766 self.keep_at(&path, bytes, OBJECT_MAX_AGE).await;
767 }
768 }
769 Ok(commits)
770 }
771
772 async fn parents(&self, commit_hash: &str) -> Result<Option<Vec<String>>> {
773 // A commit never changes.
774 if let Some(bytes) = self.cached("commit", commit_hash).await
775 && let Ok(parents) = serde_json::from_slice::<Vec<String>>(&bytes)
776 {
777 return Ok(Some(parents));
778 }
779 let commit: Option<RawCommit> = js::from_js(&self.call("readCommit", &[commit_hash.into()], true).await?)?;
780 let parents = commit.map(|commit| commit.parents);
781 if let Some(parents) = &parents
782 && let Ok(bytes) = serde_json::to_vec(parents)
783 {
784 self.keep("commit", commit_hash, bytes).await;
785 }
786 Ok(parents)
787 }
788
789 async fn read_tree(&self, tree_hash: &str) -> Result<Option<Vec<TreeEntry>>> {
790 if let Some(bytes) = self.cached("tree", tree_hash).await
791 && let Ok(entries) = serde_json::from_slice::<Vec<TreeEntry>>(&bytes)
792 {
793 return Ok(Some(entries));
794 }
795 let entries: Option<Vec<RawEntry>> = js::from_js(&self.call("readTree", &[tree_hash.into()], true).await?)?;
796 let entries: Option<Vec<TreeEntry>> = entries.map(|entries| {
797 entries
798 .into_iter()
799 .map(|entry| TreeEntry {
800 name: entry.name,
801 hash: entry.hash,
802 kind: entry.kind,
803 })
804 .collect()
805 });
806 if let Some(entries) = &entries
807 && let Ok(bytes) = serde_json::to_vec(entries)
808 {
809 self.keep("tree", tree_hash, bytes).await;
810 }
811 Ok(entries)
812 }
813
814 async fn read_blob(&self, blob_hash: &str) -> Result<Option<Vec<u8>>> {
815 if let Some(bytes) = self.cached("blob", blob_hash).await {
816 return Ok(Some(bytes));
817 }
818 let bytes = blob_bytes(self.call("readBlob", &[blob_hash.into()], true).await?).await?;
819 if let Some(bytes) = &bytes {
820 meters::record_bytes("binding.read_blob", &self.key, 0, bytes.len() as u64);
821 }
822 if let Some(bytes) = bytes.as_ref().filter(|bytes| bytes.len() <= MAX_CACHED_BLOB) {
823 self.keep("blob", blob_hash, bytes.clone()).await;
824 }
825 Ok(bytes)
826 }
827
828 async fn read_file(&self, git_ref: &str, path: &str) -> Result<Option<Vec<u8>>> {
829 let key = file_key(git_ref, path, self.refs_version);
830 if let Some(key) = &key
831 && let Some(bytes) = self.get_key(key).await
832 {
833 return Ok(Some(bytes));
834 }
835 let args = js::to_js(&serde_json::json!({ "ref": git_ref, "path": path }))?;
836 let bytes = blob_bytes(self.call("readFile", &[args], true).await?).await?;
837 if let Some(bytes) = &bytes {
838 meters::record_bytes("binding.read_file", &self.key, 0, bytes.len() as u64);
839 }
840 if let (Some(key), Some(bytes)) = (&key, bytes.as_ref().filter(|bytes| bytes.len() <= MAX_CACHED_BLOB)) {
841 self.put_key(key, bytes.clone()).await;
842 }
843 Ok(bytes)
844 }
845
846 async fn fork(&self, target_key: &str) -> Result<()> {
847 let (namespace, name) = locate(target_key);
848 if namespace != self.namespace {
849 return Err(worker::Error::RustError(format!(
850 "a fork stays in its repository's namespace: {target_key} is not in {}",
851 self.namespace
852 )));
853 }
854 let options = js::to_js(&serde_json::json!({ "defaultBranchOnly": true }))?;
855 match self.call("fork", &[name.as_str().into(), options], false).await {
856 Err(failed) if !failed.is("ALREADY_EXISTS") => Err(failed.into()),
857 _ => Ok(()),
858 }
859 }
860
861 fn at_refs_version(&mut self, version: Option<u64>) {
862 self.refs_version = version;
863 }
864}
865
866#[cfg(test)]
867mod tests {
868 use super::*;
869
870 fn access(token: &str) -> GitAccess {
871 GitAccess {
872 remote: "https://store.example/acme--rocket.git".to_owned(),
873 token: token.to_owned(),
874 }
875 }
876
877 #[test]
878 fn internal_credentials_live_an_hour_and_are_reused_for_fifty_minutes() {
879 assert_eq!(Use::Internal.ttl_seconds(), 3_600);
880 assert_eq!(Use::Internal.reuse_ms(), 50 * 60 * 1000);
881 // Handed out: five minutes, reused three, so at least two are left.
882 assert_eq!(Use::Handout.ttl_seconds(), 300);
883 assert_eq!(Use::Handout.reuse_ms(), 180_000);
884 assert_eq!(CREDENTIAL_LIFE_MS, 300_000);
885 for using in [Use::Internal, Use::Handout] {
886 assert!(u64::from(using.ttl_seconds()) * 1000 - using.reuse_ms() >= 120_000);
887 }
888 }
889
890 #[test]
891 fn a_credential_is_reused_only_while_it_has_time_left() {
892 let mut kept = Credentials::default();
893 kept.keep("acme--rocket", Scope::Read, Use::Internal, access("r1"), 1_000, 1_000);
894 assert_eq!(kept.get("acme--rocket", Scope::Read, Use::Internal, 1_000).unwrap().token, "r1");
895 let last = 1_000 + INTERNAL_REUSE_MS - 1;
896 assert_eq!(kept.get("acme--rocket", Scope::Read, Use::Internal, last).unwrap().token, "r1");
897 assert!(kept.get("acme--rocket", Scope::Read, Use::Internal, last + 1).is_none());
898 }
899
900 #[test]
901 fn a_credential_is_kept_for_its_own_repository_scope_and_use() {
902 let mut kept = Credentials::default();
903 kept.keep("acme--rocket", Scope::Read, Use::Internal, access("r1"), 1_000, 1_000);
904 // A read credential never stands in for a write one.
905 assert!(kept.get("acme--rocket", Scope::Write, Use::Internal, 1_000).is_none());
906 assert!(kept.get("acme--booster", Scope::Read, Use::Internal, 1_000).is_none());
907 // An hour-long credential is never handed out.
908 assert!(kept.get("acme--rocket", Scope::Read, Use::Handout, 1_000).is_none());
909 kept.keep("acme--rocket", Scope::Read, Use::Handout, access("h1"), 1_000, 1_000);
910 assert_eq!(kept.get("acme--rocket", Scope::Read, Use::Handout, 1_000).unwrap().token, "h1");
911 assert!(kept.get("acme--rocket", Scope::Read, Use::Handout, 1_000 + HANDOUT_REUSE_MS).is_none());
912 assert!(kept.get("acme--rocket", Scope::Read, Use::Internal, 1_000 + HANDOUT_REUSE_MS).is_some());
913 }
914
915 #[test]
916 fn a_shared_credential_is_reused_only_in_its_own_window() {
917 let value = serde_json::to_vec(&SharedCredential {
918 remote: "https://store.example/acme--rocket.git".to_owned(),
919 token: "r1".to_owned(),
920 made: 10_000,
921 })
922 .unwrap();
923 let (access, made) = shared_credential(&value, 10_000 + INTERNAL_REUSE_MS - 1, Use::Internal).unwrap();
924 assert_eq!(access.token, "r1");
925 // Kept here only for what is left of its window, not a new one.
926 assert_eq!(made, 10_000);
927 assert!(shared_credential(&value, 10_000 + INTERNAL_REUSE_MS, Use::Internal).is_none());
928 assert!(shared_credential(&value, 10_000 + HANDOUT_REUSE_MS, Use::Handout).is_none());
929 assert!(shared_credential(b"not json", 10_000, Use::Internal).is_none());
930 // Each repository, scope and use has its own key, and none is read
931 // from before uses differed.
932 assert_eq!(shared_key("acme--rocket", Scope::Read, Use::Internal), "cred2:acme--rocket:read:internal");
933 assert_ne!(shared_key("acme--rocket", Scope::Read, Use::Internal), shared_key("acme--rocket", Scope::Write, Use::Internal));
934 assert_ne!(shared_key("acme--rocket", Scope::Read, Use::Internal), shared_key("acme--rocket", Scope::Read, Use::Handout));
935 }
936
937 #[test]
938 fn a_shared_credential_kept_here_expires_with_the_original() {
939 let mut kept = Credentials::default();
940 // Made at 1_000 elsewhere, found here at 100_000.
941 kept.keep("acme--rocket", Scope::Read, Use::Internal, access("r1"), 1_000, 100_000);
942 assert!(kept.get("acme--rocket", Scope::Read, Use::Internal, 100_000).is_some());
943 assert!(kept.get("acme--rocket", Scope::Read, Use::Internal, 1_000 + INTERNAL_REUSE_MS).is_none());
944 }
945
946 #[test]
947 fn a_turned_down_credential_is_forgotten_and_old_ones_are_dropped() {
948 let mut kept = Credentials::default();
949 kept.keep("acme--rocket", Scope::Read, Use::Internal, access("r1"), 1_000, 1_000);
950 kept.keep("acme--rocket", Scope::Write, Use::Handout, access("w1"), 1_000, 1_000);
951 kept.keep("acme--booster", Scope::Read, Use::Internal, access("b1"), 1_000, 1_000);
952 kept.forget("acme--rocket");
953 assert!(kept.get("acme--rocket", Scope::Read, Use::Internal, 1_000).is_none());
954 assert!(kept.get("acme--rocket", Scope::Write, Use::Handout, 1_000).is_none());
955 assert!(kept.get("acme--booster", Scope::Read, Use::Internal, 1_000).is_some());
956 // Keeping another later drops the expired one from the map.
957 let later = 1_000 + INTERNAL_REUSE_MS;
958 kept.keep("acme--other", Scope::Read, Use::Internal, access("o1"), later, later);
959 assert_eq!(kept.kept.len(), 1);
960 }
961
962 #[test]
963 fn a_remote_is_worked_out_from_where_its_namespace_starts() {
964 // As https://developers.cloudflare.com/artifacts/api/git-protocol/ documents.
965 let remote = "https://1e6f2cffa3f445920836e8ebe446bb58.artifacts.cloudflare.net/git/g1t/acme--rocket.git";
966 let prefix = learn_prefix(remote, "acme--rocket").unwrap();
967 assert_eq!(prefix, "https://1e6f2cffa3f445920836e8ebe446bb58.artifacts.cloudflare.net/git/g1t/");
968 assert_eq!(remote_from(&prefix, "pulls--pul_1"), prefix.clone() + "pulls--pul_1.git");
969 assert_eq!(remote_from(&prefix, "acme--rocket"), remote);
970 // A remote that does not end in the name teaches nothing.
971 assert_eq!(learn_prefix(remote, "rocket"), None);
972 assert_eq!(learn_prefix("https://x/acme--rocket", "acme--rocket"), None);
973 // And back: a remote names its key.
974 assert_eq!(key_from_remote(remote).as_deref(), Some("acme--rocket"));
975 assert_eq!(
976 key_from_remote("https://a.artifacts.cloudflare.net/git/g1t-us-1/acme--rocket.git").as_deref(),
977 Some("g1t-us-1/acme--rocket")
978 );
979 assert_eq!(locate("g1t-us-1/acme--rocket"), ("g1t-us-1".to_owned(), "acme--rocket".to_owned()));
980 assert_eq!(locate("acme--rocket"), ("g1t".to_owned(), "acme--rocket".to_owned()));
981 }
982
983 #[test]
984 fn binding_methods_have_snake_case_meters() {
985 assert_eq!(meter_of("createToken"), "binding.create_token");
986 assert_eq!(meter_of("readBlob"), "binding.read_blob");
987 assert_eq!(meter_of("get"), "binding.get");
988 }
989
990 #[test]
991 fn reads_by_name_are_kept_under_the_refs_version_and_by_hash_for_good() {
992 let hash = "a".repeat(40);
993 assert_eq!(log_key(&hash, 1, Some(3)), Some(CacheKey::Forever(format!("log/{hash}-1"))));
994 assert_eq!(log_key(&hash, 1, None), Some(CacheKey::Forever(format!("log/{hash}-1"))));
995 // A branch is kept only when the version is known.
996 assert_eq!(log_key("main", 1, None), None);
997 let v3 = log_key("main", 1, Some(3)).unwrap();
998 assert!(matches!(&v3, CacheKey::Versioned(path) if path.starts_with("vlog/3/")));
999 // A push moves the version and leaves the old answer behind.
1000 assert_ne!(Some(v3), log_key("main", 1, Some(4)));
1001 assert_ne!(log_key("main", 1, Some(3)), log_key("main", 50, Some(3)));
1002 assert_ne!(log_key("main", 1, Some(3)), log_key("dev", 1, Some(3)));
1003 // Odd branch names make a usable address.
1004 assert!(matches!(log_key("fix/#1 %20", 1, Some(3)), Some(CacheKey::Versioned(path)) if !path.contains('#') && !path.contains(' ')));
1005 assert_eq!(branches_key(None), None);
1006 assert_ne!(branches_key(Some(1)), branches_key(Some(2)));
1007 assert!(matches!(file_key(&hash, "src/main.rs", None), Some(CacheKey::Forever(_))));
1008 assert_eq!(file_key("main", "src/main.rs", None), None);
1009 assert_ne!(file_key("main", "a", Some(1)), file_key("main", "b", Some(1)));
1010 assert_ne!(file_key("main", "a", Some(1)), file_key("main", "a", Some(2)));
1011 }
1012
1013 #[test]
1014 fn only_full_lowercase_hashes_are_kept() {
1015 assert!(is_commit_hash(&"a".repeat(40)));
1016 assert!(is_commit_hash(&"0123456789abcdef".repeat(4)));
1017 assert!(!is_commit_hash("main"));
1018 assert!(!is_commit_hash(&"A".repeat(40)));
1019 assert!(!is_commit_hash(&"a".repeat(39)));
1020 }
1021}