Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066//! Scope-keyed last-good catalogs with coalesced refresh and disk coherence.//!//! A source owns one path. Scope changes discard the active view. Persistent//! locks serialize publishers, while durable scoped tombstones invalidate even//! writers that fetched before a revocation. Readers observe external writes.
use std::fs::File;use std::future::Future;use std::io::Write;use std::path::{Path, PathBuf};use std::pin::Pin;use std::sync::{Arc, Mutex, OnceLock};use std::time::{Duration, SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};use sha2::{Digest, Sha256};
use crate::fetch::{CatalogFetcher, FetchError, FetchOutcome, MAX_RESPONSE_BYTES};use crate::offline::is_catalog_offline;use crate::CATALOG_REFRESH_INTERVAL_MS;
/// The unauthenticated scope shared by the aggregate catalog URLs.pub const PUBLIC_SCOPE: &str = "public";
/// Parser turning a fetched payload into the cache's value type.pub type CatalogParse<T> = Arc<dyn Fn(&serde_json::Value, &str) -> Result<T, String> + Send + Sync>;
/// Merge validated partial metadata with a previous provider-owned payload./// The callback validates the input envelope and prunes absent IDs. The cache/// validates the merged output and persists it, not just its derived models.pub type CatalogMergePayload = Arc< dyn Fn(&serde_json::Value, Option<&serde_json::Value>) -> Result<serde_json::Value, String> + Send + Sync,>;
/// An optional provider-owned fetch (for discovery protocols with multiple/// endpoints). Its result uses the same validation and publication machinery.pub type CatalogFetchPayload = Arc< dyn Fn() -> Pin<Box<dyn Future<Output = Result<serde_json::Value, FetchError>> + Send>> + Send + Sync,>;
/// Credential-free failure categories. Never contain URLs, headers, parser/// diagnostics, file paths, response bodies, or underlying transport errors.#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, thiserror::Error)]#[serde(tag = "code", rename_all = "snake_case")]pub enum CatalogRefreshError { #[error("Invalid catalog JSON")] InvalidJson, #[error("Invalid catalog payload")] InvalidPayload, #[error("Catalog request failed with status {status}")] Http { status: u16 }, #[error("Catalog refresh failed")] RequestFailed, #[error("Catalog cache persistence failed")] PersistenceFailed,}
/// Result category of one refresh (including a shared/coalesced attempt).#[derive(Clone, Debug, PartialEq, Eq, Serialize)]#[serde(tag = "kind", rename_all = "kebab-case")]pub enum CatalogRefreshOutcome { Updated, NotModified, Cached, Offline, Error { error: CatalogRefreshError }, MemoryOnly, Revoked { status: u16 }, Cancelled,}
/// A refresh result with last-good models on recoverable failure.#[derive(Clone, Debug, PartialEq, Eq)]pub struct CatalogRefreshReport<T> { pub models: Option<T>, pub outcome: CatalogRefreshOutcome,}
impl<T> CatalogRefreshReport<T> { fn cancelled() -> Self { Self { models: None, outcome: CatalogRefreshOutcome::Cancelled, } }}
/// The current state of a source's active scope.#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]#[serde(rename_all = "snake_case")]pub enum CatalogCacheState { Idle, Refreshing, Fresh, Stale, Error,}
/// Where the active snapshot came from.#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]#[serde(rename_all = "snake_case")]pub enum CatalogCacheSource { None, Disk, Network, Memory,}
/// Credential-free freshness information for one source.#[derive(Clone, Debug, PartialEq, Eq, Serialize)]#[serde(rename_all = "camelCase")]pub struct CatalogCacheStatus { pub state: CatalogCacheState, pub source: CatalogCacheSource, pub fetched_at: Option<u64>, pub checked_at: Option<u64>, pub last_attempt: Option<u64>, pub error: Option<CatalogRefreshError>,}
#[derive(Clone)]struct Snapshot<T> { file: SnapshotFile, models: T,}
/// Older snapshots have neither checkedAt nor revision. Both remain readable.#[derive(Clone, Serialize, Deserialize)]#[serde(rename_all = "camelCase")]struct SnapshotFile { url: String, scope: String, fetched_at: u64, #[serde(default, skip_serializing_if = "Option::is_none")] checked_at: Option<u64>, #[serde(default, skip_serializing_if = "Option::is_none")] revision: Option<String>, /// The tombstone epoch observed before this request began. Tombstones stay /// on disk so a writer with no prior snapshot cannot resurrect a revocation. #[serde(default, skip_serializing_if = "Option::is_none")] revocation: Option<String>, #[serde(default, skip_serializing_if = "Option::is_none")] etag: Option<String>, payload: serde_json::Value,}
#[derive(Serialize, Deserialize)]struct RevocationFile { url: String, scope: String, revision: String,}
/// Compare content, not mtime or wall-clock ordering: independent processes can/// complete in one millisecond, and revocation can happen without a snapshot.#[derive(Clone, Default, PartialEq, Eq)]struct DiskVersion { snapshot: Option<[u8; 32]>, revoked: Option<[u8; 32]>,}
struct DiskView { version: DiskVersion, snapshot: Option<Vec<u8>>, revoked: Option<Vec<u8>>, unreadable_revocation: bool,}
impl DiskView { fn read(path: &Path, url: &str, scope: &str) -> Self { let snapshot = std::fs::read(path).ok(); // Read the tombstone AFTER the payload so revocation racing a read // hides a snapshot that was just read from disk. let (revoked, unreadable_revocation) = match std::fs::read(revocation_path(path, url, scope)) { Ok(bytes) => (Some(bytes), false), Err(error) if error.kind() == std::io::ErrorKind::NotFound => (None, false), Err(_) => (None, true), }; Self { version: DiskVersion { snapshot: snapshot .as_deref() .map(|bytes| Sha256::digest(bytes).into()), revoked: if unreadable_revocation { Some([0; 32]) } else { revoked.as_deref().map(|bytes| Sha256::digest(bytes).into()) }, }, snapshot, revoked, unreadable_revocation, } }
fn revocation(&self, url: &str, scope: &str) -> Option<String> { let revoked: RevocationFile = serde_json::from_slice(self.revoked.as_deref()?).ok()?; (revoked.url == url && revoked.scope == scope).then_some(revoked.revision) }}
struct InFlight<T> { result: OnceLock<CatalogRefreshReport<T>>, notify: tokio::sync::Notify,}
impl<T> InFlight<T> { fn new() -> Self { Self { result: OnceLock::new(), notify: tokio::sync::Notify::new(), } }}
struct CacheState<T> { scope: Option<String>, snapshot: Option<Snapshot<T>>, observed_disk: DiskVersion, pending: Option<Arc<InFlight<T>>>, last_attempt: Option<u64>, generation: u64, source: CatalogCacheSource, error: Option<CatalogRefreshError>, revoked: bool,}
/// Options for [`CatalogCache::refresh`].#[derive(Default)]pub struct RefreshOptions { /// Skip the hourly gating window (startup, picker open, auth change). pub force: bool, /// Extra request headers. Never included in results or status. pub headers: Vec<(String, String)>, /// Additional staleness guard consulted before applying any result. pub is_current: Option<Arc<dyn Fn() -> bool + Send + Sync>>, /// Provider-specific validation/normalization, before parsing the merged payload. pub merge_payload: Option<CatalogMergePayload>, /// Override the ordinary single-URL HTTP request with provider discovery. pub fetch_payload: Option<CatalogFetchPayload>,}
/// The last-good cache for one catalog source and its active auth scope.pub struct CatalogCache<T> { url: Arc<str>, cache_path: Option<PathBuf>, fetcher: Arc<CatalogFetcher>, parse: CatalogParse<T>, state: Mutex<CacheState<T>>,}
fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .ok() .and_then(|since| u64::try_from(since.as_millis()).ok()) .unwrap_or_default()}
fn revision() -> String { static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .map_or(0, |since| since.as_nanos()); let count = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); format!("{}-{nanos}-{count}", std::process::id())}
fn sidecar(path: &Path, suffix: &str) -> PathBuf { let mut name = path.as_os_str().to_os_string(); name.push(suffix); PathBuf::from(name)}
fn revocation_path(path: &Path, url: &str, scope: &str) -> PathBuf { let mut digest = Sha256::new(); digest.update(url.as_bytes()); digest.update([0]); digest.update(scope.as_bytes()); sidecar( path, &format!(".revoked-{}", hex::encode(digest.finalize())), )}
impl<T: Clone + Send + Sync + 'static> CatalogCache<T> { pub fn new( url: &str, cache_path: Option<PathBuf>, fetcher: Arc<CatalogFetcher>, parse: CatalogParse<T>, ) -> Self { Self { url: Arc::from(url), cache_path, fetcher, parse, state: Mutex::new(CacheState { scope: None, snapshot: None, observed_disk: DiskVersion::default(), pending: None, last_attempt: None, generation: 0, source: CatalogCacheSource::None, error: None, revoked: false, }), } }
pub fn url(&self) -> &str { &self.url }
pub fn cache_path(&self) -> Option<&Path> { self.cache_path.as_deref() }
/// Serve validated last-good data, reloading changed disk state even when /// the scope is unchanged. Unchanged disk state preserves memory-only data. /// A changed scope or disk version invalidates old in-flight publications. /// /// # Panics /// Panics if the cache mutex is poisoned. pub fn get(&self, scope: &str) -> Option<T> { let mut state = self.state.lock().unwrap(); let disk = self .cache_path .as_deref() .map(|path| DiskView::read(path, &self.url, scope)); let changed = state.scope.as_deref() != Some(scope) || disk .as_ref() .is_some_and(|disk| disk.version != state.observed_disk); if changed { state.scope = Some(scope.to_string()); state.generation += 1; state.pending = None; state.last_attempt = None; state.error = None; state.snapshot = disk .as_ref() .and_then(|disk| self.read_snapshot(disk, scope)); state.revoked = state.snapshot.is_none() && disk .as_ref() .is_some_and(|disk| disk.unreadable_revocation || disk.revoked.is_some()); state.observed_disk = disk.map_or_else(DiskVersion::default, |disk| disk.version); state.source = if state.snapshot.is_some() { CatalogCacheSource::Disk } else { CatalogCacheSource::None }; } state .snapshot .as_ref() .map(|snapshot| snapshot.models.clone()) }
/// Force a disk reload and invalidate any prior in-flight publication. /// /// # Panics /// Panics if the cache mutex is poisoned. pub fn reload_from_disk(&self, scope: &str) -> Option<T> { self.state.lock().unwrap().scope = None; self.get(scope) }
/// The stored scope for this URL, including older-format snapshots. Used /// only to seed a fresh process's auth-change comparison. pub fn stored_scope(&self) -> Option<String> { let bytes = std::fs::read(self.cache_path.as_deref()?).ok()?; let stored: SnapshotFile = serde_json::from_slice(&bytes).ok()?; (stored.url == self.url.as_ref()).then_some(stored.scope) }
/// Whether this scope was revoked and has no validated newer snapshot. /// Use this to prevent an older private-cache format from restoring access /// after a rejection. A failed local tombstone write still returns true. /// /// # Panics /// Panics if the cache mutex is poisoned. pub fn has_revocation(&self, scope: &str) -> bool { self.get(scope); self.state.lock().unwrap().revoked }
/// Drop this scope and invalidate old requests across processes. Other /// scopes' snapshots are not removed. The durable tombstone is written /// before trying the publisher lock, so a held lock cannot hide revocation. /// /// # Panics /// Panics if the cache mutex is poisoned. pub fn clear(&self, scope: &str) { self.revoke(scope); }
fn revoke(&self, scope: &str) -> bool { let mut state = self.state.lock().unwrap(); if state.scope.as_deref() != Some(scope) { return true; } state.generation += 1; state.pending = None; state.snapshot = None; state.revoked = true; state.source = CatalogCacheSource::None; state.error = None; state.last_attempt = None; let Some(path) = &self.cache_path else { return true; }; let revocation = RevocationFile { url: self.url.to_string(), scope: scope.to_string(), revision: revision(), }; let persisted_marker = serde_json::to_vec(&revocation).is_ok_and(|bytes| { write_atomic(&revocation_path(path, &self.url, scope), &bytes).is_ok() }); // Never unlink the lock file: its inode must stay shared by all users. // A failed try-lock is safe because the tombstone hides the old payload // and any publisher bound to the previous revocation epoch. if let Some(_lock) = open_lock(path).ok().filter(|lock| lock.try_lock().is_ok()) { let disk = DiskView::read(path, &self.url, scope); let matches = disk.snapshot.as_deref().is_some_and(|bytes| { serde_json::from_slice::<SnapshotFile>(bytes) .is_ok_and(|stored| stored.url == self.url.as_ref() && stored.scope == scope) }); if matches { if let Err(error) = std::fs::remove_file(path) { tracing::debug!(%error, "catalog cache removal failed"); } } } state.observed_disk = DiskView::read(path, &self.url, scope).version; // Deletion alone is insufficient: an old fetch with no previous // snapshot also needs a durable epoch to compare before publication. if !persisted_marker { state.error = Some(CatalogRefreshError::PersistenceFailed); } persisted_marker }
/// Freshness and source information without credentials or raw errors. /// /// # Panics /// Panics if the cache mutex is poisoned. pub fn status(&self, scope: &str) -> CatalogCacheStatus { self.get(scope); let state = self.state.lock().unwrap(); let fetched_at = state .snapshot .as_ref() .map(|snapshot| snapshot.file.fetched_at); let checked_at = state .snapshot .as_ref() .map(|snapshot| snapshot.file.checked_at.unwrap_or(snapshot.file.fetched_at)); let status = if state.pending.is_some() { CatalogCacheState::Refreshing } else if state.error.is_some() { CatalogCacheState::Error } else if state.snapshot.is_none() { CatalogCacheState::Idle } else if checked_at .is_some_and(|at| now_ms().saturating_sub(at) < CATALOG_REFRESH_INTERVAL_MS) { CatalogCacheState::Fresh } else { CatalogCacheState::Stale }; CatalogCacheStatus { state: status, source: state.source, fetched_at, checked_at, last_attempt: state.last_attempt, error: state.error, } }
/// Legacy models-only API. See [`Self::refresh_report`] for attempt status. /// /// # Panics /// Panics if the cache mutex is poisoned. pub async fn refresh(&self, scope: &str, opts: RefreshOptions) -> Option<T> { self.refresh_report(scope, opts).await.models }
/// Refresh, coalescing one scope's pending request. Hourly gating, offline /// behavior and last-good fallback remain identical to `refresh`. /// Dropping a coalesced waiter does not cancel the driving request. Dropping /// the driver resolves all its waiters as cancelled and releases its gate. /// /// # Panics /// Panics if the cache mutex is poisoned or an internal gate settles twice. pub async fn refresh_report( &self, scope: &str, opts: RefreshOptions, ) -> CatalogRefreshReport<T> { enum Gate<T> { Coalesced(Arc<InFlight<T>>), Gated, Start(Arc<InFlight<T>>, Option<Box<Snapshot<T>>>, u64, DiskVersion), } let cached = self.get(scope); if opts.is_current.as_ref().is_some_and(|check| !check()) { return CatalogRefreshReport::cancelled(); } if is_catalog_offline() { return CatalogRefreshReport { models: cached, outcome: CatalogRefreshOutcome::Offline, }; } let gate = { let mut state = self.state.lock().unwrap(); if let Some(pending) = state.pending.clone() { Gate::Coalesced(pending) } else { let checked_at = state.last_attempt.or_else(|| { state.snapshot.as_ref().map(|snapshot| { snapshot.file.checked_at.unwrap_or(snapshot.file.fetched_at) }) }); let now = now_ms(); if !opts.force && checked_at .is_some_and(|at| now >= at && now - at < CATALOG_REFRESH_INTERVAL_MS) { Gate::Gated } else { state.last_attempt = Some(now); state.error = None; let inflight = Arc::new(InFlight::new()); state.pending = Some(Arc::clone(&inflight)); Gate::Start( inflight, state.snapshot.clone().map(Box::new), state.generation, state.observed_disk.clone(), ) } } }; let (inflight, previous, generation, disk_version) = match gate { Gate::Coalesced(pending) => return await_inflight(pending).await, Gate::Gated => { return CatalogRefreshReport { models: cached, outcome: CatalogRefreshOutcome::Cached, } } Gate::Start(inflight, previous, generation, version) => ( inflight, previous.map(|snapshot| *snapshot), generation, version, ), }; let _driver = SettleOnDrop { cache: self, inflight: Arc::clone(&inflight), }; let result = self .drive_refresh(scope, &opts, previous, generation, &disk_version) .await; { let mut state = self.state.lock().unwrap(); if state .pending .as_ref() .is_some_and(|pending| Arc::ptr_eq(pending, &inflight)) { state.pending = None; } } if inflight.result.set(result.clone()).is_err() { debug_assert!(false, "catalog refresh settled twice"); } inflight.notify.notify_waiters(); result }
fn is_current(&self, scope: &str, generation: u64, opts: &RefreshOptions) -> bool { if opts.is_current.as_ref().is_some_and(|check| !check()) { return false; } let state = self.state.lock().unwrap(); state.scope.as_deref() == Some(scope) && state.generation == generation }
// Keep HTTP, 304, merge validation, and revocation transitions together; // splitting this protocol into single-use helpers obscures its guards. #[allow(clippy::too_many_lines)] async fn drive_refresh( &self, scope: &str, opts: &RefreshOptions, previous: Option<Snapshot<T>>, generation: u64, version: &DiskVersion, ) -> CatalogRefreshReport<T> { let etag = previous .as_ref() .and_then(|snapshot| snapshot.file.etag.as_deref()); let result = if let Some(fetch) = &opts.fetch_payload { match fetch().await { Ok(payload) => serde_json::to_vec(&payload) .map_err(|_| FetchError::Transport { message: "Invalid catalog payload".into(), }) .and_then(|body| { if body.len() > MAX_RESPONSE_BYTES { Err(FetchError::Transport { message: "Catalog is too large".into(), }) } else { Ok(FetchOutcome::Fresh { body, etag: None }) } }), Err(error) => Err(error), } } else { self.fetcher .fetch_with(&self.url, etag, &opts.headers) .await }; if !self.is_current(scope, generation, opts) { return CatalogRefreshReport::cancelled(); } let (mut snapshot, outcome) = match result { Ok(FetchOutcome::NotModified) => { let Some(mut snapshot) = previous else { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidPayload, ); }; snapshot.file.checked_at = Some(now_ms()); (snapshot, CatalogRefreshOutcome::NotModified) } Ok(FetchOutcome::Fresh { body, etag }) => { if body.len() > MAX_RESPONSE_BYTES { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidPayload, ); } let received: serde_json::Value = match serde_json::from_slice(&body) { Ok(payload) => payload, Err(_) => { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidJson, ) } }; // The provider merge validates/normalizes its own envelope. // Partial records may not parse alone (for example an unknown // OpenRouter ID whose capabilities exist only in last-good). let payload = if let Some(merge) = &opts.merge_payload { let merged = merge( &received, previous.as_ref().map(|snapshot| &snapshot.file.payload), ); let Ok(merged) = merged else { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidPayload, ); }; if serde_json::to_vec(&merged) .map_or(true, |bytes| bytes.len() > MAX_RESPONSE_BYTES) { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidPayload, ); } merged } else { received }; let Ok(models) = (self.parse)(&payload, scope) else { return self.keep_last_good( scope, generation, opts, CatalogRefreshError::InvalidPayload, ); }; ( Snapshot { file: SnapshotFile { url: self.url.to_string(), scope: scope.to_string(), fetched_at: now_ms(), checked_at: Some(now_ms()), revision: None, revocation: None, etag, payload, }, models, }, CatalogRefreshOutcome::Updated, ) } Err(error) => { if let Some(status) = error .status() .filter(|status| scope != PUBLIC_SCOPE && matches!(status, 401 | 403)) { self.revoke(scope); let mut state = self.state.lock().unwrap(); // Keep the attempt gate after rejection; non-forced calls // must not hammer an unauthorized endpoint every frame. state.last_attempt = Some(now_ms()); if state.error.is_none() { state.error = Some(CatalogRefreshError::Http { status }); } return CatalogRefreshReport { models: None, outcome: CatalogRefreshOutcome::Revoked { status }, }; } let error = error .status() .map_or(CatalogRefreshError::RequestFailed, |status| { CatalogRefreshError::Http { status } }); return self.keep_last_good(scope, generation, opts, error); } }; snapshot.file.revision = Some(revision()); // Lock acquisition is bounded and asynchronous. The render-side get() // never waits on the publisher lock, and no mutex lives across await. let lock = if let Some(path) = &self.cache_path { acquire_lock(path).await.ok() } else { None }; if !self.is_current(scope, generation, opts) { return CatalogRefreshReport::cancelled(); } self.store_snapshot(scope, generation, version, snapshot, outcome, lock.as_ref()) }
fn keep_last_good( &self, scope: &str, generation: u64, opts: &RefreshOptions, error: CatalogRefreshError, ) -> CatalogRefreshReport<T> { // Observe revocation/external updates even on an HTTP or parse failure. self.get(scope); if !self.is_current(scope, generation, opts) { return CatalogRefreshReport::cancelled(); } let mut state = self.state.lock().unwrap(); state.error = Some(error); CatalogRefreshReport { models: state .snapshot .as_ref() .map(|snapshot| snapshot.models.clone()), outcome: CatalogRefreshOutcome::Error { error }, } }
fn store_snapshot( &self, scope: &str, generation: u64, version: &DiskVersion, mut snapshot: Snapshot<T>, outcome: CatalogRefreshOutcome, lock: Option<&File>, ) -> CatalogRefreshReport<T> { let mut state = self.state.lock().unwrap(); if state.scope.as_deref() != Some(scope) || state.generation != generation { return CatalogRefreshReport::cancelled(); } let disk = self .cache_path .as_deref() .map(|path| DiskView::read(path, &self.url, scope)); if let Some(disk) = &disk { if &disk.version != version { // A newer publication or revocation always wins, irrespective // of wall-clock granularity and even when it deleted the file. state.snapshot = self.read_snapshot(disk, scope); state.revoked = state.snapshot.is_none() && (disk.unreadable_revocation || disk.revoked.is_some()); state.observed_disk = disk.version.clone(); state.source = if state.snapshot.is_some() { CatalogCacheSource::Disk } else { CatalogCacheSource::None }; return CatalogRefreshReport { models: state .snapshot .as_ref() .map(|snapshot| snapshot.models.clone()), outcome: if state.snapshot.is_some() { CatalogRefreshOutcome::Cached } else { CatalogRefreshOutcome::Cancelled }, }; } snapshot.file.revocation = disk.revocation(&self.url, scope); } let persisted = if let (Some(path), Some(_lock)) = (&self.cache_path, lock) { serde_json::to_vec(&snapshot.file).is_ok_and(|bytes| write_atomic(path, &bytes).is_ok()) } else { false }; if let Some(path) = self.cache_path.as_ref().filter(|_| persisted) { let disk = DiskView::read(path, &self.url, scope); // Tombstones do not wait for the lock. Detect revocation that // raced the atomic rename, rather than serving a rejected result. state.snapshot = self.read_snapshot(&disk, scope); state.revoked = state.snapshot.is_none() && (disk.unreadable_revocation || disk.revoked.is_some()); state.observed_disk = disk.version; if state.snapshot.is_none() { state.source = CatalogCacheSource::None; return CatalogRefreshReport::cancelled(); } } else { state.snapshot = Some(snapshot); state.revoked = false; } state.source = if persisted { CatalogCacheSource::Network } else { CatalogCacheSource::Memory }; state.error = (!persisted).then_some(CatalogRefreshError::PersistenceFailed); CatalogRefreshReport { models: state .snapshot .as_ref() .map(|snapshot| snapshot.models.clone()), outcome: if persisted { outcome } else { CatalogRefreshOutcome::MemoryOnly }, } }
fn read_snapshot(&self, disk: &DiskView, scope: &str) -> Option<Snapshot<T>> { if disk.unreadable_revocation { return None; } let bytes = disk.snapshot.as_deref()?; let stored: SnapshotFile = serde_json::from_slice(bytes).ok()?; if stored.url != self.url.as_ref() || stored.scope != scope { return None; } if let Some(bytes) = &disk.revoked { // A malformed existing tombstone fails closed, not back to an // otherwise readable private snapshot. let revoked: RevocationFile = serde_json::from_slice(bytes).ok()?; if revoked.url == self.url.as_ref() && revoked.scope == scope && stored.revocation.as_deref() != Some(revoked.revision.as_str()) { return None; } } let models = (self.parse)(&stored.payload, scope).ok()?; Some(Snapshot { file: stored, models, }) }}
/// Persistent advisory locks are released by the OS on close/process exit./// Never remove the lock path (that would split publishers across two inodes).fn open_lock(path: &Path) -> std::io::Result<File> { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } let mut options = std::fs::OpenOptions::new(); options.read(true).write(true).create(true).truncate(false); #[cfg(unix)] { use std::os::unix::fs::OpenOptionsExt; options.mode(0o600); } options.open(sidecar(path, ".lock"))}
async fn acquire_lock(path: &Path) -> std::io::Result<File> { let file = open_lock(path)?; let deadline = tokio::time::Instant::now() + Duration::from_secs(1); loop { match file.try_lock() { Ok(()) => return Ok(file), Err(std::fs::TryLockError::Error(error)) => return Err(error), Err(std::fs::TryLockError::WouldBlock) => { if tokio::time::Instant::now() >= deadline { return Err(std::io::Error::new( std::io::ErrorKind::WouldBlock, "catalog cache lock busy", )); } tokio::time::sleep(Duration::from_millis(20)).await; } } }}
/// A dropped driver must not strand coalesced waiters or the pending gate.struct SettleOnDrop<'a, T: Clone + Send + Sync + 'static> { cache: &'a CatalogCache<T>, inflight: Arc<InFlight<T>>,}
impl<T: Clone + Send + Sync + 'static> Drop for SettleOnDrop<'_, T> { fn drop(&mut self) { if self.inflight.result.get().is_some() { return; } { let mut state = self.cache.state.lock().unwrap(); if state .pending .as_ref() .is_some_and(|pending| Arc::ptr_eq(pending, &self.inflight)) { state.pending = None; } } let _ = self.inflight.result.set(CatalogRefreshReport::cancelled()); self.inflight.notify.notify_waiters(); }}
async fn await_inflight<T: Clone>(inflight: Arc<InFlight<T>>) -> CatalogRefreshReport<T> { loop { let notified = inflight.notify.notified(); if let Some(result) = inflight.result.get() { return result.clone(); } notified.await; }}
/// Atomic snapshot write: temp file + rename at mode 0600. The temp file/// is unique per writer (`pid` + an in-process counter): one agent dir is/// shared by many processes, and a shared temp path would let concurrent/// refreshes truncate each other's temp file and publish a malformed snapshot.fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> { static TMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } let temp = path.with_extension(format!( "{}-{}.tmp", std::process::id(), TMP_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed) )); #[cfg(unix)] let write = || { use std::os::unix::fs::OpenOptionsExt; std::fs::OpenOptions::new() .write(true) .create(true) .truncate(true) .mode(0o600) .open(&temp) }; #[cfg(not(unix))] let write = || { std::fs::OpenOptions::new() .write(true) .create(true) .truncate(true) .open(&temp) }; let mut file = write()?; file.write_all(bytes)?; file.flush()?; std::fs::rename(&temp, path)}
#[cfg(test)]mod tests { use super::*;
/// The multi-process snapshot-write invariant (Macroscope on #2576): /// a reader racing concurrent `write_atomic` writers always parses a /// full document. #[test] fn concurrent_writers_never_publish_a_malformed_snapshot() { let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("snapshot.json"); write_atomic(&path, br#"{"writer":0}"#).unwrap(); std::thread::scope(|scope| { for writer in 0..4u8 { let path = &path; scope.spawn(move || { for round in 0..200usize { let payload = format!( r#"{{"writer":{writer},"round":{round},"padding":"{}"}}"#, "x".repeat(round % 7 * 512) ); write_atomic(path, payload.as_bytes()).unwrap(); } }); } scope.spawn(|| { for _ in 0..500usize { let bytes = std::fs::read(&path).unwrap(); let parsed: serde_json::Value = serde_json::from_slice(&bytes) .expect("the published snapshot always parses"); assert!(parsed.get("writer").is_some()); std::hint::spin_loop(); } }); }); // Every rename consumed its temp: no writer litter remains. let leftovers: Vec<_> = std::fs::read_dir(dir.path()) .unwrap() .filter_map(Result::ok) .map(|entry| entry.file_name()) .filter(|name| name.to_string_lossy().contains(".tmp")) .collect(); assert!( leftovers.is_empty(), "temp files left behind: {leftovers:?}" ); }}
#[cfg(test)]#[path = "cache_regression.rs"]mod regression;