Sign in

flagon-io/g1t

public

Commit

Fewer Artifacts reads: the store is asked for a handle only when needed, objects are kept in the isolate, and issue events read no workflows nobody listens for

On 2026-10-06 Cloudflare counted 107,616 read events against g1t's demo-scale traffic; reads may be billed from 2026-10-14. - repos: an ArtifactsRepo asks the store for its handle (get) on the first call that needs it, not when opened, so a cached answer or a fetch over git with a kept credential costs no get. A fork already gone still reads as gone when retired. - repos: objects named by hash are kept in the isolate (16 MB, oldest out first) ahead of the Cache API, and every look is metered: cache.memory_hit, cache.edge_hit, cache.miss. - actions: issue and comment events skip reading workflows from git when the synced table says none listens for them (a file that did not parse still counts, so its error is reported). - docs/ARTIFACTS.md: what was done and what to read next.

syntaqxcommitted Parentcfc1609Browse files
5 files+178−210/5 viewed
+8−3
413413 1,000, today's demo-scale traffic alone is about 3.2 million a month (~$480). Ask Cloudflare
414414 (Q1) before 2026-10-14. Either way the volume is mostly waste: every repos call opens a handle
415415 with `get` even when the answer is cached, and the object cache may not be hitting (no hit/miss
416− meter yet). Fixes, ranked: lazy `get`; a real cache for objects named by hash (KV or an
417− in-isolate LRU, with hit/miss meters); Actions reading workflows from the synced table instead
418− of the store on every event; caller attribution in the meters.
416+ meter yet). Done on 2026-10-06: the handle's `get` waits for the first call that needs the
417+ store (an answer from a cache, or `branches` over git, costs none); objects named by hash are
418+ kept in the isolate (16 MB, oldest out first) ahead of the Cache API, and every look is
419+ metered (`cache.memory_hit`, `cache.edge_hit`, `cache.miss`); issue and comment events start
420+ nothing and read nothing when the synced `workflows` table has no workflow listening. Next:
421+ read `cache.edge_hit` against `cache.miss` after a day; if the Cache API never hits from a
422+ Worker reached only by service bindings, put objects in KV instead. Still to do: caller
423+ attribution in the meters.
419424 - 476 client errors on 2026-10-06 are unexplained; the fetch fix below accounts for some (every
420425 failed negotiation was one).
421426
+51−2
4040 pub updated_at: String,
4141 }
4242
43+/// What a workflow's row says once its file has left the default branch.
44+pub const GONE: &str = "Its file is no longer on the default branch.";
45+
46+/// Whether a synced workflow could start on `event`: it lists the event,
47+/// or it did not parse (and is still on the branch), so reading it again
48+/// reports why.
49+pub fn could_start(events: &str, error: Option<&str>, event: &str) -> bool {
50+ match error {
51+ Some(error) => error != GONE,
52+ None => serde_json::from_str::<Vec<String>>(events).is_ok_and(|events| events.iter().any(|e| e == event)),
53+ }
54+}
55+
4356 impl Actions {
4457 /// The workflow files of `path` as of `git_ref` (the default branch
4558 /// when absent).
146159 for row in existing.iter().filter(|row| !kept.contains(&row.path)) {
147160 statements.push(
148161 self.db
149− .prepare("UPDATE workflows SET crons = '[]', error = 'Its file is no longer on the default branch.' WHERE id = ?")
150− .bind(&[row.id.as_str().into()])?,
162+ .prepare("UPDATE workflows SET crons = '[]', error = ? WHERE id = ?")
163+ .bind(&[GONE.into(), row.id.as_str().into()])?,
151164 );
152165 }
153166 statements.push(
159172 Ok(())
160173 }
161174
175+ /// Whether any of the repository's default-branch workflows could
176+ /// start on `event`, from the synced table; None before the first sync.
177+ pub async fn listens(&self, repo_id: &str, event: &str) -> Result<Option<bool>> {
178+ #[derive(Deserialize)]
179+ struct Row {
180+ events: String,
181+ error: Option<String>,
182+ }
183+ if !self.synced(repo_id).await? {
184+ return Ok(None);
185+ }
186+ let rows = self
187+ .db
188+ .prepare("SELECT events, error FROM workflows WHERE repo_id = ? AND state = 'active'")
189+ .bind(&[repo_id.into()])?
190+ .all()
191+ .await?
192+ .results::<Row>()?;
193+ Ok(Some(rows.iter().any(|row| could_start(&row.events, row.error.as_deref(), event))))
194+ }
195+
162196 pub async fn synced(&self, repo_id: &str) -> Result<bool> {
163197 Ok(self
164198 .db
198232 .ok_or_else(|| worker::Error::RustError("the workflow was not recorded".into()))
199233 }
200234 }
235+
236+#[cfg(test)]
237+mod tests {
238+ use super::*;
239+
240+ #[test]
241+ fn a_synced_workflow_starts_on_the_events_it_lists_or_when_broken() {
242+ assert!(could_start(r#"["push","issues"]"#, None, "issues"));
243+ assert!(!could_start(r#"["push","pull_request"]"#, None, "issue_comment"));
244+ // A file that does not parse is read again, so its error shows.
245+ assert!(could_start("[]", Some("bad yaml"), "issues"));
246+ // A file gone from the branch starts nothing.
247+ assert!(!could_start("[]", Some(GONE), "issues"));
248+ }
249+}
+6−0
347347 }
348348 let sender = self.username(event.actor.as_deref()).await?.unwrap_or_else(|| repo.namespace.clone());
349349 for (event_name, action) in mapped {
350+ // Issues and comments start the default branch's workflows,
351+ // which the synced table lists: when none listens, nothing is
352+ // read from git. Agents make many of these events.
353+ if matches!(event_name, "issues" | "issue_comment") && self.listens(repo_id, event_name).await? == Some(false) {
354+ continue;
355+ }
350356 let Some(mut subject) = self.subject(event, event_name, action, &repo, &ws, &sender).await? else {
351357 continue;
352358 };
+7−3
284284 Some(id) => self.registry.by_id(id).await?,
285285 None => None,
286286 };
287− let head = match self.store.open(&key).await {
288− Ok(git) => git.log(&fork.default_branch, 1).await?.into_iter().next().map(|commit| commit.hash),
289− // Already gone from the store.
287+ // Already gone from the store: it says so when first asked.
288+ let read = match self.store.open(&key).await {
289+ Ok(git) => git.log(&fork.default_branch, 1).await,
290+ Err(error) => Err(error),
291+ };
292+ let head = match read {
293+ Ok(commits) => commits.into_iter().next().map(|commit| commit.hash),
290294 Err(error) if error.to_string().contains("NOT_FOUND") => None,
291295 Err(error) => return Err(error),
292296 };
+106−13
502502
503503 async fn open(&self, key: &str) -> Result<ArtifactsRepo> {
504504 let (namespace, name) = locate(key);
505− let handle = invoke(&namespace, key, self.binding(&namespace)?, "get", &[name.as_str().into()], true).await?;
506505 Ok(ArtifactsRepo {
507− handle,
506+ handle: RefCell::new(None),
507+ binding: self.binding(&namespace)?.clone(),
508508 key: key.to_owned(),
509509 name,
510510 namespace,
525525 /// A handle to one Artifacts repository. It is an RPC stub, so it is
526526 /// released when dropped.
527527 pub struct ArtifactsRepo {
528− handle: JsValue,
528+ /// The store's handle, asked for (`get`) on the first call that needs
529+ /// the store: an answer from a cache, or a fetch over git with a kept
530+ /// credential, never costs a `get`.
531+ handle: RefCell<Option<JsValue>>,
532+ /// The namespace's binding, for that `get`.
533+ binding: JsValue,
529534 /// The repository's store key, which scopes its cached objects.
530535 key: String,
531536 /// Its name in its namespace.
584589 version.map(|version| CacheKey::Versioned(format!("branches/{version}")))
585590 }
586591
592+/// Objects named by their content, kept in the isolate ahead of the Cache
593+/// API, oldest out first past `MEMORY_CACHE_BYTES`. They never go stale,
594+/// and each is under its repository's key.
595+const MEMORY_CACHE_BYTES: usize = 16 * 1024 * 1024;
596+
597+#[derive(Default)]
598+struct MemoryCache {
599+ entries: HashMap<String, Rc<Vec<u8>>>,
600+ order: std::collections::VecDeque<String>,
601+ bytes: usize,
602+}
603+
604+impl MemoryCache {
605+ fn get(&self, url: &str) -> Option<Vec<u8>> {
606+ self.entries.get(url).map(|bytes| bytes.as_ref().clone())
607+ }
608+
609+ fn put(&mut self, url: String, bytes: &[u8]) {
610+ if bytes.len() > MEMORY_CACHE_BYTES / 16 || self.entries.contains_key(&url) {
611+ return;
612+ }
613+ self.bytes += bytes.len();
614+ self.entries.insert(url.clone(), Rc::new(bytes.to_vec()));
615+ self.order.push_back(url);
616+ while self.bytes > MEMORY_CACHE_BYTES {
617+ let Some(oldest) = self.order.pop_front() else { break };
618+ if let Some(gone) = self.entries.remove(&oldest) {
619+ self.bytes -= gone.len();
620+ }
621+ }
622+ }
623+}
624+
625+thread_local! {
626+ static MEMORY: RefCell<MemoryCache> = RefCell::new(MemoryCache::default());
627+}
628+
587629 impl ArtifactsRepo {
588630 fn cache_url(&self, path: &str) -> String {
589631 format!("{OBJECT_CACHE}{}/{path}", self.key)
590632 }
591633
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()
634+ /// A kept answer: from the isolate for one kept for good, else the
635+ /// Cache API. Each look is metered (`cache.memory_hit`, `cache.edge_hit`,
636+ /// `cache.miss`), so the usage check shows which cache answers.
637+ async fn cached_at(&self, path: &str, forever: bool) -> Option<Vec<u8>> {
638+ let url = self.cache_url(path);
639+ if forever && let Some(bytes) = MEMORY.with(|memory| memory.borrow().get(&url)) {
640+ meters::record_bytes("cache.memory_hit", &self.key, 0, bytes.len() as u64);
641+ return Some(bytes);
642+ }
643+ let found = match worker::Cache::default().get(url.clone(), false).await {
644+ Ok(Some(mut response)) => response.bytes().await.ok(),
645+ _ => None,
646+ };
647+ match &found {
648+ Some(bytes) => {
649+ meters::record_bytes("cache.edge_hit", &self.key, 0, bytes.len() as u64);
650+ if forever {
651+ MEMORY.with(|memory| memory.borrow_mut().put(url, bytes));
652+ }
653+ }
654+ None => meters::record("cache.miss", &self.key, 0, 0),
655+ }
656+ found
595657 }
596658
597659 async fn cached(&self, kind: &str, hash: &str) -> Option<Vec<u8>> {
598− self.cached_at(&format!("{kind}/{hash}")).await
660+ self.cached_at(&format!("{kind}/{hash}"), true).await
599661 }
600662
601663 async fn keep_at(&self, path: &str, bytes: Vec<u8>, max_age: &str) {
664+ if max_age == OBJECT_MAX_AGE {
665+ MEMORY.with(|memory| memory.borrow_mut().put(self.cache_url(path), &bytes));
666+ }
602667 let Ok(mut response) = worker::Response::from_bytes(bytes) else {
603668 return;
604669 };
613678
614679 async fn get_key(&self, key: &CacheKey) -> Option<Vec<u8>> {
615680 match key {
616− CacheKey::Forever(path) | CacheKey::Versioned(path) => self.cached_at(path).await,
681+ CacheKey::Forever(path) => self.cached_at(path, true).await,
682+ CacheKey::Versioned(path) => self.cached_at(path, false).await,
617683 }
618684 }
619685
625691 }
626692
627693 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
694+ let handle = self.handle().await?;
695+ invoke(&self.namespace, &self.key, &handle, method, args, retry).await
629696 }
630697
698+ /// The store's handle, asked for the first time it is needed.
699+ async fn handle(&self) -> std::result::Result<JsValue, StoreError> {
700+ if let Some(handle) = self.handle.borrow().as_ref() {
701+ return Ok(handle.clone());
702+ }
703+ let handle = invoke(&self.namespace, &self.key, &self.binding, "get", &[self.name.as_str().into()], true).await?;
704+ *self.handle.borrow_mut() = Some(handle.clone());
705+ Ok(handle)
706+ }
707+
631708 /// A new credential from the store. Its remote is worked out from the
632709 /// key once this isolate knows where the namespace's remotes start;
633710 /// until then the store is asked (`info()`) alongside the token.
638715 let token: RawToken = js::from_js(&self.call("createToken", &args, true).await?)?;
639716 return Ok(GitAccess { remote: remote_from(&prefix, &self.name), token: token.plaintext });
640717 }
718+ self.handle().await?;
641719 let (info, token) = futures_util::future::join(self.call("info", &[], true), self.call("createToken", &args, true)).await;
642720 let info: RawInfo = js::from_js(&info?)?;
643721 let token: RawToken = js::from_js(&token?)?;
655733 fn drop(&mut self) {
656734 let symbol = js::get(&worker::js_sys::global(), "Symbol");
657735 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);
736+ let Some(handle) = self.handle.borrow_mut().take() else { return };
737+ if let Ok(function) = Reflect::get(&handle, &dispose).and_then(|value| value.dyn_into::<worker::js_sys::Function>()) {
738+ let _ = function.call0(&handle);
662739 }
663740 }
664741 }
9811058 }
9821059
9831060 #[test]
1061+ fn the_memory_cache_drops_its_oldest_past_its_budget() {
1062+ let mut cache = MemoryCache::default();
1063+ let chunk = vec![7u8; MEMORY_CACHE_BYTES / 16];
1064+ for n in 0..17 {
1065+ cache.put(format!("k{n}"), &chunk);
1066+ }
1067+ // Sixteen chunks fit; the seventeenth pushed the first out.
1068+ assert!(cache.get("k0").is_none());
1069+ assert_eq!(cache.get("k16").map(|b| b.len()), Some(chunk.len()));
1070+ assert!(cache.bytes <= MEMORY_CACHE_BYTES);
1071+ // Too large to keep at all.
1072+ cache.put("big".into(), &vec![0u8; MEMORY_CACHE_BYTES / 16 + 1]);
1073+ assert!(cache.get("big").is_none());
1074+ }
1075+
1076+ #[test]
9841077 fn binding_methods_have_snake_case_meters() {
9851078 assert_eq!(meter_of("createToken"), "binding.create_token");
9861079 assert_eq!(meter_of("readBlob"), "binding.read_blob");