From 421b35ff9c4aa1346764c37afb62e204e35f95f9 Mon Sep 17 00:00:00 2001 From: Lewis Date: Thu, 12 Mar 2026 12:52:00 +0200 Subject: [PATCH] feat(lexicon): dynamic value types and schema registry --- crates/tranquil-lexicon/src/dynamic.rs | 278 ++++++++++++++++++++++++ crates/tranquil-lexicon/src/registry.rs | 211 ++++++++++++++++++ 2 files changed, 489 insertions(+) create mode 100644 crates/tranquil-lexicon/src/dynamic.rs create mode 100644 crates/tranquil-lexicon/src/registry.rs diff --git a/crates/tranquil-lexicon/src/dynamic.rs b/crates/tranquil-lexicon/src/dynamic.rs new file mode 100644 index 0000000..0942879 --- /dev/null +++ b/crates/tranquil-lexicon/src/dynamic.rs @@ -0,0 +1,278 @@ +use crate::resolve::{ResolveError, resolve_lexicon}; +use crate::schema::LexiconDoc; +use parking_lot::RwLock; +use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60); +const MAX_DYNAMIC_SCHEMAS: usize = 1024; + +struct NegativeEntry { + expires_at: Instant, +} + +struct SchemaStore { + schemas: HashMap>, + insertion_order: VecDeque, +} + +pub struct DynamicRegistry { + store: RwLock, + negative_cache: RwLock>, + network_disabled: AtomicBool, +} + +impl DynamicRegistry { + pub fn new() -> Self { + let network_disabled = + std::env::var("TRANQUIL_LEXICON_OFFLINE").is_ok_and(|v| v == "1" || v == "true"); + Self { + store: RwLock::new(SchemaStore { + schemas: HashMap::new(), + insertion_order: VecDeque::new(), + }), + negative_cache: RwLock::new(HashMap::new()), + network_disabled: AtomicBool::new(network_disabled), + } + } + + #[allow(dead_code)] + pub fn set_network_disabled(&self, disabled: bool) { + self.network_disabled.store(disabled, Ordering::Relaxed); + } + + pub fn get(&self, nsid: &str) -> Option> { + self.store.read().schemas.get(nsid).cloned() + } + + pub fn is_negative_cached(&self, nsid: &str) -> bool { + let cache = self.negative_cache.read(); + cache + .get(nsid) + .is_some_and(|entry| entry.expires_at > Instant::now()) + } + + fn insert_negative(&self, nsid: &str) { + let mut cache = self.negative_cache.write(); + if cache.len() > MAX_DYNAMIC_SCHEMAS { + let now = Instant::now(); + cache.retain(|_, entry| entry.expires_at > now); + } + cache.insert( + nsid.to_string(), + NegativeEntry { + expires_at: Instant::now() + NEGATIVE_CACHE_TTL, + }, + ); + } + + pub(crate) fn insert_schema(&self, doc: LexiconDoc) -> Arc { + let arc = Arc::new(doc); + let nsid = arc.id.clone(); + + let mut store = self.store.write(); + + if store.schemas.len() >= MAX_DYNAMIC_SCHEMAS { + tracing::warn!( + count = store.schemas.len(), + "dynamic schema registry at capacity, evicting oldest entries" + ); + let evict_count = store.schemas.len() / 4; + (0..evict_count).for_each(|_| { + if let Some(key) = store.insertion_order.pop_front() { + store.schemas.remove(&key); + } + }); + } + + if store + .schemas + .insert(nsid.clone(), Arc::clone(&arc)) + .is_some() + { + store.insertion_order.retain(|k| k != &nsid); + } + store.insertion_order.push_back(nsid.clone()); + + self.negative_cache.write().remove(&arc.id); + + arc + } + + pub async fn resolve_and_cache(&self, nsid: &str) -> Result, ResolveError> { + if let Some(doc) = self.get(nsid) { + return Ok(doc); + } + + if self.network_disabled.load(Ordering::Relaxed) { + return Err(ResolveError::NetworkDisabled); + } + + if self.is_negative_cached(nsid) { + return Err(ResolveError::NegativelyCached { + nsid: nsid.to_string(), + ttl_secs: NEGATIVE_CACHE_TTL.as_secs(), + }); + } + + match resolve_lexicon(nsid).await { + Ok(doc) => Ok(self.insert_schema(doc)), + Err(e) => { + tracing::debug!(nsid = nsid, error = %e, "caching negative resolution result"); + self.insert_negative(nsid); + Err(e) + } + } + } + + pub fn schema_count(&self) -> usize { + self.store.read().schemas.len() + } +} + +impl Default for DynamicRegistry { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_negative_cache() { + let registry = DynamicRegistry::new(); + assert!(!registry.is_negative_cached("com.example.test")); + + registry.insert_negative("com.example.test"); + assert!(registry.is_negative_cached("com.example.test")); + } + + #[tokio::test] + async fn test_negative_cache_returns_appropriate_error_variant() { + let registry = DynamicRegistry::new(); + registry.insert_negative("com.example.cached"); + + let err = registry + .resolve_and_cache("com.example.cached") + .await + .unwrap_err(); + + assert!( + !matches!(err, ResolveError::InvalidNsid(_)), + "negative cache hit should not return InvalidNsid - the NSID is valid, it just failed resolution recently. got: {}", + err + ); + } + + #[test] + fn test_empty_lookup() { + let registry = DynamicRegistry::new(); + assert!(registry.get("com.example.nonexistent").is_none()); + assert_eq!(registry.schema_count(), 0); + } + + #[test] + fn test_insert_and_retrieve() { + let registry = DynamicRegistry::new(); + let doc = LexiconDoc { + lexicon: 1, + id: "com.example.test".to_string(), + defs: HashMap::new(), + }; + + let arc = registry.insert_schema(doc); + assert_eq!(arc.id, "com.example.test"); + assert_eq!(registry.schema_count(), 1); + + let retrieved = registry.get("com.example.test"); + assert!(retrieved.is_some()); + assert_eq!(retrieved.unwrap().id, "com.example.test"); + } + + #[test] + fn test_negative_cache_cleared_on_insert() { + let registry = DynamicRegistry::new(); + + registry.insert_negative("com.example.test"); + assert!(registry.is_negative_cached("com.example.test")); + + let doc = LexiconDoc { + lexicon: 1, + id: "com.example.test".to_string(), + defs: HashMap::new(), + }; + registry.insert_schema(doc); + + assert!(!registry.is_negative_cached("com.example.test")); + } + + #[test] + fn test_eviction_is_fifo() { + let registry = DynamicRegistry::new(); + + (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| { + let doc = LexiconDoc { + lexicon: 1, + id: format!("com.example.schema{}", i), + defs: HashMap::new(), + }; + registry.insert_schema(doc); + }); + assert_eq!(registry.schema_count(), MAX_DYNAMIC_SCHEMAS); + + let trigger = LexiconDoc { + lexicon: 1, + id: "com.example.trigger".to_string(), + defs: HashMap::new(), + }; + registry.insert_schema(trigger); + + assert!( + registry.get("com.example.schema0").is_none(), + "oldest entry should be evicted" + ); + assert!( + registry.get("com.example.trigger").is_some(), + "newly inserted entry should exist" + ); + let evict_count = MAX_DYNAMIC_SCHEMAS / 4; + assert!( + registry + .get(&format!("com.example.schema{}", evict_count)) + .is_some(), + "entry after eviction window should survive" + ); + } + + #[test] + fn test_eviction_frees_memory() { + let registry = DynamicRegistry::new(); + let doc = LexiconDoc { + lexicon: 1, + id: "com.example.tracked".to_string(), + defs: HashMap::new(), + }; + let arc = registry.insert_schema(doc); + let weak = Arc::downgrade(&arc); + drop(arc); + + assert!(weak.upgrade().is_some(), "registry still holds a reference"); + + (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| { + registry.insert_schema(LexiconDoc { + lexicon: 1, + id: format!("com.example.filler{}", i), + defs: HashMap::new(), + }); + }); + + assert!( + weak.upgrade().is_none(), + "evicted Arc should be freed when no external references remain" + ); + } +} diff --git a/crates/tranquil-lexicon/src/registry.rs b/crates/tranquil-lexicon/src/registry.rs new file mode 100644 index 0000000..e6f26fe --- /dev/null +++ b/crates/tranquil-lexicon/src/registry.rs @@ -0,0 +1,211 @@ +use crate::schema::{LexDef, LexObject, LexiconDoc, ParsedRef, parse_ref}; +use std::collections::HashMap; +use std::sync::{Arc, OnceLock}; + +static REGISTRY: OnceLock = OnceLock::new(); + +pub struct LexiconRegistry { + schemas: HashMap>, + #[cfg(feature = "resolve")] + dynamic: crate::dynamic::DynamicRegistry, +} + +impl Default for LexiconRegistry { + fn default() -> Self { + Self::new() + } +} + +impl LexiconRegistry { + pub fn global() -> &'static Self { + REGISTRY.get_or_init(Self::new) + } + + pub fn new() -> Self { + Self { + schemas: HashMap::new(), + #[cfg(feature = "resolve")] + dynamic: crate::dynamic::DynamicRegistry::new(), + } + } + + pub fn register(&mut self, doc: LexiconDoc) { + let id = doc.id.clone(); + self.schemas.insert(id, Arc::new(doc)); + } + + #[cfg(feature = "resolve")] + pub fn preload(&self, doc: LexiconDoc) { + self.dynamic.insert_schema(doc); + } + + pub fn get_doc(&self, nsid: &str) -> Option> { + self.schemas.get(nsid).cloned().or_else(|| { + #[cfg(feature = "resolve")] + { + self.dynamic.get(nsid) + } + #[cfg(not(feature = "resolve"))] + { + None + } + }) + } + + pub fn get_record_def(&self, nsid: &str) -> Option> { + let doc = self.get_doc(nsid)?; + match doc.defs.get("main")? { + LexDef::Record(_) => Some(doc), + _ => None, + } + } + + pub fn resolve_ref(&self, reference: &str, context_nsid: &str) -> Option { + match parse_ref(reference) { + ParsedRef::Local(local) => { + let doc = self.get_doc(context_nsid)?; + Self::def_to_resolved(&doc, local) + } + ParsedRef::Qualified { nsid, fragment } => { + let doc = self.get_doc(nsid)?; + Self::def_to_resolved(&doc, fragment) + } + ParsedRef::Bare(nsid) => { + let doc = self.get_doc(nsid)?; + Self::def_to_resolved(&doc, "main") + } + } + } + + fn def_to_resolved(doc: &Arc, def_name: &str) -> Option { + let def = doc.defs.get(def_name)?; + match def { + LexDef::Object(_) | LexDef::Record(_) | LexDef::Token {} | LexDef::StringDef(_) => { + Some(ResolvedRef { + doc: Arc::clone(doc), + def_name: def_name.to_string(), + }) + } + _ => None, + } + } + + pub fn has_schema(&self, nsid: &str) -> bool { + self.get_doc(nsid).is_some() + } + + pub fn schema_count(&self) -> usize { + let embedded = self.schemas.len(); + #[cfg(feature = "resolve")] + { + embedded + self.dynamic.schema_count() + } + #[cfg(not(feature = "resolve"))] + { + embedded + } + } + + #[cfg(feature = "resolve")] + pub async fn resolve_dynamic( + &self, + nsid: &str, + ) -> Result, crate::resolve::ResolveError> { + self.dynamic.resolve_and_cache(nsid).await + } + + #[cfg(feature = "resolve")] + pub fn is_negative_cached(&self, nsid: &str) -> bool { + self.dynamic.is_negative_cached(nsid) + } +} + +pub struct ResolvedRef { + doc: Arc, + def_name: String, +} + +impl ResolvedRef { + pub fn as_object(&self) -> Option<&LexObject> { + match self.doc.defs.get(&self.def_name)? { + LexDef::Object(obj) => Some(obj), + LexDef::Record(rec) => Some(&rec.record), + _ => None, + } + } + + pub fn is_token(&self) -> bool { + self.doc + .defs + .get(&self.def_name) + .is_some_and(|def| matches!(def, LexDef::Token {} | LexDef::StringDef(_))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_empty_registry() { + let registry = LexiconRegistry::new(); + assert_eq!(registry.schema_count(), 0); + assert!(!registry.has_schema("app.bsky.feed.post")); + } + + #[test] + fn test_register_and_lookup() { + let mut registry = LexiconRegistry::new(); + let doc = LexiconDoc { + lexicon: 1, + id: "com.example.test".to_string(), + defs: HashMap::new(), + }; + registry.register(doc); + assert_eq!(registry.schema_count(), 1); + assert!(registry.has_schema("com.example.test")); + assert!(!registry.has_schema("com.example.other")); + } + + #[test] + fn test_get_record_def() { + let registry = crate::test_schemas::test_registry(); + let doc = registry.get_record_def("com.test.basic"); + assert!(doc.is_some()); + let doc = doc.unwrap(); + match doc.defs.get("main").unwrap() { + LexDef::Record(rec) => { + assert!(rec.record.required.contains(&"text".to_string())); + assert!(rec.record.required.contains(&"createdAt".to_string())); + } + _ => panic!("expected record def"), + } + } + + #[test] + fn test_get_record_def_unknown() { + let registry = LexiconRegistry::new(); + assert!(registry.get_record_def("com.example.nonexistent").is_none()); + } + + #[test] + fn test_resolve_ref_cross_schema() { + let registry = crate::test_schemas::test_registry(); + let resolved = registry.resolve_ref("com.test.strongref", "com.test.withref"); + assert!(resolved.is_some_and(|r| r.as_object().is_some())); + } + + #[test] + fn test_resolve_local_ref() { + let registry = crate::test_schemas::test_registry(); + let resolved = registry.resolve_ref("#replyRef", "com.test.withreply"); + assert!(resolved.is_some_and(|r| r.as_object().is_some())); + } + + #[test] + fn test_has_schema() { + let registry = crate::test_schemas::test_registry(); + assert!(registry.has_schema("com.test.basic")); + assert!(!registry.has_schema("com.example.nonexistent")); + } +}