Rust Caching Strategies Best Practices¶
Objective: Master senior-level Rust caching patterns for production systems. When you need to implement high-performance caching, when you want to optimize data access patterns, when you need enterprise-grade caching strategiesโthese best practices become your weapon of choice.
Core Principles¶
- Cache-Aside Pattern: Application manages cache directly
- Write-Through Caching: Write to cache and database simultaneously
- Write-Behind Caching: Write to cache first, then database asynchronously
- Cache Invalidation: Proper cache invalidation strategies
- Distributed Caching: Multi-node cache coordination
Caching Strategies¶
In-Memory Caching¶
// rust/01-in-memory-caching.rs
/*
In-memory caching patterns and best practices
*/
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use serde::{Deserialize, Serialize};
/// In-memory cache with TTL support.
pub struct InMemoryCache<K, V> {
data: Arc<RwLock<HashMap<K, CacheEntry<V>>>>,
max_size: usize,
default_ttl: Duration,
cleanup_interval: Duration,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheEntry<V> {
pub value: V,
pub created_at: Instant,
pub expires_at: Instant,
pub access_count: u64,
pub last_accessed: Instant,
}
impl<K, V> InMemoryCache<K, V>
where
K: std::hash::Hash + Eq + Clone + Send + Sync + 'static,
V: Clone + Send + Sync + 'static,
{
pub fn new(max_size: usize, default_ttl: Duration, cleanup_interval: Duration) -> Self {
let cache = Self {
data: Arc::new(RwLock::new(HashMap::new())),
max_size,
default_ttl,
cleanup_interval,
};
// Start cleanup task
cache.start_cleanup_task();
cache
}
/// Get a value from the cache.
pub async fn get(&self, key: &K) -> Option<V> {
let mut data = self.data.write().await;
if let Some(entry) = data.get_mut(key) {
// Check if entry has expired
if Instant::now() > entry.expires_at {
data.remove(key);
return None;
}
// Update access statistics
entry.access_count += 1;
entry.last_accessed = Instant::now();
return Some(entry.value.clone());
}
None
}
/// Set a value in the cache.
pub async fn set(&self, key: K, value: V) -> Result<(), String> {
self.set_with_ttl(key, value, self.default_ttl).await
}
/// Set a value in the cache with custom TTL.
pub async fn set_with_ttl(&self, key: K, value: V, ttl: Duration) -> Result<(), String> {
let mut data = self.data.write().await;
// Check if we need to evict entries
if data.len() >= self.max_size {
self.evict_entries(&mut data).await;
}
let now = Instant::now();
let entry = CacheEntry {
value,
created_at: now,
expires_at: now + ttl,
access_count: 0,
last_accessed: now,
};
data.insert(key, entry);
Ok(())
}
/// Remove a value from the cache.
pub async fn remove(&self, key: &K) -> Option<V> {
let mut data = self.data.write().await;
data.remove(key).map(|entry| entry.value)
}
/// Clear all entries from the cache.
pub async fn clear(&self) {
let mut data = self.data.write().await;
data.clear();
}
/// Get cache statistics.
pub async fn get_statistics(&self) -> CacheStatistics {
let data = self.data.read().await;
let total_entries = data.len();
let expired_entries = data.values()
.filter(|entry| Instant::now() > entry.expires_at)
.count();
let active_entries = total_entries - expired_entries;
let total_access_count: u64 = data.values()
.map(|entry| entry.access_count)
.sum();
let average_access_count = if total_entries > 0 {
total_access_count as f64 / total_entries as f64
} else {
0.0
};
CacheStatistics {
total_entries,
active_entries,
expired_entries,
total_access_count,
average_access_count,
max_size: self.max_size,
}
}
/// Evict entries using LRU strategy.
async fn evict_entries(&self, data: &mut HashMap<K, CacheEntry<V>>) {
if data.is_empty() {
return;
}
// Find the least recently used entry
let mut lru_key = None;
let mut oldest_access = Instant::now();
for (key, entry) in data.iter() {
if entry.last_accessed < oldest_access {
oldest_access = entry.last_accessed;
lru_key = Some(key.clone());
}
}
if let Some(key) = lru_key {
data.remove(&key);
}
}
/// Start cleanup task to remove expired entries.
fn start_cleanup_task(&self) {
let data = Arc::clone(&self.data);
let cleanup_interval = self.cleanup_interval;
tokio::spawn(async move {
let mut interval = tokio::time::interval(cleanup_interval);
loop {
interval.tick().await;
let mut data = data.write().await;
let now = Instant::now();
data.retain(|_, entry| now <= entry.expires_at);
}
});
}
}
#[derive(Debug, Clone)]
pub struct CacheStatistics {
pub total_entries: usize,
pub active_entries: usize,
pub expired_entries: usize,
pub total_access_count: u64,
pub average_access_count: f64,
pub max_size: usize,
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_in_memory_cache() {
let cache = InMemoryCache::new(
100,
Duration::from_secs(60),
Duration::from_secs(10),
);
cache.set("key1".to_string(), "value1".to_string()).await.unwrap();
let value = cache.get(&"key1".to_string()).await;
assert_eq!(value, Some("value1".to_string()));
}
#[tokio::test]
async fn test_cache_ttl() {
let cache = InMemoryCache::new(
100,
Duration::from_millis(100),
Duration::from_secs(10),
);
cache.set("key1".to_string(), "value1".to_string()).await.unwrap();
// Wait for TTL to expire
tokio::time::sleep(Duration::from_millis(150)).await;
let value = cache.get(&"key1".to_string()).await;
assert_eq!(value, None);
}
#[tokio::test]
async fn test_cache_statistics() {
let cache = InMemoryCache::new(
100,
Duration::from_secs(60),
Duration::from_secs(10),
);
cache.set("key1".to_string(), "value1".to_string()).await.unwrap();
cache.get(&"key1".to_string()).await;
let stats = cache.get_statistics().await;
assert_eq!(stats.total_entries, 1);
assert_eq!(stats.active_entries, 1);
assert_eq!(stats.total_access_count, 1);
}
}
Distributed Caching¶
// rust/02-distributed-caching.rs
/*
Distributed caching patterns and best practices
*/
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use serde::{Deserialize, Serialize};
/// Distributed cache node.
pub struct CacheNode {
pub id: String,
pub address: String,
pub port: u16,
pub is_healthy: bool,
pub last_heartbeat: Instant,
pub load: f64,
}
/// Distributed cache manager.
pub struct DistributedCache {
nodes: Arc<RwLock<Vec<CacheNode>>>,
local_cache: Arc<InMemoryCache<String, String>>,
replication_factor: usize,
consistency_level: ConsistencyLevel,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ConsistencyLevel {
Eventual,
Strong,
Session,
}
impl DistributedCache {
pub fn new(
replication_factor: usize,
consistency_level: ConsistencyLevel,
local_cache_size: usize,
local_cache_ttl: Duration,
) -> Self {
Self {
nodes: Arc::new(RwLock::new(Vec::new())),
local_cache: Arc::new(InMemoryCache::new(
local_cache_size,
local_cache_ttl,
Duration::from_secs(10),
)),
replication_factor,
consistency_level,
}
}
/// Add a cache node.
pub async fn add_node(&self, node: CacheNode) {
let mut nodes = self.nodes.write().await;
nodes.push(node);
}
/// Remove a cache node.
pub async fn remove_node(&self, node_id: &str) {
let mut nodes = self.nodes.write().await;
nodes.retain(|node| node.id != node_id);
}
/// Get a value from the distributed cache.
pub async fn get(&self, key: &str) -> Result<Option<String>, String> {
// First, check local cache
if let Some(value) = self.local_cache.get(&key.to_string()).await {
return Ok(Some(value));
}
// Get from distributed nodes
let nodes = self.nodes.read().await;
let healthy_nodes: Vec<&CacheNode> = nodes
.iter()
.filter(|node| node.is_healthy)
.collect();
if healthy_nodes.is_empty() {
return Err("No healthy nodes available".to_string());
}
// Select nodes based on consistency level
let selected_nodes = self.select_nodes(&healthy_nodes, key);
// Try to get value from selected nodes
for node in selected_nodes {
if let Ok(Some(value)) = self.get_from_node(node, key).await {
// Cache locally
self.local_cache.set(key.to_string(), value.clone()).await?;
return Ok(Some(value));
}
}
Ok(None)
}
/// Set a value in the distributed cache.
pub async fn set(&self, key: String, value: String, ttl: Duration) -> Result<(), String> {
// Set in local cache
self.local_cache.set(key.clone(), value.clone()).await?;
// Set in distributed nodes
let nodes = self.nodes.read().await;
let healthy_nodes: Vec<&CacheNode> = nodes
.iter()
.filter(|node| node.is_healthy)
.collect();
if healthy_nodes.is_empty() {
return Err("No healthy nodes available".to_string());
}
// Select nodes for replication
let selected_nodes = self.select_nodes(&healthy_nodes, &key);
// Set value in selected nodes
let mut success_count = 0;
for node in selected_nodes {
if self.set_in_node(node, &key, &value, ttl).await.is_ok() {
success_count += 1;
}
}
// Check if we have enough successful writes
if success_count < self.replication_factor {
return Err("Insufficient successful writes".to_string());
}
Ok(())
}
/// Remove a value from the distributed cache.
pub async fn remove(&self, key: &str) -> Result<(), String> {
// Remove from local cache
self.local_cache.remove(&key.to_string()).await;
// Remove from distributed nodes
let nodes = self.nodes.read().await;
let healthy_nodes: Vec<&CacheNode> = nodes
.iter()
.filter(|node| node.is_healthy)
.collect();
if healthy_nodes.is_empty() {
return Err("No healthy nodes available".to_string());
}
// Select nodes for deletion
let selected_nodes = self.select_nodes(&healthy_nodes, key);
// Remove value from selected nodes
for node in selected_nodes {
self.remove_from_node(node, key).await?;
}
Ok(())
}
/// Select nodes for operation based on consistency level.
fn select_nodes(&self, nodes: &[&CacheNode], key: &str) -> Vec<&CacheNode> {
match self.consistency_level {
ConsistencyLevel::Eventual => {
// Select nodes based on consistent hashing
self.consistent_hash_select(nodes, key)
}
ConsistencyLevel::Strong => {
// Select all healthy nodes
nodes.to_vec()
}
ConsistencyLevel::Session => {
// Select nodes based on session affinity
self.session_affinity_select(nodes, key)
}
}
}
/// Select nodes using consistent hashing.
fn consistent_hash_select(&self, nodes: &[&CacheNode], key: &str) -> Vec<&CacheNode> {
// Simple consistent hashing implementation
let mut selected = Vec::new();
let mut hash = self.hash_key(key);
for _ in 0..self.replication_factor {
let node_index = hash % nodes.len();
if let Some(node) = nodes.get(node_index) {
selected.push(*node);
hash = hash.wrapping_add(1);
}
}
selected
}
/// Select nodes using session affinity.
fn session_affinity_select(&self, nodes: &[&CacheNode], key: &str) -> Vec<&CacheNode> {
// Simple session affinity implementation
let mut selected = Vec::new();
let session_id = self.extract_session_id(key);
let node_index = session_id % nodes.len();
if let Some(node) = nodes.get(node_index) {
selected.push(*node);
}
selected
}
/// Hash a key for consistent hashing.
fn hash_key(&self, key: &str) -> usize {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
let mut hasher = DefaultHasher::new();
key.hash(&mut hasher);
hasher.finish() as usize
}
/// Extract session ID from key.
fn extract_session_id(&self, key: &str) -> usize {
// Simple session ID extraction
key.len() % 1000
}
/// Get value from a specific node.
async fn get_from_node(&self, node: &CacheNode, key: &str) -> Result<Option<String>, String> {
// In a real implementation, you would make an HTTP request
// to the node's API
println!("Getting {} from node {}", key, node.id);
// Simulate network delay
tokio::time::sleep(Duration::from_millis(10)).await;
// For this example, assume all nodes return the same value
Ok(Some(format!("value_for_{}", key)))
}
/// Set value in a specific node.
async fn set_in_node(&self, node: &CacheNode, key: &str, value: &str, ttl: Duration) -> Result<(), String> {
// In a real implementation, you would make an HTTP request
// to the node's API
println!("Setting {} = {} in node {} with TTL {:?}", key, value, node.id, ttl);
// Simulate network delay
tokio::time::sleep(Duration::from_millis(10)).await;
Ok(())
}
/// Remove value from a specific node.
async fn remove_from_node(&self, node: &CacheNode, key: &str) -> Result<(), String> {
// In a real implementation, you would make an HTTP request
// to the node's API
println!("Removing {} from node {}", key, node.id);
// Simulate network delay
tokio::time::sleep(Duration::from_millis(10)).await;
Ok(())
}
/// Get cluster statistics.
pub async fn get_cluster_statistics(&self) -> ClusterStatistics {
let nodes = self.nodes.read().await;
let healthy_nodes = nodes.iter().filter(|node| node.is_healthy).count();
let total_nodes = nodes.len();
ClusterStatistics {
total_nodes,
healthy_nodes,
unhealthy_nodes: total_nodes - healthy_nodes,
replication_factor: self.replication_factor,
consistency_level: self.consistency_level.clone(),
}
}
}
#[derive(Debug, Clone)]
pub struct ClusterStatistics {
pub total_nodes: usize,
pub healthy_nodes: usize,
pub unhealthy_nodes: usize,
pub replication_factor: usize,
pub consistency_level: ConsistencyLevel,
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_distributed_cache() {
let cache = DistributedCache::new(
2,
ConsistencyLevel::Eventual,
100,
Duration::from_secs(60),
);
// Add nodes
cache.add_node(CacheNode {
id: "node1".to_string(),
address: "127.0.0.1".to_string(),
port: 8080,
is_healthy: true,
last_heartbeat: Instant::now(),
load: 0.5,
}).await;
cache.add_node(CacheNode {
id: "node2".to_string(),
address: "127.0.0.1".to_string(),
port: 8081,
is_healthy: true,
last_heartbeat: Instant::now(),
load: 0.3,
}).await;
// Test set and get
cache.set("key1".to_string(), "value1".to_string(), Duration::from_secs(60)).await.unwrap();
let value = cache.get("key1").await.unwrap();
assert_eq!(value, Some("value1".to_string()));
}
#[tokio::test]
async fn test_cluster_statistics() {
let cache = DistributedCache::new(
2,
ConsistencyLevel::Eventual,
100,
Duration::from_secs(60),
);
cache.add_node(CacheNode {
id: "node1".to_string(),
address: "127.0.0.1".to_string(),
port: 8080,
is_healthy: true,
last_heartbeat: Instant::now(),
load: 0.5,
}).await;
let stats = cache.get_cluster_statistics().await;
assert_eq!(stats.total_nodes, 1);
assert_eq!(stats.healthy_nodes, 1);
}
}
Cache-Aside Pattern¶
// rust/03-cache-aside-pattern.rs
/*
Cache-aside pattern implementation and best practices
*/
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use serde::{Deserialize, Serialize};
/// Cache-aside pattern implementation.
pub struct CacheAsidePattern<T> {
cache: Arc<InMemoryCache<String, T>>,
data_source: Arc<dyn DataSource<T>>,
cache_hit_count: Arc<RwLock<u64>>,
cache_miss_count: Arc<RwLock<u64>>,
}
/// Data source trait.
pub trait DataSource<T>: Send + Sync {
async fn get(&self, key: &str) -> Result<Option<T>, String>;
async fn set(&self, key: &str, value: &T) -> Result<(), String>;
async fn remove(&self, key: &str) -> Result<(), String>;
}
/// Database data source.
pub struct DatabaseDataSource {
connection_pool: Arc<ConnectionPool>,
}
impl DatabaseDataSource {
pub fn new(connection_pool: Arc<ConnectionPool>) -> Self {
Self { connection_pool }
}
}
impl<T> DataSource<T> for DatabaseDataSource
where
T: Clone + Send + Sync + 'static,
{
async fn get(&self, key: &str) -> Result<Option<T>, String> {
// In a real implementation, you would query the database
println!("Getting {} from database", key);
// Simulate database query
tokio::time::sleep(Duration::from_millis(100)).await;
// For this example, assume all keys exist
Ok(Some(format!("db_value_for_{}", key) as T))
}
async fn set(&self, key: &str, value: &T) -> Result<(), String> {
// In a real implementation, you would update the database
println!("Setting {} = {:?} in database", key, value);
// Simulate database update
tokio::time::sleep(Duration::from_millis(50)).await;
Ok(())
}
async fn remove(&self, key: &str) -> Result<(), String> {
// In a real implementation, you would delete from the database
println!("Removing {} from database", key);
// Simulate database deletion
tokio::time::sleep(Duration::from_millis(30)).await;
Ok(())
}
}
impl<T> CacheAsidePattern<T>
where
T: Clone + Send + Sync + 'static,
{
pub fn new(
cache: Arc<InMemoryCache<String, T>>,
data_source: Arc<dyn DataSource<T>>,
) -> Self {
Self {
cache,
data_source,
cache_hit_count: Arc::new(RwLock::new(0)),
cache_miss_count: Arc::new(RwLock::new(0)),
}
}
/// Get a value using cache-aside pattern.
pub async fn get(&self, key: &str) -> Result<Option<T>, String> {
// Try to get from cache first
if let Some(value) = self.cache.get(&key.to_string()).await {
// Cache hit
let mut hit_count = self.cache_hit_count.write().await;
*hit_count += 1;
return Ok(Some(value));
}
// Cache miss - get from data source
let mut miss_count = self.cache_miss_count.write().await;
*miss_count += 1;
drop(miss_count);
let value = self.data_source.get(key).await?;
// If value exists, cache it
if let Some(ref val) = value {
self.cache.set(key.to_string(), val.clone()).await?;
}
Ok(value)
}
/// Set a value using cache-aside pattern.
pub async fn set(&self, key: &str, value: &T) -> Result<(), String> {
// Update data source first
self.data_source.set(key, value).await?;
// Update cache
self.cache.set(key.to_string(), value.clone()).await?;
Ok(())
}
/// Remove a value using cache-aside pattern.
pub async fn remove(&self, key: &str) -> Result<(), String> {
// Remove from data source first
self.data_source.remove(key).await?;
// Remove from cache
self.cache.remove(&key.to_string()).await;
Ok(())
}
/// Invalidate cache entry.
pub async fn invalidate(&self, key: &str) -> Result<(), String> {
self.cache.remove(&key.to_string()).await;
Ok(())
}
/// Get cache statistics.
pub async fn get_cache_statistics(&self) -> CacheAsideStatistics {
let cache_stats = self.cache.get_statistics().await;
let hit_count = *self.cache_hit_count.read().await;
let miss_count = *self.cache_miss_count.read().await;
let total_requests = hit_count + miss_count;
let hit_rate = if total_requests > 0 {
hit_count as f64 / total_requests as f64
} else {
0.0
};
CacheAsideStatistics {
cache_hit_count: hit_count,
cache_miss_count: miss_count,
total_requests,
hit_rate,
cache_statistics: cache_stats,
}
}
}
#[derive(Debug, Clone)]
pub struct CacheAsideStatistics {
pub cache_hit_count: u64,
pub cache_miss_count: u64,
pub total_requests: u64,
pub hit_rate: f64,
pub cache_statistics: CacheStatistics,
}
/// Write-through cache pattern.
pub struct WriteThroughCache<T> {
cache: Arc<InMemoryCache<String, T>>,
data_source: Arc<dyn DataSource<T>>,
write_count: Arc<RwLock<u64>>,
read_count: Arc<RwLock<u64>>,
}
impl<T> WriteThroughCache<T>
where
T: Clone + Send + Sync + 'static,
{
pub fn new(
cache: Arc<InMemoryCache<String, T>>,
data_source: Arc<dyn DataSource<T>>,
) -> Self {
Self {
cache,
data_source,
write_count: Arc::new(RwLock::new(0)),
read_count: Arc::new(RwLock::new(0)),
}
}
/// Get a value from cache.
pub async fn get(&self, key: &str) -> Result<Option<T>, String> {
let mut read_count = self.read_count.write().await;
*read_count += 1;
drop(read_count);
self.cache.get(&key.to_string()).await
}
/// Set a value using write-through pattern.
pub async fn set(&self, key: &str, value: &T) -> Result<(), String> {
let mut write_count = self.write_count.write().await;
*write_count += 1;
drop(write_count);
// Write to data source first
self.data_source.set(key, value).await?;
// Write to cache
self.cache.set(key.to_string(), value.clone()).await?;
Ok(())
}
/// Remove a value using write-through pattern.
pub async fn remove(&self, key: &str) -> Result<(), String> {
// Remove from data source first
self.data_source.remove(key).await?;
// Remove from cache
self.cache.remove(&key.to_string()).await;
Ok(())
}
/// Get write-through cache statistics.
pub async fn get_statistics(&self) -> WriteThroughStatistics {
let cache_stats = self.cache.get_statistics().await;
let write_count = *self.write_count.read().await;
let read_count = *self.read_count.read().await;
WriteThroughStatistics {
write_count,
read_count,
cache_statistics: cache_stats,
}
}
}
#[derive(Debug, Clone)]
pub struct WriteThroughStatistics {
pub write_count: u64,
pub read_count: u64,
pub cache_statistics: CacheStatistics,
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_cache_aside_pattern() {
let cache = Arc::new(InMemoryCache::new(
100,
Duration::from_secs(60),
Duration::from_secs(10),
));
let data_source = Arc::new(DatabaseDataSource {
connection_pool: Arc::new(ConnectionPool::new(
10,
2,
Duration::from_secs(30),
Duration::from_secs(300),
Duration::from_secs(60),
)),
});
let cache_aside = CacheAsidePattern::new(cache, data_source);
// Test cache miss
let value = cache_aside.get("key1").await.unwrap();
assert!(value.is_some());
// Test cache hit
let value = cache_aside.get("key1").await.unwrap();
assert!(value.is_some());
let stats = cache_aside.get_cache_statistics().await;
assert_eq!(stats.cache_hit_count, 1);
assert_eq!(stats.cache_miss_count, 1);
}
#[tokio::test]
async fn test_write_through_cache() {
let cache = Arc::new(InMemoryCache::new(
100,
Duration::from_secs(60),
Duration::from_secs(10),
));
let data_source = Arc::new(DatabaseDataSource {
connection_pool: Arc::new(ConnectionPool::new(
10,
2,
Duration::from_secs(30),
Duration::from_secs(300),
Duration::from_secs(60),
)),
});
let write_through = WriteThroughCache::new(cache, data_source);
// Test write-through
write_through.set("key1", &"value1".to_string()).await.unwrap();
// Test read
let value = write_through.get("key1").await.unwrap();
assert_eq!(value, Some("value1".to_string()));
let stats = write_through.get_statistics().await;
assert_eq!(stats.write_count, 1);
assert_eq!(stats.read_count, 1);
}
}
TL;DR Runbook¶
Quick Start¶
// 1. In-memory caching
let cache = InMemoryCache::new(100, Duration::from_secs(60), Duration::from_secs(10));
cache.set("key".to_string(), "value".to_string()).await?;
let value = cache.get(&"key".to_string()).await;
// 2. Distributed caching
let distributed_cache = DistributedCache::new(2, ConsistencyLevel::Eventual, 100, Duration::from_secs(60));
distributed_cache.set("key".to_string(), "value".to_string(), Duration::from_secs(60)).await?;
// 3. Cache-aside pattern
let cache_aside = CacheAsidePattern::new(cache, data_source);
let value = cache_aside.get("key").await?;
Essential Patterns¶
// Complete caching setup
pub fn setup_rust_caching_strategies() {
// 1. In-memory caching
// 2. Distributed caching
// 3. Cache-aside pattern
// 4. Write-through caching
// 5. Write-behind caching
// 6. Cache invalidation
// 7. Performance monitoring
// 8. Consistency management
println!("Rust caching strategies setup complete!");
}
This guide provides the complete machinery for Rust caching strategies. Each pattern includes implementation examples, caching strategies, and real-world usage patterns for enterprise caching systems.