warpdrive_proxy/cache/

coordinator.rs

//! Cache coordinator with distributed + local fallback strategy
//!
//! This module provides a high-level cache coordinator that manages both
//! distributed (Redis) and local (Memory) caches with automatic fallback.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────┐
//! │ Coordinator │
//! └──────┬──────┘
//!        │
//!    ┌───┴───┐
//!    │       │
//!    ▼       ▼
//! ┌─────┐ ┌────────┐
//! │Redis│ │ Memory │
//! └─────┘ └────────┘
//!   (L2)     (L1)
//! ```
//!
//! **Strategy**:
//! - GET: L1 → L2 → upstream (populate L1 + L2 on miss)
//! - SET: L1 + L2 (write-through)
//! - DELETE: L1 + L2 (invalidate both)
//!
//! **Fallback**:
//! - If Redis unavailable, use Memory only
//! - If Redis operation fails, continue with Memory
//! - Log degraded mode for monitoring
//!
//! # Example
//!
//! ```no_run
//! use warpdrive::cache::coordinator::CacheCoordinator;
//! use warpdrive::config::Config;
//!
//! # async fn example() -> anyhow::Result<()> {
//! let config = Config::from_env()?;
//! let cache = CacheCoordinator::from_config(&config).await?;
//!
//! // Use cache (automatic Redis → Memory fallback)
//! cache.set("key", b"value", 60).await?;
//! let value = cache.get("key").await?;
//! # Ok(())
//! # }
//! ```

use async_trait::async_trait;
use std::sync::Arc;
use tracing::{debug, warn};

use crate::cache::Cache;
use crate::cache::memory::MemoryCache;
use crate::cache::redis::RedisCache;
use crate::config::Config;

/// Cache backend selection
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CacheBackend {
    /// Redis + Memory (L1 + L2 strategy)
    RedisMemory,
    /// Memory only (Redis unavailable or not configured)
    MemoryOnly,
}

/// Cache coordinator managing distributed and local caches
///
/// Provides automatic fallback from Redis (L2) to Memory (L1) on errors.
/// Memory cache acts as both L1 fast-path and fallback for Redis failures.
#[derive(Clone)]
pub struct CacheCoordinator {
    /// Local memory cache (L1) - always present
    memory: Arc<MemoryCache>,

    /// Distributed Redis cache (L2) - optional
    redis: Option<Arc<RedisCache>>,

    /// Current backend mode
    backend: CacheBackend,
}

impl CacheCoordinator {
    /// Create cache coordinator from config
    ///
    /// Initializes both Redis and Memory caches based on configuration:
    /// - If `redis_url` is set, initializes Redis (degraded mode if fails)
    /// - Always initializes Memory cache with configured size
    ///
    /// # Errors
    ///
    /// Returns error only if Memory cache creation fails (should never happen).
    /// Redis connection failures are logged but don't fail initialization.
    pub async fn from_config(config: &Config) -> anyhow::Result<Self> {
        // Always create memory cache
        let memory = Arc::new(MemoryCache::new(
            config.cache_size_bytes,
            config.max_cache_item_size_bytes,
        ));

        debug!(
            cache_size_mb = config.cache_size_bytes / crate::config::MB,
            max_item_size_kb = config.max_cache_item_size_bytes / crate::config::KB,
            "Initialized memory cache (L1)"
        );

        // Try to create Redis cache if configured
        let redis = if let Some(ref redis_url) = config.redis_url {
            match RedisCache::from_url(redis_url, "warpdrive:cache:".to_string()).await {
                Ok(cache) => {
                    debug!(redis_url = %redis_url, "Initialized Redis cache (L2)");
                    Some(Arc::new(cache))
                }
                Err(e) => {
                    warn!(
                        error = %e,
                        "Failed to connect to Redis, running in memory-only mode"
                    );
                    None
                }
            }
        } else {
            debug!("No Redis URL configured, running in memory-only mode");
            None
        };

        let backend = if redis.is_some() {
            CacheBackend::RedisMemory
        } else {
            CacheBackend::MemoryOnly
        };

        Ok(CacheCoordinator {
            memory,
            redis,
            backend,
        })
    }

    /// Get current cache backend mode
    pub fn backend(&self) -> CacheBackend {
        self.backend
    }

    /// Get memory cache statistics
    ///
    /// Returns (current_size, capacity, item_count)
    pub fn memory_stats(&self) -> (usize, usize, usize) {
        self.memory.stats()
    }

    /// Clear memory cache (for testing/admin operations)
    pub fn clear_memory(&self) {
        self.memory.clear()
    }

    /// Get Redis cache for advanced operations
    ///
    /// Returns `None` if Redis is not configured or unavailable.
    pub fn redis(&self) -> Option<&Arc<RedisCache>> {
        self.redis.as_ref()
    }
}

#[async_trait]
impl Cache for CacheCoordinator {
    async fn get(&self, key: &str) -> anyhow::Result<Option<Vec<u8>>> {
        // Try L1 (Memory) first
        if let Some(value) = self.memory.get(key).await? {
            debug!(key = %key, backend = "memory", "Cache hit (L1)");
            crate::metrics::CACHE_HITS
                .with_label_values(&["memory"])
                .inc();
            return Ok(Some(value));
        }

        // Try L2 (Redis) if available
        if let Some(ref redis) = self.redis {
            match redis.get(key).await {
                Ok(Some(value)) => {
                    debug!(key = %key, backend = "redis", "Cache hit (L2)");
                    crate::metrics::CACHE_HITS
                        .with_label_values(&["redis"])
                        .inc();

                    // Populate L1 for next access (fire and forget)
                    let memory = Arc::clone(&self.memory);
                    let key = key.to_string();
                    let value_clone = value.clone();
                    tokio::spawn(async move {
                        // Use default TTL of 60s for L1 backfill
                        let _ = memory.set(&key, &value_clone, 60).await;
                    });

                    return Ok(Some(value));
                }
                Ok(None) => {
                    debug!(key = %key, backend = "redis", "Cache miss (L2)");
                    crate::metrics::CACHE_MISSES
                        .with_label_values(&["redis"])
                        .inc();
                }
                Err(e) => {
                    warn!(
                        key = %key,
                        error = %e,
                        "Redis get failed, continuing with memory-only"
                    );
                    crate::metrics::CACHE_ERRORS
                        .with_label_values(&["redis", "get"])
                        .inc();
                }
            }
        }

        // Miss on both L1 and L2
        crate::metrics::CACHE_MISSES
            .with_label_values(&["memory"])
            .inc();
        Ok(None)
    }

    async fn set(&self, key: &str, value: &[u8], ttl_seconds: u64) -> anyhow::Result<()> {
        // Write to L1 (Memory) - always succeeds
        self.memory.set(key, value, ttl_seconds).await?;
        debug!(key = %key, ttl = ttl_seconds, backend = "memory", "Cache set (L1)");

        // Write to L2 (Redis) if available
        if let Some(ref redis) = self.redis {
            match redis.set(key, value, ttl_seconds).await {
                Ok(_) => {
                    debug!(key = %key, ttl = ttl_seconds, backend = "redis", "Cache set (L2)");
                }
                Err(e) => {
                    warn!(
                        key = %key,
                        error = %e,
                        "Redis set failed, data only in L1 cache"
                    );
                    crate::metrics::CACHE_ERRORS
                        .with_label_values(&["redis", "set"])
                        .inc();
                }
            }
        }

        Ok(())
    }

    async fn delete(&self, key: &str) -> anyhow::Result<()> {
        // Delete from L1 (Memory)
        self.memory.delete(key).await?;
        debug!(key = %key, backend = "memory", "Cache delete (L1)");

        // Delete from L2 (Redis) if available
        if let Some(ref redis) = self.redis {
            match redis.delete(key).await {
                Ok(_) => {
                    debug!(key = %key, backend = "redis", "Cache delete (L2)");
                }
                Err(e) => {
                    warn!(
                        key = %key,
                        error = %e,
                        "Redis delete failed, L1 cleared but L2 may still have stale data"
                    );
                    crate::metrics::CACHE_ERRORS
                        .with_label_values(&["redis", "delete"])
                        .inc();
                }
            }
        }

        crate::metrics::CACHE_INVALIDATIONS
            .with_label_values(&["explicit"])
            .inc();
        Ok(())
    }

    async fn exists(&self, key: &str) -> anyhow::Result<bool> {
        // Check L1 first
        if self.memory.exists(key).await? {
            return Ok(true);
        }

        // Check L2 if available
        if let Some(ref redis) = self.redis {
            match redis.exists(key).await {
                Ok(exists) => return Ok(exists),
                Err(e) => {
                    warn!(
                        key = %key,
                        error = %e,
                        "Redis exists check failed"
                    );
                    crate::metrics::CACHE_ERRORS
                        .with_label_values(&["redis", "exists"])
                        .inc();
                }
            }
        }

        Ok(false)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_memory_only_mode() {
        let config = Config {
            redis_url: None,
            cache_size_bytes: 1024 * 1024,
            max_cache_item_size_bytes: 1024,
            ..Default::default()
        };

        let cache = CacheCoordinator::from_config(&config).await.unwrap();
        assert_eq!(cache.backend(), CacheBackend::MemoryOnly);
        assert!(cache.redis().is_none());
    }

    #[tokio::test]
    async fn test_set_and_get_memory_only() {
        let config = Config {
            redis_url: None,
            cache_size_bytes: 1024 * 1024,
            max_cache_item_size_bytes: 1024,
            ..Default::default()
        };

        let cache = CacheCoordinator::from_config(&config).await.unwrap();

        cache.set("test", b"value", 60).await.unwrap();
        let value = cache.get("test").await.unwrap();
        assert_eq!(value, Some(b"value".to_vec()));
    }

    #[tokio::test]
    async fn test_delete() {
        let config = Config {
            redis_url: None,
            cache_size_bytes: 1024 * 1024,
            max_cache_item_size_bytes: 1024,
            ..Default::default()
        };

        let cache = CacheCoordinator::from_config(&config).await.unwrap();

        cache.set("test", b"value", 60).await.unwrap();
        assert!(cache.exists("test").await.unwrap());

        cache.delete("test").await.unwrap();
        assert!(!cache.exists("test").await.unwrap());
    }

    #[tokio::test]
    async fn test_memory_stats() {
        let config = Config {
            redis_url: None,
            cache_size_bytes: 1024 * 1024,
            max_cache_item_size_bytes: 1024,
            ..Default::default()
        };

        let cache = CacheCoordinator::from_config(&config).await.unwrap();

        let (size_before, capacity, count_before) = cache.memory_stats();
        assert_eq!(size_before, 0);
        assert_eq!(count_before, 0);
        assert_eq!(capacity, 1024 * 1024);

        cache.set("test", b"value", 60).await.unwrap();

        let (size_after, _, count_after) = cache.memory_stats();
        assert!(size_after > 0);
        assert_eq!(count_after, 1);
    }
}