Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668use std::collections::{BTreeMap, HashMap};use std::sync::{Arc, Mutex};use std::time::{Duration, Instant};
use bobbin_knot_proxy::MirrorProxy;use chrono::{DateTime, Datelike, Utc};use futures::StreamExt;use http::HeaderMap;use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use serde::Deserialize;use tokio::sync::{Notify, Semaphore};
const COMMIT_STATS_NSID: &str = "sh.tangled.git.temp2.getCommitStats";const FETCH_MONTHS: u32 = 24;const MAX_RESPONSE_BYTES: usize = 64 * 1024;const DEFAULT_CAPACITY: usize = 4_096;const DEFAULT_FRESHNESS: Duration = Duration::from_secs(30);const FAILURE_RETRY: Duration = Duration::from_secs(5);const MAX_CONCURRENT_REFRESHES: usize = 16;
#[derive(Clone, Debug, Deserialize)]struct CommitMonth { start: String, commits: i64,}
#[derive(Debug, Deserialize)]struct CommitStats { months: Vec<CommitMonth>,}
#[derive(Clone)]struct Entry { counts: BTreeMap<i32, i64>, refresh_after: Option<Instant>, refreshing: bool, loaded: bool, notify: Arc<Notify>, access: u64,}
struct CacheState { entries: HashMap<String, Entry>, access: u64,}
pub(crate) struct CommitStatsCache { state: Mutex<CacheState>, capacity: usize, freshness: Duration, refreshes: Arc<Semaphore>, capacity_notify: Arc<Notify>,}
impl Default for CommitStatsCache { fn default() -> Self { Self::new(DEFAULT_CAPACITY, DEFAULT_FRESHNESS) }}
impl CommitStatsCache { fn new(capacity: usize, freshness: Duration) -> Self { Self { state: Mutex::new(CacheState { entries: HashMap::new(), access: 0, }), capacity: capacity.max(1), freshness, refreshes: Arc::new(Semaphore::new(MAX_CONCURRENT_REFRESHES)), capacity_notify: Arc::new(Notify::new()), } }
pub(crate) async fn get( self: &Arc<Self>, mirror: Option<Arc<MirrorProxy>>, actor: &Did<DefaultStr>, ) -> BTreeMap<i32, i64> { let Some(mirror) = mirror else { return BTreeMap::new(); }; let key = actor.to_string();
loop { let (counts, loaded, should_refresh, waiter) = { let mut state = self .state .lock() .expect("commit stats cache mutex poisoned"); state.access = state.access.wrapping_add(1); let access = state.access; let capacity_waiter = if !state.entries.contains_key(&key) { if evict_one(&mut state.entries, self.capacity) { state.entries.insert( key.clone(), Entry { counts: BTreeMap::new(), refresh_after: None, refreshing: false, loaded: false, notify: Arc::new(Notify::new()), access, }, ); None } else { Some(Arc::clone(&self.capacity_notify).notified_owned()) } } else { None }; if let Some(waiter) = capacity_waiter { (BTreeMap::new(), false, false, Some(waiter)) } else { let entry = state.entries.get_mut(&key).expect("entry inserted above"); entry.access = access; if !entry.loaded && entry.refreshing { ( BTreeMap::new(), false, false, Some(Arc::clone(&entry.notify).notified_owned()), ) } else { let fresh = entry .refresh_after .is_some_and(|refresh_after| Instant::now() < refresh_after); let should_refresh = !fresh && !entry.refreshing; if should_refresh { entry.refreshing = true; } (entry.counts.clone(), entry.loaded, should_refresh, None) } } };
if loaded { if should_refresh { match Arc::clone(&self.refreshes).try_acquire_owned() { Ok(permit) => { let cache = Arc::clone(self); let mirror = Arc::clone(&mirror); let key = key.clone(); let mut guard = RefreshGuard::new(Arc::clone(&cache), key.clone()); tokio::spawn(async move { let result = fetch_commit_stats(&mirror, &key).await; drop(permit); match result { Ok(counts) => cache.finish(key, Some(counts)), Err(error) => { tracing::warn!(actor = %key, %error, "refreshing commit stats failed"); cache.finish(key, None); } } guard.disarm(); }); } Err(_) => self.finish(key.clone(), None), } } return counts; }
if let Some(waiter) = waiter { waiter.await; continue; }
let mut guard = RefreshGuard::new(Arc::clone(self), key.clone()); let Ok(permit) = Arc::clone(&self.refreshes).acquire_owned().await else { self.finish(key, None); guard.disarm(); return BTreeMap::new(); }; let result = fetch_commit_stats(&mirror, &key).await; drop(permit); return match result { Ok(counts) => { self.finish(key, Some(counts.clone())); guard.disarm(); counts } Err(error) => { tracing::warn!(actor = %key, %error, "loading commit stats failed"); self.finish(key, None); guard.disarm(); BTreeMap::new() } }; } }
fn finish(&self, key: String, counts: Option<BTreeMap<i32, i64>>) { let mut state = self .state .lock() .expect("commit stats cache mutex poisoned"); state.access = state.access.wrapping_add(1); let access = state.access; let Some(entry) = state.entries.get_mut(&key) else { return; }; entry.refreshing = false; entry.loaded = true; entry.access = access; match counts { Some(counts) => { entry.counts = counts; entry.refresh_after = Some(Instant::now() + self.freshness); } None => entry.refresh_after = Some(Instant::now() + FAILURE_RETRY), } let notify = Arc::clone(&entry.notify); drop(state); notify.notify_waiters(); self.capacity_notify.notify_waiters(); }
fn cancel_refresh(&self, key: &str) { let mut state = self .state .lock() .expect("commit stats cache mutex poisoned"); let Some(entry) = state.entries.get_mut(key) else { return; }; entry.refreshing = false; let notify = Arc::clone(&entry.notify); drop(state); notify.notify_waiters(); self.capacity_notify.notify_waiters(); }
#[cfg(test)] fn seed(&self, actor: &str, counts: BTreeMap<i32, i64>, refresh_after: Option<Instant>) { self.state .lock() .expect("commit stats cache mutex poisoned") .entries .insert( actor.to_owned(), Entry { counts, refresh_after, refreshing: false, loaded: true, notify: Arc::new(Notify::new()), access: 0, }, ); }}
struct RefreshGuard { cache: Arc<CommitStatsCache>, key: String, active: bool,}
impl RefreshGuard { fn new(cache: Arc<CommitStatsCache>, key: String) -> Self { Self { cache, key, active: true, } }
fn disarm(&mut self) { self.active = false; }}
impl Drop for RefreshGuard { fn drop(&mut self) { if self.active { self.cache.cancel_refresh(&self.key); } }}
fn evict_one(entries: &mut HashMap<String, Entry>, capacity: usize) -> bool { if entries.len() < capacity { return true; } let Some(key) = entries .iter() .filter(|(_, entry)| !entry.refreshing) .min_by_key(|(_, entry)| entry.access) .map(|(key, _)| key.clone()) else { return false; }; entries.remove(&key); true}
async fn fetch_commit_stats( mirror: &MirrorProxy, actor: &str,) -> Result<BTreeMap<i32, i64>, String> { let nsid = Nsid::new_static(COMMIT_STATS_NSID).expect("commit stats NSID is valid"); let months = FETCH_MONTHS.to_string(); let response = mirror .forward_raw( &nsid, &[("actor", actor), ("months", months.as_str())], HeaderMap::new(), ) .await .map_err(|error| error.to_string())?; if !response.status().is_success() { let status = response.status(); response.discard().await; return Err(format!("mirror returned {status}")); }
let mut body = response.into_body_stream(); let mut bytes = Vec::new(); while let Some(chunk) = body.next().await { let chunk = chunk.map_err(|error| error.to_string())?; if bytes.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { return Err("mirror commit stats response is too large".to_owned()); } bytes.extend_from_slice(&chunk); } let output: CommitStats = serde_json::from_slice(&bytes).map_err(|error| error.to_string())?; if output.months.len() > FETCH_MONTHS as usize { return Err("mirror returned too many commit stat months".to_owned()); } output .months .into_iter() .try_fold(BTreeMap::new(), |mut counts, month| -> Result<_, String> { let start = DateTime::parse_from_rfc3339(&month.start) .map_err(|error| format!("invalid month start: {error}"))? .with_timezone(&Utc); if month.commits < 0 { return Err("mirror returned a negative commit count".to_owned()); } counts.insert(month_number(start), month.commits); Ok(counts) })}
pub(crate) fn month_number(value: DateTime<Utc>) -> i32 { value.year() * 12 + value.month0() as i32}
#[cfg(test)]mod tests { use super::*; use bobbin_runtime::{RuntimeHasher, SystemClock}; use serde_json::json; use wiremock::matchers::{method, path, query_param}; use wiremock::{Mock, MockServer, ResponseTemplate};
fn actor() -> Did<DefaultStr> { Did::new_static("did:plc:pushcacheactor").unwrap() }
fn mirror(server: &MockServer) -> Arc<MirrorProxy> { Arc::new( MirrorProxy::new( &url::Url::parse(&server.uri()).unwrap(), Arc::new(SystemClock::new()), RuntimeHasher::default(), ) .unwrap(), ) }
#[tokio::test] async fn a_cold_read_waits_for_the_first_value() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats")) .and(query_param("actor", actor().as_ref())) .and(query_param("months", "24")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "months": [{"start": "2026-09-01T00:00:00Z", "commits": 7}] }))) .expect(1) .mount(&server) .await;
let cache = Arc::new(CommitStatsCache::new(4, Duration::from_secs(60))); let counts = cache.get(Some(mirror(&server)), &actor()).await; assert_eq!(counts.get(&(2026 * 12 + 8)), Some(&7)); assert_eq!( cache .get(Some(mirror(&server)), &actor()) .await .get(&(2026 * 12 + 8)), Some(&7), ); server.verify().await; }
#[tokio::test] async fn rejects_more_months_than_requested() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "months": (0..=FETCH_MONTHS) .map(|month| json!({ "start": format!("2024-{:02}-01T00:00:00Z", month % 12 + 1), "commits": 1, })) .collect::<Vec<_>>() }))) .mount(&server) .await;
let error = fetch_commit_stats(&mirror(&server), actor().as_ref()) .await .unwrap_err(); assert!(error.contains("too many commit stat months")); }
#[tokio::test] async fn concurrent_cold_reads_share_one_request() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats")) .respond_with( ResponseTemplate::new(200) .set_delay(Duration::from_millis(25)) .set_body_json(json!({ "months": [{"start": "2026-09-01T00:00:00Z", "commits": 7}] })), ) .expect(1) .mount(&server) .await; let cache = Arc::new(CommitStatsCache::new(4, Duration::from_secs(60))); let actor = actor(); let first = cache.get(Some(mirror(&server)), &actor); let second = cache.get(Some(mirror(&server)), &actor);
let (first, second) = tokio::join!(first, second); assert_eq!(first.get(&(2026 * 12 + 8)), Some(&7)); assert_eq!(second.get(&(2026 * 12 + 8)), Some(&7)); server.verify().await; }
#[tokio::test] async fn a_cold_read_waits_for_a_bounded_refresh_permit() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({"months": []}))) .expect(1) .mount(&server) .await; let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO)); let permits = Arc::clone(&cache.refreshes) .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32) .await .unwrap(); let read = tokio::spawn({ let cache = Arc::clone(&cache); let mirror = mirror(&server); async move { cache.get(Some(mirror), &actor()).await } });
for _ in 0..50 { if cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| entry.refreshing) { break; } tokio::task::yield_now().await; } assert!( cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| entry.refreshing && !entry.loaded) ); assert!(server.received_requests().await.unwrap().is_empty()); drop(permits); assert!(read.await.unwrap().is_empty()); server.verify().await; }
#[tokio::test] async fn cancelling_a_cold_read_releases_waiters() { let server = MockServer::start().await; let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO)); let permits = Arc::clone(&cache.refreshes) .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32) .await .unwrap(); let read = tokio::spawn({ let cache = Arc::clone(&cache); let mirror = mirror(&server); async move { cache.get(Some(mirror), &actor()).await } }); for _ in 0..50 { if cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| entry.refreshing) { break; } tokio::task::yield_now().await; } assert!( cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| entry.refreshing && !entry.loaded) ); read.abort(); assert!(read.await.unwrap_err().is_cancelled()); assert!( cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| !entry.refreshing && !entry.loaded) ); drop(permits); }
#[tokio::test] async fn cold_entries_do_not_exceed_capacity_while_refreshes_wait() { let server = MockServer::start().await; let cache = Arc::new(CommitStatsCache::new(1, Duration::ZERO)); let permits = Arc::clone(&cache.refreshes) .acquire_many_owned(MAX_CONCURRENT_REFRESHES as u32) .await .unwrap(); let first_actor = Did::new_static("did:plc:firstcoldactor").unwrap(); let second_actor = Did::new_static("did:plc:secondcoldactor").unwrap(); let first = tokio::spawn({ let cache = Arc::clone(&cache); let mirror = mirror(&server); async move { cache.get(Some(mirror), &first_actor).await } }); for _ in 0..50 { if cache.state.lock().unwrap().entries.len() == 1 { break; } tokio::task::yield_now().await; } let second = tokio::spawn({ let cache = Arc::clone(&cache); let mirror = mirror(&server); async move { cache.get(Some(mirror), &second_actor).await } }); for _ in 0..50 { tokio::task::yield_now().await; } assert_eq!(cache.state.lock().unwrap().entries.len(), 1);
first.abort(); assert!(first.await.unwrap_err().is_cancelled()); for _ in 0..50 { if cache .state .lock() .unwrap() .entries .contains_key("did:plc:secondcoldactor") { break; } tokio::task::yield_now().await; } let state = cache.state.lock().unwrap(); assert_eq!(state.entries.len(), 1); assert!(state.entries.contains_key("did:plc:secondcoldactor")); drop(state); second.abort(); assert!(second.await.unwrap_err().is_cancelled()); drop(permits); }
#[test] fn dropping_a_refresh_guard_clears_stale_refresh_state() { let cache = Arc::new(CommitStatsCache::new(1, Duration::ZERO)); cache.seed(actor().as_ref(), BTreeMap::new(), Some(Instant::now())); cache .state .lock() .unwrap() .entries .get_mut(actor().as_ref()) .unwrap() .refreshing = true; drop(RefreshGuard::new(Arc::clone(&cache), actor().to_string())); assert!( !cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .unwrap() .refreshing ); }
#[tokio::test] async fn a_failed_refresh_keeps_stale_counts() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/xrpc/sh.tangled.git.temp2.getCommitStats")) .respond_with(ResponseTemplate::new(503)) .expect(1) .mount(&server) .await;
let cache = Arc::new(CommitStatsCache::new(4, Duration::ZERO)); cache.seed( actor().as_ref(), BTreeMap::from([(2026 * 12 + 8, 5)]), Some(Instant::now()), ); let counts = cache.get(Some(mirror(&server)), &actor()).await; assert_eq!(counts.get(&(2026 * 12 + 8)), Some(&5)); for _ in 0..50 { let refreshing = cache .state .lock() .unwrap() .entries .get(actor().as_ref()) .is_some_and(|entry| entry.refreshing); if !refreshing { break; } tokio::time::sleep(Duration::from_millis(10)).await; } assert_eq!( cache .get(Some(mirror(&server)), &actor()) .await .get(&(2026 * 12 + 8)), Some(&5), ); server.verify().await; }}