multi-tier-cache 0.4.0

High-performance multi-tier cache with L1 (Moka) + L2 (Redis), cross-instance invalidation, and stampede protection
Documentation

🚀 multi-tier-cache

Crates.io Documentation License

A high-performance, production-ready multi-tier caching library for Rust featuring L1 (in-memory) + L2 (Redis) caches, automatic stampede protection, and built-in Redis Streams support.

✨ Features

  • 🔥 Multi-Tier Architecture: Combines fast in-memory (Moka) with persistent distributed (Redis) caching
  • 🔄 Cross-Instance Cache Invalidation (v0.4.0+): Real-time cache synchronization across all instances via Redis Pub/Sub
  • 🔌 Pluggable Backends (v0.3.0+): Swap Moka/Redis with custom implementations (DashMap, Memcached, etc.)
  • 🛡️ Cache Stampede Protection: DashMap + Mutex request coalescing prevents duplicate computations (99.6% latency reduction: 534ms → 5.2ms)
  • 📊 Redis Streams: Built-in publish/subscribe with automatic trimming for event streaming
  • ⚡ Automatic L2-to-L1 Promotion: Intelligent cache tier promotion for frequently accessed data with TTL preservation
  • 📈 Comprehensive Statistics: Hit rates, promotions, in-flight request tracking, invalidation metrics
  • 🎯 Zero-Config: Sensible defaults, works out of the box
  • ✅ Production-Proven: Battle-tested at 16,829+ RPS with 5.2ms latency and 95% hit rate

🏗️ Architecture

Request → L1 Cache (Moka) → L2 Cache (Redis) → Compute/Fetch
          ↓ Hit (90%)       ↓ Hit (75%)        ↓ Miss (5%)
          Return            Promote to L1       Store in L1+L2

Cache Flow

  1. Fast Path: Check L1 cache (sub-millisecond, 90% hit rate)
  2. Fallback: Check L2 cache (2-5ms, 75% hit rate) + auto-promote to L1
  3. Compute: Fetch/compute fresh data with stampede protection, store in both tiers

📦 Installation

Add to your Cargo.toml:

[dependencies]
multi-tier-cache = "0.4"
tokio = { version = "1.28", features = ["full"] }
serde_json = "1.0"

Version Guide:

  • v0.4.0+: Cross-instance cache invalidation via Redis Pub/Sub
  • v0.3.0+: Pluggable backends, trait-based architecture
  • v0.2.0+: Type-safe database caching with get_or_compute_typed()
  • v0.1.0+: Core multi-tier caching with stampede protection

🚀 Quick Start

use multi_tier_cache::{CacheSystem, CacheStrategy};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Initialize cache system (uses REDIS_URL env var)
    let cache = CacheSystem::new().await?;

    // Store data with cache strategy
    let data = serde_json::json!({"user": "alice", "score": 100});
    cache.cache_manager()
        .set_with_strategy("user:1", data, CacheStrategy::ShortTerm)
        .await?;

    // Retrieve data (L1 first, then L2 fallback)
    if let Some(cached) = cache.cache_manager().get("user:1").await? {
        println!("Cached data: {}", cached);
    }

    // Get statistics
    let stats = cache.cache_manager().get_stats();
    println!("Hit rate: {:.2}%", stats.hit_rate);

    Ok(())
}

💡 Usage Patterns

1. Cache Strategies

Choose the right TTL for your use case:

use std::time::Duration;

// RealTime (10s) - Fast-changing data
cache.cache_manager()
    .set_with_strategy("live_price", data, CacheStrategy::RealTime)
    .await?;

// ShortTerm (5min) - Frequently accessed data
cache.cache_manager()
    .set_with_strategy("session:123", data, CacheStrategy::ShortTerm)
    .await?;

// MediumTerm (1hr) - Moderately stable data
cache.cache_manager()
    .set_with_strategy("catalog", data, CacheStrategy::MediumTerm)
    .await?;

// LongTerm (3hr) - Stable data
cache.cache_manager()
    .set_with_strategy("config", data, CacheStrategy::LongTerm)
    .await?;

// Custom - Specific requirements
cache.cache_manager()
    .set_with_strategy("metrics", data, CacheStrategy::Custom(Duration::from_secs(30)))
    .await?;

2. Compute-on-Miss Pattern

Fetch data only when cache misses, with stampede protection:

async fn fetch_from_database(id: u32) -> anyhow::Result<serde_json::Value> {
    // Expensive operation...
    Ok(serde_json::json!({"id": id, "data": "..."}))
}

// Only ONE request will compute, others wait and read from cache
let product = cache.cache_manager()
    .get_or_compute_with(
        "product:42",
        CacheStrategy::MediumTerm,
        || fetch_from_database(42)
    )
    .await?;

3. Redis Streams Integration

Publish and consume events:

// Publish to stream
let fields = vec![
    ("event_id".to_string(), "123".to_string()),
    ("event_type".to_string(), "user_action".to_string()),
    ("timestamp".to_string(), "2025-01-01T00:00:00Z".to_string()),
];
let entry_id = cache.cache_manager()
    .publish_to_stream("events_stream", fields, Some(1000)) // Auto-trim to 1000 entries
    .await?;

// Read latest entries
let entries = cache.cache_manager()
    .read_stream_latest("events_stream", 10)
    .await?;

// Blocking read for new entries
let new_entries = cache.cache_manager()
    .read_stream("events_stream", "$", 10, Some(5000)) // Block for 5s
    .await?;

4. Type-Safe Database Caching (New in 0.2.0! 🎉)

Eliminate boilerplate with automatic serialization/deserialization for database queries:

use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize)]
struct User {
    id: i64,
    name: String,
    email: String,
}

// ❌ OLD WAY: Manual cache + serialize + deserialize (40+ lines)
let cached = cache.cache_manager().get("user:123").await?;
let user: User = match cached {
    Some(json) => serde_json::from_value(json)?,
    None => {
        let user = sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
            .bind(123)
            .fetch_one(&pool)
            .await?;
        let json = serde_json::to_value(&user)?;
        cache.cache_manager().set_with_strategy("user:123", json, CacheStrategy::MediumTerm).await?;
        user
    }
};

// ✅ NEW WAY: Type-safe automatic caching (5 lines)
let user: User = cache.cache_manager()
    .get_or_compute_typed(
        "user:123",
        CacheStrategy::MediumTerm,
        || async {
            sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
                .bind(123)
                .fetch_one(&pool)
                .await
        }
    )
    .await?;

Benefits:

  • ✅ Type-Safe: Compiler checks types, no runtime surprises
  • ✅ Zero Boilerplate: Automatic serialize/deserialize
  • ✅ Full Cache Features: L1→L2 fallback, stampede protection, auto-promotion
  • ✅ Generic: Works with any type implementing Serialize + DeserializeOwned

More Examples:

// PostgreSQL Reports
#[derive(Serialize, Deserialize)]
struct Report {
    id: i64,
    title: String,
    data: serde_json::Value,
}

let report: Report = cache.cache_manager()
    .get_or_compute_typed(
        &format!("report:{}", id),
        CacheStrategy::LongTerm,
        || async {
            sqlx::query_as("SELECT * FROM reports WHERE id = $1")
                .bind(id)
                .fetch_one(&pool)
                .await
        }
    )
    .await?;

// API Responses
#[derive(Serialize, Deserialize)]
struct ApiData {
    status: String,
    items: Vec<String>,
}

let data: ApiData = cache.cache_manager()
    .get_or_compute_typed(
        "api:external",
        CacheStrategy::RealTime,
        || async {
            reqwest::get("https://api.example.com/data")
                .await?
                .json::<ApiData>()
                .await
        }
    )
    .await?;

// Complex Computations
#[derive(Serialize, Deserialize)]
struct AnalyticsResult {
    total: i64,
    average: f64,
    breakdown: HashMap<String, i64>,
}

let analytics: AnalyticsResult = cache.cache_manager()
    .get_or_compute_typed(
        "analytics:monthly",
        CacheStrategy::Custom(Duration::from_secs(6 * 3600)),
        || async {
            // Expensive computation...
            compute_monthly_analytics(&pool).await
        }
    )
    .await?;

Performance:

  • L1 Hit: <1ms + deserialization (~10-50μs)
  • L2 Hit: 2-5ms + deserialization + L1 promotion
  • Cache Miss: Your query time + serialization + L1+L2 storage

5. Cross-Instance Cache Invalidation (New in 0.4.0! 🎉)

Keep caches synchronized across multiple servers/instances using Redis Pub/Sub:

Why Invalidation?

In distributed systems with multiple cache instances, stale data is a common problem:

  • User updates profile on Server A → Cache on Server B still has old data
  • Admin changes product price → Other servers show outdated prices
  • TTL-only expiration → Users see stale data until timeout

Solution: Real-time cache invalidation across ALL instances!

Two Invalidation Strategies

1. Remove Strategy (Lazy Reload)

use multi_tier_cache::{CacheManager, L1Cache, L2Cache, InvalidationConfig};

// Initialize with invalidation support
let config = InvalidationConfig::default();
let cache_manager = CacheManager::new_with_invalidation(
    Arc::new(L1Cache::new().await?),
    Arc::new(L2Cache::new().await?),
    "redis://localhost",
    config
).await?;

// Update database
database.update_user(123, new_data).await?;

// Invalidate cache across ALL instances
// → Cache removed, next access triggers reload
cache_manager.invalidate("user:123").await?;

2. Update Strategy (Zero Cache Miss)

// Update database
database.update_user(123, new_data).await?;

// Push new data directly to ALL instances' L1 caches
// → No cache miss, instant update!
cache_manager.update_cache(
    "user:123",
    serde_json::to_value(&new_data)?,
    Some(Duration::from_secs(3600))
).await?;

Pattern-Based Invalidation

Invalidate multiple related keys at once:

// Update product category in database
database.update_category(42, new_price).await?;

// Invalidate ALL products in category across ALL instances
cache_manager.invalidate_pattern("product:category:42:*").await?;

Write-Through Caching

Cache and broadcast in one operation:

let report = generate_monthly_report().await?;

// Cache locally AND broadcast to all other instances
cache_manager.set_with_broadcast(
    "report:monthly",
    serde_json::to_value(&report)?,
    CacheStrategy::LongTerm
).await?;

How It Works

Instance A              Redis Pub/Sub           Instance B
    │                        │                       │
    │  1. Update data        │                       │
    │  2. Broadcast msg  ───>│                       │
    │                        │  3. Receive msg  ───>│
    │                        │  4. Update L1    ────┘
    │                        │                       ✓

Performance:

  • Latency: ~1-5ms invalidation propagation
  • Overhead: Negligible (<0.1% CPU for subscriber)
  • Production-Safe: Auto-reconnection, error recovery

Configuration

use multi_tier_cache::InvalidationConfig;

let config = InvalidationConfig {
    channel: "my_app:cache:invalidate".to_string(),
    auto_broadcast_on_write: false,  // Manual control
    enable_audit_stream: true,       // Enable audit trail
    audit_stream: "cache:invalidations".to_string(),
    audit_stream_maxlen: Some(10000),
};

When to Use:

  • ✅ Multi-server deployments (load balancers, horizontal scaling)
  • ✅ Data that changes frequently (user profiles, prices, inventory)
  • ✅ Real-time requirements (instant consistency)
  • ❌ Single-server deployments (unnecessary overhead)
  • ❌ Rarely-changing data (TTL is sufficient)

Comparison:

Strategy Bandwidth Cache Miss Use Case
Remove Low Yes (on next access) Large values, infrequent access
Update Higher No (instant) Small values, frequent access
Pattern Medium Yes Bulk invalidation (categories)

6. Custom Cache Backends (New in 0.3.0! 🎉)

Starting from v0.3.0, you can replace the default Moka (L1) and Redis (L2) backends with your own custom implementations!

Use Cases:

  • Replace Redis with Memcached, DragonflyDB, or KeyDB
  • Use DashMap instead of Moka for L1
  • Implement no-op caches for testing
  • Add custom cache eviction policies
  • Integrate with proprietary caching systems

Basic Example: Custom HashMap L1 Cache

use multi_tier_cache::{CacheBackend, CacheSystemBuilder, async_trait};
use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};
use anyhow::Result;

struct HashMapCache {
    store: Arc<RwLock<HashMap<String, (serde_json::Value, Instant)>>>,
}

#[async_trait]
impl CacheBackend for HashMapCache {
    async fn get(&self, key: &str) -> Option<serde_json::Value> {
        let store = self.store.read().unwrap();
        store.get(key).and_then(|(value, expiry)| {
            if *expiry > Instant::now() {
                Some(value.clone())
            } else {
                None
            }
        })
    }

    async fn set_with_ttl(
        &self,
        key: &str,
        value: serde_json::Value,
        ttl: Duration,
    ) -> Result<()> {
        let mut store = self.store.write().unwrap();
        store.insert(key.to_string(), (value, Instant::now() + ttl));
        Ok(())
    }

    async fn remove(&self, key: &str) -> Result<()> {
        self.store.write().unwrap().remove(key);
        Ok(())
    }

    async fn health_check(&self) -> bool {
        true
    }

    fn name(&self) -> &str {
        "HashMap"
    }
}

// Use custom backend
let custom_l1 = Arc::new(HashMapCache::new());

let cache = CacheSystemBuilder::new()
    .with_l1(custom_l1 as Arc<dyn CacheBackend>)
    .build()
    .await?;

Advanced: Custom L2 Backend with TTL

For L2 caches, implement L2CacheBackend which extends CacheBackend with get_with_ttl():

use multi_tier_cache::{L2CacheBackend, async_trait};

#[async_trait]
impl CacheBackend for MyCustomL2 {
    // ... implement CacheBackend methods
}

#[async_trait]
impl L2CacheBackend for MyCustomL2 {
    async fn get_with_ttl(
        &self,
        key: &str,
    ) -> Option<(serde_json::Value, Option<Duration>)> {
        // Return value with remaining TTL
        Some((value, Some(remaining_ttl)))
    }
}

Builder API

use multi_tier_cache::CacheSystemBuilder;

let cache = CacheSystemBuilder::new()
    .with_l1(custom_l1)        // Custom L1 backend
    .with_l2(custom_l2)        // Custom L2 backend
    .with_streams(kafka)       // Optional: Custom streaming backend
    .build()
    .await?;

Mix and Match:

  • Use custom L1 with default Redis L2
  • Use default Moka L1 with custom L2
  • Replace both L1 and L2 backends

See: examples/custom_backends.rs for complete working examples including:

  • HashMap L1 cache
  • In-memory L2 cache with TTL
  • No-op cache (for testing)
  • Mixed backend configurations

⚖️ Feature Compatibility

Invalidation + Custom Backends

✅ Compatible:

  • Cache invalidation works with default Redis L2 backend
  • Single-key operations (invalidate, update_cache) work with any backend
  • Type-safe caching works with all backends
  • Stampede protection works with all backends

⚠️ Limited Support:

  • Pattern-based invalidation (invalidate_pattern) requires concrete Redis L2Cache
  • Custom L2 backends: Single-key invalidation works, but pattern invalidation not available
  • Workaround: Implement pattern matching in your custom backend

Example:

// ✅ Works: Default Redis + Invalidation
let cache = CacheManager::new_with_invalidation(
    Arc::new(L1Cache::new().await?),
    Arc::new(L2Cache::new().await?),  // Concrete Redis L2
    "redis://localhost",
    InvalidationConfig::default()
).await?;

cache.invalidate("key").await?;           // ✅ Works
cache.invalidate_pattern("user:*").await?; // ✅ Works (has scan_keys)

// ⚠️ Limited: Custom L2 + Invalidation
let cache = CacheManager::new_with_backends(
    custom_l1,
    custom_l2,  // Custom trait-based L2
    None
).await?;

// Pattern invalidation not available without concrete L2Cache
// Use single-key invalidation instead

Combining All Features

All features work together seamlessly:

use multi_tier_cache::*;

// v0.4.0: Invalidation
let config = InvalidationConfig::default();

// v0.3.0: Custom backends (or use defaults)
let l1 = Arc::new(L1Cache::new().await?);
let l2 = Arc::new(L2Cache::new().await?);

// Initialize with invalidation
let cache_manager = CacheManager::new_with_invalidation(
    l1, l2, "redis://localhost", config
).await?;

// v0.2.0: Type-safe caching
let user: User = cache_manager.get_or_compute_typed(
    "user:123",
    CacheStrategy::MediumTerm,
    || fetch_user(123)
).await?;

// v0.4.0: Invalidate across instances
cache_manager.invalidate("user:123").await?;

// v0.1.0: All core features work
let stats = cache_manager.get_stats();
println!("Hit rate: {:.2}%", stats.hit_rate);

No Conflicts: All features are designed to work together without interference.

📊 Performance Benchmarks

Tested in production environment:

Metric Value
Throughput 16,829+ requests/second
Latency (p50) 5.2ms
Cache Hit Rate 95% (L1: 90%, L2: 75%)
Stampede Protection 99.6% latency reduction (534ms → 5.2ms)
Success Rate 100% (zero failures under load)

Comparison with Other Libraries

Library Multi-Tier Stampede Protection Redis Support Streams Invalidation
multi-tier-cache ✅ L1+L2 ✅ Full ✅ Full ✅ Built-in ✅ Pub/Sub
cached ❌ Single ❌ No ❌ No ❌ No ❌ No
moka ❌ L1 only ✅ L1 only ❌ No ❌ No ❌ No
redis-rs ❌ No cache ❌ Manual ✅ Low-level ✅ Manual ❌ Manual

🔧 Configuration

Redis Connection (REDIS_URL)

The library connects to Redis using the REDIS_URL environment variable. Configuration priority (highest to lowest):

1. Programmatic Configuration (Highest Priority)

// Set custom Redis URL before initialization
let cache = CacheSystem::with_redis_url("redis://production:6379").await?;

2. Environment Variable

# Set in shell
export REDIS_URL="redis://your-redis-host:6379"
cargo run

3. .env File (Recommended for Development)

# Create .env file in project root
REDIS_URL="redis://localhost:6379"

4. Default Fallback

If not configured, defaults to: redis://127.0.0.1:6379

Use Cases

Development (Local Redis)

# .env
REDIS_URL="redis://127.0.0.1:6379"

Production (Cloud Redis with Authentication)

# Railway, Render, AWS ElastiCache, etc.
REDIS_URL="redis://:your-password@redis-host.cloud:6379"

Docker Compose

services:
  app:
    environment:
      - REDIS_URL=redis://redis:6379
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

Testing (Separate Instance)

#[tokio::test]
async fn test_cache() {
    let cache = CacheSystem::with_redis_url("redis://localhost:6380").await?;
    // Test logic...
}

Redis URL Format

redis://[username]:[password]@[host]:[port]/[database]

Examples:

  • redis://localhost:6379 - Local Redis, no authentication
  • redis://:mypassword@localhost:6379 - Local with password only
  • redis://user:pass@redis.example.com:6379/0 - Remote with username, password, and database 0
  • rediss://redis.cloud:6380 - SSL/TLS connection (note the rediss://)

Troubleshooting Redis Connection

Connection Refused

# Check if Redis is running
redis-cli ping  # Should return "PONG"

# Check the port
netstat -an | grep 6379

# Verify REDIS_URL
echo $REDIS_URL

Authentication Failed

# Ensure password is in the URL
REDIS_URL="redis://:YOUR_PASSWORD@host:6379"

# Test connection with redis-cli
redis-cli -h host -p 6379 -a YOUR_PASSWORD ping

Timeout Errors

  • Check network connectivity: ping your-redis-host
  • Verify firewall rules allow port 6379
  • Check Redis maxclients setting (may be full)
  • Review Redis logs: redis-cli INFO clients

DNS Resolution Issues

# Test DNS resolution
nslookup your-redis-host.com

# Use IP address as fallback
REDIS_URL="redis://192.168.1.100:6379"

Cache Tuning

Default settings (configurable in library source):

  • L1 Capacity: 2000 entries
  • L1 TTL: 5 minutes (per key)
  • L2 TTL: 1 hour (per key)
  • Stream Max Length: 1000 entries

📚 Examples

Run examples with:

# Basic usage
cargo run --example basic_usage

# Stampede protection demonstration
cargo run --example stampede_protection

# Redis Streams
cargo run --example redis_streams

# Cache strategies
cargo run --example cache_strategies

# Advanced patterns
cargo run --example advanced_usage

# Health monitoring
cargo run --example health_monitoring

🏛️ Architecture Details

Cache Stampede Protection

When multiple requests hit an expired cache key simultaneously:

  1. First request acquires DashMap mutex lock and computes value
  2. Subsequent requests wait on the same mutex
  3. After computation, all requests read from cache
  4. Result: Only ONE computation instead of N

Performance Impact:

  • Without protection: 10 requests × 500ms = 5000ms total
  • With protection: 1 request × 500ms = 500ms total (90% faster)

L2-to-L1 Promotion

When data is found in L2 but not L1:

  1. Retrieve from Redis (L2)
  2. Automatically store in Moka (L1) with fresh TTL
  3. Future requests hit fast L1 cache
  4. Result: Self-optimizing cache that adapts to access patterns

🛠️ Development

Build

# Development
cargo build

# Release (optimized)
cargo build --release

# Run tests
cargo test

Documentation

# Generate and open docs
cargo doc --open

📖 Migration Guide

From cached crate

// Before (cached)
use cached::proc_macro::cached;

#[cached(time = 60)]
fn expensive_function(arg: String) -> String {
    // ...
}

// After (multi-tier-cache)
async fn expensive_function(cache: &CacheManager, arg: String) -> Result<String> {
    cache.get_or_compute_with(
        &format!("func:{}", arg),
        CacheStrategy::ShortTerm,
        || async { /* computation */ }
    ).await
}

From direct Redis usage

// Before (redis-rs)
let mut conn = client.get_connection()?;
let value: String = conn.get("key")?;
conn.set_ex("key", value, 3600)?;

// After (multi-tier-cache)
if let Some(value) = cache.cache_manager().get("key").await? {
    // Use cached value
}
cache.cache_manager()
    .set_with_strategy("key", value, CacheStrategy::MediumTerm)
    .await?;

🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

📄 License

Licensed under either of:

at your option.

🙏 Acknowledgments

Built with:

  • Moka - High-performance concurrent cache library
  • Redis-rs - Redis client for Rust
  • DashMap - Blazingly fast concurrent map
  • Tokio - Asynchronous runtime

📞 Contact

Made with ❤️ in Rust | Production-proven in crypto trading dashboard serving 16,829+ RPS