eidetica/backend/database/in_memory/
mod.rs

1//! In-memory database backend implementation
2//!
3//! This module provides an in-memory implementation of the Database trait,
4//! suitable for testing, development, or scenarios where data persistence
5//! is not strictly required or is handled externally.
6
7mod persistence;
8mod storage;
9mod traversal;
10
11use std::{
12    any::Any,
13    collections::{BTreeMap, HashMap, HashSet},
14    path::Path,
15    sync::{
16        RwLock,
17        atomic::{AtomicUsize, Ordering},
18    },
19};
20
21use async_trait::async_trait;
22use serde::{Deserialize, Serialize};
23
24use crate::{
25    Result,
26    backend::{
27        BackendImpl, InstanceMetadata, InstanceSecrets, RecordMutations, RecordPage, RecordRange,
28        RecordView, StagingToken, StoreStateLifecycle, StoreStateRequest, VerificationStatus,
29        errors::BackendError,
30    },
31    entry::{Entry, ID},
32    snapshot::Snapshot,
33};
34
35use crate::backend::database::sorting;
36
37/// Grouped tree tips cache: (tree_tips, subtree_name -> subtree_tips)
38#[derive(Debug, Clone, Default, Serialize, Deserialize)]
39pub(crate) struct TreeTipsCache {
40    pub(crate) tree_tips: HashSet<ID>,
41    pub(crate) subtree_tips: HashMap<String, HashSet<ID>>,
42}
43
44/// Core data protected by a single lock.
45///
46/// All fields that participate in entry storage and tip tracking are grouped
47/// together to eliminate lock ordering concerns. A single `RwLock` on the
48/// outer `InMemory` struct protects all fields atomically.
49#[derive(Debug)]
50pub(crate) struct InMemoryInner {
51    pub(crate) entries: HashMap<ID, Entry>,
52    pub(crate) store_state_namespaces: HashMap<String, RecordNamespace>,
53    pub(crate) verification_status: HashMap<ID, VerificationStatus>,
54    /// Instance metadata containing device public key and system database IDs.
55    ///
56    /// When `None`, the backend is uninitialized. When `Some`, contains the
57    /// device public key and root IDs for system databases.
58    pub(crate) instance_metadata: Option<InstanceMetadata>,
59    /// Instance secrets containing the device signing key.
60    ///
61    /// **Security Warning**: The signing key is stored in memory without encryption.
62    /// This is suitable for development/testing only. Production systems should use
63    /// proper key management with encryption at rest.
64    pub(crate) instance_secrets: Option<InstanceSecrets>,
65    /// Cached tips grouped by tree: tree_id -> (tree_tips, subtree_name -> subtree_tips)
66    pub(crate) tips: HashMap<ID, TreeTipsCache>,
67}
68
69#[derive(Debug)]
70pub(crate) struct RecordNamespace {
71    request: StoreStateRequest,
72    ready: bool,
73    /// Unlinked by a derived clear: no longer resolvable, still readable
74    /// through views resolved before the clear.
75    unlinked: bool,
76    records: BTreeMap<Vec<u8>, Option<Vec<u8>>>,
77}
78
79/// A simple in-memory database implementation using a `HashMap` for storage.
80///
81/// This database is suitable for testing, development, or scenarios where
82/// data persistence is not strictly required or is handled externally
83/// (e.g., by saving/loading the entire state to/from a file).
84///
85/// It provides basic persistence capabilities via `save_to_file` and
86/// `load_from_file`, serializing the `HashMap` to JSON.
87///
88/// **Security Note**: The device key is stored in memory in plaintext in this implementation.
89/// This is acceptable for development and testing but should not be used in production
90/// without proper encryption or hardware security module integration.
91#[derive(Debug)]
92pub struct InMemory {
93    /// Core data protected by a single lock for atomic access and
94    /// to eliminate lock ordering concerns between entries, verification
95    /// status, and tips.
96    pub(crate) inner: RwLock<InMemoryInner>,
97    store_state_point_reads: AtomicUsize,
98    store_state_scan_reads: AtomicUsize,
99    #[cfg(feature = "testing")]
100    store_history_reads: AtomicUsize,
101}
102
103impl InMemory {
104    #[cfg(feature = "testing")]
105    pub fn store_state_record_count(&self, database: &ID, store: &str) -> usize {
106        self.inner
107            .read()
108            .unwrap()
109            .store_state_namespaces
110            .values()
111            .filter(|record_set| {
112                record_set.ready
113                    && record_set.request.database == *database
114                    && record_set.request.store == store
115                    && matches!(
116                        record_set.request.projection.name.as_str(),
117                        "eidetica/table/rows" | "eidetica/password/eidetica/table/rows"
118                    )
119            })
120            .map(|record_set| record_set.records.len())
121            .max()
122            .unwrap_or(0)
123    }
124
125    #[cfg(feature = "testing")]
126    pub fn store_state_records(
127        &self,
128        database: &ID,
129        store: &str,
130    ) -> Option<crate::backend::RecordMutations> {
131        self.inner
132            .read()
133            .unwrap()
134            .store_state_namespaces
135            .values()
136            .find(|namespace| {
137                namespace.ready
138                    && namespace.request.database == *database
139                    && namespace.request.store == store
140            })
141            .map(|namespace| namespace.records.clone())
142    }
143
144    #[cfg(feature = "testing")]
145    pub fn store_state_read_counts(&self) -> (usize, usize) {
146        (
147            self.store_state_point_reads.load(Ordering::Relaxed),
148            self.store_state_scan_reads.load(Ordering::Relaxed),
149        )
150    }
151
152    #[cfg(feature = "testing")]
153    pub fn store_history_read_count(&self) -> usize {
154        self.store_history_reads.load(Ordering::Relaxed)
155    }
156}
157
158impl InMemory {
159    /// Creates a new, empty `InMemory` database.
160    pub fn new() -> Self {
161        Self {
162            inner: RwLock::new(InMemoryInner {
163                entries: HashMap::new(),
164                store_state_namespaces: HashMap::new(),
165                verification_status: HashMap::new(),
166                instance_metadata: None,
167                instance_secrets: None,
168                tips: HashMap::new(),
169            }),
170            store_state_point_reads: AtomicUsize::new(0),
171            store_state_scan_reads: AtomicUsize::new(0),
172            #[cfg(feature = "testing")]
173            store_history_reads: AtomicUsize::new(0),
174        }
175    }
176
177    /// Returns a vector containing the IDs of all entries currently stored in the database.
178    pub async fn all_ids(&self) -> Vec<ID> {
179        let inner = self.inner.read().unwrap();
180        inner.entries.keys().cloned().collect()
181    }
182
183    /// Saves the entire database state (all entries) to a specified file as JSON.
184    ///
185    /// The write is atomic on POSIX (writes to `<path>.tmp` then renames
186    /// into place). On Windows the final rename is not atomic when the
187    /// destination already exists.
188    ///
189    /// This is synchronous — the body is just `std::fs::write` + `rename`
190    /// with no await points — so it's safe to call from `Drop` impls and
191    /// other non-async contexts. Callers on a tokio runtime should be
192    /// aware that the write briefly blocks the calling worker thread.
193    ///
194    /// # Arguments
195    /// * `path` - The path to the file where the state should be saved.
196    ///
197    /// # Returns
198    /// A `Result` indicating success or an I/O or serialization error.
199    pub fn save_to_file<P: AsRef<Path>>(&self, path: P) -> Result<()> {
200        persistence::save_to_file(self, path)
201    }
202
203    /// Loads the database state from a specified JSON file.
204    ///
205    /// If the file does not exist, a new, empty `InMemory` database is returned.
206    /// Callers that need to tell "missing" apart from "loaded empty" should
207    /// use [`Self::try_load_from_file`].
208    ///
209    /// # Arguments
210    /// * `path` - The path to the file from which to load the state.
211    ///
212    /// # Returns
213    /// A `Result` containing the loaded `InMemory` database or an I/O or deserialization error.
214    pub async fn load_from_file<P: AsRef<Path>>(path: P) -> Result<Self> {
215        persistence::load_from_file(path)
216    }
217
218    /// Like [`Self::load_from_file`], but returns `Ok(None)` when the file
219    /// does not exist instead of falling back to an empty backend. Lets
220    /// callers distinguish "no snapshot yet" from "snapshot loaded as
221    /// empty" without a separate `path.exists()` round-trip (and the TOCTOU
222    /// window that comes with it).
223    pub async fn try_load_from_file<P: AsRef<Path>>(path: P) -> Result<Option<Self>> {
224        persistence::try_load_from_file(path)
225    }
226
227    /// Sort entries by their height within a tree (exposed for testing)
228    ///
229    /// Heights are stored directly in entries, so this just reads and sorts.
230    ///
231    /// # Arguments
232    /// * `_tree` - The ID of the tree context (unused, kept for API compatibility)
233    /// * `entries` - The vector of entries to be sorted in place
234    pub fn sort_entries_by_height(&self, _tree: &ID, entries: &mut [Entry]) {
235        sorting::sort_entries_by_height(entries)
236    }
237
238    /// Sort entries by their height within a subtree (exposed for testing)
239    ///
240    /// Heights are stored directly in entries, so this just reads and sorts.
241    ///
242    /// # Arguments
243    /// * `_tree` - The ID of the tree context (unused, kept for API compatibility)
244    /// * `subtree` - The name of the subtree context
245    /// * `entries` - The vector of entries to be sorted in place
246    pub fn sort_entries_by_subtree_height(&self, _tree: &ID, subtree: &str, entries: &mut [Entry]) {
247        sorting::sort_entries_by_store_height(subtree, entries)
248    }
249
250    /// Check if an entry is a tip within its tree (exposed for benchmarks)
251    ///
252    /// An entry is a tip if no other entry in the same tree lists it as a parent.
253    ///
254    /// # Arguments
255    /// * `tree` - The ID of the tree to check within
256    /// * `entry_id` - The ID of the entry to check
257    ///
258    /// # Returns
259    /// `true` if the entry is a tip, `false` otherwise
260    pub async fn is_tip(&self, tree: &ID, entry_id: &ID) -> bool {
261        let inner = self.inner.read().unwrap();
262        storage::is_tip(&inner.entries, tree, entry_id)
263    }
264}
265
266impl Default for InMemory {
267    fn default() -> Self {
268        Self::new()
269    }
270}
271
272#[async_trait]
273impl BackendImpl for InMemory {
274    async fn resolve_store_state(&self, request: &StoreStateRequest) -> Result<Option<RecordView>> {
275        let inner = self.inner.read().unwrap();
276        Ok(inner
277            .store_state_namespaces
278            .iter()
279            .find(|(_, namespace)| {
280                namespace.ready && !namespace.unlinked && namespace.request == *request
281            })
282            .map(|(namespace_id, _)| RecordView {
283                namespace_id: namespace_id.clone(),
284            }))
285    }
286
287    async fn begin_store_state_staging(
288        &self,
289        mut request: StoreStateRequest,
290    ) -> Result<StagingToken> {
291        if request.lifecycle == StoreStateLifecycle::Staging {
292            return Err(BackendError::InvalidStoreStateStagingToken.into());
293        }
294        let target = request.clone();
295        request.lifecycle = StoreStateLifecycle::Staging;
296        let namespace_id = uuid::Uuid::new_v4().to_string();
297        self.inner.write().unwrap().store_state_namespaces.insert(
298            namespace_id.clone(),
299            RecordNamespace {
300                request,
301                ready: false,
302                unlinked: false,
303                records: BTreeMap::new(),
304            },
305        );
306        Ok(StagingToken {
307            namespace_id,
308            target,
309        })
310    }
311
312    async fn stage_store_state_records(
313        &self,
314        token: &StagingToken,
315        records: RecordMutations,
316    ) -> Result<()> {
317        let mut inner = self.inner.write().unwrap();
318        let namespace = inner
319            .store_state_namespaces
320            .get_mut(&token.namespace_id)
321            .ok_or(BackendError::InvalidStoreStateStagingToken)?;
322        if namespace.ready || namespace.request.lifecycle != StoreStateLifecycle::Staging {
323            return Err(BackendError::StoreStateNamespaceImmutable.into());
324        }
325        namespace.records.extend(records);
326        Ok(())
327    }
328
329    async fn publish_store_state(&self, token: StagingToken) -> Result<RecordView> {
330        let mut inner = self.inner.write().unwrap();
331        // A repeat publish of an already-published token is idempotent: the
332        // namespace is still ready for the same target, so hand back its view.
333        // Removing it first would turn a retry into data loss.
334        if inner
335            .store_state_namespaces
336            .get(&token.namespace_id)
337            .is_some_and(|namespace| namespace.ready && namespace.request == token.target)
338        {
339            return Ok(RecordView {
340                namespace_id: token.namespace_id,
341            });
342        }
343        let staged = inner.store_state_namespaces.remove(&token.namespace_id);
344        // A concurrent materializer may have made this exact target ready
345        // first. Both derived the same state from the same source, so adopt the
346        // winner and discard this namespace rather than failing the loser.
347        if let Some((winner_id, _)) = inner
348            .store_state_namespaces
349            .iter()
350            .find(|(_, ready)| ready.ready && !ready.unlinked && ready.request == token.target)
351        {
352            return Ok(RecordView {
353                namespace_id: winner_id.clone(),
354            });
355        }
356        let mut namespace = staged.ok_or(BackendError::InvalidStoreStateStagingToken)?;
357        // A ready namespace is never removed: reaching here with one means the
358        // token's target no longer matches, which is an error that must not
359        // destroy published state.
360        if namespace.ready {
361            inner
362                .store_state_namespaces
363                .insert(token.namespace_id.clone(), namespace);
364            return Err(BackendError::InvalidStoreStateStagingToken.into());
365        }
366        if namespace.request.lifecycle != StoreStateLifecycle::Staging {
367            return Err(BackendError::InvalidStoreStateStagingToken.into());
368        }
369        if namespace.records.values().any(Option::is_none) {
370            return Err(BackendError::InvalidStoreStateStagingToken.into());
371        }
372        namespace.request = token.target;
373        namespace.ready = true;
374        inner
375            .store_state_namespaces
376            .insert(token.namespace_id.clone(), namespace);
377        Ok(RecordView {
378            namespace_id: token.namespace_id,
379        })
380    }
381
382    async fn abort_store_state(&self, token: StagingToken) -> Result<()> {
383        let mut inner = self.inner.write().unwrap();
384        if inner
385            .store_state_namespaces
386            .get(&token.namespace_id)
387            .is_some_and(|namespace| !namespace.ready)
388        {
389            inner.store_state_namespaces.remove(&token.namespace_id);
390        }
391        Ok(())
392    }
393
394    async fn store_state_record_get(
395        &self,
396        view: &RecordView,
397        key: &[u8],
398    ) -> Result<Option<Vec<u8>>> {
399        self.store_state_point_reads.fetch_add(1, Ordering::Relaxed);
400        let inner = self.inner.read().unwrap();
401        let namespace = inner
402            .store_state_namespaces
403            .get(&view.namespace_id)
404            .filter(|namespace| namespace.ready)
405            .ok_or(BackendError::InvalidStoreStateView)?;
406        Ok(namespace.records.get(key).and_then(Clone::clone))
407    }
408
409    async fn store_state_record_scan(
410        &self,
411        view: &RecordView,
412        range: &RecordRange,
413        after: Option<&[u8]>,
414        limit: usize,
415    ) -> Result<RecordPage> {
416        let inner = self.inner.read().unwrap();
417        let namespace = inner
418            .store_state_namespaces
419            .get(&view.namespace_id)
420            .filter(|namespace| namespace.ready)
421            .ok_or(BackendError::InvalidStoreStateView)?;
422        if limit == 0 {
423            return Ok(RecordPage::default());
424        }
425        self.store_state_scan_reads.fetch_add(1, Ordering::Relaxed);
426        let mut records = namespace
427            .records
428            .iter()
429            .filter(|(key, value)| {
430                value.is_some()
431                    && range
432                        .start
433                        .as_deref()
434                        .is_none_or(|start| key.as_slice() >= start)
435                    && range.end.as_deref().is_none_or(|end| key.as_slice() < end)
436                    && after.is_none_or(|after| key.as_slice() > after)
437            })
438            .take(limit.saturating_add(1))
439            .filter_map(|(key, value)| value.clone().map(|value| (key.clone(), value)))
440            .collect::<Vec<_>>();
441        let has_more = records.len() > limit;
442        records.truncate(limit);
443        let next = if has_more {
444            records.last().map(|record| record.0.clone())
445        } else {
446            None
447        };
448        Ok(RecordPage { records, next })
449    }
450
451    async fn reset_local_verification(&self) -> Result<()> {
452        let mut inner = self.inner.write().unwrap();
453        for status in inner.verification_status.values_mut() {
454            *status = VerificationStatus::Unverified;
455        }
456        // The ordinary clear unlinks a generation for active readers. A reset
457        // instead requires exclusive offline ownership and drops all disposable
458        // namespaces, including incomplete builds from a prior run.
459        inner.store_state_namespaces.retain(|_, namespace| {
460            namespace.request.lifecycle == StoreStateLifecycle::Authoritative
461        });
462        Ok(())
463    }
464
465    /// Unlink every ready derived namespace and reclaim the previously
466    /// unlinked generation.
467    ///
468    /// Clearing is two-phase because a reader that already resolved a view
469    /// keeps reading through it: unlinking removes the namespace from
470    /// resolution, so the next miss rebuilds, while the records stay readable
471    /// until the following clear reclaims them. Authoritative namespaces are
472    /// never selected.
473    async fn clear_derived_store_state(&self) -> Result<()> {
474        let mut inner = self.inner.write().unwrap();
475        inner.store_state_namespaces.retain(|_, namespace| {
476            !(namespace.unlinked && namespace.request.lifecycle == StoreStateLifecycle::Derived)
477        });
478        for namespace in inner.store_state_namespaces.values_mut() {
479            if namespace.ready && namespace.request.lifecycle == StoreStateLifecycle::Derived {
480                namespace.unlinked = true;
481            }
482        }
483        Ok(())
484    }
485
486    /// Retrieves an entry by its unique content-addressable ID.
487    ///
488    /// # Arguments
489    /// * `id` - The ID of the entry to retrieve.
490    ///
491    /// # Returns
492    /// A `Result` containing the `Entry` if found, or a `DatabaseError::EntryNotFound` otherwise.
493    /// Returns an owned copy to support concurrent access with internal synchronization.
494    async fn get(&self, id: &ID) -> Result<Entry> {
495        let inner = self.inner.read().unwrap();
496        storage::get(&inner, id)
497    }
498
499    /// Gets the verification status of an entry.
500    ///
501    /// # Arguments
502    /// * `id` - The ID of the entry to check.
503    ///
504    /// # Returns
505    /// A `Result` containing the `VerificationStatus` if the entry exists, or a `DatabaseError::VerificationStatusNotFound` otherwise.
506    async fn get_verification_status(&self, id: &ID) -> Result<VerificationStatus> {
507        let inner = self.inner.read().unwrap();
508        inner
509            .verification_status
510            .get(id)
511            .copied()
512            .ok_or_else(|| BackendError::VerificationStatusNotFound { id: id.clone() }.into())
513    }
514
515    async fn put(&self, entry: Entry) -> Result<()> {
516        // Validate before acquiring write lock to fail fast
517        entry.validate()?;
518        let mut inner = self.inner.write().unwrap();
519        storage::put(&mut inner, entry)
520    }
521
522    /// Updates the verification status of an existing entry.
523    ///
524    /// This allows the authentication system to mark entries as verified or failed
525    /// after they have been stored. Useful for batch verification operations.
526    ///
527    /// # Arguments
528    /// * `id` - The ID of the entry to update
529    /// * `verification_status` - The new verification status
530    ///
531    /// # Returns
532    /// A `Result` indicating success or `DatabaseError::EntryNotFound` if the entry doesn't exist.
533    async fn update_verification_status(
534        &self,
535        id: &ID,
536        verification_status: VerificationStatus,
537    ) -> Result<()> {
538        let mut inner = self.inner.write().unwrap();
539        if inner.verification_status.contains_key(id) {
540            inner
541                .verification_status
542                .insert(id.clone(), verification_status);
543            Ok(())
544        } else {
545            Err(BackendError::EntryNotFound { id: id.clone() }.into())
546        }
547    }
548
549    /// Gets all entries with a specific verification status.
550    ///
551    /// This is useful for finding unverified entries that need authentication
552    /// or for security audits.
553    ///
554    /// # Arguments
555    /// * `status` - The verification status to filter by
556    ///
557    /// # Returns
558    /// A `Result` containing a vector of entry IDs with the specified status.
559    async fn get_entries_by_verification_status(
560        &self,
561        status: VerificationStatus,
562    ) -> Result<Vec<ID>> {
563        let inner = self.inner.read().unwrap();
564        let ids = inner
565            .verification_status
566            .iter()
567            .filter(|&(_, entry_status)| *entry_status == status)
568            .map(|(id, _)| id.clone())
569            .collect();
570        Ok(ids)
571    }
572
573    async fn snapshot(&self, tree: &ID) -> Result<Snapshot> {
574        // Fast path: check cache with read lock
575        {
576            let inner = self.inner.read().unwrap();
577            if let Some(cache) = inner.tips.get(tree) {
578                return Ok(Snapshot::new(cache.tree_tips.iter().cloned().collect()));
579            }
580        }
581        // Slow path: compute and cache with write lock
582        let mut inner = self.inner.write().unwrap();
583        traversal::snapshot(&mut inner, tree).map(Snapshot::new)
584    }
585
586    async fn store_snapshot(&self, tree: &ID, subtree: &str) -> Result<Snapshot> {
587        // Fast path: check cache with read lock
588        {
589            let inner = self.inner.read().unwrap();
590            if let Some(cache) = inner.tips.get(tree)
591                && let Some(subtree_tips) = cache.subtree_tips.get(subtree)
592            {
593                return Ok(Snapshot::new(subtree_tips.iter().cloned().collect()));
594            }
595        }
596        // Slow path: compute and cache with write lock
597        let mut inner = self.inner.write().unwrap();
598        traversal::store_snapshot(&mut inner, tree, subtree).map(Snapshot::new)
599    }
600
601    async fn store_snapshot_at(
602        &self,
603        tree: &ID,
604        subtree: &str,
605        main_snapshot: &Snapshot,
606    ) -> Result<Snapshot> {
607        let mut inner = self.inner.write().unwrap();
608        traversal::store_snapshot_at(&mut inner, tree, subtree, main_snapshot.tips())
609            .map(Snapshot::new)
610    }
611
612    /// Retrieves the IDs of all top-level root entries stored in the database.
613    ///
614    /// Top-level roots are entries that are themselves roots of a tree
615    /// (i.e., `entry.is_root()` is true) and are not part of a larger tree structure
616    /// tracked by the backend. These represent the starting points
617    /// of distinct trees managed by the database.
618    ///
619    /// # Returns
620    /// A `Result` containing a vector of top-level root entry IDs or an error.
621    async fn all_roots(&self) -> Result<Vec<ID>> {
622        let inner = self.inner.read().unwrap();
623        let roots: Vec<ID> = inner
624            .entries
625            .values()
626            .filter(|entry| entry.is_root())
627            .map(|entry| entry.id())
628            .collect();
629        Ok(roots)
630    }
631
632    async fn find_merge_base(
633        &self,
634        tree: &ID,
635        subtree: &str,
636        entry_ids: &[ID],
637    ) -> Result<Option<ID>> {
638        let inner = self.inner.read().unwrap();
639        traversal::find_merge_base(&inner, tree, subtree, entry_ids)
640    }
641
642    fn as_any(&self) -> &dyn Any {
643        self
644    }
645
646    async fn get_tree(&self, tree: &ID) -> Result<Vec<Entry>> {
647        let inner = self.inner.read().unwrap();
648        storage::get_tree(&inner, tree)
649    }
650
651    async fn get_store(&self, tree: &ID, subtree: &str) -> Result<Vec<Entry>> {
652        let inner = self.inner.read().unwrap();
653        storage::get_store(&inner, tree, subtree)
654    }
655
656    async fn get_tree_from_tips(&self, tree: &ID, tips: &[ID]) -> Result<Vec<Entry>> {
657        let inner = self.inner.read().unwrap();
658        storage::get_tree_from_tips(&inner, tree, tips)
659    }
660
661    async fn store_at(&self, tree: &ID, subtree: &str, snapshot: &Snapshot) -> Result<Vec<Entry>> {
662        #[cfg(feature = "testing")]
663        self.store_history_reads.fetch_add(1, Ordering::Relaxed);
664        let inner = self.inner.read().unwrap();
665        storage::store_at(&inner, tree, subtree, snapshot.tips())
666    }
667
668    async fn get_instance_metadata(&self) -> Result<Option<InstanceMetadata>> {
669        let inner = self.inner.read().unwrap();
670        Ok(inner.instance_metadata.clone())
671    }
672
673    async fn set_instance_metadata(&self, metadata: &InstanceMetadata) -> Result<()> {
674        let mut inner = self.inner.write().unwrap();
675        inner.instance_metadata = Some(metadata.clone());
676        Ok(())
677    }
678
679    async fn get_instance_secrets(&self) -> Result<Option<InstanceSecrets>> {
680        let inner = self.inner.read().unwrap();
681        Ok(inner.instance_secrets.clone())
682    }
683
684    async fn set_instance_secrets(&self, secrets: &InstanceSecrets) -> Result<()> {
685        let mut inner = self.inner.write().unwrap();
686        inner.instance_secrets = Some(secrets.clone());
687        Ok(())
688    }
689
690    async fn get_sorted_store_parents(
691        &self,
692        tree_id: &ID,
693        entry_id: &ID,
694        subtree: &str,
695    ) -> Result<Vec<ID>> {
696        let inner = self.inner.read().unwrap();
697        traversal::get_sorted_store_parents(&inner, tree_id, entry_id, subtree)
698    }
699
700    async fn get_path_from_to(
701        &self,
702        tree_id: &ID,
703        subtree: &str,
704        from_id: Option<&ID>,
705        to_ids: &[ID],
706    ) -> Result<Vec<ID>> {
707        let inner = self.inner.read().unwrap();
708        traversal::get_path_from_to(&inner, tree_id, subtree, from_id, to_ids)
709    }
710}
711
712/// A publish carrying a token whose target does not match the namespace it
713/// names must never disturb a ready namespace.
714///
715/// `StagingToken` is `Clone` with crate-visible fields, so a caller can hold
716/// a token for one target while naming another namespace (or vice versa).
717/// Publishing such a malformed clone either adopts the already-ready winner
718/// for the claimed target or fails with the ready namespace re-inserted —
719/// the ready snapshot and its records always survive.
720#[cfg(test)]
721mod store_state_token_tests {
722    use std::collections::BTreeMap;
723
724    use super::InMemory;
725    use crate::backend::{
726        BackendImpl, CacheScope, ProjectionDescriptor, StagingToken, StoreStateLifecycle,
727        StoreStateRequest,
728    };
729    use crate::entry::ID;
730
731    fn request(store: &str) -> StoreStateRequest {
732        StoreStateRequest {
733            database: ID::from_bytes("db"),
734            store: store.to_string(),
735            lifecycle: StoreStateLifecycle::Derived,
736            scope: CacheScope::Shared,
737            projection: ProjectionDescriptor {
738                name: "test/opaque".to_string(),
739                version: 0,
740            },
741            source_key: b"snapshot".to_vec(),
742        }
743    }
744
745    #[tokio::test]
746    async fn mismatched_target_publish_preserves_ready_namespace() {
747        let backend = InMemory::new();
748
749        // Ready namespace for target A.
750        let request_a = request("store-a");
751        let token_a = backend
752            .begin_store_state_staging(request_a.clone())
753            .await
754            .unwrap();
755        backend
756            .stage_store_state_records(
757                &token_a,
758                BTreeMap::from([(b"key".to_vec(), Some(b"value-a".to_vec()))]),
759            )
760            .await
761            .unwrap();
762        let view_a = backend.publish_store_state(token_a).await.unwrap();
763
764        // Staging namespace for target B.
765        let request_b = request("store-b");
766        let token_b = backend
767            .begin_store_state_staging(request_b.clone())
768            .await
769            .unwrap();
770        backend
771            .stage_store_state_records(
772                &token_b,
773                BTreeMap::from([(b"key".to_vec(), Some(b"value-b".to_vec()))]),
774            )
775            .await
776            .unwrap();
777
778        // Malformed clone: B's namespace id, A's target. The ready winner for
779        // A is adopted and A's snapshot is untouched.
780        let bad = StagingToken {
781            namespace_id: token_b.namespace_id.clone(),
782            target: request_a.clone(),
783        };
784        let adopted = backend.publish_store_state(bad).await.unwrap();
785        assert_eq!(adopted, view_a);
786        assert_eq!(
787            backend
788                .store_state_record_get(&view_a, b"key")
789                .await
790                .unwrap(),
791            Some(b"value-a".to_vec())
792        );
793        assert_eq!(backend.resolve_store_state(&request_b).await.unwrap(), None);
794
795        // Malformed clone: A's (ready) namespace id with a target that
796        // resolves nowhere. The publish fails and the ready namespace is
797        // re-inserted with its records intact.
798        let nowhere = request("store-nowhere");
799        let bad_ready = StagingToken {
800            namespace_id: view_a.namespace_id.clone(),
801            target: nowhere.clone(),
802        };
803        assert!(backend.publish_store_state(bad_ready).await.is_err());
804        assert_eq!(
805            backend.resolve_store_state(&request_a).await.unwrap(),
806            Some(view_a.clone())
807        );
808        assert_eq!(
809            backend
810                .store_state_record_get(&view_a, b"key")
811                .await
812                .unwrap(),
813            Some(b"value-a".to_vec())
814        );
815        assert_eq!(backend.resolve_store_state(&nowhere).await.unwrap(), None);
816    }
817}
eidetica/backend/database/in_memory/mod.rs

eidetica/backend/database/in_memory/
mod.rs
