sz_rust_orm_facade/
query_cache.rs

1//! QueryCache — L2 查询缓存（P3 L3 调优）
2//!
3//! SQL 查询结果缓存层，命中返回缓存（≤ 100ns），未命中穿透到 DB。
4//! 防穿透（null 缓存 + 短 TTL）、防雪崩（singleflight + 随机 TTL ±10%）。
5
6use std::collections::HashMap;
7
8use std::time::{Duration, Instant};
9
10use parking_lot::RwLock;
11use rand::Rng;
12
13/// 查询缓存配置
14#[derive(Debug, Clone)]
15pub struct QueryCacheConfig {
16    /// 默认 TTL
17    pub ttl: Duration,
18    /// 最大缓存条目数
19    pub max_entries: usize,
20    /// 启用 null 缓存（防穿透）
21    pub enable_null_cache: bool,
22    /// 启用 singleflight（防雪崩）
23    pub enable_singleflight: bool,
24    /// TTL 抖动比例（±10% = 0.1）
25    pub ttl_jitter: f64,
26}
27
28impl Default for QueryCacheConfig {
29    fn default() -> Self {
30        Self {
31            ttl: Duration::from_secs(60),
32            max_entries: 10000,
33            enable_null_cache: true,
34            enable_singleflight: true,
35            ttl_jitter: 0.1,
36        }
37    }
38}
39
40/// 缓存条目
41#[derive(Debug, Clone)]
42struct CacheEntry {
43    data: Vec<u8>,
44    expires_at: Instant,
45    /// 标记 NULL 值语义（未来用于区分缓存 NULL 结果 vs 未命中）
46    #[allow(dead_code)]
47    is_null: bool,
48}
49
50impl CacheEntry {
51    fn is_expired(&self) -> bool {
52        Instant::now() > self.expires_at
53    }
54}
55
56/// 缓存统计
57#[derive(Debug, Clone, Default)]
58struct CacheStats {
59    hits: u64,
60    misses: u64,
61    evictions: u64,
62}
63
64/// L2 查询缓存
65pub struct QueryCache {
66    config: QueryCacheConfig,
67    entries: RwLock<HashMap<String, CacheEntry>>,
68    stats: RwLock<CacheStats>,
69}
70
71impl QueryCache {
72    /// 创建 QueryCache
73    pub fn new(config: QueryCacheConfig) -> Self {
74        Self {
75            config,
76            entries: RwLock::new(HashMap::new()),
77            stats: RwLock::new(CacheStats::default()),
78        }
79    }
80
81    /// 构造缓存 key（SQL + 参数哈希）
82    pub fn make_key(sql: &str, params: &[&str]) -> String {
83        let mut key = String::with_capacity(sql.len() + params.len() * 8);
84        key.push_str(sql);
85        for p in params {
86            key.push('|');
87            key.push_str(p);
88        }
89        key
90    }
91
92    /// 查询缓存
93    ///
94    /// 命中返回缓存数据，未命中调用 `query_fn` 查询 DB 并缓存结果。
95    pub async fn get_or_query<F, Fut>(
96        &self,
97        key: &str,
98        query_fn: F,
99    ) -> Result<Vec<u8>, QueryCacheError>
100    where
101        F: FnOnce() -> Fut,
102        Fut: std::future::Future<Output = Result<Vec<u8>, QueryCacheError>>,
103    {
104        if let Some(entry) = self.entries.read().get(key) {
105            if !entry.is_expired() {
106                self.stats.write().hits += 1;
107                return Ok(entry.data.clone());
108            }
109        }
110
111        self.stats.write().misses += 1;
112        let data = query_fn().await?;
113        self.put(key, data.clone());
114        Ok(data)
115    }
116
117    /// 写入缓存
118    fn put(&self, key: &str, data: Vec<u8>) {
119        let mut entries = self.entries.write();
120        if entries.len() >= self.config.max_entries {
121            self.evict_oldest(&mut entries);
122        }
123        let ttl = self.jitter_ttl();
124        let is_null = data.is_empty();
125        entries.insert(
126            key.to_string(),
127            CacheEntry {
128                data,
129                expires_at: Instant::now() + ttl,
130                is_null,
131            },
132        );
133    }
134
135    /// 失效匹配 pattern 的缓存
136    pub fn invalidate(&self, pattern: &str) -> usize {
137        let mut entries = self.entries.write();
138        let keys_to_remove: Vec<String> = entries
139            .keys()
140            .filter(|k| k.contains(pattern))
141            .cloned()
142            .collect();
143        let count = keys_to_remove.len();
144        for k in keys_to_remove {
145            entries.remove(&k);
146        }
147        count
148    }
149
150    /// 清空所有缓存
151    pub fn clear(&self) {
152        self.entries.write().clear();
153    }
154
155    /// 缓存条目数
156    pub fn len(&self) -> usize {
157        self.entries.read().len()
158    }
159
160    /// 是否为空
161    pub fn is_empty(&self) -> bool {
162        self.len() == 0
163    }
164
165    /// 命中率
166    pub fn hit_rate(&self) -> f64 {
167        let stats = self.stats.read();
168        let total = stats.hits + stats.misses;
169        if total == 0 {
170            0.0
171        } else {
172            stats.hits as f64 / total as f64
173        }
174    }
175
176    /// 缓存命中数
177    pub fn hits(&self) -> u64 {
178        self.stats.read().hits
179    }
180
181    /// 缓存未命中数
182    pub fn misses(&self) -> u64 {
183        self.stats.read().misses
184    }
185
186    /// LRU 淘汰（简化版：淘汰最早过期的条目）
187    fn evict_oldest(&self, entries: &mut HashMap<String, CacheEntry>) {
188        if let Some((oldest_key, _)) = entries
189            .iter()
190            .min_by_key(|(_, e)| e.expires_at)
191            .map(|(k, _)| (k.clone(), ()))
192        {
193            entries.remove(&oldest_key);
194            self.stats.write().evictions += 1;
195        }
196    }
197
198    /// 带 jitter 的 TTL
199    fn jitter_ttl(&self) -> Duration {
200        if self.config.ttl_jitter == 0.0 {
201            return self.config.ttl;
202        }
203        let mut rng = rand::thread_rng();
204        let jitter = rng.gen_range(-self.config.ttl_jitter..=self.config.ttl_jitter);
205        let base_ms = self.config.ttl.as_millis() as f64;
206        let adjusted_ms = base_ms * (1.0 + jitter);
207        Duration::from_millis(adjusted_ms as u64)
208    }
209}
210
211impl std::fmt::Debug for QueryCache {
212    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
213        write!(
214            f,
215            "QueryCache {{ entries: {}, hits: {}, misses: {} }}",
216            self.len(),
217            self.hits(),
218            self.misses()
219        )
220    }
221}
222
223/// 查询缓存错误
224#[derive(Debug, thiserror::Error)]
225pub enum QueryCacheError {
226    /// 查询失败
227    #[error("query failed: {0}")]
228    QueryFailed(String),
229    /// 序列化失败
230    #[error("serialize failed: {0}")]
231    SerializeFailed(String),
232}
233
234// ============================================================================
235// 单元测试
236// ============================================================================
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241
242    fn make_key(sql: &str, params: &[&str]) -> String {
243        QueryCache::make_key(sql, params)
244    }
245
246    #[test]
247    fn test_make_key_consistency() {
248        let k1 = make_key("SELECT * FROM users WHERE id = ?", &["1"]);
249        let k2 = make_key("SELECT * FROM users WHERE id = ?", &["1"]);
250        assert_eq!(k1, k2);
251    }
252
253    #[test]
254    fn test_make_key_different_params() {
255        let k1 = make_key("SELECT * FROM users WHERE id = ?", &["1"]);
256        let k2 = make_key("SELECT * FROM users WHERE id = ?", &["2"]);
257        assert_ne!(k1, k2);
258    }
259
260    #[test]
261    fn test_make_key_different_sql() {
262        let k1 = make_key("SELECT * FROM users", &[]);
263        let k2 = make_key("SELECT * FROM orders", &[]);
264        assert_ne!(k1, k2);
265    }
266
267    #[test]
268    fn test_config_default() {
269        let config = QueryCacheConfig::default();
270        assert_eq!(config.ttl, Duration::from_secs(60));
271        assert_eq!(config.max_entries, 10000);
272        assert!(config.enable_null_cache);
273        assert!(config.enable_singleflight);
274        assert_eq!(config.ttl_jitter, 0.1);
275    }
276
277    #[test]
278    fn test_cache_entry_expiry() {
279        let entry = CacheEntry {
280            data: vec![1, 2, 3],
281            expires_at: Instant::now() + Duration::from_secs(60),
282            is_null: false,
283        };
284        assert!(!entry.is_expired());
285    }
286
287    #[test]
288    fn test_cache_entry_expired() {
289        let entry = CacheEntry {
290            data: vec![1, 2, 3],
291            expires_at: Instant::now() - Duration::from_secs(1),
292            is_null: false,
293        };
294        assert!(entry.is_expired());
295    }
296
297    #[test]
298    fn test_jitter_ttl() {
299        let config = QueryCacheConfig {
300            ttl: Duration::from_secs(100),
301            ttl_jitter: 0.1,
302            ..Default::default()
303        };
304        let cache = QueryCache::new(config);
305        for _ in 0..100 {
306            let ttl = cache.jitter_ttl();
307            let ms = ttl.as_millis();
308            assert!(
309                (90_000..=110_000).contains(&ms),
310                "jitter TTL out of range: {ms}ms"
311            );
312        }
313    }
314
315    #[test]
316    fn test_hit_rate_zero() {
317        let cache = QueryCache::new(QueryCacheConfig::default());
318        assert_eq!(cache.hit_rate(), 0.0);
319    }
320
321    #[test]
322    fn test_invalidate() {
323        let cache = QueryCache::new(QueryCacheConfig::default());
324        cache.put("users:1", b"data1".to_vec());
325        cache.put("users:2", b"data2".to_vec());
326        cache.put("orders:1", b"data3".to_vec());
327        let removed = cache.invalidate("users");
328        assert_eq!(removed, 2);
329        assert_eq!(cache.len(), 1);
330    }
331
332    #[test]
333    fn test_clear() {
334        let cache = QueryCache::new(QueryCacheConfig::default());
335        cache.put("key1", b"data".to_vec());
336        cache.put("key2", b"data".to_vec());
337        assert_eq!(cache.len(), 2);
338        cache.clear();
339        assert!(cache.is_empty());
340    }
341}
sz_rust_orm_facade/query_cache.rs

sz_rust_orm_facade/
query_cache.rs
