use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use parking_lot::RwLock;
use tracing::{debug, info};
use super::tan_curve::{CacheEntry, TanCurvePolicy};
#[derive(Debug, Clone)]
pub struct RedisMemoryProfile {
pub used_bytes: u64,
pub max_bytes: u64,
pub last_updated: Instant,
pub pending_writes_estimate: u64,
}
impl Default for RedisMemoryProfile {
fn default() -> Self {
Self {
used_bytes: 0,
max_bytes: 100 * 1024 * 1024, last_updated: Instant::now(),
pending_writes_estimate: 0,
}
}
}
impl RedisMemoryProfile {
pub fn pressure(&self) -> f64 {
if self.max_bytes == 0 {
return 0.0;
}
(self.used_bytes + self.pending_writes_estimate) as f64 / self.max_bytes as f64
}
pub fn needs_refresh(&self, max_staleness: Duration, max_drift_bytes: u64) -> bool {
self.last_updated.elapsed() > max_staleness
|| self.pending_writes_estimate > max_drift_bytes
}
pub fn add_pending_writes(&mut self, bytes: u64) {
self.pending_writes_estimate += bytes;
}
pub fn refresh_from_info(&mut self, used_bytes: u64, max_bytes: u64) {
self.used_bytes = used_bytes;
self.max_bytes = max_bytes;
self.pending_writes_estimate = 0;
self.last_updated = Instant::now();
}
}
#[derive(Debug, Clone)]
pub struct RedisKeyMeta {
pub last_access: Instant,
pub access_count: u64,
pub size_bytes: usize,
pub protected: bool,
}
impl RedisKeyMeta {
pub fn new(size_bytes: usize, protected: bool) -> Self {
Self {
last_access: Instant::now(),
access_count: 1,
size_bytes,
protected,
}
}
pub fn touch(&mut self) {
self.last_access = Instant::now();
self.access_count += 1;
}
pub fn to_cache_entry(&self, id: String) -> CacheEntry {
CacheEntry {
id,
size_bytes: self.size_bytes,
created_at: self.last_access, last_access: self.last_access,
access_count: self.access_count,
is_dirty: self.protected, }
}
}
#[derive(Debug, Clone)]
pub struct RedisEvictionConfig {
pub eviction_start_pressure: f64,
pub eviction_target_pressure: f64,
pub max_profile_staleness: Duration,
pub max_profile_drift_bytes: u64,
pub eviction_batch_size: usize,
pub protected_prefixes: Vec<String>,
}
impl Default for RedisEvictionConfig {
fn default() -> Self {
Self {
eviction_start_pressure: 0.75, eviction_target_pressure: 0.60, max_profile_staleness: Duration::from_secs(5),
max_profile_drift_bytes: 1024 * 1024, eviction_batch_size: 100,
protected_prefixes: vec![
"merkle:".to_string(),
"idx:".to_string(),
],
}
}
}
pub struct RedisEvictionManager {
config: RedisEvictionConfig,
policy: TanCurvePolicy,
memory_profile: Arc<RwLock<RedisMemoryProfile>>,
key_metadata: Arc<RwLock<HashMap<String, RedisKeyMeta>>>,
prefix: Option<String>,
}
impl RedisEvictionManager {
pub fn new(config: RedisEvictionConfig, prefix: Option<String>) -> Self {
Self {
config,
policy: TanCurvePolicy::default(),
memory_profile: Arc::new(RwLock::new(RedisMemoryProfile::default())),
key_metadata: Arc::new(RwLock::new(HashMap::new())),
prefix,
}
}
pub fn pressure(&self) -> f64 {
self.memory_profile.read().pressure()
}
pub fn needs_profile_refresh(&self) -> bool {
let profile = self.memory_profile.read();
profile.needs_refresh(
self.config.max_profile_staleness,
self.config.max_profile_drift_bytes,
)
}
pub fn refresh_profile(&self, used_bytes: u64, max_bytes: u64) {
let mut profile = self.memory_profile.write();
profile.refresh_from_info(used_bytes, max_bytes);
debug!(
used_mb = used_bytes / 1024 / 1024,
max_mb = max_bytes / 1024 / 1024,
pressure = format!("{:.1}%", profile.pressure() * 100.0),
"Redis memory profile refreshed"
);
}
pub fn record_batch_write(&self, bytes: u64) {
self.memory_profile.write().add_pending_writes(bytes);
}
pub fn record_key_write(&self, key: &str, size_bytes: usize) {
let protected = self.is_protected(key);
let mut metadata = self.key_metadata.write();
metadata.insert(key.to_string(), RedisKeyMeta::new(size_bytes, protected));
}
pub fn record_key_access(&self, key: &str) {
let mut metadata = self.key_metadata.write();
if let Some(meta) = metadata.get_mut(key) {
meta.touch();
}
}
pub fn remove_key(&self, key: &str) {
self.key_metadata.write().remove(key);
}
pub fn needs_eviction(&self) -> bool {
self.pressure() >= self.config.eviction_start_pressure
}
pub fn get_eviction_candidates(&self) -> Vec<String> {
let pressure = self.pressure();
if pressure < self.config.eviction_start_pressure {
return vec![];
}
let metadata = self.key_metadata.read();
let entries: Vec<CacheEntry> = metadata
.iter()
.map(|(key, meta)| meta.to_cache_entry(key.clone()))
.collect();
if entries.is_empty() {
return vec![];
}
let victims = self.policy.select_victims(
&entries,
self.config.eviction_batch_size,
pressure
);
if !victims.is_empty() {
info!(
candidates = victims.len(),
pressure = format!("{:.1}%", pressure * 100.0),
"Selected Redis eviction candidates"
);
}
victims
}
fn is_protected(&self, key: &str) -> bool {
let key_without_prefix = if let Some(ref prefix) = self.prefix {
key.strip_prefix(prefix).unwrap_or(key)
} else {
key
};
self.config.protected_prefixes.iter().any(|p| key_without_prefix.starts_with(p))
}
pub fn stats(&self) -> RedisEvictionStats {
let profile = self.memory_profile.read();
let metadata = self.key_metadata.read();
let protected_count = metadata.values().filter(|m| m.protected).count();
let data_count = metadata.len() - protected_count;
RedisEvictionStats {
used_bytes: profile.used_bytes,
max_bytes: profile.max_bytes,
pending_writes_estimate: profile.pending_writes_estimate,
pressure: profile.pressure(),
tracked_keys: metadata.len(),
protected_keys: protected_count,
data_keys: data_count,
profile_age_secs: profile.last_updated.elapsed().as_secs_f64(),
}
}
}
#[derive(Debug, Clone)]
pub struct RedisEvictionStats {
pub used_bytes: u64,
pub max_bytes: u64,
pub pending_writes_estimate: u64,
pub pressure: f64,
pub tracked_keys: usize,
pub protected_keys: usize,
pub data_keys: usize,
pub profile_age_secs: f64,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_memory_profile_pressure() {
let mut profile = RedisMemoryProfile {
used_bytes: 50 * 1024 * 1024, max_bytes: 100 * 1024 * 1024, ..Default::default()
};
assert!((profile.pressure() - 0.5).abs() < 0.01);
profile.add_pending_writes(25 * 1024 * 1024); assert!((profile.pressure() - 0.75).abs() < 0.01);
}
#[test]
fn test_profile_needs_refresh() {
let mut profile = RedisMemoryProfile::default();
assert!(!profile.needs_refresh(Duration::from_secs(5), 1024 * 1024));
profile.add_pending_writes(2 * 1024 * 1024);
assert!(profile.needs_refresh(Duration::from_secs(5), 1024 * 1024));
}
#[test]
fn test_protected_keys() {
let config = RedisEvictionConfig::default();
let manager = RedisEvictionManager::new(config, Some("sync:".to_string()));
assert!(manager.is_protected("sync:merkle:hash:user.alice")); assert!(manager.is_protected("sync:merkle:children:user"));
assert!(manager.is_protected("sync:idx:users"));
assert!(!manager.is_protected("sync:user.alice")); assert!(!manager.is_protected("sync:config.app"));
assert!(manager.is_protected("merkle:hash:test")); }
#[test]
fn test_no_eviction_under_threshold() {
let config = RedisEvictionConfig::default();
let manager = RedisEvictionManager::new(config, None);
manager.refresh_profile(30 * 1024 * 1024, 100 * 1024 * 1024); manager.record_key_write("data:key", 10_000);
let candidates = manager.get_eviction_candidates();
assert!(candidates.is_empty(), "Should not evict under threshold");
}
#[test]
fn test_eviction_uses_tan_curve_policy() {
let config = RedisEvictionConfig {
eviction_start_pressure: 0.5,
eviction_batch_size: 10,
..Default::default()
};
let manager = RedisEvictionManager::new(config, None);
manager.refresh_profile(80 * 1024 * 1024, 100 * 1024 * 1024);
manager.record_key_write("data:old", 10_000);
manager.record_key_write("data:new", 10_000);
{
let mut meta = manager.key_metadata.write();
if let Some(m) = meta.get_mut("data:old") {
m.last_access = Instant::now() - Duration::from_secs(3600);
}
}
let candidates = manager.get_eviction_candidates();
assert!(!candidates.is_empty());
if candidates.len() >= 2 {
assert_eq!(candidates[0], "data:old");
}
}
}