siderite_cache/
memory.rs

1//! Process-local LRU cache with per-entry TTL.
2
3type CacheResult<T> = Result<T, CacheError>;
4
5use crate::cache::ttl_millis;
6use crate::{Admission, AdmitMarker};
7use crate::{Cache, CacheError};
8use async_trait::async_trait;
9use lru::LruCache;
10use std::num::NonZeroUsize;
11use std::ops::{Deref, DerefMut};
12use std::sync::{Arc, Mutex, PoisonError};
13use std::time::{Duration, Instant};
14
15/// Hard byte-admission limits for process-local cache storage.
16///
17/// Key and value lengths count toward total bytes. Fixed entry/map overhead
18/// is additionally bounded by entry capacity; reads clone values and can
19/// temporarily allocate outside the stored-byte budget.
20#[derive(Debug, Clone, Copy)]
21pub struct MemoryCacheLimits {
22    /// Total retained key/value bytes (default: 64 MiB).
23    pub max_total_bytes: usize,
24    /// Maximum bytes in one value (default: 2 MiB).
25    pub max_value_bytes: usize,
26    /// Maximum bytes in one key (default: 4 KiB).
27    pub max_key_bytes: usize,
28}
29
30impl Default for MemoryCacheLimits {
31    fn default() -> Self {
32        Self {
33            max_total_bytes: 64 * 1024 * 1024,
34            max_value_bytes: 2 * 1024 * 1024,
35            max_key_bytes: 4096,
36        }
37    }
38}
39
40struct Store {
41    entries: LruCache<String, Entry>,
42    bytes: usize,
43    limits: MemoryCacheLimits,
44    /// Origin of the budget windows (this cache's own clock).
45    started: Instant,
46}
47
48impl Deref for Store {
49    type Target = LruCache<String, Entry>;
50    fn deref(&self) -> &Self::Target {
51        &self.entries
52    }
53}
54
55impl DerefMut for Store {
56    fn deref_mut(&mut self) -> &mut Self::Target {
57        &mut self.entries
58    }
59}
60
61impl Store {
62    /// The live value at `key`, refreshing its LRU position.
63    fn live(&mut self, key: &str, now: Instant) -> Option<&[u8]> {
64        if self.entries.peek(key).is_some_and(|e| e.is_expired(now)) {
65            self.pop(key);
66            return None;
67        }
68        self.entries.get(key).map(|entry| entry.value.as_slice())
69    }
70
71    fn pop(&mut self, key: &str) -> Option<Entry> {
72        let (key, entry) = self.entries.pop_entry(key)?;
73        self.bytes -= key.len() + entry.value.len();
74        Some(entry)
75    }
76
77    fn clear(&mut self) {
78        self.entries.clear();
79        self.bytes = 0;
80    }
81
82    fn put(&mut self, key: String, entry: Entry) -> Result<(), CacheError> {
83        let limits = self.limits;
84        let size = key
85            .len()
86            .checked_add(entry.value.len())
87            .ok_or(CacheError::SizeLimit)?;
88        if key.len() > limits.max_key_bytes
89            || entry.value.len() > limits.max_value_bytes
90            || size > limits.max_total_bytes
91        {
92            return Err(CacheError::SizeLimit);
93        }
94        self.pop(&key);
95        while self.bytes > limits.max_total_bytes - size
96            || self.entries.len() >= self.entries.cap().get()
97        {
98            let Some((key, entry)) = self.entries.pop_lru() else {
99                break;
100            };
101            self.bytes -= key.len() + entry.value.len();
102        }
103        self.bytes += size;
104        self.entries.put(key, entry);
105        Ok(())
106    }
107}
108
109struct Entry {
110    value: Vec<u8>,
111    expires_at: Option<Instant>,
112}
113
114impl Entry {
115    fn is_expired(&self, now: Instant) -> bool {
116        self.expires_at.is_some_and(|deadline| deadline <= now)
117    }
118}
119
120/// In-memory LRU cache with per-entry time-to-live.
121///
122/// Cloning shares the same map. A poisoned mutex is recovered (`into_inner`)
123/// so a panic in one caller does not disable the cache. `capacity` of `0` is
124/// treated as `1`. Retained key/value bytes are bounded by
125/// [`MemoryCacheLimits`]; writes beyond admission limits fail explicitly.
126/// Expiry is lazy on access; pressure evicts LRU entries without full scans.
127///
128/// [`set_if_absent_with_ttl`](Cache::set_if_absent_with_ttl) is atomic: the
129/// check, deadline and insertion happen under one lock acquisition. TTLs
130/// round up to whole milliseconds and zero is [`CacheError::InvalidTtl`].
131///
132/// [`increment`](Cache::increment) matches Redis `INCRBY`: a missing or
133/// expired key starts at `0`, the stored form is a decimal integer, and a
134/// non-integer value is [`CacheError::NotInteger`].
135#[derive(Clone)]
136pub struct MemoryCache {
137    inner: Arc<Mutex<Store>>,
138}
139
140impl std::fmt::Debug for MemoryCache {
141    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
142        let inner = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
143        f.debug_struct("MemoryCache")
144            .field("len", &inner.len())
145            .field("capacity", &inner.cap())
146            .field("stored_bytes", &inner.bytes)
147            .field("byte_limits", &inner.limits)
148            .finish()
149    }
150}
151
152impl MemoryCache {
153    /// Bound the cache to at most `capacity` live entries (minimum 1).
154    #[must_use]
155    pub fn new(capacity: usize) -> Self {
156        Self::build(capacity, MemoryCacheLimits::default())
157    }
158
159    /// Configure entry count and retained key/value byte budgets.
160    ///
161    /// Args:
162    ///     n: Maximum entry count, with zero treated as one.
163    ///     lim: Positive key/value/total byte limits.
164    ///
165    /// Returns:
166    ///     Cache with size-aware LRU eviction and hard write admission.
167    ///
168    /// # Errors
169    /// Invalid zero limits or key/value maxima above the total budget.
170    pub fn with_limits(n: usize, lim: MemoryCacheLimits) -> CacheResult<Self> {
171        if lim.max_total_bytes == 0
172            || lim.max_key_bytes == 0
173            || lim.max_value_bytes == 0
174            || lim.max_key_bytes > lim.max_total_bytes
175            || lim.max_value_bytes > lim.max_total_bytes
176        {
177            return Err(CacheError::InvalidLimits);
178        }
179        Ok(Self::build(n, lim))
180    }
181
182    fn build(capacity: usize, limits: MemoryCacheLimits) -> Self {
183        let size = NonZeroUsize::new(capacity.max(1));
184        let cap = size.unwrap_or(NonZeroUsize::MIN);
185        Self {
186            inner: Arc::new(Mutex::new(Store {
187                entries: LruCache::new(cap),
188                bytes: 0,
189                limits,
190                started: Instant::now(),
191            })),
192        }
193    }
194
195    /// Retained key and value bytes currently counted against
196    /// [`MemoryCacheLimits::max_total_bytes`].
197    ///
198    /// Expired entries still count until they are touched or evicted.
199    #[must_use]
200    pub fn stored_bytes(&self) -> usize {
201        self.lock().bytes
202    }
203
204    fn lock(&self) -> std::sync::MutexGuard<'_, Store> {
205        self.inner.lock().unwrap_or_else(PoisonError::into_inner)
206    }
207}
208
209fn expires_at(ttl: Option<Duration>) -> Result<Option<Instant>, CacheError> {
210    match ttl {
211        Some(ttl) => Instant::now()
212            .checked_add(ttl)
213            .map(Some)
214            .ok_or(CacheError::InvalidTtl),
215        None => Ok(None),
216    }
217}
218
219/// Redis `INCRBY` accepts an optional ASCII minus and one or more digits.
220fn parse_incr_int(bytes: &[u8]) -> Option<i64> {
221    let text = std::str::from_utf8(bytes).ok()?;
222    if text.is_empty() {
223        return None;
224    }
225    let digits = match text.as_bytes()[0] {
226        b'-' => &text.as_bytes()[1..],
227        b'0'..=b'9' => text.as_bytes(),
228        _ => return None,
229    };
230    if digits.is_empty() || !digits.iter().all(|b| b.is_ascii_digit()) {
231        return None;
232    }
233    text.parse().ok()
234}
235
236impl MemoryCache {
237    /// Increment under one lock; `expiry` applies only to a key that has
238    /// no deadline (new, or live without one).
239    fn incr_locked(&self, key: &str, by: i64, expiry: Option<Duration>) -> CacheResult<i64> {
240        let mut cache = self.lock();
241        let now = Instant::now();
242        if cache.peek(key).is_some_and(|entry| entry.is_expired(now)) {
243            cache.pop(key);
244        }
245        let invalid = || CacheError::NotInteger {
246            key: key.to_owned(),
247        };
248        let (next, deadline) = match cache.get(key) {
249            Some(entry) => {
250                let current = parse_incr_int(&entry.value).ok_or_else(invalid)?;
251                let next = current.checked_add(by).ok_or_else(invalid)?;
252                (next, entry.expires_at)
253            }
254            None => (by, None),
255        };
256        let deadline = match (deadline, expiry) {
257            (None, Some(ttl)) => Some(now.checked_add(ttl).ok_or(CacheError::InvalidTtl)?),
258            (existing, _) => existing,
259        };
260        cache.put(
261            key.to_owned(),
262            Entry {
263                value: next.to_string().into_bytes(),
264                expires_at: deadline,
265            },
266        )?;
267        Ok(next)
268    }
269}
270
271#[async_trait]
272impl Cache for MemoryCache {
273    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, CacheError> {
274        let mut cache = self.lock();
275        let now = Instant::now();
276        let expired = cache.peek(key).is_some_and(|e| e.is_expired(now));
277        if expired {
278            cache.pop(key);
279            return Ok(None);
280        }
281        Ok(cache.get(key).map(|entry| entry.value.clone()))
282    }
283
284    async fn set(
285        &self,
286        key: &str,
287        value: Vec<u8>,
288        ttl: Option<Duration>,
289    ) -> Result<(), CacheError> {
290        let mut cache = self.lock();
291        cache.put(
292            key.to_owned(),
293            Entry {
294                value,
295                expires_at: expires_at(ttl)?,
296            },
297        )
298    }
299
300    async fn set_if_absent(&self, key: &str, v: Vec<u8>) -> CacheResult<bool> {
301        let mut cache = self.lock();
302        let now = Instant::now();
303        if cache.peek(key).is_some_and(|entry| !entry.is_expired(now)) {
304            return Ok(false);
305        }
306        cache.put(
307            key.to_owned(),
308            Entry {
309                value: v,
310                expires_at: None,
311            },
312        )?;
313        Ok(true)
314    }
315
316    async fn set_if_absent_with_ttl(
317        &self,
318        key: &str,
319        value: Vec<u8>,
320        ttl: Duration,
321    ) -> CacheResult<bool> {
322        let ttl = Duration::from_millis(ttl_millis(ttl)?);
323        // One lock section: liveness check, deadline and insertion share a
324        // single clock reading, so no creator can observe a half-state.
325        let mut cache = self.lock();
326        let now = Instant::now();
327        if cache.peek(key).is_some_and(|entry| !entry.is_expired(now)) {
328            return Ok(false);
329        }
330        let deadline = now.checked_add(ttl).ok_or(CacheError::InvalidTtl)?;
331        cache.put(
332            key.to_owned(),
333            Entry {
334                value,
335                expires_at: Some(deadline),
336            },
337        )?;
338        Ok(true)
339    }
340
341    async fn delete(&self, key: &str) -> Result<bool, CacheError> {
342        let mut cache = self.lock();
343        let now = Instant::now();
344        match cache.peek(key) {
345            Some(entry) if entry.is_expired(now) => {
346                cache.pop(key);
347                Ok(false)
348            }
349            Some(_) => {
350                cache.pop(key);
351                Ok(true)
352            }
353            None => Ok(false),
354        }
355    }
356
357    async fn increment(&self, key: &str, by: i64) -> Result<i64, CacheError> {
358        self.incr_locked(key, by, None)
359    }
360
361    async fn increment_with_expiry(
362        &self,
363        key: &str,
364        by: i64,
365        ttl: Duration,
366    ) -> Result<i64, CacheError> {
367        let ttl = Duration::from_millis(ttl_millis(ttl)?);
368        self.incr_locked(key, by, Some(ttl))
369    }
370
371    async fn clear(&self) -> Result<(), CacheError> {
372        self.lock().clear();
373        Ok(())
374    }
375
376    async fn admit_marker(&self, req: &AdmitMarker<'_>) -> CacheResult<Admission> {
377        let window = Duration::from_millis(ttl_millis(req.window)?);
378        let counter_ttl = window.checked_mul(2).ok_or(CacheError::InvalidTtl)?;
379        let mut cache = self.lock();
380        let now = Instant::now();
381        let epoch = cache.live(req.epoch_key, now);
382        if epoch.is_none_or(|current| current != req.epoch) {
383            return Ok(Admission::Stale);
384        }
385        if let Some(live) = cache.live(req.marker, now) {
386            return Ok(Admission::Marker(live.to_vec()));
387        }
388        // The window comes from this cache's own clock (one process).
389        let ms = window.as_millis().max(1);
390        let elapsed = now.saturating_duration_since(cache.started).as_millis();
391        let counter = format!("{}{}", req.budget_prefix, elapsed / ms);
392        let used = match cache.live(&counter, now) {
393            Some(value) => parse_incr_int(value).ok_or_else(|| CacheError::NotInteger {
394                key: counter.clone(),
395            })?,
396            None => 0,
397        };
398        let used = u64::try_from(used).unwrap_or(u64::MAX);
399        if used >= req.max_new {
400            let left = u64::try_from(ms - elapsed % ms).unwrap_or(u64::MAX);
401            let resets_in = Duration::from_millis(left);
402            return Ok(Admission::Exhausted { resets_in });
403        }
404        let counted = Entry {
405            value: (used + 1).to_string().into_bytes(),
406            expires_at: Some(now.checked_add(counter_ttl).ok_or(CacheError::InvalidTtl)?),
407        };
408        // Keep the live counter's deadline: only a new counter gets one.
409        let counted = match cache.peek(&counter) {
410            Some(old) if !old.is_expired(now) => Entry {
411                expires_at: old.expires_at,
412                ..counted
413            },
414            _ => counted,
415        };
416        let deadline = now.checked_add(window).ok_or(CacheError::InvalidTtl)?;
417        let marker = Entry {
418            value: req.token.to_vec(),
419            expires_at: Some(deadline),
420        };
421        // Marker first: a refused marker must not consume budget.
422        cache.put(req.marker.to_owned(), marker)?;
423        if let Err(error) = cache.put(counter, counted) {
424            cache.pop(req.marker);
425            return Err(error);
426        }
427        Ok(Admission::Marker(req.token.to_vec()))
428    }
429
430    async fn invalidate_marker(
431        &self,
432        marker: &str,
433        epoch_key: &str,
434        epoch: &[u8],
435        ttl: Duration,
436    ) -> CacheResult<()> {
437        let ttl = Duration::from_millis(ttl_millis(ttl)?);
438        let mut cache = self.lock();
439        let deadline = Instant::now().checked_add(ttl);
440        let deadline = deadline.ok_or(CacheError::InvalidTtl)?;
441        let entry = Entry {
442            value: epoch.to_vec(),
443            expires_at: Some(deadline),
444        };
445        // Marker first: a refused epoch put cannot leave the marker.
446        cache.pop(marker);
447        cache.put(epoch_key.to_owned(), entry)?;
448        Ok(())
449    }
450}
451
452#[cfg(test)]
453mod tests {
454    use super::*;
455    use crate::CacheExt;
456    use std::thread;
457    use std::time::Duration;
458
459    #[tokio::test]
460    async fn get_set_delete_clear() {
461        let cache = MemoryCache::new(8);
462        assert_eq!(cache.get("k").await.unwrap(), None);
463        cache.set("k", b"v".to_vec(), None).await.unwrap();
464        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"v"[..]));
465        assert!(cache.delete("k").await.unwrap());
466        assert!(!cache.delete("k").await.unwrap());
467        cache.set("a", b"1".to_vec(), None).await.unwrap();
468        cache.set("b", b"2".to_vec(), None).await.unwrap();
469        cache.clear().await.unwrap();
470        assert_eq!(cache.get("a").await.unwrap(), None);
471        assert_eq!(cache.get("b").await.unwrap(), None);
472    }
473
474    #[tokio::test]
475    async fn increment_with_expiry_sets_a_deadline_once() {
476        let cache = MemoryCache::new(8);
477        let ttl = Duration::from_millis(60);
478        assert_eq!(cache.increment_with_expiry("c", 1, ttl).await.unwrap(), 1);
479        thread::sleep(Duration::from_millis(40));
480        // A later increment keeps the original deadline.
481        let long = Duration::from_secs(60);
482        assert_eq!(cache.increment_with_expiry("c", 2, long).await.unwrap(), 3);
483        thread::sleep(Duration::from_millis(40));
484        assert_eq!(cache.get("c").await.unwrap(), None);
485        // After expiry the counter restarts at zero with a fresh deadline.
486        assert_eq!(cache.increment_with_expiry("c", 1, long).await.unwrap(), 1);
487        // A key without a deadline receives one.
488        cache.set("p", b"5".to_vec(), None).await.unwrap();
489        assert_eq!(cache.increment_with_expiry("p", 1, ttl).await.unwrap(), 6);
490        thread::sleep(Duration::from_millis(80));
491        assert_eq!(cache.get("p").await.unwrap(), None);
492    }
493
494    #[tokio::test]
495    async fn increment_with_expiry_validates_ttl_and_value() {
496        let cache = MemoryCache::new(8);
497        let zero = cache.increment_with_expiry("c", 1, Duration::ZERO).await;
498        assert!(matches!(zero, Err(CacheError::InvalidTtl)));
499        assert_eq!(cache.get("c").await.unwrap(), None);
500        cache.set("s", b"x".to_vec(), None).await.unwrap();
501        let bad = cache.increment_with_expiry("s", 1, Duration::from_secs(1));
502        assert!(matches!(bad.await, Err(CacheError::NotInteger { .. })));
503    }
504
505    fn request<'a>(marker: &'a str, token: &'a [u8], epoch: &'a [u8]) -> AdmitMarker<'a> {
506        let window = Duration::from_millis(80);
507        AdmitMarker::new(marker, token, "epoch", epoch).budget("budget:", 2, window)
508    }
509
510    #[tokio::test]
511    async fn admit_marker_checks_epoch_counts_creations_and_rolls_windows() {
512        let cache = MemoryCache::new(64);
513        let epoch: &[u8] = &[7];
514        cache.set("epoch", epoch.to_vec(), None).await.unwrap();
515        let stale = cache.admit_marker(&request("a", &[1], &[8])).await;
516        assert_eq!(stale.unwrap(), Admission::Stale);
517        // A missing epoch counts as changed.
518        let gone = AdmitMarker::new("a", &[1], "elsewhere", epoch);
519        assert_eq!(cache.admit_marker(&gone).await.unwrap(), Admission::Stale);
520        assert_eq!(cache.get("a").await.unwrap(), None, "stale creates nothing");
521        let a = cache
522            .admit_marker(&request("a", &[1], epoch))
523            .await
524            .unwrap();
525        assert_eq!(a, Admission::Marker(vec![1]));
526        // An existing marker is returned and costs no budget.
527        let again = cache
528            .admit_marker(&request("a", &[9], epoch))
529            .await
530            .unwrap();
531        assert_eq!(again, Admission::Marker(vec![1]));
532        let b = cache
533            .admit_marker(&request("b", &[2], epoch))
534            .await
535            .unwrap();
536        assert_eq!(b, Admission::Marker(vec![2]));
537        let over = cache
538            .admit_marker(&request("c", &[3], epoch))
539            .await
540            .unwrap();
541        let Admission::Exhausted { resets_in } = over else {
542            panic!("expected exhaustion, got {over:?}");
543        };
544        assert!(resets_in <= Duration::from_millis(80));
545        assert_eq!(cache.get("c").await.unwrap(), None, "no marker over budget");
546        // The next window has a fresh budget.
547        thread::sleep(Duration::from_millis(90));
548        let c = cache
549            .admit_marker(&request("c", &[3], epoch))
550            .await
551            .unwrap();
552        assert_eq!(c, Admission::Marker(vec![3]));
553    }
554
555    #[tokio::test]
556    async fn admit_marker_is_exact_under_concurrency() {
557        let cache = MemoryCache::new(1024);
558        cache.set("epoch", vec![7], None).await.unwrap();
559        let window = Duration::from_secs(3600);
560        let mut tasks = Vec::new();
561        for n in 0..100u8 {
562            let cache = cache.clone();
563            tasks.push(tokio::spawn(async move {
564                let marker = format!("m{n}");
565                let token = [n];
566                let base = AdmitMarker::new(&marker, &token, "epoch", &[7]);
567                let request = base.budget("budget:", 10, window);
568                cache.admit_marker(&request).await.unwrap()
569            }));
570        }
571        let mut admitted = 0;
572        for task in tasks {
573            admitted += usize::from(matches!(task.await.unwrap(), Admission::Marker(_)));
574        }
575        assert_eq!(admitted, 10);
576    }
577
578    #[tokio::test]
579    async fn invalidate_marker_replaces_epoch_and_deletes_marker() {
580        let cache = MemoryCache::new(8);
581        cache.set("marker", vec![1], None).await.unwrap();
582        let ttl = Duration::from_millis(40);
583        cache
584            .invalidate_marker("marker", "epoch", &[5], ttl)
585            .await
586            .unwrap();
587        assert_eq!(cache.get("marker").await.unwrap(), None);
588        assert_eq!(cache.get("epoch").await.unwrap(), Some(vec![5]));
589        thread::sleep(Duration::from_millis(60));
590        assert_eq!(cache.get("epoch").await.unwrap(), None, "epoch expires");
591        let zero = cache
592            .invalidate_marker("m", "e", &[1], Duration::ZERO)
593            .await;
594        assert!(matches!(zero, Err(CacheError::InvalidTtl)));
595    }
596
597    #[tokio::test]
598    async fn stored_bytes_tracks_keys_and_values() {
599        let cache = MemoryCache::new(8);
600        assert_eq!(cache.stored_bytes(), 0);
601        cache.set("key", b"value".to_vec(), None).await.unwrap();
602        assert_eq!(cache.stored_bytes(), 8);
603        cache.delete("key").await.unwrap();
604        assert_eq!(cache.stored_bytes(), 0);
605    }
606
607    #[tokio::test]
608    async fn clones_share_storage() {
609        let cache = MemoryCache::new(4);
610        let clone = cache.clone();
611        clone.set("k", b"v".to_vec(), None).await.unwrap();
612        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"v"[..]));
613    }
614
615    #[tokio::test]
616    async fn expired_entries_are_missing() {
617        let cache = MemoryCache::new(4);
618        cache
619            .set("k", b"v".to_vec(), Some(Duration::from_millis(15)))
620            .await
621            .unwrap();
622        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"v"[..]));
623        tokio::time::sleep(Duration::from_millis(40)).await;
624        assert_eq!(cache.get("k").await.unwrap(), None);
625        assert!(!cache.delete("k").await.unwrap());
626    }
627
628    #[tokio::test]
629    async fn lru_evicts_the_least_recently_used() {
630        let cache = MemoryCache::new(2);
631        cache.set("a", b"1".to_vec(), None).await.unwrap();
632        cache.set("b", b"2".to_vec(), None).await.unwrap();
633        assert_eq!(cache.get("a").await.unwrap().as_deref(), Some(&b"1"[..]));
634        cache.set("c", b"3".to_vec(), None).await.unwrap();
635        assert_eq!(cache.get("b").await.unwrap(), None);
636        assert_eq!(cache.get("a").await.unwrap().as_deref(), Some(&b"1"[..]));
637        assert_eq!(cache.get("c").await.unwrap().as_deref(), Some(&b"3"[..]));
638    }
639
640    #[tokio::test]
641    async fn zero_capacity_is_one() {
642        let cache = MemoryCache::new(0);
643        cache.set("a", b"1".to_vec(), None).await.unwrap();
644        cache.set("b", b"2".to_vec(), None).await.unwrap();
645        assert_eq!(cache.get("a").await.unwrap(), None);
646        assert_eq!(cache.get("b").await.unwrap().as_deref(), Some(&b"2"[..]));
647    }
648
649    #[tokio::test]
650    async fn increment_matches_redis_incrby() {
651        let cache = MemoryCache::new(8);
652        assert_eq!(cache.increment("n", 3).await.unwrap(), 3);
653        assert_eq!(cache.increment("n", 2).await.unwrap(), 5);
654        assert_eq!(cache.increment("n", -8).await.unwrap(), -3);
655        assert_eq!(cache.get("n").await.unwrap().as_deref(), Some(&b"-3"[..]));
656        cache.set("n", b"9".to_vec(), None).await.unwrap();
657        assert_eq!(cache.increment("n", 1).await.unwrap(), 10);
658    }
659
660    #[tokio::test]
661    async fn increment_rejects_non_integers_and_overflow() {
662        let cache = MemoryCache::new(8);
663        cache.set("s", b"hello".to_vec(), None).await.unwrap();
664        assert!(matches!(
665            cache.increment("s", 1).await.unwrap_err(),
666            CacheError::NotInteger { ref key } if key == "s"
667        ));
668        cache.set("plus", b"+1".to_vec(), None).await.unwrap();
669        assert!(matches!(
670            cache.increment("plus", 1).await.unwrap_err(),
671            CacheError::NotInteger { .. }
672        ));
673        cache.set("ws", b" 1".to_vec(), None).await.unwrap();
674        assert!(matches!(
675            cache.increment("ws", 1).await.unwrap_err(),
676            CacheError::NotInteger { .. }
677        ));
678        cache
679            .set("max", i64::MAX.to_string().into_bytes(), None)
680            .await
681            .unwrap();
682        assert!(matches!(
683            cache.increment("max", 1).await.unwrap_err(),
684            CacheError::NotInteger { ref key } if key == "max"
685        ));
686    }
687
688    #[tokio::test]
689    async fn increment_on_expired_key_starts_at_zero() {
690        let cache = MemoryCache::new(4);
691        cache
692            .set("n", b"9".to_vec(), Some(Duration::from_millis(15)))
693            .await
694            .unwrap();
695        tokio::time::sleep(Duration::from_millis(40)).await;
696        assert_eq!(cache.increment("n", 4).await.unwrap(), 4);
697    }
698
699    #[tokio::test]
700    async fn increment_preserves_ttl_of_a_live_key() {
701        let cache = MemoryCache::new(4);
702        cache
703            .set("n", b"1".to_vec(), Some(Duration::from_millis(80)))
704            .await
705            .unwrap();
706        assert_eq!(cache.increment("n", 1).await.unwrap(), 2);
707        tokio::time::sleep(Duration::from_millis(120)).await;
708        assert_eq!(cache.get("n").await.unwrap(), None);
709    }
710
711    #[tokio::test]
712    async fn concurrent_increments_are_serialized() {
713        let cache = MemoryCache::new(4);
714        let mut joins = Vec::new();
715        for _ in 0..32 {
716            let cache = cache.clone();
717            let worker = async move { cache.increment("n", 1).await };
718            joins.push(tokio::spawn(worker));
719        }
720        let mut total = 0;
721        for join in joins {
722            total += join.await.unwrap().unwrap();
723        }
724        assert_eq!(cache.get("n").await.unwrap().as_deref(), Some(&b"32"[..]));
725        assert_eq!(total, (1..=32).sum::<i64>());
726    }
727
728    #[tokio::test]
729    async fn poisoned_mutex_is_recovered() {
730        let cache = MemoryCache::new(4);
731        cache.set("k", b"v".to_vec(), None).await.unwrap();
732        let poison = cache.clone();
733        let _ = thread::spawn(move || {
734            let _guard = poison.lock();
735            panic!("poison");
736        })
737        .join();
738        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"v"[..]));
739        cache.set("k", b"w".to_vec(), None).await.unwrap();
740        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"w"[..]));
741    }
742
743    #[tokio::test]
744    async fn binary_values_roundtrip() {
745        let cache = MemoryCache::new(2);
746        let value = vec![0, 127, 128, 255];
747        cache.set("bin", value.clone(), None).await.unwrap();
748        assert_eq!(cache.get("bin").await.unwrap(), Some(value));
749    }
750
751    #[tokio::test]
752    async fn set_replaces_ttl() {
753        let cache = MemoryCache::new(2);
754        cache
755            .set("k", b"1".to_vec(), Some(Duration::from_millis(15)))
756            .await
757            .unwrap();
758        cache.set("k", b"2".to_vec(), None).await.unwrap();
759        tokio::time::sleep(Duration::from_millis(40)).await;
760        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"2"[..]));
761    }
762
763    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
764    async fn expiring_creation_has_one_winner() {
765        let cache = MemoryCache::new(8);
766        let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(64));
767        let mut joins = Vec::new();
768        for v in 0..64u8 {
769            let cache = cache.clone();
770            let barrier = std::sync::Arc::clone(&barrier);
771            joins.push(tokio::spawn(async move {
772                let ttl = Duration::from_secs(30);
773                barrier.wait().await;
774                cache.set_if_absent_with_ttl("k", vec![v], ttl).await
775            }));
776        }
777        let mut winners = Vec::new();
778        for (v, join) in joins.into_iter().enumerate() {
779            if join.await.unwrap().unwrap() {
780                winners.push(v as u8);
781            }
782        }
783        assert_eq!(winners.len(), 1);
784        assert_eq!(cache.get("k").await.unwrap(), Some(winners));
785    }
786
787    #[tokio::test]
788    async fn expiring_creation_rejects_zero_before_touching_state() {
789        let cache = MemoryCache::new(4);
790        cache.set("k", b"live".to_vec(), None).await.unwrap();
791        let err = cache.set_if_absent_with_ttl("k", vec![1], Duration::ZERO);
792        assert!(matches!(err.await, Err(CacheError::InvalidTtl)));
793        let err = cache.set_if_absent_with_ttl("new", vec![1], Duration::ZERO);
794        assert!(matches!(err.await, Err(CacheError::InvalidTtl)));
795        assert_eq!(cache.get("k").await.unwrap().as_deref(), Some(&b"live"[..]));
796        assert_eq!(cache.get("new").await.unwrap(), None);
797        let huge = cache.set_if_absent_with_ttl("new", vec![1], Duration::MAX);
798        assert!(matches!(huge.await, Err(CacheError::InvalidTtl)));
799    }
800
801    #[tokio::test]
802    async fn expiring_creation_rounds_subunit_ttl_up() {
803        let cache = MemoryCache::new(4);
804        let before = Instant::now();
805        let tiny = Duration::from_nanos(1);
806        assert!(
807            cache
808                .set_if_absent_with_ttl("k", vec![1], tiny)
809                .await
810                .unwrap()
811        );
812        let deadline = cache.lock().peek("k").and_then(|e| e.expires_at);
813        let deadline = deadline.expect("entry has a deadline");
814        assert!(deadline >= before + Duration::from_millis(1));
815        let odd = Duration::from_micros(1500);
816        assert!(
817            cache
818                .set_if_absent_with_ttl("j", vec![1], odd)
819                .await
820                .unwrap()
821        );
822        let before = Instant::now();
823        let deadline = cache.lock().peek("j").and_then(|e| e.expires_at);
824        assert!(deadline.expect("deadline") <= before + Duration::from_millis(2));
825    }
826
827    #[tokio::test]
828    async fn expiring_creation_replaces_expired_and_expires() {
829        let cache = MemoryCache::new(4);
830        let ms = Duration::from_millis(5);
831        assert!(
832            cache
833                .set_if_absent_with_ttl("k", vec![1], ms)
834                .await
835                .unwrap()
836        );
837        assert!(
838            !cache
839                .set_if_absent_with_ttl("k", vec![2], ms)
840                .await
841                .unwrap()
842        );
843        tokio::time::sleep(Duration::from_millis(30)).await;
844        // Expired entries are absent for creation even before a get evicts.
845        let long = Duration::from_secs(30);
846        assert!(
847            cache
848                .set_if_absent_with_ttl("k", vec![3], long)
849                .await
850                .unwrap()
851        );
852        assert_eq!(cache.get("k").await.unwrap(), Some(vec![3]));
853        // The loser never extends the winner's deadline.
854        let one = Duration::from_millis(5);
855        assert!(
856            !cache
857                .set_if_absent_with_ttl("k", vec![4], one)
858                .await
859                .unwrap()
860        );
861        tokio::time::sleep(Duration::from_millis(30)).await;
862        assert_eq!(cache.get("k").await.unwrap(), Some(vec![3]));
863    }
864
865    #[tokio::test]
866    async fn expiring_creation_obeys_byte_admission() {
867        let lim = MemoryCacheLimits {
868            max_total_bytes: 8,
869            max_value_bytes: 4,
870            max_key_bytes: 4,
871        };
872        let cache = MemoryCache::with_limits(4, lim).unwrap();
873        let ttl = Duration::from_secs(1);
874        let err = cache.set_if_absent_with_ttl("k", vec![0; 5], ttl).await;
875        assert!(matches!(err, Err(CacheError::SizeLimit)));
876        assert_eq!(cache.get("k").await.unwrap(), None);
877    }
878
879    #[test]
880    fn parse_incr_int_is_strict() {
881        assert_eq!(parse_incr_int(b"0"), Some(0));
882        assert_eq!(parse_incr_int(b"01"), Some(1));
883        assert_eq!(parse_incr_int(b"-12"), Some(-12));
884        assert_eq!(parse_incr_int(b""), None);
885        assert_eq!(parse_incr_int(b"-"), None);
886        assert_eq!(parse_incr_int(b"+1"), None);
887        assert_eq!(parse_incr_int(b"1 "), None);
888        assert_eq!(parse_incr_int(b"1.0"), None);
889        assert_eq!(parse_incr_int(b"abc"), None);
890    }
891
892    #[test]
893    fn types_are_send_sync() {
894        fn assert_send_sync<T: Send + Sync>() {}
895        assert_send_sync::<MemoryCache>();
896        assert_send_sync::<Box<dyn Cache>>();
897    }
898
899    #[tokio::test]
900    async fn json_helper_uses_compact_encoding() {
901        let cache = MemoryCache::new(2);
902        cache.set_json("n", &42u32, None).await.unwrap();
903        assert_eq!(cache.get("n").await.unwrap().as_deref(), Some(&b"42"[..]));
904    }
905}
906
907#[cfg(test)]
908#[path = "memory_limits_tests.rs"]
909mod limits_tests;
siderite_cache/memory.rs

siderite_cache/
memory.rs
