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