pingora_cache/
memory.rs

1// Copyright 2026 Cloudflare, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Hash map based in memory cache
16//!
17//! For testing only, not for production use
18
19//TODO: Mark this module #[test] only
20
21use super::*;
22use crate::eviction::CacheEntryKey;
23use crate::storage::{
24    streaming_write::U64WriteId, HandleHit, HandleMiss, PurgeOutcome, PurgeTarget,
25};
26use crate::trace::SpanHandle;
27
28use async_trait::async_trait;
29use bytes::Bytes;
30use parking_lot::RwLock;
31use pingora_error::*;
32use std::any::Any;
33use std::collections::HashMap;
34use std::sync::atomic::{AtomicU64, Ordering};
35use std::sync::Arc;
36use tokio::sync::watch;
37
38type BinaryMeta = (Vec<u8>, Vec<u8>);
39
40pub(crate) struct CacheObject {
41    pub meta: BinaryMeta,
42    pub body: Arc<Vec<u8>>,
43}
44
45pub(crate) struct TempObject {
46    pub meta: BinaryMeta,
47    // these are Arc because they need to continue to exist after this TempObject is removed
48    pub body: Arc<RwLock<Vec<u8>>>,
49    bytes_written: Arc<watch::Sender<PartialState>>, // this should match body.len()
50}
51
52impl TempObject {
53    fn new(meta: BinaryMeta) -> Self {
54        let (tx, _rx) = watch::channel(PartialState::Partial(0));
55        TempObject {
56            meta,
57            body: Arc::new(RwLock::new(Vec::new())),
58            bytes_written: Arc::new(tx),
59        }
60    }
61    // this is not at all optimized
62    fn make_cache_object(&self) -> CacheObject {
63        let meta = self.meta.clone();
64        let body = Arc::new(self.body.read().clone());
65        CacheObject { meta, body }
66    }
67}
68
69/// Hash map based in memory cache
70///
71/// For testing only, not for production use.
72pub struct MemCache {
73    pub(crate) cached: Arc<RwLock<HashMap<String, CacheObject>>>,
74    pub(crate) temp: Arc<RwLock<HashMap<String, HashMap<u64, TempObject>>>>,
75    pub(crate) last_temp_id: AtomicU64,
76}
77
78impl MemCache {
79    /// Create a new [MemCache]
80    pub fn new() -> Self {
81        MemCache {
82            cached: Arc::new(RwLock::new(HashMap::new())),
83            temp: Arc::new(RwLock::new(HashMap::new())),
84            last_temp_id: AtomicU64::new(0),
85        }
86    }
87}
88
89pub enum MemHitHandler {
90    Complete(CompleteHit),
91    Partial(PartialHit),
92}
93
94#[derive(Copy, Clone)]
95enum PartialState {
96    Partial(usize),
97    Complete(usize),
98}
99
100pub struct CompleteHit {
101    body: Arc<Vec<u8>>,
102    done: bool,
103    range_start: usize,
104    range_end: usize,
105}
106
107impl CompleteHit {
108    fn get(&mut self) -> Option<Bytes> {
109        if self.done {
110            None
111        } else {
112            self.done = true;
113            Some(Bytes::copy_from_slice(
114                &self.body.as_slice()[self.range_start..self.range_end],
115            ))
116        }
117    }
118
119    fn seek(&mut self, start: usize, end: Option<usize>) -> Result<()> {
120        if start >= self.body.len() {
121            return Error::e_explain(
122                ErrorType::InternalError,
123                format!("seek start out of range {start} >= {}", self.body.len()),
124            );
125        }
126        self.range_start = start;
127        if let Some(end) = end {
128            // end over the actual last byte is allowed, we just need to return the actual bytes
129            self.range_end = std::cmp::min(self.body.len(), end);
130        }
131        // seek resets read so that one handler can be used for multiple ranges
132        self.done = false;
133        Ok(())
134    }
135}
136
137pub struct PartialHit {
138    body: Arc<RwLock<Vec<u8>>>,
139    bytes_written: watch::Receiver<PartialState>,
140    bytes_read: usize,
141}
142
143impl PartialHit {
144    async fn read(&mut self) -> Option<Bytes> {
145        loop {
146            let bytes_written = *self.bytes_written.borrow_and_update();
147            let bytes_end = match bytes_written {
148                PartialState::Partial(s) => s,
149                PartialState::Complete(c) => {
150                    // no more data will arrive
151                    if c == self.bytes_read {
152                        return None;
153                    }
154                    c
155                }
156            };
157            assert!(bytes_end >= self.bytes_read);
158
159            // more data available to read
160            if bytes_end > self.bytes_read {
161                let new_bytes =
162                    Bytes::copy_from_slice(&self.body.read()[self.bytes_read..bytes_end]);
163                self.bytes_read = bytes_end;
164                return Some(new_bytes);
165            }
166
167            // wait for more data
168            if self.bytes_written.changed().await.is_err() {
169                // err: sender dropped, body is finished
170                // FIXME: sender could drop because of an error
171                return None;
172            }
173        }
174    }
175}
176
177#[async_trait]
178impl HandleHit for MemHitHandler {
179    async fn read_body(&mut self) -> Result<Option<Bytes>> {
180        match self {
181            Self::Complete(c) => Ok(c.get()),
182            Self::Partial(p) => Ok(p.read().await),
183        }
184    }
185    async fn finish(
186        self: Box<Self>, // because self is always used as a trait object
187        _storage: &'static (dyn storage::Storage + Sync),
188        _key: &CacheKey,
189        _trace: &SpanHandle,
190    ) -> Result<()> {
191        Ok(())
192    }
193
194    fn can_seek(&self) -> bool {
195        match self {
196            Self::Complete(_) => true,
197            Self::Partial(_) => false, // TODO: support seeking in partial reads
198        }
199    }
200
201    fn seek(&mut self, start: usize, end: Option<usize>) -> Result<()> {
202        match self {
203            Self::Complete(c) => c.seek(start, end),
204            Self::Partial(_) => Error::e_explain(
205                ErrorType::InternalError,
206                "seek not supported for partial cache",
207            ),
208        }
209    }
210
211    fn should_count_access(&self) -> bool {
212        match self {
213            // avoid counting accesses for partial reads to keep things simple
214            Self::Complete(_) => true,
215            Self::Partial(_) => false,
216        }
217    }
218
219    fn get_eviction_weight(&self) -> usize {
220        match self {
221            // FIXME: just body size, also track meta size
222            Self::Complete(c) => c.body.len(),
223            // partial read cannot be estimated since body size is unknown
224            Self::Partial(_) => 0,
225        }
226    }
227
228    fn as_any(&self) -> &(dyn Any + Send + Sync) {
229        self
230    }
231
232    fn as_any_mut(&mut self) -> &mut (dyn Any + Send + Sync) {
233        self
234    }
235}
236
237pub struct MemMissHandler {
238    body: Arc<RwLock<Vec<u8>>>,
239    bytes_written: Arc<watch::Sender<PartialState>>,
240    // these are used only in finish() to data from temp to cache
241    key: String,
242    temp_id: U64WriteId,
243    // key -> cache object
244    cache: Arc<RwLock<HashMap<String, CacheObject>>>,
245    // key -> (temp writer id -> temp object) to support concurrent writers
246    temp: Arc<RwLock<HashMap<String, HashMap<u64, TempObject>>>>,
247}
248
249#[async_trait]
250impl HandleMiss for MemMissHandler {
251    async fn write_body(&mut self, data: bytes::Bytes, eof: bool) -> Result<()> {
252        let current_bytes = match *self.bytes_written.borrow() {
253            PartialState::Partial(p) => p,
254            PartialState::Complete(_) => panic!("already EOF"),
255        };
256        self.body.write().extend_from_slice(&data);
257        let written = current_bytes + data.len();
258        let new_state = if eof {
259            PartialState::Complete(written)
260        } else {
261            PartialState::Partial(written)
262        };
263        self.bytes_written.send_replace(new_state);
264        Ok(())
265    }
266
267    async fn finish(self: Box<Self>) -> Result<MissFinishType> {
268        // safe, the temp object is inserted when the miss handler is created
269        let cache_object = self
270            .temp
271            .read()
272            .get(&self.key)
273            .unwrap()
274            .get(&self.temp_id.into())
275            .unwrap()
276            .make_cache_object();
277        let size = cache_object.body.len(); // FIXME: this just body size, also track meta size
278        self.cache.write().insert(self.key.clone(), cache_object);
279        self.temp
280            .write()
281            .get_mut(&self.key)
282            .and_then(|map| map.remove(&self.temp_id.into()));
283        Ok(MissFinishType::Created(size))
284    }
285
286    fn streaming_write_tag(&self) -> Option<&[u8]> {
287        Some(self.temp_id.as_bytes())
288    }
289}
290
291impl Drop for MemMissHandler {
292    fn drop(&mut self) {
293        self.temp
294            .write()
295            .get_mut(&self.key)
296            .and_then(|map| map.remove(&self.temp_id.into()));
297    }
298}
299
300fn hit_from_temp_obj(temp_obj: &TempObject) -> Result<Option<(CacheMeta, HitHandler)>> {
301    let meta = CacheMeta::deserialize(&temp_obj.meta.0, &temp_obj.meta.1)?;
302    let partial = PartialHit {
303        body: temp_obj.body.clone(),
304        bytes_written: temp_obj.bytes_written.subscribe(),
305        bytes_read: 0,
306    };
307    let hit_handler = MemHitHandler::Partial(partial);
308    Ok(Some((meta, Box::new(hit_handler))))
309}
310
311#[async_trait]
312impl Storage for MemCache {
313    async fn lookup(
314        &'static self,
315        key: &CacheKey,
316        _trace: &SpanHandle,
317    ) -> Result<Option<(CacheMeta, HitHandler)>> {
318        let hash = key.combined();
319        // always prefer partial read otherwise fresh asset will not be visible on expired asset
320        // until it is fully updated
321        // no preference on which partial read we get (if there are multiple writers)
322        if let Some((_, temp_obj)) = self
323            .temp
324            .read()
325            .get(&hash)
326            .and_then(|map| map.iter().next())
327        {
328            hit_from_temp_obj(temp_obj)
329        } else if let Some(obj) = self.cached.read().get(&hash) {
330            let meta = CacheMeta::deserialize(&obj.meta.0, &obj.meta.1)?;
331            let hit_handler = CompleteHit {
332                body: obj.body.clone(),
333                done: false,
334                range_start: 0,
335                range_end: obj.body.len(),
336            };
337            let hit_handler = MemHitHandler::Complete(hit_handler);
338            Ok(Some((meta, Box::new(hit_handler))))
339        } else {
340            Ok(None)
341        }
342    }
343
344    async fn lookup_streaming_write(
345        &'static self,
346        key: &CacheKey,
347        streaming_write_tag: Option<&[u8]>,
348        _trace: &SpanHandle,
349    ) -> Result<Option<(CacheMeta, HitHandler)>> {
350        let hash = key.combined();
351        let write_tag: U64WriteId = streaming_write_tag
352            .expect("tag must be set during streaming write")
353            .try_into()
354            .expect("tag must be correct length");
355        hit_from_temp_obj(
356            self.temp
357                .read()
358                .get(&hash)
359                .and_then(|map| map.get(&write_tag.into()))
360                .expect("must have partial write in progress"),
361        )
362    }
363
364    async fn get_miss_handler(
365        &'static self,
366        key: &CacheKey,
367        meta: &CacheMeta,
368        _trace: &SpanHandle,
369    ) -> Result<MissHandler> {
370        let hash = key.combined();
371        let meta = meta.serialize()?;
372        let temp_obj = TempObject::new(meta);
373        let temp_id = self.last_temp_id.fetch_add(1, Ordering::Relaxed);
374        let miss_handler = MemMissHandler {
375            body: temp_obj.body.clone(),
376            bytes_written: temp_obj.bytes_written.clone(),
377            key: hash.clone(),
378            cache: self.cached.clone(),
379            temp: self.temp.clone(),
380            temp_id: temp_id.into(),
381        };
382        self.temp
383            .write()
384            .entry(hash)
385            .or_default()
386            .insert(miss_handler.temp_id.into(), temp_obj);
387        Ok(Box::new(miss_handler))
388    }
389
390    async fn purge(
391        &'static self,
392        target: PurgeTarget<'_>,
393        _type: PurgeType,
394        _trace: &SpanHandle,
395    ) -> Result<PurgeOutcome> {
396        // This test store does not retain entry IDs, so it cannot safely match identified entries.
397        if matches!(target, PurgeTarget::Exact(CacheEntryKey::Identified { .. })) {
398            return Ok(PurgeOutcome::NotFound);
399        }
400        // This usually purges the primary key because, without a lookup, the variance key is usually
401        // empty
402        let hash = target.key().combined();
403        let temp_removed = self.temp.write().remove(&hash).is_some();
404        let cache_removed = self.cached.write().remove(&hash).is_some();
405        if temp_removed || cache_removed {
406            // MemCache does not assign entry IDs, so active purges select key-only entries.
407            Ok(PurgeOutcome::Purged(None))
408        } else {
409            Ok(PurgeOutcome::NotFound)
410        }
411    }
412
413    async fn expire(
414        &'static self,
415        target: PurgeTarget<'_>,
416        _trace: &SpanHandle,
417    ) -> Result<PurgeOutcome> {
418        // This test store does not retain entry IDs, so it cannot safely match identified entries.
419        if matches!(target, PurgeTarget::Exact(CacheEntryKey::Identified { .. })) {
420            return Ok(PurgeOutcome::NotFound);
421        }
422        let hash = target.key().combined();
423        // Unlike purge this leaves `temp` alone. An in-flight miss has no committed meta to
424        // rewrite, and expiring what it is about to store would discard a fresh response.
425        // The guard covers the whole read-modify-write so concurrent expires cannot clobber
426        // each other.
427        let mut cached = self.cached.write();
428        let Some(obj) = cached.get_mut(&hash) else {
429            return Ok(PurgeOutcome::NotFound);
430        };
431        let mut meta = CacheMeta::deserialize(&obj.meta.0, &obj.meta.1)?;
432        meta.expire_at(SystemTime::now());
433        obj.meta = meta.serialize()?;
434        Ok(PurgeOutcome::Expired)
435    }
436
437    async fn update_meta(
438        &'static self,
439        key: &CacheKey,
440        meta: &CacheMeta,
441        _trace: &SpanHandle,
442    ) -> Result<bool> {
443        let hash = key.combined();
444        if let Some(obj) = self.cached.write().get_mut(&hash) {
445            obj.meta = meta.serialize()?;
446            Ok(true)
447        } else {
448            panic!("no meta found")
449        }
450    }
451
452    fn support_streaming_partial_write(&self) -> bool {
453        true
454    }
455
456    fn as_any(&self) -> &(dyn Any + Send + Sync) {
457        self
458    }
459}
460
461#[cfg(test)]
462mod test {
463    use super::*;
464    use crate::trace::Span;
465    use once_cell::sync::Lazy;
466
467    fn gen_meta() -> CacheMeta {
468        let mut header = ResponseHeader::build(200, None).unwrap();
469        header.append_header("foo1", "bar1").unwrap();
470        header.append_header("foo2", "bar2").unwrap();
471        header.append_header("foo3", "bar3").unwrap();
472        header.append_header("Server", "Pingora").unwrap();
473        let internal = crate::meta::InternalMeta::default();
474        CacheMeta(Box::new(crate::meta::CacheMetaInner {
475            internal,
476            header,
477            extensions: http::Extensions::new(),
478        }))
479    }
480
481    #[tokio::test]
482    async fn test_write_then_read() {
483        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
484        let span = &Span::inactive().handle();
485
486        let key1 = CacheKey::new("a", "1");
487        let res = MEM_CACHE.lookup(&key1, span).await.unwrap();
488        assert!(res.is_none());
489
490        let cache_meta = gen_meta();
491
492        let mut miss_handler = MEM_CACHE
493            .get_miss_handler(&key1, &cache_meta, span)
494            .await
495            .unwrap();
496        miss_handler
497            .write_body(b"test1"[..].into(), false)
498            .await
499            .unwrap();
500        miss_handler
501            .write_body(b"test2"[..].into(), false)
502            .await
503            .unwrap();
504        miss_handler.finish().await.unwrap();
505
506        let (cache_meta2, mut hit_handler) = MEM_CACHE.lookup(&key1, span).await.unwrap().unwrap();
507        assert_eq!(
508            cache_meta.0.internal.fresh_until,
509            cache_meta2.0.internal.fresh_until
510        );
511
512        let data = hit_handler.read_body().await.unwrap().unwrap();
513        assert_eq!("test1test2", data);
514        let data = hit_handler.read_body().await.unwrap();
515        assert!(data.is_none());
516    }
517
518    #[tokio::test]
519    async fn test_read_range() {
520        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
521        let span = &Span::inactive().handle();
522
523        let key1 = CacheKey::new("a", "1");
524        let res = MEM_CACHE.lookup(&key1, span).await.unwrap();
525        assert!(res.is_none());
526
527        let cache_meta = gen_meta();
528
529        let mut miss_handler = MEM_CACHE
530            .get_miss_handler(&key1, &cache_meta, span)
531            .await
532            .unwrap();
533        miss_handler
534            .write_body(b"test1test2"[..].into(), false)
535            .await
536            .unwrap();
537        miss_handler.finish().await.unwrap();
538
539        let (cache_meta2, mut hit_handler) = MEM_CACHE.lookup(&key1, span).await.unwrap().unwrap();
540        assert_eq!(
541            cache_meta.0.internal.fresh_until,
542            cache_meta2.0.internal.fresh_until
543        );
544
545        // out of range
546        assert!(hit_handler.seek(10000, None).is_err());
547
548        assert!(hit_handler.seek(5, None).is_ok());
549        let data = hit_handler.read_body().await.unwrap().unwrap();
550        assert_eq!("test2", data);
551        let data = hit_handler.read_body().await.unwrap();
552        assert!(data.is_none());
553
554        assert!(hit_handler.seek(4, Some(5)).is_ok());
555        let data = hit_handler.read_body().await.unwrap().unwrap();
556        assert_eq!("1", data);
557        let data = hit_handler.read_body().await.unwrap();
558        assert!(data.is_none());
559    }
560
561    #[tokio::test]
562    async fn test_write_while_read() {
563        use futures::FutureExt;
564
565        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
566        let span = &Span::inactive().handle();
567
568        let key1 = CacheKey::new("a", "1");
569        let res = MEM_CACHE.lookup(&key1, span).await.unwrap();
570        assert!(res.is_none());
571
572        let cache_meta = gen_meta();
573
574        let mut miss_handler = MEM_CACHE
575            .get_miss_handler(&key1, &cache_meta, span)
576            .await
577            .unwrap();
578
579        // first reader
580        let (cache_meta1, mut hit_handler1) = MEM_CACHE.lookup(&key1, span).await.unwrap().unwrap();
581        assert_eq!(
582            cache_meta.0.internal.fresh_until,
583            cache_meta1.0.internal.fresh_until
584        );
585
586        // No body to read
587        let res = hit_handler1.read_body().now_or_never();
588        assert!(res.is_none());
589
590        miss_handler
591            .write_body(b"test1"[..].into(), false)
592            .await
593            .unwrap();
594
595        let data = hit_handler1.read_body().await.unwrap().unwrap();
596        assert_eq!("test1", data);
597        let res = hit_handler1.read_body().now_or_never();
598        assert!(res.is_none());
599
600        miss_handler
601            .write_body(b"test2"[..].into(), false)
602            .await
603            .unwrap();
604        let data = hit_handler1.read_body().await.unwrap().unwrap();
605        assert_eq!("test2", data);
606
607        // second reader
608        let (cache_meta2, mut hit_handler2) = MEM_CACHE.lookup(&key1, span).await.unwrap().unwrap();
609        assert_eq!(
610            cache_meta.0.internal.fresh_until,
611            cache_meta2.0.internal.fresh_until
612        );
613
614        let data = hit_handler2.read_body().await.unwrap().unwrap();
615        assert_eq!("test1test2", data);
616        let res = hit_handler2.read_body().now_or_never();
617        assert!(res.is_none());
618
619        let res = hit_handler1.read_body().now_or_never();
620        assert!(res.is_none());
621
622        miss_handler.finish().await.unwrap();
623
624        let data = hit_handler1.read_body().await.unwrap();
625        assert!(data.is_none());
626        let data = hit_handler2.read_body().await.unwrap();
627        assert!(data.is_none());
628    }
629
630    #[tokio::test]
631    async fn test_purge_partial() {
632        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
633        let cache = &MEM_CACHE;
634
635        let key = CacheKey::new("a", "1").to_compact();
636        let hash = key.combined();
637        let meta = (
638            "meta_key".as_bytes().to_vec(),
639            "meta_value".as_bytes().to_vec(),
640        );
641
642        let temp_obj = TempObject::new(meta);
643        let mut map = HashMap::new();
644        map.insert(0, temp_obj);
645        cache.temp.write().insert(hash.clone(), map);
646
647        assert!(cache.temp.read().contains_key(&hash));
648
649        let result = cache
650            .purge(
651                crate::storage::PurgeTarget::Active(&key),
652                PurgeType::Invalidation,
653                &Span::inactive().handle(),
654            )
655            .await;
656        assert!(result.is_ok());
657
658        assert!(!cache.temp.read().contains_key(&hash));
659    }
660
661    #[tokio::test]
662    async fn test_purge_complete() {
663        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
664        let cache = &MEM_CACHE;
665
666        let key = CacheKey::new("a", "1").to_compact();
667        let hash = key.combined();
668        let meta = (
669            "meta_key".as_bytes().to_vec(),
670            "meta_value".as_bytes().to_vec(),
671        );
672        let body = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 0];
673        let cache_obj = CacheObject {
674            meta,
675            body: Arc::new(body),
676        };
677        cache.cached.write().insert(hash.clone(), cache_obj);
678
679        assert!(cache.cached.read().contains_key(&hash));
680
681        let result = cache
682            .purge(
683                crate::storage::PurgeTarget::Active(&key),
684                PurgeType::Invalidation,
685                &Span::inactive().handle(),
686            )
687            .await;
688        assert!(result.is_ok());
689
690        assert!(!cache.cached.read().contains_key(&hash));
691    }
692
693    #[tokio::test]
694    async fn expiring_keeps_the_entry_but_makes_it_stale() {
695        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
696        let cache = &MEM_CACHE;
697
698        let key = CacheKey::new("expire-me", "1");
699        let compact = key.to_compact();
700        let hash = compact.combined();
701        let body = vec![1, 2, 3];
702        let fresh = CacheMeta::new(
703            SystemTime::now() + std::time::Duration::from_secs(300),
704            SystemTime::now(),
705            30,
706            30,
707            ResponseHeader::build(200, None).unwrap(),
708        );
709        cache.cached.write().insert(
710            hash.clone(),
711            CacheObject {
712                meta: fresh.serialize().unwrap(),
713                body: Arc::new(body.clone()),
714            },
715        );
716
717        let outcome = cache
718            .expire(
719                crate::storage::PurgeTarget::Active(&compact),
720                &Span::inactive().handle(),
721            )
722            .await
723            .unwrap();
724
725        assert_eq!(outcome, crate::storage::PurgeOutcome::Expired);
726        let (meta, mut hit) = cache
727            .lookup(&key, &Span::inactive().handle())
728            .await
729            .unwrap()
730            .expect("the entry is still stored");
731        assert!(!meta.is_fresh(SystemTime::now() + std::time::Duration::from_secs(1)));
732        assert!(
733            meta.serve_stale_while_revalidate(SystemTime::now()),
734            "expiring must not close the serve stale windows the response set"
735        );
736        assert_eq!(
737            hit.read_body().await.unwrap().as_deref(),
738            Some(body.as_slice()),
739            "the body stays readable so a revalidation can reuse it"
740        );
741    }
742
743    #[tokio::test]
744    async fn expiring_an_absent_entry_finds_nothing() {
745        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
746        let key = CacheKey::new("never-stored", "1").to_compact();
747
748        let outcome = MEM_CACHE
749            .expire(
750                crate::storage::PurgeTarget::Active(&key),
751                &Span::inactive().handle(),
752            )
753            .await
754            .unwrap();
755
756        assert_eq!(outcome, crate::storage::PurgeOutcome::NotFound);
757    }
758
759    #[tokio::test]
760    async fn test_exact_identified_purge_does_not_remove_key_only_entry() {
761        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
762        let cache = &MEM_CACHE;
763        let key = CacheKey::new("identified", "1").to_compact();
764        let hash = key.combined();
765        cache.cached.write().insert(
766            hash.clone(),
767            CacheObject {
768                meta: (Vec::new(), Vec::new()),
769                body: Arc::new(Vec::new()),
770            },
771        );
772        let entry =
773            crate::eviction::CacheEntryKey::identified(key, crate::eviction::CacheEntryId::new(1));
774        let target = crate::storage::PurgeTarget::Exact(&entry);
775
776        let outcome = cache
777            .purge(target, PurgeType::Eviction, &Span::inactive().handle())
778            .await
779            .unwrap();
780
781        assert_eq!(outcome, crate::storage::PurgeOutcome::NotFound);
782        assert!(cache.cached.read().contains_key(&hash));
783    }
784
785    #[tokio::test]
786    async fn test_exact_key_only_purge_succeeds_and_removes_entry() {
787        static MEM_CACHE: Lazy<MemCache> = Lazy::new(MemCache::new);
788        let cache = &MEM_CACHE;
789        let key = CacheKey::new("key-only", "1").to_compact();
790        let hash = key.combined();
791        cache.cached.write().insert(
792            hash,
793            CacheObject {
794                meta: (Vec::new(), Vec::new()),
795                body: Arc::new(Vec::new()),
796            },
797        );
798
799        let entry = CacheEntryKey::key_only(key);
800        let outcome = cache
801            .purge(
802                crate::storage::PurgeTarget::Exact(&entry),
803                PurgeType::Eviction,
804                &Span::inactive().handle(),
805            )
806            .await
807            .unwrap();
808
809        assert_eq!(outcome, crate::storage::PurgeOutcome::Purged(None));
810        assert!(cache.cached.read().is_empty());
811    }
812}
pingora_cache/memory.rs

pingora_cache/
memory.rs
