Skip to main content

cubecl_environment/persistence/
store.rs

1use core::{fmt::Display, hash::Hash};
2
3use alloc::boxed::Box;
4use alloc::string::String;
5use alloc::vec::Vec;
6
7use hashbrown::{HashMap, HashSet};
8use serde::{Serialize, de::DeserializeOwned};
9
10use super::namespace::Namespace;
11use super::storage::{Insertion, Origin, Storage};
12use crate::bytes::Bytes;
13
14/// Error related to a [`Store`].
15#[derive(Debug)]
16pub enum StoreError<K, V> {
17    /// This process already stored a different value under that key: the same
18    /// function was computed twice with disagreeing results.
19    #[allow(missing_docs)]
20    DuplicatedKey {
21        key: K,
22        value_previous: V,
23        value_updated: V,
24    },
25    /// The durable entry was written by someone else — another process sharing
26    /// the cache root, or a bundle import — before this insert reached it.
27    /// Benign: the two values are equally valid and the stored one stays.
28    #[allow(missing_docs)]
29    KeyOutOfSync {
30        key: K,
31        value_previous: V,
32        value_updated: V,
33    },
34    /// The storage backend refused the write, so the entry is not durable and
35    /// will be recomputed on the next run.
36    #[allow(missing_docs)]
37    Backend { key: K, error: String },
38}
39
40impl<K, V> StoreError<K, V> {
41    /// Why the write failed, without the key or the value.
42    ///
43    /// [`Display`] renders both, which is right for small entries and wrong
44    /// for a compiled kernel: those values are megabytes of binary, and a log
45    /// line must not carry one. Callers storing large values report this
46    /// instead.
47    pub fn reason(&self) -> &str {
48        match self {
49            Self::DuplicatedKey { .. } => "the key was already stored with a different value",
50            Self::KeyOutOfSync { .. } => "another process stored the key first",
51            Self::Backend { error, .. } => error,
52        }
53    }
54}
55
56impl<K: core::fmt::Debug, V: core::fmt::Debug> Display for StoreError<K, V> {
57    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
58        match self {
59            Self::DuplicatedKey {
60                key,
61                value_previous,
62                value_updated,
63            } => write!(
64                f,
65                "key {key:?} was already stored with a different value: \
66                 kept {value_previous:?}, dropped {value_updated:?}"
67            ),
68            Self::KeyOutOfSync {
69                key,
70                value_previous,
71                value_updated,
72            } => write!(
73                f,
74                "key {key:?} was stored concurrently: kept {value_previous:?}, \
75                 dropped {value_updated:?}"
76            ),
77            Self::Backend { key, error } => {
78                write!(f, "storing key {key:?} failed: {error}")
79            }
80        }
81    }
82}
83
84impl<K: core::fmt::Debug, V: core::fmt::Debug> core::error::Error for StoreError<K, V> {}
85
86/// Trait to be implemented for store keys.
87pub trait StoreKey: Serialize + DeserializeOwned + PartialEq + Eq + Hash + Clone {}
88/// Trait to be implemented for store values.
89pub trait StoreValue: Serialize + DeserializeOwned + PartialEq + Eq + Clone {}
90
91impl<T: Serialize + DeserializeOwned + PartialEq + Eq + Clone + Hash> StoreKey for T {}
92impl<T: Serialize + DeserializeOwned + PartialEq + Eq + Clone> StoreValue for T {}
93
94/// How a [`Store`] populates its in-memory map from its storage.
95#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
96pub enum CacheOption {
97    /// Scan everything at open. Reads are complete: a miss is a miss and
98    /// never queries the storage.
99    #[default]
100    Eager,
101    /// Fault entries in from the storage on first access. Right when the
102    /// values are large enough that loading all of them eagerly is wasteful,
103    /// typically compilation artifacts.
104    Lazy,
105}
106
107/// Where a [`Store`] persists its entries.
108#[derive(Debug, Default)]
109enum StorageOption {
110    /// Nowhere: the in-memory map is all there is.
111    #[default]
112    InMemory,
113    /// The active environment's storage for this namespace.
114    Environment(Namespace),
115    /// An explicitly provided storage, addressing this namespace.
116    Explicit(Box<dyn Storage>, Namespace),
117}
118
119/// Defines how to create a [`Store`].
120#[derive(Debug, Default)]
121pub struct StoreOptions {
122    storage: StorageOption,
123    cache: CacheOption,
124}
125
126impl StoreOptions {
127    /// Options for a store with no persistence and everything in memory.
128    pub fn new() -> Self {
129        Self::default()
130    }
131
132    /// Persist entries under `namespace` in the active environment.
133    pub fn storage<N: Into<Namespace>>(mut self, namespace: N) -> Self {
134        self.storage = StorageOption::Environment(namespace.into());
135        self
136    }
137
138    /// Persist entries to an explicit storage addressing `namespace`,
139    /// bypassing the active environment. Mostly for tests and benches.
140    pub fn storage_with<N: Into<Namespace>>(
141        mut self,
142        storage: Box<dyn Storage>,
143        namespace: N,
144    ) -> Self {
145        self.storage = StorageOption::Explicit(storage, namespace.into());
146        self
147    }
148
149    /// How the in-memory map is populated from the storage. Meaningless
150    /// without one: an in-memory store keeps everything regardless.
151    pub fn cache(mut self, cache: CacheOption) -> Self {
152        self.cache = cache;
153        self
154    }
155}
156
157/// A typed key-value store over an optional persistence [`Storage`]: a Turso
158/// database on native targets and in the browser (feature `persistence`), or
159/// nothing at all.
160///
161/// Reads follow `HashMap`'s shape and mutation rules, with no interior
162/// mutability: [`get`](Store::get) serves shared references from memory,
163/// while everything that may change the store — [`insert`](Store::insert),
164/// [`get_mut`](Store::get_mut), [`remove`](Store::remove),
165/// [`sync`](Store::sync) — requires `&mut self`. Share a store by wrapping it
166/// in a lock, not by cloning it.
167///
168/// [`remove`](Store::remove) and [`clear`](Store::clear) act on memory alone;
169/// their durable mirrors [`purge_key`](Store::purge_key) and
170/// [`purge`](Store::purge) also delete from the storage.
171///
172/// [`CacheOption`] decides how the map is populated: [`Eager`
173/// ](CacheOption::Eager) ingests the whole namespace at open and serves every
174/// read from memory, [`Lazy`](CacheOption::Lazy) reads one key at a time on
175/// demand.
176///
177/// # No Edits
178///
179/// A stored value never changes: there is no update, and reinserting a key
180/// with a different value is an error. The one exception is imported entries:
181/// a locally computed value replaces a value that came from a bundle, because
182/// a shipped bundle must never be able to wedge the application that imported
183/// it. See [`Store::insert`]. Mutating a value through
184/// [`get_mut`](Store::get_mut) changes only the in-memory copy, never the
185/// storage.
186///
187/// # Environment switches
188///
189/// A store opened on the active environment stays bound to *the environment*,
190/// not to the storage it opened: when [`crate::environment`] switches
191/// ([`activate`](crate::environment::activate),
192/// [`load`](crate::environment::load), ...), the store detects it and resets —
193/// reads miss instead of serving the old environment's entries, and the next
194/// `&mut` operation drops the in-memory state and reopens the storage.
195/// Detection is one relaxed atomic load, so it costs nothing while the
196/// environment stays put. Stores on an explicit or absent storage are not
197/// bound and never reset.
198pub struct Store<K, V> {
199    entries: HashMap<K, V>,
200    /// Keys this process interacted with without the map holding their value:
201    /// [`CacheOption::Lazy`] inserts (which drop the value rather than retain
202    /// megabytes of compiled artifacts never read again) and
203    /// [`remove`](Store::remove)d entries. What tells a reinsert collision
204    /// [`StoreError::DuplicatedKey`] apart from [`StoreError::KeyOutOfSync`].
205    known: HashSet<K>,
206    storage: Option<Box<dyn Storage>>,
207    namespace: Option<Namespace>,
208    cache: CacheOption,
209    /// The environment generation the state belongs to, for stores bound to
210    /// the active environment; `None` for unbound ones (explicit storage, or
211    /// none). A mismatch with the current generation means everything here
212    /// describes an environment that is no longer active.
213    generation: Option<u32>,
214}
215
216impl<K: StoreKey, V: StoreValue> Store<K, V> {
217    /// Create a new store from the options.
218    ///
219    /// With an [`Eager`](CacheOption::Eager) cache over a storage, everything
220    /// the storage holds is ingested before returning.
221    #[cfg_attr(
222        feature = "tracing",
223        tracing::instrument(level = "trace", skip_all, fields(options = ?options))
224    )]
225    pub fn new(options: StoreOptions) -> Self {
226        let (storage, namespace, generation) = match options.storage {
227            StorageOption::InMemory => (None, None, None),
228            StorageOption::Environment(namespace) => {
229                // Sampled before the storage opens: a switch landing in
230                // between leaves a stale generation, which reads as "reset",
231                // never as "this storage belongs to the new environment".
232                let generation = crate::environment::generation();
233                (
234                    Some(super::storage::open(namespace.as_str())),
235                    Some(namespace),
236                    Some(generation),
237                )
238            }
239            StorageOption::Explicit(storage, namespace) => (Some(storage), Some(namespace), None),
240        };
241
242        let mut store = Self {
243            entries: HashMap::new(),
244            known: HashSet::new(),
245            storage,
246            namespace,
247            cache: options.cache,
248            generation,
249        };
250
251        if matches!(store.cache, CacheOption::Eager) && store.storage.is_some() {
252            store.ingest();
253        }
254
255        store
256    }
257
258    /// The namespace this store addresses, if it persists anywhere.
259    pub fn namespace(&self) -> Option<&Namespace> {
260        self.namespace.as_ref()
261    }
262
263    /// Fetch an item from memory.
264    ///
265    /// Never touches the storage: on an eager store the map is complete, so a
266    /// miss is a miss. On a lazy store this only serves entries a previous
267    /// [`get_mut`](Store::get_mut) faulted in; use `get_mut` to read through.
268    ///
269    /// After an environment switch everything in memory belongs to the old
270    /// environment, so this misses rather than serve it — a miss costs a
271    /// recompute, a stale hit would be wrong.
272    pub fn get(&self, key: &K) -> Option<&V> {
273        if self.stale() {
274            return None;
275        }
276
277        self.entries.get(key)
278    }
279
280    /// Fetch an item, reading it from the storage on the first lookup of a
281    /// lazy store and memoizing it afterwards.
282    ///
283    /// Mutating the value changes only the in-memory copy, never the storage.
284    pub fn get_mut(&mut self, key: &K) -> Option<&mut V> {
285        self.reset_if_stale();
286
287        if matches!(self.cache, CacheOption::Lazy)
288            && !self.entries.contains_key(key)
289            && let Some(value) = self.fetch(key)
290        {
291            self.entries.insert(key.clone(), value);
292        }
293
294        self.entries.get_mut(key)
295    }
296
297    /// Take an item out of memory, reading through to the storage on a lazy
298    /// store, and hand it out owned.
299    ///
300    /// The storage is untouched — deleting durably is
301    /// [`purge_key`](Store::purge_key). The key stays
302    /// [`known`](StoreError::DuplicatedKey) to the store. This is the read
303    /// for a value consumed once per process — a compiled kernel about to be
304    /// loaded — because nothing is cloned and nothing stays memoized.
305    pub fn remove(&mut self, key: &K) -> Option<V> {
306        self.reset_if_stale();
307
308        let value = match self.entries.remove(key) {
309            Some(value) => Some(value),
310            None => match self.cache {
311                CacheOption::Eager => None,
312                CacheOption::Lazy => self.fetch(key),
313            },
314        };
315
316        // Only when durable: for a purely in-memory store the entry is simply
317        // gone, and reinserting the key is a fresh insert.
318        if value.is_some() && self.storage.is_some() {
319            self.known.insert(key.clone());
320        }
321
322        value
323    }
324
325    /// Insert a new item into the store.
326    ///
327    /// - Key absent: the entry is written to the storage.
328    /// - Present with the same value: `Ok`, nothing written.
329    /// - Present with a different value: an error, and the stored value is
330    ///   left untouched — [`StoreError::DuplicatedKey`] when this process
331    ///   wrote or read it, [`StoreError::KeyOutOfSync`] when another process
332    ///   sharing the environment got there first. Both are routine, not bugs.
333    ///
334    /// The exception is an entry that came from a bundle: the storage lets a
335    /// locally computed value replace it, so a stale bundle can never wedge
336    /// the application that imported it.
337    pub fn insert(&mut self, key: K, value: V) -> Result<(), StoreError<K, V>> {
338        self.reset_if_stale();
339
340        let known = match self.entries.get(&key) {
341            Some(existing) if existing == &value => return Ok(()),
342            existing => existing.is_some() || self.known.contains(&key),
343        };
344
345        let Some(storage) = self.storage.as_deref() else {
346            return match self.entries.get(&key) {
347                Some(existing) => Err(StoreError::DuplicatedKey {
348                    value_previous: existing.clone(),
349                    value_updated: value,
350                    key,
351                }),
352                None => {
353                    self.entries.insert(key, value);
354                    Ok(())
355                }
356            };
357        };
358
359        // Only memory is consulted above: `write_through` asks the storage
360        // atomically, so reading it first would cost a second round trip and
361        // still not know whether the existing entry is imported.
362        match write_through(storage, &key, &value) {
363            Written::Stored => {
364                self.record(key, value);
365                Ok(())
366            }
367            Written::Failed(error) => Err(StoreError::Backend { key, error }),
368            Written::Conflict(existing) => {
369                // A later read must serve the durable value, so the eager map
370                // memoizes it; the clone only happens on this cold path.
371                if matches!(self.cache, CacheOption::Eager) {
372                    self.entries.insert(key.clone(), existing.clone());
373                } else {
374                    self.known.insert(key.clone());
375                }
376
377                // `known` is what tells the two conflicts apart: this process
378                // saw the key before, or someone else stored it first. The
379                // second is a routine multi-process race, not a bug.
380                let (value_previous, value_updated) = (existing, value);
381                Err(if known {
382                    StoreError::DuplicatedKey {
383                        key,
384                        value_previous,
385                        value_updated,
386                    }
387                } else {
388                    StoreError::KeyOutOfSync {
389                        key,
390                        value_previous,
391                        value_updated,
392                    }
393                })
394            }
395        }
396    }
397
398    /// Deletes one entry, in memory and durably, and hands it out owned.
399    ///
400    /// [`remove`](Store::remove)'s durable mirror, with the same signature:
401    /// where `remove` only evicts the in-memory copy, this also deletes the
402    /// storage entry, so the key is gone for every process sharing the
403    /// environment and free to reinsert.
404    pub fn purge_key(&mut self, key: &K) -> Option<V> {
405        self.reset_if_stale();
406
407        let value = match self.entries.remove(key) {
408            Some(value) => Some(value),
409            None => match self.cache {
410                CacheOption::Eager => None,
411                CacheOption::Lazy => self.fetch(key),
412            },
413        };
414
415        // A purged key is a fresh key, and the storage is asked regardless of
416        // what memory held: another process may have written it since.
417        self.known.remove(key);
418        if let Some(storage) = self.storage.as_deref() {
419            storage.purge_key(&encode(key));
420        }
421
422        value
423    }
424
425    /// Evicts every in-memory entry, leaving the storage untouched.
426    ///
427    /// `clear` is to [`purge`](Store::purge) what [`remove`](Store::remove)
428    /// is to [`purge_key`](Store::purge_key): the in-memory half only. On an
429    /// eager store the entries come back on the next [`sync`](Store::sync);
430    /// the keys stay known, exactly as with `remove`.
431    pub fn clear(&mut self) {
432        self.reset_if_stale();
433
434        if self.storage.is_some() {
435            self.known.extend(self.entries.drain().map(|(key, _)| key));
436        } else {
437            self.entries.clear();
438        }
439    }
440
441    /// Deletes everything this store addresses, in memory and durably.
442    ///
443    /// [`clear`](Store::clear)'s durable mirror: after a purge the namespace
444    /// is empty for every process sharing the environment, and every key is
445    /// free to reinsert. Only this namespace is touched, never the rest of
446    /// the environment. Other stores on the same namespace keep what they
447    /// already ingested until they sync.
448    pub fn purge(&mut self) {
449        self.reset_if_stale();
450
451        self.entries.clear();
452        self.known.clear();
453
454        if let Some(storage) = self.storage.as_deref() {
455            storage.purge();
456        }
457    }
458
459    /// Ingest everything the storage holds into memory.
460    ///
461    /// This is what makes an eager store complete, and [`new`](Store::new)
462    /// performs it; call it again to ingest content another store or process
463    /// wrote to the storage since.
464    #[cfg_attr(
465        feature = "tracing",
466        tracing::instrument(level = "trace", skip_all, fields(namespace = ?self.namespace))
467    )]
468    pub fn sync(&mut self) {
469        // A reset re-ingests an eager store on its own.
470        if self.reset_if_stale() && matches!(self.cache, CacheOption::Eager) {
471            return;
472        }
473        self.ingest();
474    }
475
476    /// Reads every entry of the storage into memory.
477    fn ingest(&mut self) {
478        let Some(storage) = self.storage.as_deref() else {
479            return;
480        };
481        let entries = &mut self.entries;
482
483        storage.scan(&mut |key, value| {
484            if let Some((key, value)) = decode_entry::<K, V>(key, value) {
485                entries.insert(key, value);
486            }
487        });
488    }
489
490    /// Visits every entry the storage holds, decoded and handed out owned,
491    /// retaining none of them in memory.
492    ///
493    /// The read-through counterpart of [`for_each`](Store::for_each): where
494    /// `for_each` walks the in-memory map, this walks the storage. It is the
495    /// hydration read for a caller keeping its own index over a
496    /// [`Lazy`](CacheOption::Lazy) store — everything is visited once, and
497    /// nothing stays resident afterwards.
498    pub fn scan<F: FnMut(K, V)>(&mut self, mut func: F) {
499        self.reset_if_stale();
500
501        let Some(storage) = self.storage.as_deref() else {
502            for (key, value) in self.entries.iter() {
503                func(key.clone(), value.clone());
504            }
505            return;
506        };
507
508        storage.scan(&mut |key, value| {
509            if let Some((key, value)) = decode_entry::<K, V>(key, value) {
510                func(key, value);
511            }
512        });
513    }
514
515    /// Iterate over all in-memory entries of the store.
516    pub fn for_each<F: FnMut(&K, &V)>(&self, mut func: F) {
517        if self.stale() {
518            return;
519        }
520
521        for (key, value) in self.entries.iter() {
522            func(key, value);
523        }
524    }
525
526    /// How many entries are in memory.
527    pub fn len(&self) -> usize {
528        if self.stale() {
529            return 0;
530        }
531
532        self.entries.len()
533    }
534
535    /// If nothing is in memory.
536    pub fn is_empty(&self) -> bool {
537        self.len() == 0
538    }
539
540    /// One entry read straight from the storage.
541    fn fetch(&self, key: &K) -> Option<V> {
542        let bytes = self.storage.as_deref()?.get(&encode(key))?;
543        decode::<V>(&bytes)
544    }
545
546    /// Records a value the storage accepted, according to the cache option.
547    fn record(&mut self, key: K, value: V) {
548        match self.cache {
549            CacheOption::Eager => {
550                self.entries.insert(key, value);
551            }
552            // The value is dropped rather than memoized: freshly compiled
553            // artifacts are typically never read again by this process.
554            CacheOption::Lazy => {
555                self.known.insert(key);
556            }
557        }
558    }
559
560    /// Whether the in-memory state belongs to an environment that is no
561    /// longer active. One relaxed atomic load for bound stores; unbound ones
562    /// are never stale.
563    fn stale(&self) -> bool {
564        match self.generation {
565            Some(generation) => generation != crate::environment::generation(),
566            None => false,
567        }
568    }
569
570    /// Drops everything belonging to the previous environment and reopens the
571    /// storage against the active one, re-ingesting it when eager. Reports
572    /// whether a reset happened.
573    fn reset_if_stale(&mut self) -> bool {
574        if !self.stale() {
575            return false;
576        }
577
578        // `stale` implies `generation` and an environment-bound namespace.
579        let (Some(namespace), Some(_)) = (&self.namespace, self.generation) else {
580            return false;
581        };
582
583        log::debug!("Environment switched, resetting the store for {namespace}");
584
585        // Generation first, storage second, mirroring `new`: a switch landing
586        // in between reads as stale again, never as up to date.
587        self.generation = Some(crate::environment::generation());
588        self.storage = Some(super::storage::open(namespace.as_str()));
589        self.entries.clear();
590        self.known.clear();
591        if matches!(self.cache, CacheOption::Eager) {
592            self.ingest();
593        }
594
595        true
596    }
597}
598
599impl<K, V> core::fmt::Debug for Store<K, V> {
600    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
601        f.debug_struct("Store")
602            .field("namespace", &self.namespace)
603            .field("cache", &self.cache)
604            .field("entries", &self.entries.len())
605            .field("known", &self.known.len())
606            .field("storage", &self.storage)
607            .finish()
608    }
609}
610
611impl<K: StoreKey, V: StoreValue> Display for Store<K, V> {
612    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
613        match (&self.namespace, &self.storage) {
614            (Some(namespace), Some(storage)) => write!(
615                f,
616                "{namespace} ({} entries in {})",
617                self.len(),
618                storage.describe()
619            ),
620            _ => write!(f, "in-memory ({} entries)", self.len()),
621        }
622    }
623}
624
625/// The outcome of writing an entry through to the storage.
626pub(crate) enum Written<V> {
627    /// The storage now holds this value, or already held an identical one.
628    Stored,
629    /// The storage kept a different value, which is the durable one.
630    Conflict(V),
631    /// The backend refused the write; nothing is durable.
632    Failed(String),
633}
634
635/// Writes `value` through to `storage` and lets it arbitrate.
636///
637/// The storage decides what happens on a collision: it lets a local value
638/// replace an imported one, and otherwise refuses to overwrite. Both stores go
639/// through here, so the rule is identical for eager and lazy caches rather
640/// than reimplemented per store.
641pub(crate) fn write_through<K: StoreKey, V: StoreValue>(
642    storage: &dyn Storage,
643    key: &K,
644    value: &V,
645) -> Written<V> {
646    let key_bytes = encode(key);
647
648    match storage.insert(&key_bytes, encode(value), Origin::Local) {
649        Insertion::Stored => Written::Stored,
650        Insertion::Failed(error) => Written::Failed(error),
651        Insertion::Conflict(existing) => match decode::<V>(&existing) {
652            Some(existing) if &existing != value => Written::Conflict(existing),
653            Some(_) => Written::Stored,
654            // Bytes that don't decode are bytes no later insert could ever
655            // agree with, so leaving them in place would refuse every write
656            // for this key forever — a permanent recompile for a lazily read
657            // key. Repair the row instead.
658            None => match storage.replace(&key_bytes, encode(value), Origin::Local) {
659                Insertion::Failed(error) => Written::Failed(error),
660                _ => Written::Stored,
661            },
662        },
663    }
664}
665
666/// Serializes a key or a value to its stored representation.
667///
668/// Unlike [`decode`], this panics on failure rather than degrading. The two
669/// are not symmetric: `decode` reads bytes from outside this process (a shared
670/// database, an imported bundle) that a version skew or corruption can make
671/// unreadable, which is a routine recompute. `encode` serializes a value the
672/// caller just constructed with a derived `Serialize`, and writing CBOR into a
673/// `Vec` has no I/O to fail on: a failure here is a broken `StoreKey`/
674/// `StoreValue` impl, a bug to surface loudly, not a cache miss to swallow.
675pub(crate) fn encode<T: Serialize>(value: &T) -> Bytes {
676    let mut bytes = Vec::new();
677    ciborium::ser::into_writer(value, &mut bytes).expect("Can serialize data");
678    Bytes::from_bytes_vec(bytes)
679}
680
681/// Deserializes a key or a value, reporting corrupted content instead of
682/// failing: a cache entry we can't read is one we recompute.
683pub(crate) fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Option<T> {
684    match ciborium::de::from_reader(bytes) {
685        Ok(value) => Some(value),
686        Err(err) => {
687            log::warn!("Corrupted cache entry, ignoring it: {err}");
688            None
689        }
690    }
691}
692
693fn decode_entry<K: StoreKey, V: StoreValue>(key: &[u8], value: &[u8]) -> Option<(K, V)> {
694    Some((decode::<K>(key)?, decode::<V>(value)?))
695}
696
697#[cfg(all(test, native_cache))]
698mod tests {
699    use std::string::ToString;
700    use std::vec;
701
702    use super::*;
703
704    fn eager(path: &str) -> StoreOptions {
705        StoreOptions::new().storage(Namespace::new(path))
706    }
707
708    fn lazy(path: &str) -> StoreOptions {
709        eager(path).cache(CacheOption::Lazy)
710    }
711
712    #[test_log::test]
713    #[serial_test::serial]
714    #[cfg_attr(miri, ignore)]
715    fn test_cache_simple() {
716        let dir = tempfile::tempdir().unwrap();
717        crate::environment::set_root(dir.path());
718
719        let key1 = || "key1".to_string();
720        let key2 = || "key2".to_string();
721
722        let value1 = || "value1".to_string();
723        let value2 = || "value2".to_string();
724
725        let mut cache = Store::<String, String>::new(eager("test"));
726        cache.insert(key1(), value1()).unwrap();
727        cache.insert(key2(), value2()).unwrap();
728
729        let result = cache.insert(key1(), value2());
730        assert!(
731            result.is_err(),
732            "Can't reinsert the same key with a different value."
733        );
734
735        assert_eq!(cache.len(), 2);
736
737        let value1_actual = cache.get(&key1()).unwrap();
738        assert_eq!(value1_actual, &value1());
739
740        let value2_actual = cache.get(&key2()).unwrap();
741        assert_eq!(value2_actual, &value2());
742    }
743
744    /// Guards the on-disk contract: the database file location and the exact
745    /// namespace a given set of options resolves to. Breaking either
746    /// invalidates every existing cache on users' machines.
747    #[test_log::test]
748    #[serial_test::serial]
749    #[cfg_attr(miri, ignore)]
750    fn test_on_disk_format_is_stable() {
751        let dir = tempfile::tempdir().unwrap();
752        crate::environment::set_root(dir.path());
753        let namespace = Namespace::scoped("golden", "device0/matmul");
754
755        let mut cache = Store::<String, u32>::new(StoreOptions::new().storage(namespace));
756        cache.insert("shape=2x2".to_string(), 42).unwrap();
757
758        let expected_namespace =
759            std::format!("golden/{}/device0/matmul", env!("CARGO_PKG_VERSION"));
760        assert_eq!(cache.namespace().unwrap().as_str(), expected_namespace);
761
762        let path = crate::environment::path();
763        assert!(path.exists(), "Database missing at {path:?}");
764
765        // Read it back through a fresh store: the entry must be addressable
766        // by namespace and encoded key alone.
767        let reopened = Store::<String, u32>::new(
768            StoreOptions::new().storage(Namespace::scoped("golden", "device0/matmul")),
769        );
770        assert_eq!(reopened.get(&"shape=2x2".to_string()), Some(&42));
771    }
772
773    /// A store reopened over the same root must see what the previous one
774    /// wrote, without any bundle involved.
775    #[test_log::test]
776    #[serial_test::serial]
777    #[cfg_attr(miri, ignore)]
778    fn test_entries_survive_reopen() {
779        let dir = tempfile::tempdir().unwrap();
780        crate::environment::set_root(dir.path());
781
782        let mut cache = Store::<String, u32>::new(eager("reopen"));
783        cache.insert("key".to_string(), 7).unwrap();
784        drop(cache);
785
786        let cache = Store::<String, u32>::new(eager("reopen"));
787        assert_eq!(cache.get(&"key".to_string()), Some(&7));
788    }
789
790    /// Two namespaces in the same root must not see each other's entries.
791    #[test_log::test]
792    #[serial_test::serial]
793    #[cfg_attr(miri, ignore)]
794    fn test_stores_are_isolated() {
795        let dir = tempfile::tempdir().unwrap();
796        crate::environment::set_root(dir.path());
797
798        let mut first = Store::<String, u32>::new(eager("device0/matmul"));
799        first.insert("key".to_string(), 1).unwrap();
800
801        let mut second = Store::<String, u32>::new(eager("device1/matmul"));
802        assert_eq!(second.get(&"key".to_string()), None);
803        second.insert("key".to_string(), 2).unwrap();
804
805        assert_eq!(first.get(&"key".to_string()), Some(&1));
806        assert_eq!(second.get(&"key".to_string()), Some(&2));
807    }
808
809    #[test_log::test]
810    #[serial_test::serial]
811    #[cfg_attr(miri, ignore)]
812    fn lazy_values_survive_reopen_and_load_lazily() {
813        let dir = tempfile::tempdir().unwrap();
814        crate::environment::set_root(dir.path());
815
816        let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
817        cache
818            .insert(
819                "kernel_a".to_string(),
820                Bytes::from_bytes_vec(std::vec![1, 2, 3]),
821            )
822            .unwrap();
823        cache
824            .insert(
825                "kernel_b".to_string(),
826                Bytes::from_bytes_vec(std::vec![4, 5]),
827            )
828            .unwrap();
829        // A lazy insert records the key but never retains the artifact.
830        assert!(cache.is_empty());
831        drop(cache);
832
833        let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
834        // Nothing is read until a key is asked for.
835        assert!(cache.is_empty());
836
837        assert_eq!(
838            cache.get_mut(&"kernel_a".to_string()).map(|v| v.to_vec()),
839            Some(std::vec![1, 2, 3])
840        );
841        assert_eq!(cache.len(), 1, "get_mut memoizes");
842        assert_eq!(
843            cache.remove(&"kernel_b".to_string()).map(|v| v.to_vec()),
844            Some(std::vec![4, 5])
845        );
846        assert_eq!(cache.len(), 1, "remove reads through without memoizing");
847        assert_eq!(cache.get_mut(&"missing".to_string()), None);
848    }
849
850    #[test_log::test]
851    #[serial_test::serial]
852    #[cfg_attr(miri, ignore)]
853    fn lazy_reinserting_a_different_value_errors() {
854        let dir = tempfile::tempdir().unwrap();
855        crate::environment::set_root(dir.path());
856
857        let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
858        let kernel = |byte: u8| Bytes::from_bytes_vec(std::vec![byte]);
859        cache.insert("kernel".to_string(), kernel(1)).unwrap();
860
861        assert!(cache.insert("kernel".to_string(), kernel(1)).is_ok());
862        let error = cache.insert("kernel".to_string(), kernel(2));
863        assert!(matches!(error, Err(StoreError::DuplicatedKey { .. })));
864
865        // Taking the value out doesn't forget the key: the entry is still
866        // durable, so a disagreeing reinsert stays a duplicate.
867        assert!(cache.remove(&"kernel".to_string()).is_some());
868        let error = cache.insert("kernel".to_string(), kernel(2));
869        assert!(matches!(error, Err(StoreError::DuplicatedKey { .. })));
870    }
871
872    /// A bound store follows the environment: a switch makes reads miss
873    /// instead of serving the old environment, and the next `&mut` access
874    /// reopens against the new one.
875    #[test_log::test]
876    #[serial_test::serial]
877    #[cfg_attr(miri, ignore)]
878    fn switching_environments_resets_bound_stores() {
879        let first = tempfile::tempdir().unwrap();
880        let second = tempfile::tempdir().unwrap();
881
882        crate::environment::set_root(first.path());
883        let mut store = Store::<String, u32>::new(eager("reset"));
884        store.insert("key".to_string(), 1).unwrap();
885        assert_eq!(store.get(&"key".to_string()), Some(&1));
886
887        // The old environment's entries are never served after the switch.
888        crate::environment::set_root(second.path());
889        assert_eq!(store.get(&"key".to_string()), None);
890        assert_eq!(store.len(), 0);
891
892        // The next write lands in the new environment, with no conflict
893        // against the value the old one holds.
894        store.insert("key".to_string(), 2).unwrap();
895        assert_eq!(store.get(&"key".to_string()), Some(&2));
896
897        // Switching back serves the first environment's value again.
898        crate::environment::set_root(first.path());
899        store.sync();
900        assert_eq!(store.get(&"key".to_string()), Some(&1));
901    }
902
903    /// Stores on an explicit or absent storage are not bound to the
904    /// environment and must not reset on a switch.
905    #[test_log::test]
906    #[serial_test::serial]
907    #[cfg_attr(miri, ignore)]
908    fn unbound_stores_survive_environment_switches() {
909        let root = tempfile::tempdir().unwrap();
910
911        let mut store = Store::<String, u32>::new(StoreOptions::new());
912        store.insert("key".to_string(), 1).unwrap();
913
914        crate::environment::set_root(root.path());
915        assert_eq!(store.get(&"key".to_string()), Some(&1));
916    }
917
918    /// `purge_key` is `remove` with a durable delete: the entry is handed out
919    /// owned, gone from the storage, and the key is fresh again.
920    #[test_log::test]
921    #[serial_test::serial]
922    #[cfg_attr(miri, ignore)]
923    fn purge_key_deletes_one_entry_durably() {
924        let dir = tempfile::tempdir().unwrap();
925        crate::environment::set_root(dir.path());
926
927        let mut store = Store::<String, u32>::new(eager("purge_key"));
928        store.insert("gone".to_string(), 1).unwrap();
929        store.insert("kept".to_string(), 2).unwrap();
930
931        assert_eq!(store.purge_key(&"gone".to_string()), Some(1));
932        // Fresh key: a different value is a plain insert, not a duplicate.
933        store.insert("gone".to_string(), 3).unwrap();
934        assert_eq!(store.purge_key(&"gone".to_string()), Some(3));
935        drop(store);
936
937        let store = Store::<String, u32>::new(eager("purge_key"));
938        assert_eq!(store.get(&"gone".to_string()), None);
939        assert_eq!(store.get(&"kept".to_string()), Some(&2));
940    }
941
942    /// `clear` evicts memory only: the storage keeps everything, the keys
943    /// stay known, and a sync brings the entries back.
944    #[test_log::test]
945    #[serial_test::serial]
946    #[cfg_attr(miri, ignore)]
947    fn clear_evicts_memory_but_not_the_storage() {
948        let dir = tempfile::tempdir().unwrap();
949        crate::environment::set_root(dir.path());
950
951        let mut store = Store::<String, u32>::new(eager("clear"));
952        store.insert("key".to_string(), 1).unwrap();
953
954        store.clear();
955        assert!(store.is_empty());
956        // Still durable and still known: a disagreeing reinsert is a
957        // duplicate, not a fresh insert.
958        assert!(matches!(
959            store.insert("key".to_string(), 2),
960            Err(StoreError::DuplicatedKey { .. })
961        ));
962
963        store.sync();
964        assert_eq!(store.get(&"key".to_string()), Some(&1));
965    }
966
967    /// Unlike `remove`, which only evicts the in-memory copy, `purge` deletes
968    /// the whole namespace durably and frees every key for reinsertion.
969    #[test_log::test]
970    #[serial_test::serial]
971    #[cfg_attr(miri, ignore)]
972    fn purge_deletes_durably_and_frees_the_keys() {
973        let dir = tempfile::tempdir().unwrap();
974        crate::environment::set_root(dir.path());
975
976        let mut store = Store::<String, u32>::new(eager("purge"));
977        store.insert("kept".to_string(), 1).unwrap();
978        store.insert("gone".to_string(), 2).unwrap();
979
980        // An isolated namespace must survive its neighbor's purge.
981        let mut other = Store::<String, u32>::new(eager("other"));
982        other.insert("kept".to_string(), 9).unwrap();
983
984        store.purge();
985        assert!(store.is_empty());
986
987        // A purged key is a fresh key, even with a different value.
988        store.insert("kept".to_string(), 3).unwrap();
989        drop(store);
990
991        let store = Store::<String, u32>::new(eager("purge"));
992        assert_eq!(store.get(&"kept".to_string()), Some(&3));
993        assert_eq!(store.get(&"gone".to_string()), None);
994        assert_eq!(
995            Store::<String, u32>::new(eager("other")).get(&"kept".to_string()),
996            Some(&9)
997        );
998    }
999
1000    /// `scan` visits the whole storage owned, without retaining anything —
1001    /// the hydration read for consumers keeping their own index.
1002    #[test_log::test]
1003    #[serial_test::serial]
1004    #[cfg_attr(miri, ignore)]
1005    fn scan_visits_the_storage_without_retaining() {
1006        let dir = tempfile::tempdir().unwrap();
1007        crate::environment::set_root(dir.path());
1008
1009        let mut store = Store::<String, u32>::new(lazy("scan"));
1010        store.insert("a".to_string(), 1).unwrap();
1011        store.insert("b".to_string(), 2).unwrap();
1012
1013        let mut seen = std::vec::Vec::new();
1014        store.scan(|key, value| seen.push((key, value)));
1015        seen.sort();
1016
1017        assert_eq!(seen, std::vec![("a".to_string(), 1), ("b".to_string(), 2)]);
1018        assert!(store.is_empty(), "nothing stays resident after a scan");
1019    }
1020
1021    #[test]
1022    fn in_memory_store_needs_no_storage() {
1023        let mut store = Store::<String, u32>::new(StoreOptions::new());
1024
1025        store.insert("key".to_string(), 1).unwrap();
1026        assert_eq!(store.get(&"key".to_string()), Some(&1));
1027        assert!(store.insert("key".to_string(), 1).is_ok());
1028        assert!(matches!(
1029            store.insert("key".to_string(), 2),
1030            Err(StoreError::DuplicatedKey { .. })
1031        ));
1032
1033        // Nothing durable behind the map: a removed entry is simply gone and
1034        // the key is free again.
1035        assert_eq!(store.remove(&"key".to_string()), Some(1));
1036        store.insert("key".to_string(), 2).unwrap();
1037        assert_eq!(store.get(&"key".to_string()), Some(&2));
1038    }
1039}