use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};
use crate::l2_cache::{CacheKey, CacheKeyKind, L2Cache};
use crate::value::Value;
#[derive(Debug, Clone)]
pub struct ProcessL1Config {
pub capacity: usize,
pub ttl_ms: u64,
pub enable_coherence: bool,
pub tenant_isolated: bool,
}
impl Default for ProcessL1Config {
fn default() -> Self {
Self {
capacity: 10_000,
ttl_ms: 300_000,
enable_coherence: true,
tenant_isolated: true,
}
}
}
impl ProcessL1Config {
pub fn new() -> Self {
Self::default()
}
pub fn with_capacity(mut self, capacity: usize) -> Self {
self.capacity = capacity;
self
}
pub fn with_ttl_ms(mut self, ttl_ms: u64) -> Self {
self.ttl_ms = ttl_ms;
self
}
pub fn with_coherence(mut self, enable: bool) -> Self {
self.enable_coherence = enable;
self
}
pub fn with_tenant_isolated(mut self, isolated: bool) -> Self {
self.tenant_isolated = isolated;
self
}
}
#[derive(Debug, Default)]
pub struct ProcessL1Stats {
pub hits: AtomicU64,
pub misses: AtomicU64,
pub entry_count: AtomicU64,
pub evict_count: AtomicU64,
}
#[derive(Debug, Clone, Default)]
pub struct ProcessL1StatsSnapshot {
pub hits: u64,
pub misses: u64,
pub entry_count: u64,
pub evict_count: u64,
pub hit_rate: f64,
}
impl ProcessL1StatsSnapshot {
pub fn total_lookups(&self) -> u64 {
self.hits + self.misses
}
}
struct CacheEntry<T> {
value: Arc<T>,
inserted_at: u64,
last_accessed_at: u64,
}
struct ProcessL1Inner<T> {
entries: HashMap<CacheKey, CacheEntry<T>>,
lru_order: VecDeque<CacheKey>,
}
pub struct ProcessL1Cache<T: Clone + Send + Sync + 'static> {
inner: RwLock<ProcessL1Inner<T>>,
config: ProcessL1Config,
stats: ProcessL1Stats,
l2: Option<Arc<L2Cache>>,
}
impl<T: Clone + Send + Sync + 'static> ProcessL1Cache<T> {
pub fn new(config: ProcessL1Config) -> Self {
Self {
inner: RwLock::new(ProcessL1Inner {
entries: HashMap::new(),
lru_order: VecDeque::new(),
}),
config,
stats: ProcessL1Stats::default(),
l2: None,
}
}
pub fn with_l2(mut self, l2: Arc<L2Cache>) -> Self {
self.l2 = Some(l2);
self
}
fn build_key(&self, table: &str, pk: &Value) -> CacheKey {
CacheKey::by_pk(table, pk)
}
fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
fn is_expired(&self, entry: &CacheEntry<T>) -> bool {
if self.config.ttl_ms == 0 {
return false;
}
Self::now_ms().saturating_sub(entry.inserted_at) > self.config.ttl_ms
}
pub async fn get(&self, table: &str, pk: &Value) -> Option<Arc<T>> {
if self.config.capacity == 0 {
self.stats.misses.fetch_add(1, Ordering::Relaxed);
return None;
}
let key = self.build_key(table, pk);
let now = Self::now_ms();
let mut inner = self.inner.write().unwrap();
if let Some(entry) = inner.entries.get_mut(&key) {
if self.is_expired(entry) {
inner.entries.remove(&key);
inner.lru_order.retain(|k| k != &key);
self.stats.misses.fetch_add(1, Ordering::Relaxed);
return None;
}
entry.last_accessed_at = now;
let value = Arc::clone(&entry.value);
inner.lru_order.retain(|k| k != &key);
inner.lru_order.push_back(key);
self.stats.hits.fetch_add(1, Ordering::Relaxed);
return Some(value);
}
self.stats.misses.fetch_add(1, Ordering::Relaxed);
None
}
pub async fn put(&self, table: &str, pk: Value, value: Arc<T>) {
if self.config.capacity == 0 {
return;
}
let key = self.build_key(table, &pk);
let now = Self::now_ms();
let mut inner = self.inner.write().unwrap();
if inner.entries.contains_key(&key) {
inner.lru_order.retain(|k| k != &key);
}
inner.entries.insert(
key.clone(),
CacheEntry {
value: Arc::clone(&value),
inserted_at: now,
last_accessed_at: now,
},
);
inner.lru_order.push_back(key.clone());
self.evict_lru(&mut inner);
self.stats
.entry_count
.store(inner.entries.len() as u64, Ordering::Relaxed);
}
fn evict_lru(&self, inner: &mut ProcessL1Inner<T>) {
while inner.entries.len() > self.config.capacity {
if let Some(old_key) = inner.lru_order.pop_front() {
inner.entries.remove(&old_key);
self.stats.evict_count.fetch_add(1, Ordering::Relaxed);
} else {
break;
}
}
}
pub async fn invalidate(&self, table: &str, pk: &Value) {
let key = self.build_key(table, pk);
let mut inner = self.inner.write().unwrap();
inner.entries.remove(&key);
inner.lru_order.retain(|k| k != &key);
self.stats
.entry_count
.store(inner.entries.len() as u64, Ordering::Relaxed);
if self.config.enable_coherence {
if let Some(l2) = &self.l2 {
l2.invalidate(&key);
}
}
}
pub async fn invalidate_table(&self, table: &str) {
let mut inner = self.inner.write().unwrap();
let keys_to_remove: Vec<CacheKey> = inner
.entries
.keys()
.filter(|k| k.table == table)
.cloned()
.collect();
for key in &keys_to_remove {
inner.entries.remove(key);
}
inner.lru_order.retain(|k| !keys_to_remove.contains(k));
self.stats
.entry_count
.store(inner.entries.len() as u64, Ordering::Relaxed);
if self.config.enable_coherence {
if let Some(l2) = &self.l2 {
l2.invalidate_table(table);
}
}
}
pub fn stats(&self) -> ProcessL1StatsSnapshot {
let hits = self.stats.hits.load(Ordering::Relaxed);
let misses = self.stats.misses.load(Ordering::Relaxed);
let entry_count = self.stats.entry_count.load(Ordering::Relaxed);
let evict_count = self.stats.evict_count.load(Ordering::Relaxed);
let total = hits + misses;
let hit_rate = if total == 0 {
0.0
} else {
hits as f64 / total as f64
};
ProcessL1StatsSnapshot {
hits,
misses,
entry_count,
evict_count,
hit_rate,
}
}
pub fn len(&self) -> usize {
self.inner.read().unwrap().entries.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
#[cfg(feature = "cache-warmup-protection")]
impl<T: Clone + Send + Sync + 'static> ProcessL1Cache<T> {
pub async fn warmup<F, Fut>(
self: &Arc<Self>,
config: &crate::cache_warmup_protection::WarmupConfig,
loader: F,
) -> Result<
crate::cache_warmup_protection::WarmupResult,
crate::cache_warmup_protection::CacheError,
>
where
F: Fn(&str) -> Fut,
Fut: std::future::Future<
Output = Result<Vec<(Value, T)>, crate::cache_warmup_protection::CacheError>,
>,
{
let warmer = crate::cache_warmup_protection::CacheWarmer::new(Arc::clone(self));
warmer.warmup(config, loader).await
}
}
pub struct CrossSessionIdentityMap<T: Clone + Send + Sync + 'static> {
cache: Arc<ProcessL1Cache<T>>,
}
impl<T: Clone + Send + Sync + 'static> CrossSessionIdentityMap<T> {
pub fn new(cache: Arc<ProcessL1Cache<T>>) -> Self {
Self { cache }
}
pub async fn get_or_load<F, Fut, E>(
&self,
table: &str,
pk: Value,
loader: F,
) -> Result<Arc<T>, E>
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<T, E>>,
{
if let Some(cached) = self.cache.get(table, &pk).await {
return Ok(cached);
}
let value = loader().await?;
let arc = Arc::new(value);
self.cache.put(table, pk, Arc::clone(&arc)).await;
Ok(arc)
}
pub async fn invalidate(&self, table: &str, pk: &Value) {
self.cache.invalidate(table, pk).await;
}
pub async fn invalidate_table(&self, table: &str) {
self.cache.invalidate_table(table).await;
}
pub fn stats(&self) -> ProcessL1StatsSnapshot {
self.cache.stats()
}
}
pub fn tenant_cache_key(
tenant_id: &str,
table: &str,
pk: &Value,
tenant_isolated: bool,
) -> CacheKey {
if tenant_isolated {
CacheKey {
table: format!("{}:{}", tenant_id, table),
kind: CacheKeyKind::ByPk,
identifier: pk.to_string(),
}
} else {
CacheKey::by_pk(table, pk)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_config() -> ProcessL1Config {
ProcessL1Config::new()
}
#[test]
fn test_config_default() {
let config = ProcessL1Config::new();
assert_eq!(config.capacity, 10_000);
assert_eq!(config.ttl_ms, 300_000);
assert!(config.enable_coherence);
assert!(config.tenant_isolated);
}
#[test]
fn test_config_builder() {
let config = ProcessL1Config::new()
.with_capacity(500)
.with_ttl_ms(60_000)
.with_coherence(false)
.with_tenant_isolated(false);
assert_eq!(config.capacity, 500);
assert_eq!(config.ttl_ms, 60_000);
assert!(!config.enable_coherence);
assert!(!config.tenant_isolated);
}
#[tokio::test]
async fn test_get_hit() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
let result = cache.get("users", &Value::I64(1)).await;
assert!(result.is_some());
assert_eq!(result.unwrap().as_str(), "Alice");
let stats = cache.stats();
assert_eq!(stats.hits, 1);
}
#[tokio::test]
async fn test_get_miss() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
let result = cache.get("users", &Value::I64(1)).await;
assert!(result.is_none());
let stats = cache.stats();
assert_eq!(stats.misses, 1);
}
#[tokio::test]
async fn test_identity_map() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
let a = cache.get("users", &Value::I64(1)).await.unwrap();
let b = cache.get("users", &Value::I64(1)).await.unwrap();
assert!(Arc::ptr_eq(&a, &b));
}
#[tokio::test]
async fn test_lru_eviction() {
let config = ProcessL1Config::new().with_capacity(3);
let cache: ProcessL1Cache<i64> = ProcessL1Cache::new(config);
for i in 1..=4i64 {
cache.put("t", Value::I64(i), Arc::new(i)).await;
}
assert_eq!(cache.len(), 3);
assert!(cache.get("t", &Value::I64(1)).await.is_none());
assert!(cache.get("t", &Value::I64(4)).await.is_some());
let stats = cache.stats();
assert!(stats.evict_count >= 1);
}
#[tokio::test]
async fn test_ttl_expiry() {
let config = ProcessL1Config::new().with_capacity(100).with_ttl_ms(50);
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(config);
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
assert!(cache.get("users", &Value::I64(1)).await.is_some());
tokio::time::sleep(std::time::Duration::from_millis(60)).await;
assert!(cache.get("users", &Value::I64(1)).await.is_none());
}
#[tokio::test]
async fn test_capacity_zero() {
let config = ProcessL1Config::new().with_capacity(0);
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(config);
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
assert!(cache.get("users", &Value::I64(1)).await.is_none());
}
#[tokio::test]
async fn test_invalidate() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
assert!(cache.get("users", &Value::I64(1)).await.is_some());
cache.invalidate("users", &Value::I64(1)).await;
assert!(cache.get("users", &Value::I64(1)).await.is_none());
}
#[tokio::test]
async fn test_invalidate_table() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
cache
.put("users", Value::I64(2), Arc::new("Bob".to_string()))
.await;
cache
.put("orders", Value::I64(1), Arc::new("Order1".to_string()))
.await;
cache.invalidate_table("users").await;
assert!(cache.get("users", &Value::I64(1)).await.is_none());
assert!(cache.get("users", &Value::I64(2)).await.is_none());
assert!(cache.get("orders", &Value::I64(1)).await.is_some());
}
#[tokio::test]
async fn test_invalidate_nonexistent() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache.invalidate("users", &Value::I64(999)).await;
cache.invalidate_table("nonexistent").await;
}
#[tokio::test]
async fn test_cross_session_identity_map() {
let cache: Arc<ProcessL1Cache<String>> = Arc::new(ProcessL1Cache::new(make_config()));
let id_map = CrossSessionIdentityMap::new(Arc::clone(&cache));
let result1 = id_map
.get_or_load("users", Value::I64(1), || async {
Ok::<String, String>("Alice".to_string())
})
.await
.unwrap();
let result2 = id_map
.get_or_load("users", Value::I64(1), || async {
Err::<String, String>("should not be called".to_string())
})
.await
.unwrap();
assert!(Arc::ptr_eq(&result1, &result2));
}
#[tokio::test]
async fn test_cross_session_load_from_db() {
let cache: Arc<ProcessL1Cache<String>> = Arc::new(ProcessL1Cache::new(make_config()));
let id_map = CrossSessionIdentityMap::new(Arc::clone(&cache));
let call_count = Arc::new(AtomicU64::new(0));
let cc = Arc::clone(&call_count);
let result1 = id_map
.get_or_load("users", Value::I64(1), || {
let cc = Arc::clone(&cc);
async move {
cc.fetch_add(1, Ordering::Relaxed);
Ok::<String, String>("Alice".to_string())
}
})
.await
.unwrap();
let result2 = id_map
.get_or_load("users", Value::I64(1), || {
let cc = Arc::clone(&cc);
async move {
cc.fetch_add(1, Ordering::Relaxed);
Ok::<String, String>("Alice".to_string())
}
})
.await
.unwrap();
assert_eq!(result1.as_str(), "Alice");
assert_eq!(result2.as_str(), "Alice");
assert!(Arc::ptr_eq(&result1, &result2));
assert_eq!(call_count.load(Ordering::Relaxed), 1);
}
#[tokio::test]
async fn test_cross_session_loader_error() {
let cache: Arc<ProcessL1Cache<String>> = Arc::new(ProcessL1Cache::new(make_config()));
let id_map = CrossSessionIdentityMap::new(Arc::clone(&cache));
let result: Result<Arc<String>, String> = id_map
.get_or_load("users", Value::I64(1), || async {
Err("db error".to_string())
})
.await;
assert!(result.is_err());
assert!(cache.get("users", &Value::I64(1)).await.is_none());
}
#[tokio::test]
async fn test_l2_invalidation_sync() {
let l2 = Arc::new(L2Cache::new());
let config = ProcessL1Config::new().with_coherence(true);
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(config).with_l2(Arc::clone(&l2));
let key = CacheKey::by_pk("users", "1");
l2.put(&key, Value::String("Alice".to_string()), None);
assert!(l2.get(&key).is_some());
cache.invalidate("users", &Value::I64(1)).await;
assert!(l2.get(&key).is_none());
}
#[tokio::test]
async fn test_l2_invalidate_table_sync() {
let l2 = Arc::new(L2Cache::new());
let config = ProcessL1Config::new().with_coherence(true);
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(config).with_l2(Arc::clone(&l2));
let key1 = CacheKey::by_pk("users", "1");
let key2 = CacheKey::by_pk("users", "2");
l2.put(&key1, Value::String("Alice".to_string()), None);
l2.put(&key2, Value::String("Bob".to_string()), None);
cache.invalidate_table("users").await;
assert!(l2.get(&key1).is_none());
assert!(l2.get(&key2).is_none());
}
#[tokio::test]
async fn test_coherence_disabled() {
let l2 = Arc::new(L2Cache::new());
let config = ProcessL1Config::new().with_coherence(false);
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(config).with_l2(Arc::clone(&l2));
let key = CacheKey::by_pk("users", "1");
l2.put(&key, Value::String("Alice".to_string()), None);
cache.invalidate("users", &Value::I64(1)).await;
assert!(l2.get(&key).is_some());
}
#[tokio::test]
async fn test_tenant_cache_key_isolated() {
let key_a = tenant_cache_key("tenant_a", "users", &Value::I64(1), true);
let key_b = tenant_cache_key("tenant_b", "users", &Value::I64(1), true);
assert_ne!(key_a, key_b);
}
#[tokio::test]
async fn test_tenant_cache_key_shared() {
let key_a = tenant_cache_key("tenant_a", "users", &Value::I64(1), false);
let key_b = tenant_cache_key("tenant_b", "users", &Value::I64(1), false);
assert_eq!(key_a, key_b);
}
#[tokio::test]
async fn test_concurrent_access() {
let cache: Arc<ProcessL1Cache<i64>> =
Arc::new(ProcessL1Cache::new(make_config().with_capacity(1000)));
let mut handles = Vec::new();
for i in 0..10i64 {
let cache = Arc::clone(&cache);
handles.push(tokio::spawn(async move {
for j in 0..100i64 {
let v = i * 100 + j;
cache.put("t", Value::I64(v), Arc::new(v)).await;
cache.get("t", &Value::I64(v)).await;
}
}));
}
for h in handles {
h.await.unwrap();
}
assert!(cache.len() <= 1000);
}
#[tokio::test]
async fn test_stats_snapshot() {
let cache: ProcessL1Cache<String> = ProcessL1Cache::new(make_config());
cache
.put("users", Value::I64(1), Arc::new("Alice".to_string()))
.await;
cache.get("users", &Value::I64(1)).await;
cache.get("users", &Value::I64(2)).await;
let stats = cache.stats();
assert_eq!(stats.hits, 1);
assert_eq!(stats.misses, 1);
assert_eq!(stats.total_lookups(), 2);
assert!((stats.hit_rate - 0.5).abs() < 0.001);
}
}