engram/
engine.rs

1use std::{
2    any::{type_name, Any},
3    collections::HashMap,
4    future::Future,
5    marker::PhantomData,
6    sync::{
7        atomic::{AtomicBool, Ordering},
8        Arc, Weak,
9    },
10    time::{Duration, SystemTime, UNIX_EPOCH},
11};
12
13use serde::{de::DeserializeOwned, Serialize};
14use tokio::sync::{Mutex, Notify, RwLock};
15
16use crate::{
17    compression,
18    config::EngramBuilder,
19    memory::{self, MemoryEntry},
20    storage::{SqliteStore, StoredWrite},
21    Compression, Error, Result,
22};
23
24/// The tier that satisfied a cache read.
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26pub enum CacheTier {
27    Memory,
28    Persistent,
29}
30
31/// Expiration and source information for a cached value.
32#[derive(Clone, Debug, Eq, PartialEq)]
33pub struct CacheMetadata {
34    /// `None` means the value does not expire automatically.
35    pub expires_at: Option<SystemTime>,
36    /// The tier in which the entry was found for this lookup.
37    pub tier: CacheTier,
38}
39
40impl CacheMetadata {
41    /// Returns the remaining TTL. `None` means the entry is non-expiring.
42    pub fn remaining_ttl(&self) -> Option<Duration> {
43        self.expires_at
44            .and_then(|expires_at| expires_at.duration_since(SystemTime::now()).ok())
45    }
46}
47
48/// A cached value returned together with its metadata.
49#[derive(Clone, Debug, Eq, PartialEq)]
50pub struct CacheEntry<T> {
51    pub value: T,
52    pub metadata: CacheMetadata,
53}
54
55/// A cache key with a caller-controlled, stable schema tag.
56///
57/// Use this when values are intended to survive releases. Keep `schema` stable
58/// across Rust type/module renames, and change it when the serialized format is
59/// intentionally incompatible.
60#[derive(Clone, Debug, Eq, Hash, PartialEq)]
61pub struct CacheKey<T> {
62    key: String,
63    schema: String,
64    marker: PhantomData<fn() -> T>,
65}
66
67impl<T> CacheKey<T> {
68    /// Creates a key with a durable, application-defined schema tag.
69    pub fn new(key: impl Into<String>, schema: impl Into<String>) -> Self {
70        Self {
71            key: key.into(),
72            schema: schema.into(),
73            marker: PhantomData,
74        }
75    }
76
77    /// The storage key.
78    pub fn key(&self) -> &str {
79        &self.key
80    }
81
82    /// The application-defined schema tag persisted with this key.
83    pub fn schema(&self) -> &str {
84        &self.schema
85    }
86}
87
88/// A reusable asynchronous cache with Moka memory and SQLite persistence tiers.
89/// `Engram` is cheap to clone; clones share memory, persistence, and synchronization.
90#[derive(Clone)]
91pub struct Engram {
92    memory: memory::Memory,
93    storage: Arc<SqliteStore>,
94    compression: Compression,
95    compression_threshold: usize,
96    compress_only_if_smaller: bool,
97    // `clear` takes this write lock; all other operations take a shared lock.
98    clear_gate: Arc<RwLock<()>>,
99    // Operations on the same key serialize, while unrelated keys proceed concurrently.
100    key_locks: Arc<Mutex<HashMap<String, Weak<Mutex<()>>>>>,
101    in_flight: Arc<Mutex<HashMap<String, Arc<Flight>>>>,
102}
103
104struct Flight {
105    notify: Arc<Notify>,
106    completed: AtomicBool,
107}
108
109struct PendingWrite {
110    write: StoredWrite,
111    value: Arc<dyn Any + Send + Sync>,
112    expires_at: Option<SystemTime>,
113}
114
115impl Engram {
116    /// Starts configuration of an Engram cache.
117    pub fn builder() -> EngramBuilder {
118        EngramBuilder::new()
119    }
120
121    pub(crate) async fn from_builder(config: EngramBuilder) -> Result<Self> {
122        let storage = Arc::new(
123            SqliteStore::open(
124                &config.storage,
125                config.sqlite_connections,
126                config.sqlite_synchronous,
127            )
128            .await?,
129        );
130        let clear_gate = Arc::new(RwLock::new(()));
131        if let Some(interval) = config.cleanup_interval {
132            spawn_expiry_cleaner(
133                Arc::downgrade(&storage),
134                Arc::downgrade(&clear_gate),
135                interval,
136            );
137        }
138        Ok(Self {
139            memory: memory::new(config.memory_capacity),
140            storage,
141            compression: config.compression,
142            compression_threshold: config.compression_threshold,
143            compress_only_if_smaller: config.compress_only_if_smaller,
144            clear_gate,
145            key_locks: Arc::new(Mutex::new(HashMap::new())),
146            in_flight: Arc::new(Mutex::new(HashMap::new())),
147        })
148    }
149
150    /// Retrieves a value, checking Moka before SQLite.
151    pub async fn get<T>(&self, key: impl AsRef<str>) -> Result<Option<T>>
152    where
153        T: DeserializeOwned + Clone + Send + Sync + 'static,
154    {
155        Ok(self
156            .get_arc_with_metadata::<T>(key.as_ref(), type_name::<T>())
157            .await?
158            .map(|(value, _)| (*value).clone()))
159    }
160
161    /// Retrieves a value using a caller-controlled schema tag.
162    pub async fn get_key<T>(&self, key: &CacheKey<T>) -> Result<Option<T>>
163    where
164        T: DeserializeOwned + Clone + Send + Sync + 'static,
165    {
166        Ok(self
167            .get_arc_with_metadata::<T>(key.key(), key.schema())
168            .await?
169            .map(|(value, _)| (*value).clone()))
170    }
171
172    /// Retrieves a value with expiration/source metadata.
173    pub async fn get_with_metadata<T>(&self, key: impl AsRef<str>) -> Result<Option<CacheEntry<T>>>
174    where
175        T: DeserializeOwned + Clone + Send + Sync + 'static,
176    {
177        Ok(self
178            .get_arc_with_metadata::<T>(key.as_ref(), type_name::<T>())
179            .await?
180            .map(|(value, metadata)| CacheEntry {
181                value: (*value).clone(),
182                metadata,
183            }))
184    }
185
186    /// Retrieves a value and metadata using a caller-controlled schema tag.
187    pub async fn get_key_with_metadata<T>(&self, key: &CacheKey<T>) -> Result<Option<CacheEntry<T>>>
188    where
189        T: DeserializeOwned + Clone + Send + Sync + 'static,
190    {
191        Ok(self
192            .get_arc_with_metadata::<T>(key.key(), key.schema())
193            .await?
194            .map(|(value, metadata)| CacheEntry {
195                value: (*value).clone(),
196                metadata,
197            }))
198    }
199
200    /// Retrieves a shared L1 value without cloning it. This is useful for large values.
201    pub async fn get_arc<T>(&self, key: impl AsRef<str>) -> Result<Option<Arc<T>>>
202    where
203        T: DeserializeOwned + Send + Sync + 'static,
204    {
205        Ok(self
206            .get_arc_with_metadata::<T>(key.as_ref(), type_name::<T>())
207            .await?
208            .map(|(value, _)| value))
209    }
210
211    /// Retrieves a shared value using a caller-controlled schema tag.
212    pub async fn get_arc_key<T>(&self, key: &CacheKey<T>) -> Result<Option<Arc<T>>>
213    where
214        T: DeserializeOwned + Send + Sync + 'static,
215    {
216        Ok(self
217            .get_arc_with_metadata::<T>(key.key(), key.schema())
218            .await?
219            .map(|(value, _)| value))
220    }
221
222    /// Returns expiry and source information without deserializing an L2 value.
223    pub async fn metadata(&self, key: impl AsRef<str>) -> Result<Option<CacheMetadata>> {
224        let key = key.as_ref();
225        let _clear_guard = self.clear_gate.read().await;
226        let key_lock = self.key_lock(key).await;
227        let _key_guard = key_lock.lock().await;
228        if let Some(entry) = self.memory.get(key).await {
229            if !expired(entry.expires_at) {
230                return Ok(Some(CacheMetadata {
231                    expires_at: entry.expires_at,
232                    tier: CacheTier::Memory,
233                }));
234            }
235            self.memory.invalidate(key).await;
236        }
237        let Some(expires_at_ms) = self.storage.metadata(key).await? else {
238            return Ok(None);
239        };
240        if expired_ms(expires_at_ms) {
241            self.storage.remove(key).await?;
242            return Ok(None);
243        }
244        Ok(Some(CacheMetadata {
245            expires_at: expires_at_ms.map(system_time_from_ms),
246            tier: CacheTier::Persistent,
247        }))
248    }
249
250    /// Stores a value in both tiers using serialized byte length as its L1 weight.
251    pub async fn set<T>(&self, key: impl AsRef<str>, value: &T, ttl: Option<Duration>) -> Result<()>
252    where
253        T: Serialize + Clone + Send + Sync + 'static,
254    {
255        self.set_arc(key, Arc::new(value.clone()), ttl).await
256    }
257
258    /// Stores a shared value without cloning it for L1.
259    pub async fn set_arc<T>(
260        &self,
261        key: impl AsRef<str>,
262        value: Arc<T>,
263        ttl: Option<Duration>,
264    ) -> Result<()>
265    where
266        T: Serialize + Send + Sync + 'static,
267    {
268        self.set_arc_inner(
269            key.as_ref().to_owned(),
270            value,
271            ttl,
272            None,
273            type_name::<T>().to_owned(),
274        )
275        .await
276    }
277
278    /// Stores a value with a caller-controlled schema tag.
279    pub async fn set_key<T>(
280        &self,
281        key: &CacheKey<T>,
282        value: &T,
283        ttl: Option<Duration>,
284    ) -> Result<()>
285    where
286        T: Serialize + Clone + Send + Sync + 'static,
287    {
288        self.set_arc_key(key, Arc::new(value.clone()), ttl).await
289    }
290
291    /// Stores a shared value with a caller-controlled schema tag.
292    pub async fn set_arc_key<T>(
293        &self,
294        key: &CacheKey<T>,
295        value: Arc<T>,
296        ttl: Option<Duration>,
297    ) -> Result<()>
298    where
299        T: Serialize + Send + Sync + 'static,
300    {
301        self.set_arc_inner(
302            key.key().to_owned(),
303            value,
304            ttl,
305            None,
306            key.schema().to_owned(),
307        )
308        .await
309    }
310
311    /// Stores a value with an explicit L1 weight, useful when serialized size is not
312    /// representative of its in-memory footprint.
313    pub async fn set_with_weight<T>(
314        &self,
315        key: impl AsRef<str>,
316        value: &T,
317        ttl: Option<Duration>,
318        weight: u32,
319    ) -> Result<()>
320    where
321        T: Serialize + Clone + Send + Sync + 'static,
322    {
323        self.set_arc_inner(
324            key.as_ref().to_owned(),
325            Arc::new(value.clone()),
326            ttl,
327            Some(weight),
328            type_name::<T>().to_owned(),
329        )
330        .await
331    }
332
333    /// Stores an `Arc` with an explicit L1 weight.
334    pub async fn set_arc_with_weight<T>(
335        &self,
336        key: impl AsRef<str>,
337        value: Arc<T>,
338        ttl: Option<Duration>,
339        weight: u32,
340    ) -> Result<()>
341    where
342        T: Serialize + Send + Sync + 'static,
343    {
344        self.set_arc_inner(
345            key.as_ref().to_owned(),
346            value,
347            ttl,
348            Some(weight),
349            type_name::<T>().to_owned(),
350        )
351        .await
352    }
353
354    /// Persists many values in one SQLite transaction.
355    pub async fn set_many<T, I, K>(&self, values: I, ttl: Option<Duration>) -> Result<()>
356    where
357        T: Serialize + Send + Sync + 'static,
358        I: IntoIterator<Item = (K, T)>,
359        K: Into<String>,
360    {
361        let mut pending = Vec::new();
362        for (key, value) in values {
363            pending.push(
364                self.prepare_write(
365                    key.into(),
366                    Arc::new(value),
367                    ttl,
368                    None,
369                    type_name::<T>().to_owned(),
370                )
371                .await?,
372            );
373        }
374        if pending.is_empty() {
375            return Ok(());
376        }
377
378        let _clear_guard = self.clear_gate.read().await;
379        let mut keys: Vec<String> = pending
380            .iter()
381            .map(|pending| pending.write.key.clone())
382            .collect();
383        keys.sort_unstable();
384        keys.dedup();
385        let mut _key_guards = Vec::with_capacity(keys.len());
386        for key in &keys {
387            _key_guards.push(self.key_lock(key).await.lock_owned().await);
388        }
389
390        let writes: Vec<StoredWrite> = pending
391            .iter()
392            .map(|pending| pending.write.clone())
393            .collect();
394        self.storage.set_many(&writes).await?;
395        for pending_write in &pending {
396            self.insert_memory(pending_write).await;
397        }
398        Ok(())
399    }
400
401    /// Removes a key from both memory and persistent storage.
402    pub async fn remove(&self, key: impl AsRef<str>) -> Result<()> {
403        let key = key.as_ref();
404        let _clear_guard = self.clear_gate.read().await;
405        let key_lock = self.key_lock(key).await;
406        let _key_guard = key_lock.lock().await;
407        self.memory.invalidate(key).await;
408        self.storage.remove(key).await
409    }
410
411    /// Clears both cache tiers.
412    pub async fn clear(&self) -> Result<()> {
413        let _clear_guard = self.clear_gate.write().await;
414        self.memory.invalidate_all();
415        self.memory.run_pending_tasks().await;
416        self.storage.clear().await
417    }
418
419    /// Deletes expired SQLite rows and returns the number removed.
420    pub async fn purge_expired(&self) -> Result<u64> {
421        let _clear_guard = self.clear_gate.read().await;
422        self.storage
423            .purge_expired(system_time_ms(SystemTime::now()))
424            .await
425    }
426
427    /// Reads a key, calling `init` only on a miss and coalescing concurrent calls for
428    /// the same key. A failed initialization is not cached.
429    pub async fn get_or_try_init<T, F, Fut>(
430        &self,
431        key: impl AsRef<str>,
432        ttl: Option<Duration>,
433        init: F,
434    ) -> Result<T>
435    where
436        T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
437        F: FnOnce() -> Fut,
438        Fut: Future<Output = Result<T>>,
439    {
440        let key = key.as_ref().to_owned();
441        let mut init = Some(init);
442        loop {
443            if let Some(value) = self.get(&key).await? {
444                return Ok(value);
445            }
446            let waiting_on = {
447                let mut in_flight = self.in_flight.lock().await;
448                if let Some(flight) = in_flight.get(&key) {
449                    Some(flight.clone())
450                } else {
451                    in_flight.insert(
452                        key.clone(),
453                        Arc::new(Flight {
454                            notify: Arc::new(Notify::new()),
455                            completed: AtomicBool::new(false),
456                        }),
457                    );
458                    None
459                }
460            };
461            if let Some(flight) = waiting_on {
462                let notified = flight.notify.clone().notified_owned();
463                if !flight.completed.load(Ordering::Acquire) {
464                    notified.await;
465                }
466                continue;
467            }
468
469            let result =
470                init.take()
471                    .expect("only the initialization leader invokes its closure")()
472                .await;
473            if let Ok(value) = &result {
474                if let Err(error) = self.set(&key, value, ttl).await {
475                    self.finish_initialization(&key).await;
476                    return Err(error);
477                }
478            }
479            self.finish_initialization(&key).await;
480            return result;
481        }
482    }
483
484    async fn get_arc_with_metadata<T>(
485        &self,
486        key: &str,
487        type_tag: &str,
488    ) -> Result<Option<(Arc<T>, CacheMetadata)>>
489    where
490        T: DeserializeOwned + Send + Sync + 'static,
491    {
492        let _clear_guard = self.clear_gate.read().await;
493        let key_lock = self.key_lock(key).await;
494        let _key_guard = key_lock.lock().await;
495        if let Some(entry) = self.memory.get(key).await {
496            if !expired(entry.expires_at) {
497                if entry.type_name != type_tag {
498                    return Err(Error::TypeMismatch {
499                        key: key.to_owned(),
500                        requested: type_tag.to_owned(),
501                        actual: entry.type_name.clone(),
502                    });
503                }
504                let value = downcast_arc(key, entry.value.clone(), &entry.type_name)?;
505                return Ok(Some((
506                    value,
507                    CacheMetadata {
508                        expires_at: entry.expires_at,
509                        tier: CacheTier::Memory,
510                    },
511                )));
512            }
513            self.memory.invalidate(key).await;
514        }
515
516        let Some(stored) = self.storage.get(key).await? else {
517            return Ok(None);
518        };
519        if expired_ms(stored.expires_at_ms) {
520            self.storage.remove(key).await?;
521            return Ok(None);
522        }
523        if stored.type_name != type_tag {
524            return Err(Error::TypeMismatch {
525                key: key.to_owned(),
526                requested: type_tag.to_owned(),
527                actual: stored.type_name,
528            });
529        }
530        let bytes = compression::decode(stored.bytes, stored.encoding).await?;
531        let value = Arc::new(bincode::deserialize::<T>(&bytes).map_err(Error::Deserialize)?);
532        let expires_at = stored.expires_at_ms.map(system_time_from_ms);
533        let memory_value: Arc<dyn Any + Send + Sync> = value.clone();
534        self.memory
535            .insert(
536                key.to_owned(),
537                Arc::new(MemoryEntry {
538                    value: memory_value,
539                    type_name: type_tag.to_owned(),
540                    expires_at,
541                    weight: stored.memory_weight,
542                }),
543            )
544            .await;
545        Ok(Some((
546            value,
547            CacheMetadata {
548                expires_at,
549                tier: CacheTier::Persistent,
550            },
551        )))
552    }
553
554    async fn set_arc_inner<T>(
555        &self,
556        key: String,
557        value: Arc<T>,
558        ttl: Option<Duration>,
559        weight: Option<u32>,
560        type_tag: String,
561    ) -> Result<()>
562    where
563        T: Serialize + Send + Sync + 'static,
564    {
565        let pending = self
566            .prepare_write(key.clone(), value, ttl, weight, type_tag)
567            .await?;
568        let _clear_guard = self.clear_gate.read().await;
569        let key_lock = self.key_lock(&key).await;
570        let _key_guard = key_lock.lock().await;
571        self.insert_memory(&pending).await;
572        if let Err(error) = self.storage.set(&pending.write).await {
573            self.memory.invalidate(&key).await;
574            return Err(error);
575        }
576        Ok(())
577    }
578
579    async fn prepare_write<T>(
580        &self,
581        key: String,
582        value: Arc<T>,
583        ttl: Option<Duration>,
584        requested_weight: Option<u32>,
585        type_tag: String,
586    ) -> Result<PendingWrite>
587    where
588        T: Serialize + Send + Sync + 'static,
589    {
590        let serialized = bincode::serialize(value.as_ref()).map_err(Error::Serialize)?;
591        let weight = requested_weight
592            .unwrap_or_else(|| cache_weight(serialized.len()))
593            .max(1);
594        let (bytes, encoding) = compression::encode(
595            serialized,
596            &self.compression,
597            self.compression_threshold,
598            self.compress_only_if_smaller,
599        )
600        .await?;
601        let expires_at = ttl.map(|ttl| SystemTime::now() + ttl);
602        let memory_value: Arc<dyn Any + Send + Sync> = value;
603        Ok(PendingWrite {
604            write: StoredWrite {
605                key,
606                bytes,
607                expires_at_ms: expires_at.map(system_time_ms),
608                encoding,
609                type_name: type_tag,
610                memory_weight: weight,
611                created_at_ms: system_time_ms(SystemTime::now()),
612            },
613            value: memory_value,
614            expires_at,
615        })
616    }
617
618    async fn insert_memory(&self, pending: &PendingWrite) {
619        let entry = Arc::new(MemoryEntry {
620            value: pending.value.clone(),
621            type_name: pending.write.type_name.clone(),
622            expires_at: pending.expires_at,
623            weight: pending.write.memory_weight,
624        });
625        self.memory.insert(pending.write.key.clone(), entry).await;
626    }
627
628    async fn key_lock(&self, key: &str) -> Arc<Mutex<()>> {
629        let mut locks = self.key_locks.lock().await;
630        if locks.len() >= 1024 {
631            locks.retain(|_, lock| lock.strong_count() > 0);
632        }
633        if let Some(lock) = locks.get(key).and_then(Weak::upgrade) {
634            return lock;
635        }
636        let lock = Arc::new(Mutex::new(()));
637        locks.insert(key.to_owned(), Arc::downgrade(&lock));
638        lock
639    }
640
641    async fn finish_initialization(&self, key: &str) {
642        if let Some(flight) = self.in_flight.lock().await.remove(key) {
643            flight.completed.store(true, Ordering::Release);
644            flight.notify.notify_waiters();
645        }
646    }
647}
648
649fn spawn_expiry_cleaner(
650    storage: Weak<SqliteStore>,
651    clear_gate: Weak<RwLock<()>>,
652    interval: Duration,
653) {
654    tokio::spawn(async move {
655        loop {
656            tokio::time::sleep(interval).await;
657            let (Some(storage), Some(clear_gate)) = (storage.upgrade(), clear_gate.upgrade())
658            else {
659                return;
660            };
661            let _clear_guard = clear_gate.read().await;
662            let _ = storage
663                .purge_expired(system_time_ms(SystemTime::now()))
664                .await;
665        }
666    });
667}
668
669fn downcast_arc<T>(key: &str, value: Arc<dyn Any + Send + Sync>, actual: &str) -> Result<Arc<T>>
670where
671    T: Send + Sync + 'static,
672{
673    value.downcast::<T>().map_err(|_| Error::TypeMismatch {
674        key: key.to_owned(),
675        requested: type_name::<T>().to_owned(),
676        actual: actual.to_owned(),
677    })
678}
679
680fn cache_weight(serialized_len: usize) -> u32 {
681    serialized_len.clamp(1, u32::MAX as usize) as u32
682}
683
684fn expired(expires_at: Option<SystemTime>) -> bool {
685    expires_at.is_some_and(|expires_at| expires_at <= SystemTime::now())
686}
687
688fn expired_ms(expires_at_ms: Option<i64>) -> bool {
689    expires_at_ms.is_some_and(|expires_at| expires_at <= system_time_ms(SystemTime::now()))
690}
691
692fn system_time_ms(value: SystemTime) -> i64 {
693    value
694        .duration_since(UNIX_EPOCH)
695        .unwrap_or_default()
696        .as_millis()
697        .min(i64::MAX as u128) as i64
698}
699
700fn system_time_from_ms(value: i64) -> SystemTime {
701    UNIX_EPOCH + Duration::from_millis(value.max(0) as u64)
702}
703
704#[cfg(test)]
705mod tests {
706    use tempfile::TempDir;
707
708    use super::*;
709
710    async fn cache(directory: &TempDir, capacity: u64) -> Engram {
711        Engram::builder()
712            .storage(directory.path().join("engram.db"))
713            .memory_capacity(capacity)
714            .build()
715            .await
716            .unwrap()
717    }
718
719    #[tokio::test]
720    async fn l1_hit_does_not_need_a_persistent_row() {
721        let directory = TempDir::new().unwrap();
722        let cache = cache(&directory, 1_024).await;
723        cache
724            .set("key", &"memory value".to_owned(), None)
725            .await
726            .unwrap();
727        cache.storage.remove("key").await.unwrap();
728        assert_eq!(
729            cache.get::<String>("key").await.unwrap(),
730            Some("memory value".to_owned())
731        );
732    }
733
734    #[tokio::test]
735    async fn l2_hit_promotes_an_entry_back_to_l1() {
736        let directory = TempDir::new().unwrap();
737        let cache = cache(&directory, 1_024).await;
738        cache.set("key", &99_u64, None).await.unwrap();
739        cache.memory.invalidate("key").await;
740        assert_eq!(cache.get::<u64>("key").await.unwrap(), Some(99));
741        cache.storage.remove("key").await.unwrap();
742        assert_eq!(cache.get::<u64>("key").await.unwrap(), Some(99));
743    }
744
745    #[tokio::test]
746    async fn l1_eviction_never_removes_l2() {
747        let directory = TempDir::new().unwrap();
748        // Each value exceeds L1's capacity and is evicted by Moka.
749        let cache = cache(&directory, 1).await;
750        cache
751            .set("first", &"first entry".to_owned(), None)
752            .await
753            .unwrap();
754        cache
755            .set("second", &"second entry".to_owned(), None)
756            .await
757            .unwrap();
758        cache.memory.run_pending_tasks().await;
759        assert_eq!(
760            cache.get::<String>("first").await.unwrap(),
761            Some("first entry".to_owned())
762        );
763    }
764
765    #[tokio::test]
766    async fn clear_waits_for_an_active_key_operation() {
767        let directory = TempDir::new().unwrap();
768        let cache = cache(&directory, 1_024).await;
769        let clear_read_guard = cache.clear_gate.read().await;
770        let key_lock = cache.key_lock("promoting").await;
771        let key_guard = key_lock.lock().await;
772        let clearing_cache = cache.clone();
773        let clearing = tokio::spawn(async move { clearing_cache.clear().await });
774
775        tokio::task::yield_now().await;
776        assert!(
777            !clearing.is_finished(),
778            "clear must wait until the operation-wide gate is released"
779        );
780
781        drop(key_guard);
782        drop(clear_read_guard);
783        clearing.await.unwrap().unwrap();
784    }
785
786    #[tokio::test]
787    async fn failed_batch_does_not_partially_update_l1() {
788        let directory = TempDir::new().unwrap();
789        let cache = cache(&directory, 1_024).await;
790        cache.set("good", &"old".to_owned(), None).await.unwrap();
791        cache.storage.abort_writes_for_test().await.unwrap();
792
793        assert!(cache
794            .set_many(
795                [("good", "new".to_owned()), ("bad", "new".to_owned())],
796                None,
797            )
798            .await
799            .is_err());
800        assert_eq!(
801            cache.get::<String>("good").await.unwrap(),
802            Some("old".to_owned())
803        );
804        assert_eq!(cache.get::<String>("bad").await.unwrap(), None);
805    }
806}
engram/engine.rs

engram/
engine.rs
