d5939db
use bincache::strategies::{Hybrid, Limits};
use bincache::{Cache as Bincache, Noop};
use crate::config::cache::AppCacheSettings;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::RwLock;
use tracing::debug;
pub struct Cache {
pub hybrid: RwLock<Bincache<String, Hybrid, Noop>>,
pub expires_at: RwLock<HashMap<String, u64>>,
pub l3: Option<Arc<super::RedisCache>>,
pub stats: Arc<CacheStats>,
pub default_ttl: u64,
}
pub struct CacheStats {
pub hybrid_hits: AtomicU64,
pub l3_hits: AtomicU64,
pub misses: AtomicU64,
pub errors: AtomicU64,
}
impl Cache {
pub(super) fn new(
hybrid: Bincache<String, Hybrid, Noop>,
l3: Option<super::RedisCache>,
default_ttl: u64,
) -> Self {
Self {
hybrid: RwLock::new(hybrid),
expires_at: RwLock::new(HashMap::new()),
l3: l3.map(Arc::new),
stats: Arc::new(CacheStats {
hybrid_hits: AtomicU64::new(0),
l3_hits: AtomicU64::new(0),
misses: AtomicU64::new(0),
errors: AtomicU64::new(0),
}),
default_ttl,
}
}
pub async fn get(&self, key: &str) -> Option<Vec<u8>> {
let expired = {
let expirations = self.expires_at.read().await;
expirations
.get(key)
.is_some_and(|expiry| *expiry <= unix_timestamp_millis())
};
if expired {
let mut expirations = self.expires_at.write().await;
if expirations
.get(key)
.is_some_and(|expiry| *expiry <= unix_timestamp_millis())
{
let mut hybrid = self.hybrid.write().await;
let _ = hybrid.delete(key.to_string()).await;
expirations.remove(key);
}
} else {
let hybrid = self.hybrid.read().await;
if let Ok(val) = hybrid.get(key.to_string()).await {
self.stats.hybrid_hits.fetch_add(1, Ordering::Relaxed);
debug!(key, tier = "hybrid", "cache hit");
return Some(val.into_owned());
}
}
if let Some(ref l3) = self.l3 {
match l3.get_bytes_with_ttl(key).await {
Ok(Some((val, remaining_ttl_millis))) => {
self.stats.l3_hits.fetch_add(1, Ordering::Relaxed);
debug!(key, tier = 3, "cache hit");
if remaining_ttl_millis != Some(0) {
let mut expirations = self.expires_at.write().await;
let mut hybrid = self.hybrid.write().await;
if expirations.contains_key(key) {
let _ = hybrid.delete(key.to_string()).await;
}
let _ = hybrid.put(key.to_string(), val.clone()).await;
let local_expiry = match remaining_ttl_millis {
Some(remaining) if self.default_ttl > 0 => unix_timestamp_millis()
.saturating_add(
remaining.min(self.default_ttl.saturating_mul(1000)),
),
Some(remaining) => unix_timestamp_millis().saturating_add(remaining),
None => expires_at(self.default_ttl),
};
expirations.insert(key.to_string(), local_expiry);
}
return Some(val);
}
Ok(None) => {}
Err(e) => {
self.stats.errors.fetch_add(1, Ordering::Relaxed);
debug!(key, error = %e, "L3 cache error");
}
}
}
self.stats.misses.fetch_add(1, Ordering::Relaxed);
debug!(key, "cache miss");
None
}
pub fn redis(&self) -> Option<Arc<super::RedisCache>> {
self.l3.clone()
}
pub async fn set(&self, key: &str, value: Vec<u8>, ttl_seconds: u64) {
let ttl = if ttl_seconds == 0 {
self.default_ttl
} else {
ttl_seconds
};
let mut expirations = self.expires_at.write().await;
let mut hybrid = self.hybrid.write().await;
if expirations.contains_key(key) {
let _ = hybrid.delete(key.to_string()).await;
}
if let Err(e) = hybrid.put(key.to_string(), value.clone()).await {
debug!(key, error = %e, "hybrid cache put error");
} else {
expirations.insert(key.to_string(), expires_at(ttl));
}
drop(hybrid);
drop(expirations);
if let Some(ref l3) = self.l3
&& let Err(e) = l3.set_bytes(key, value, ttl).await
{
self.stats.errors.fetch_add(1, Ordering::Relaxed);
debug!(key, error = %e, "L3 cache set error");
}
}
pub async fn delete(&self, key: &str) {
let mut expirations = self.expires_at.write().await;
let mut hybrid = self.hybrid.write().await;
let _ = hybrid.delete(key.to_string()).await;
expirations.remove(key);
drop(hybrid);
drop(expirations);
if let Some(ref l3) = self.l3
&& let Err(e) = l3.delete(key).await
{
debug!(key, error = %e, "L3 cache delete error");
}
}
pub async fn clear(&self) {
let mut expirations = self.expires_at.write().await;
let mut hybrid = self.hybrid.write().await;
let keys = expirations.keys().cloned().collect::<Vec<_>>();
for key in keys {
let _ = hybrid.delete(key).await;
}
expirations.clear();
drop(hybrid);
drop(expirations);
if let Some(ref l3) = self.l3
&& let Err(e) = l3.clear().await
{
debug!(error = %e, "L3 cache clear error");
}
}
pub fn hybrid_hits(&self) -> u64 {
self.stats.hybrid_hits.load(Ordering::Relaxed)
}
pub fn l3_hits(&self) -> u64 {
self.stats.l3_hits.load(Ordering::Relaxed)
}
pub fn misses(&self) -> u64 {
self.stats.misses.load(Ordering::Relaxed)
}
pub fn errors(&self) -> u64 {
self.stats.errors.load(Ordering::Relaxed)
}
pub fn hit_ratio(&self) -> Option<f64> {
let hits = self.hybrid_hits() + self.l3_hits();
let total = hits + self.misses();
if total == 0 {
None
} else {
Some(hits as f64 / total as f64)
}
}
}
fn unix_timestamp_millis() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
fn expires_at(ttl_seconds: u64) -> u64 {
if ttl_seconds == 0 {
return u64::MAX;
}
unix_timestamp_millis().saturating_add(ttl_seconds.saturating_mul(1000))
}
pub(super) async fn new_hybrid(
cfg: &AppCacheSettings,
) -> Result<Bincache<String, Hybrid, Noop>, bincache::Error> {
let disk_path = cfg.disk_path.as_deref().unwrap_or("./data/cache/hybrid");
let hybrid = Bincache::new(
Hybrid::new(
Path::new(disk_path),
Limits::new(Some(cfg.max_size), Some(cfg.max_capacity)),
if cfg.enable_disk {
Limits::new(
Some(cfg.disk_cache_max_size),
Some(cfg.disk_cache_max_capacity),
)
} else {
Limits::new(Some(0), Some(0))
},
),
Some(Noop),
)
.await?;
debug!(path = %disk_path, capacity = %cfg.max_capacity, "hybrid cache initialised");
Ok(hybrid)
}