icydb_core/db/index/
store.rs

1//! Module: index::store
2//! Responsibility: journaled-or-heap index-entry storage behind the index-store boundary.
3//! Does not own: range-scan resolution, continuation semantics, or predicate execution.
4//! Boundary: scan/executor layers depend on this storage boundary.
5
6use crate::db::index::{IndexId, IndexKeyKind};
7use crate::db::{
8    direction::Direction,
9    index::{IndexEntryValue, cardinality::IndexPrefixCardinality, key::RawIndexStoreKey},
10    ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
11    positioned_overlay::{
12        JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
13    },
14};
15
16use candid::CandidType;
17use ic_stable_structures::{
18    BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
19};
20use serde::Deserialize;
21#[cfg(any(test, all(feature = "sql", feature = "diagnostics")))]
22use std::cell::Cell;
23use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
24use std::ops::Bound;
25
26#[cfg(test)]
27thread_local! {
28    static JOURNALED_SNAPSHOT_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
29}
30
31#[cfg(all(feature = "sql", feature = "diagnostics"))]
32thread_local! {
33    static INDEX_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
34    static INDEX_STORE_RANGE_SCAN_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
35    static INDEX_STORE_ENTRY_READ_COUNT: Cell<u64> = const { Cell::new(0) };
36}
37
38#[cfg(all(feature = "sql", feature = "diagnostics"))]
39fn record_index_store_get_call() {
40    INDEX_STORE_GET_CALL_COUNT.with(|count| {
41        count.set(count.get().saturating_add(1));
42    });
43}
44
45#[cfg(all(feature = "sql", feature = "diagnostics"))]
46fn record_index_store_range_scan_call() {
47    INDEX_STORE_RANGE_SCAN_CALL_COUNT.with(|count| {
48        count.set(count.get().saturating_add(1));
49    });
50}
51
52#[cfg(all(feature = "sql", feature = "diagnostics"))]
53fn record_index_store_entry_read() {
54    INDEX_STORE_ENTRY_READ_COUNT.with(|count| {
55        count.set(count.get().saturating_add(1));
56    });
57}
58
59fn visit_index_store_entry<E>(
60    key: &RawIndexStoreKey,
61    value: &IndexEntryValue,
62    visit: &mut impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<bool, E>,
63) -> Result<bool, E> {
64    #[cfg(all(feature = "sql", feature = "diagnostics"))]
65    record_index_store_entry_read();
66
67    visit(key, value)
68}
69
70#[cfg(test)]
71fn record_journaled_snapshot_call() {
72    JOURNALED_SNAPSHOT_CALL_COUNT.with(|count| {
73        count.set(count.get().saturating_add(1));
74    });
75}
76
77#[cfg(test)]
78fn reset_journaled_snapshot_call_count_for_tests() {
79    JOURNALED_SNAPSHOT_CALL_COUNT.with(|count| count.set(0));
80}
81
82#[cfg(test)]
83fn journaled_snapshot_call_count_for_tests() -> u64 {
84    JOURNALED_SNAPSHOT_CALL_COUNT.with(Cell::get)
85}
86
87//
88// IndexState
89//
90// Explicit lifecycle visibility state for one index store.
91// Visibility matters because planner-visible indexes must already be complete:
92// the index contents are fully built and query-visible for reads.
93//
94#[derive(CandidType, Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
95pub enum IndexState {
96    Building,
97    #[default]
98    Ready,
99}
100
101impl IndexState {
102    /// Return the stable lowercase text label for this lifecycle state.
103    #[must_use]
104    pub const fn as_str(self) -> &'static str {
105        match self {
106            Self::Building => "building",
107            Self::Ready => "ready",
108        }
109    }
110}
111
112///
113/// IndexStore
114///
115/// Thin persistence wrapper over one journaled or heap BTreeMap.
116///
117/// Invariant: callers provide already-validated `RawIndexStoreKey`/`IndexEntryValue`.
118///
119
120pub struct IndexStore {
121    pub(super) backend: IndexStoreBackend,
122    generation: u64,
123    state: IndexState,
124    access_state_revision: u64,
125    prefix_cardinality: IndexPrefixCardinality,
126}
127
128pub(super) enum IndexStoreBackend {
129    Heap(HeapBTreeMap<RawIndexStoreKey, IndexEntryValue>),
130    Journaled {
131        canonical:
132            StableBTreeMap<RawIndexStoreKey, IndexEntryValue, VirtualMemory<DefaultMemoryImpl>>,
133        live: HeapBTreeMap<RawIndexStoreKey, IndexEntryValue>,
134        tombstones: BTreeSet<RawIndexStoreKey>,
135        positions: PositionedOverlayMetadata<RawIndexStoreKey>,
136    },
137}
138
139/// Preflighted provenance publication for explicit journal index records.
140pub(in crate::db) struct PreparedIndexPositionPublication {
141    keys: Vec<RawIndexStoreKey>,
142    position: JournalOverlayPosition,
143}
144
145/// Preflighted exact retirement for one complete journal batch.
146pub(in crate::db) struct PreparedIndexPositionRetirement {
147    entries: Vec<(RawIndexStoreKey, PositionedOverlayRetirement)>,
148}
149
150/// Control-flow result for index-store traversal visitors.
151#[derive(Clone, Copy, Debug, Eq, PartialEq)]
152pub(in crate::db) enum IndexStoreVisit {
153    Continue,
154    Stop,
155}
156
157impl IndexStoreVisit {
158    const fn should_stop(self) -> bool {
159        matches!(self, Self::Stop)
160    }
161}
162
163impl IndexStore {
164    /// Initialize a volatile heap-backed index store.
165    #[must_use]
166    pub const fn init_heap() -> Self {
167        Self {
168            backend: IndexStoreBackend::Heap(HeapBTreeMap::new()),
169            generation: 0,
170            state: IndexState::Ready,
171            access_state_revision: 1,
172            prefix_cardinality: IndexPrefixCardinality::synchronized_empty(),
173        }
174    }
175
176    /// Initialize a journaled cached-stable index store.
177    ///
178    /// Normal writes update only the live materialized projection. The
179    /// canonical stable index is updated by future fold/rebuild paths.
180    #[must_use]
181    pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
182        let canonical = StableBTreeMap::init(memory);
183        let prefix_cardinality = if canonical.is_empty() {
184            IndexPrefixCardinality::synchronized_empty()
185        } else {
186            IndexPrefixCardinality::unavailable()
187        };
188        Self {
189            backend: IndexStoreBackend::Journaled {
190                canonical,
191                live: HeapBTreeMap::new(),
192                tombstones: BTreeSet::new(),
193                positions: PositionedOverlayMetadata::new(),
194            },
195            generation: 0,
196            state: IndexState::Ready,
197            access_state_revision: 1,
198            // Exact zero cardinality is known for an empty canonical map;
199            // populated maps remain unavailable without a startup scan.
200            prefix_cardinality,
201        }
202    }
203
204    /// Visit all index entries in canonical store order without exposing the
205    /// backing stable-map iterator.
206    pub(in crate::db) fn visit_entries<E>(
207        &self,
208        mut visitor: impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<IndexStoreVisit, E>,
209    ) -> Result<(), E> {
210        match &self.backend {
211            IndexStoreBackend::Heap(map) => {
212                for (key, value) in map {
213                    #[cfg(all(feature = "sql", feature = "diagnostics"))]
214                    record_index_store_entry_read();
215
216                    if visitor(key, value)?.should_stop() {
217                        return Ok(());
218                    }
219                }
220            }
221            IndexStoreBackend::Journaled { .. } => self.visit_journaled_entries_in_range(
222                (&Bound::Unbounded, &Bound::Unbounded),
223                Direction::Asc,
224                |key, value| visitor(key, value).map(IndexStoreVisit::should_stop),
225            )?,
226        }
227
228        Ok(())
229    }
230
231    pub(in crate::db) fn get(&self, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
232        #[cfg(all(feature = "sql", feature = "diagnostics"))]
233        record_index_store_get_call();
234
235        match &self.backend {
236            IndexStoreBackend::Heap(map) => map.get(key).cloned(),
237            IndexStoreBackend::Journaled { .. } => Self::journaled_get(&self.backend, key),
238        }
239    }
240
241    /// Load one index entry from the canonical predecessor view.
242    pub(in crate::db) fn get_canonical(&self, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
243        match &self.backend {
244            IndexStoreBackend::Heap(map) => map.get(key).cloned(),
245            IndexStoreBackend::Journaled { canonical, .. } => canonical.get(key),
246        }
247    }
248
249    pub fn len(&self) -> u64 {
250        match &self.backend {
251            IndexStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
252            IndexStoreBackend::Journaled { .. } => {
253                let mut count = 0_u64;
254                let _: Result<(), std::convert::Infallible> = self.visit_entries(|_key, _value| {
255                    count = count.saturating_add(1);
256                    Ok(IndexStoreVisit::Continue)
257                });
258                count
259            }
260        }
261    }
262
263    pub fn is_empty(&self) -> bool {
264        match &self.backend {
265            IndexStoreBackend::Heap(map) => map.is_empty(),
266            IndexStoreBackend::Journaled { .. } => {
267                let mut empty = true;
268                let _: Result<(), std::convert::Infallible> = self.visit_entries(|_key, _value| {
269                    empty = false;
270                    Ok(IndexStoreVisit::Stop)
271                });
272                empty
273            }
274        }
275    }
276
277    #[must_use]
278    pub(in crate::db) const fn generation(&self) -> u64 {
279        self.generation
280    }
281
282    /// Return the explicit lifecycle state for this index store.
283    #[must_use]
284    pub(in crate::db) const fn state(&self) -> IndexState {
285        self.state
286    }
287
288    /// Return the current physical access-readiness revision.
289    #[must_use]
290    pub(in crate::db) const fn access_state_revision(&self) -> u64 {
291        self.access_state_revision
292    }
293
294    /// Return an exact user-index prefix count when the index metadata is
295    /// synchronized with the caller's authoritative row-store generation.
296    #[must_use]
297    pub(in crate::db) fn exact_prefix_cardinality(
298        &self,
299        data_generation: u64,
300        key_kind: IndexKeyKind,
301        index_id: IndexId,
302        components: &[Vec<u8>],
303    ) -> Option<u64> {
304        self.prefix_cardinality
305            .exact_count(data_generation, key_kind, index_id, components)
306    }
307
308    /// Return the sum of exact prefix counts for prefixes on the same index
309    /// when synchronized metadata can prove all requested counts.
310    #[must_use]
311    pub(in crate::db) fn exact_prefix_cardinality_sum<'a>(
312        &self,
313        data_generation: u64,
314        key_kind: IndexKeyKind,
315        index_id: IndexId,
316        component_prefixes: impl IntoIterator<Item = &'a [Vec<u8>]>,
317        stop_after: Option<u64>,
318    ) -> Option<u64> {
319        self.prefix_cardinality.exact_count_sum(
320            data_generation,
321            key_kind,
322            index_id,
323            component_prefixes,
324            stop_after,
325        )
326    }
327
328    /// Return non-empty exact child prefixes under a sparse set of already-encoded
329    /// parent prefixes when synchronized metadata can prove the bounded child set.
330    #[must_use]
331    pub(in crate::db) fn exact_child_prefixes_for_parent_set<'a>(
332        &self,
333        data_generation: u64,
334        key_kind: IndexKeyKind,
335        index_id: IndexId,
336        parent_component_prefixes: impl IntoIterator<Item = &'a [Vec<u8>]>,
337        max_children: usize,
338    ) -> Option<Vec<Vec<Vec<u8>>>> {
339        self.prefix_cardinality.exact_child_prefixes_for_parent_set(
340            data_generation,
341            key_kind,
342            index_id,
343            parent_component_prefixes,
344            max_children,
345        )
346    }
347
348    /// Mark prefix-cardinality metadata synchronized with the authoritative
349    /// row-store generation after a committed row/index transition.
350    pub(in crate::db) const fn mark_prefix_cardinality_data_generation(&mut self, generation: u64) {
351        self.prefix_cardinality.mark_synchronized(generation);
352    }
353
354    /// Mark this index store as in-progress and therefore ineligible for
355    /// planner visibility until a full authoritative rebuild ends.
356    pub(in crate::db) const fn set_access_state(&mut self, state: IndexState, revision: u64) {
357        self.state = state;
358        self.access_state_revision = revision;
359    }
360
361    pub(crate) fn insert(
362        &mut self,
363        key: RawIndexStoreKey,
364        entry: IndexEntryValue,
365    ) -> Option<IndexEntryValue> {
366        let previous_journaled = if matches!(self.backend, IndexStoreBackend::Journaled { .. }) {
367            self.get(&key)
368        } else {
369            None
370        };
371        let cardinality_key = key.clone();
372        let previous = match &mut self.backend {
373            IndexStoreBackend::Heap(map) => map.insert(key, entry.clone()),
374            IndexStoreBackend::Journaled {
375                live, tombstones, ..
376            } => {
377                tombstones.remove(&key);
378                live.insert(key, entry.clone());
379                previous_journaled
380            }
381        };
382        self.prefix_cardinality
383            .apply_insert(&cardinality_key, previous.as_ref(), &entry);
384        self.bump_generation();
385        previous
386    }
387
388    /// Insert one key whose absence was proved by complete-domain staging.
389    ///
390    /// Accepted-schema replacement first removes its complete current user
391    /// domain. Raw index identity makes every final key owner-local, so no
392    /// canonical point lookup can add information during mechanical Apply.
393    pub(in crate::db) fn insert_preflighted_absent(
394        &mut self,
395        key: RawIndexStoreKey,
396        entry: IndexEntryValue,
397    ) {
398        let cardinality_key = key.clone();
399        match &mut self.backend {
400            IndexStoreBackend::Heap(map) => {
401                map.insert(key, entry.clone());
402            }
403            IndexStoreBackend::Journaled {
404                live, tombstones, ..
405            } => {
406                tombstones.remove(&key);
407                live.insert(key, entry.clone());
408            }
409        }
410        self.prefix_cardinality
411            .apply_insert(&cardinality_key, None, &entry);
412        self.bump_generation();
413    }
414
415    pub(crate) fn remove(&mut self, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
416        let previous_journaled = if matches!(self.backend, IndexStoreBackend::Journaled { .. }) {
417            self.get(key)
418        } else {
419            None
420        };
421        let previous = match &mut self.backend {
422            IndexStoreBackend::Heap(map) => map.remove(key),
423            IndexStoreBackend::Journaled {
424                live, tombstones, ..
425            } => {
426                live.remove(key);
427                tombstones.insert(key.clone());
428                previous_journaled
429            }
430        };
431        self.prefix_cardinality.apply_remove(key, previous.as_ref());
432        self.bump_generation();
433        previous
434    }
435
436    /// Reset the disposable journaled index overlay without traversing or
437    /// mutating the canonical stable index.
438    pub(in crate::db) fn reset_journaled_live_projection(
439        &mut self,
440        data_generation: u64,
441    ) -> Result<(), crate::error::InternalError> {
442        let IndexStoreBackend::Journaled {
443            canonical,
444            live,
445            tombstones,
446            positions,
447        } = &mut self.backend
448        else {
449            return Err(crate::error::InternalError::store_invariant());
450        };
451
452        live.clear();
453        tombstones.clear();
454        positions.clear();
455        self.prefix_cardinality = if canonical.is_empty() {
456            let mut cardinality = IndexPrefixCardinality::synchronized_empty();
457            cardinality.mark_synchronized(data_generation);
458            cardinality
459        } else {
460            IndexPrefixCardinality::unavailable()
461        };
462        self.bump_generation();
463
464        Ok(())
465    }
466
467    /// Publish one preflighted positioned derived or explicit index effect.
468    pub(in crate::db) fn publish_preflighted_journal_entry(
469        &mut self,
470        key: RawIndexStoreKey,
471        value: Option<IndexEntryValue>,
472        position: JournalOverlayPosition,
473    ) -> Result<Option<IndexEntryValue>, crate::error::InternalError> {
474        let IndexStoreBackend::Journaled {
475            canonical,
476            live,
477            tombstones,
478            positions,
479        } = &mut self.backend
480        else {
481            return Err(crate::error::InternalError::store_invariant());
482        };
483        let previous = if tombstones.contains(&key) {
484            None
485        } else {
486            live.get(&key).cloned().or_else(|| canonical.get(&key))
487        };
488        let cardinality_key = key.clone();
489
490        if let Some(value) = value {
491            tombstones.remove(&key);
492            live.insert(key.clone(), value.clone());
493            self.prefix_cardinality
494                .apply_insert(&cardinality_key, previous.as_ref(), &value);
495        } else {
496            live.remove(&key);
497            tombstones.insert(key.clone());
498            self.prefix_cardinality
499                .apply_remove(&cardinality_key, previous.as_ref());
500        }
501        positions.publish_preflighted(key, position);
502        self.bump_generation();
503
504        Ok(previous)
505    }
506
507    /// Validate and publish one positioned index effect for direct store tests.
508    #[cfg(test)]
509    pub(in crate::db) fn publish_positioned_journal_entry(
510        &mut self,
511        key: RawIndexStoreKey,
512        value: Option<IndexEntryValue>,
513        position: JournalOverlayPosition,
514    ) -> Result<Option<IndexEntryValue>, crate::error::InternalError> {
515        self.preflight_positioned_journal_entry(&key, position)?;
516        self.publish_preflighted_journal_entry(key, value, position)
517    }
518
519    /// Preflight index provenance before marker publication.
520    pub(in crate::db) fn preflight_positioned_journal_entry(
521        &self,
522        key: &RawIndexStoreKey,
523        position: JournalOverlayPosition,
524    ) -> Result<(), crate::error::InternalError> {
525        let IndexStoreBackend::Journaled { positions, .. } = &self.backend else {
526            return Err(crate::error::InternalError::store_invariant());
527        };
528        positions.preflight_publish(key, position)
529    }
530
531    /// Preflight explicit index provenance before marker publication.
532    pub(in crate::db) fn prepare_position_publication(
533        &self,
534        keys: impl IntoIterator<Item = RawIndexStoreKey>,
535        position: JournalOverlayPosition,
536    ) -> Result<PreparedIndexPositionPublication, crate::error::InternalError> {
537        let IndexStoreBackend::Journaled { positions, .. } = &self.backend else {
538            return Err(crate::error::InternalError::store_invariant());
539        };
540        let keys = keys.into_iter().collect::<BTreeSet<_>>();
541        for key in &keys {
542            positions.preflight_publish(key, position)?;
543        }
544        Ok(PreparedIndexPositionPublication {
545            keys: keys.into_iter().collect(),
546            position,
547        })
548    }
549
550    /// Publish explicit index provenance after its values have been applied.
551    pub(in crate::db) fn publish_prepared_positions(
552        &mut self,
553        prepared: PreparedIndexPositionPublication,
554    ) {
555        let IndexStoreBackend::Journaled { positions, .. } = &mut self.backend else {
556            debug_assert!(
557                false,
558                "preflighted index positions require a journaled store"
559            );
560            return;
561        };
562        for key in prepared.keys {
563            positions.publish_preflighted(key, prepared.position);
564        }
565    }
566
567    /// Preflight exact index-overlay retirement before canonical mutation.
568    pub(in crate::db) fn prepare_position_retirement(
569        &self,
570        keys: impl IntoIterator<Item = RawIndexStoreKey>,
571        position: JournalOverlayPosition,
572    ) -> Result<PreparedIndexPositionRetirement, crate::error::InternalError> {
573        let IndexStoreBackend::Journaled { positions, .. } = &self.backend else {
574            return Err(crate::error::InternalError::store_invariant());
575        };
576        let entries = keys
577            .into_iter()
578            .collect::<BTreeSet<_>>()
579            .into_iter()
580            .map(|key| {
581                positions
582                    .preflight_retirement(&key, position)
583                    .map(|retirement| (key, retirement))
584            })
585            .collect::<Result<Vec<_>, _>>()?;
586        Ok(PreparedIndexPositionRetirement { entries })
587    }
588
589    /// Retire only exact index overlays after canonical mutation succeeds.
590    pub(in crate::db) fn apply_prepared_position_retirement(
591        &mut self,
592        prepared: PreparedIndexPositionRetirement,
593    ) {
594        let IndexStoreBackend::Journaled {
595            live,
596            tombstones,
597            positions,
598            ..
599        } = &mut self.backend
600        else {
601            debug_assert!(
602                false,
603                "preflighted index retirement requires a journaled store"
604            );
605            return;
606        };
607        for (key, retirement) in prepared.entries {
608            if retirement == PositionedOverlayRetirement::Exact {
609                live.remove(&key);
610                tombstones.remove(&key);
611                positions.retire_preflighted(&key, retirement);
612            }
613        }
614    }
615
616    #[cfg(test)]
617    fn retire_positioned_journal_effect(
618        &mut self,
619        key: &RawIndexStoreKey,
620        position: JournalOverlayPosition,
621    ) -> Result<PositionedOverlayRetirement, crate::error::InternalError> {
622        let IndexStoreBackend::Journaled { positions, .. } = &self.backend else {
623            return Err(crate::error::InternalError::store_invariant());
624        };
625        let retirement = positions.preflight_retirement(key, position)?;
626        let prepared = PreparedIndexPositionRetirement {
627            entries: vec![(key.clone(), retirement)],
628        };
629        self.apply_prepared_position_retirement(prepared);
630        Ok(retirement)
631    }
632
633    /// Apply one recovered index entry directly to canonical stable storage.
634    pub(in crate::db) fn fold_recovered_journal_entry(
635        &mut self,
636        key: RawIndexStoreKey,
637        value: Option<IndexEntryValue>,
638    ) -> Result<(), crate::error::InternalError> {
639        let IndexStoreBackend::Journaled {
640            canonical,
641            live,
642            tombstones,
643            ..
644        } = &mut self.backend
645        else {
646            return Err(crate::error::InternalError::store_invariant());
647        };
648
649        let visible = !live.contains_key(&key) && !tombstones.contains(&key);
650        let cardinality_key = key.clone();
651        let previous = if let Some(value) = value.as_ref() {
652            canonical.insert(key, value.clone())
653        } else {
654            canonical.remove(&key)
655        };
656        if visible {
657            if let Some(value) = value.as_ref() {
658                self.prefix_cardinality
659                    .apply_insert(&cardinality_key, previous.as_ref(), value);
660            } else {
661                self.prefix_cardinality
662                    .apply_remove(&cardinality_key, previous.as_ref());
663            }
664        }
665        self.bump_generation();
666
667        Ok(())
668    }
669
670    /// Prove that recovered journal entries can be folded into canonical storage.
671    pub(in crate::db) fn preflight_fold_recovered_journal(
672        &self,
673    ) -> Result<(), crate::error::InternalError> {
674        match self.backend {
675            IndexStoreBackend::Journaled { .. } => Ok(()),
676            IndexStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
677        }
678    }
679
680    pub fn clear(&mut self) {
681        match &mut self.backend {
682            IndexStoreBackend::Heap(map) => map.clear(),
683            IndexStoreBackend::Journaled {
684                canonical,
685                live,
686                tombstones,
687                ..
688            } => {
689                live.clear();
690                tombstones.clear();
691                for entry in canonical.iter() {
692                    tombstones.insert(entry.key().clone());
693                }
694            }
695        }
696        self.prefix_cardinality.clear_unsynchronized();
697        self.bump_generation();
698    }
699
700    /// Fold the current journaled materialized index view into the canonical
701    /// stable base and clear volatile projection state.
702    #[cfg(any(test, feature = "migration"))]
703    pub(in crate::db) fn fold_journaled_materialized_view(
704        &mut self,
705    ) -> Result<(), crate::error::InternalError> {
706        let entries = Self::journaled_entries_snapshot_for_fold(&self.backend);
707        let IndexStoreBackend::Journaled {
708            canonical,
709            live,
710            tombstones,
711            ..
712        } = &mut self.backend
713        else {
714            return Err(crate::error::InternalError::store_invariant());
715        };
716
717        canonical.clear_new();
718        for (key, value) in entries {
719            canonical.insert(key, value);
720        }
721        live.clear();
722        tombstones.clear();
723        let data_generation = self.prefix_cardinality.synchronized_generation();
724        self.rebuild_prefix_cardinality_from_entries(data_generation);
725        self.bump_generation();
726
727        Ok(())
728    }
729
730    /// Sum of bytes used by all stored index entries.
731    pub fn memory_bytes(&self) -> u64 {
732        let mut bytes = 0u64;
733        let _: Result<(), std::convert::Infallible> = self.visit_entries(|key, value| {
734            bytes = bytes.saturating_add(key.as_bytes().len() as u64 + value.len() as u64);
735            Ok(IndexStoreVisit::Continue)
736        });
737        bytes
738    }
739
740    /// Return the monotonic perf-only count of index-entry fetches seen by this process.
741    #[cfg(all(feature = "sql", feature = "diagnostics"))]
742    pub(in crate::db) fn current_get_call_count() -> u64 {
743        INDEX_STORE_GET_CALL_COUNT.with(Cell::get)
744    }
745
746    /// Return the monotonic perf-only count of index range traversal probes seen by this process.
747    #[cfg(all(feature = "sql", feature = "diagnostics"))]
748    pub(in crate::db) fn current_range_scan_call_count() -> u64 {
749        INDEX_STORE_RANGE_SCAN_CALL_COUNT.with(Cell::get)
750    }
751
752    /// Return the monotonic perf-only count of index entries yielded by traversal.
753    #[cfg(all(feature = "sql", feature = "diagnostics"))]
754    pub(in crate::db) fn current_entry_read_count() -> u64 {
755        INDEX_STORE_ENTRY_READ_COUNT.with(Cell::get)
756    }
757
758    #[cfg(all(feature = "sql", feature = "diagnostics"))]
759    pub(in crate::db::index) fn record_range_scan_call() {
760        record_index_store_range_scan_call();
761    }
762
763    #[cfg(all(feature = "sql", feature = "diagnostics"))]
764    pub(in crate::db::index) fn record_merged_entry_reads(count: u64) {
765        INDEX_STORE_ENTRY_READ_COUNT.with(|total| {
766            total.set(total.get().saturating_add(count));
767        });
768    }
769
770    const fn bump_generation(&mut self) {
771        self.generation = self.generation.saturating_add(1);
772    }
773
774    #[cfg(any(test, feature = "migration"))]
775    fn rebuild_prefix_cardinality_from_entries(&mut self, data_generation: Option<u64>) {
776        self.prefix_cardinality.clear_unsynchronized();
777        let entries = Self::entries_snapshot_for_cardinality(&self.backend);
778        for (key, value) in &entries {
779            self.prefix_cardinality.apply_insert(key, None, value);
780        }
781        if let Some(data_generation) = data_generation {
782            self.prefix_cardinality.mark_synchronized(data_generation);
783        }
784    }
785
786    #[cfg(any(test, feature = "migration"))]
787    fn entries_snapshot_for_cardinality(
788        backend: &IndexStoreBackend,
789    ) -> HeapBTreeMap<RawIndexStoreKey, IndexEntryValue> {
790        match backend {
791            IndexStoreBackend::Heap(map) => map.clone(),
792            IndexStoreBackend::Journaled { .. } => {
793                Self::journaled_entries_snapshot_for_fold(backend)
794            }
795        }
796    }
797
798    fn journaled_get(
799        backend: &IndexStoreBackend,
800        key: &RawIndexStoreKey,
801    ) -> Option<IndexEntryValue> {
802        let IndexStoreBackend::Journaled {
803            canonical,
804            live,
805            tombstones,
806            ..
807        } = backend
808        else {
809            return None;
810        };
811
812        if tombstones.contains(key) {
813            return None;
814        }
815        live.get(key).cloned().or_else(|| canonical.get(key))
816    }
817
818    #[cfg(any(test, feature = "migration"))]
819    pub(super) fn journaled_entries_snapshot_for_fold(
820        backend: &IndexStoreBackend,
821    ) -> HeapBTreeMap<RawIndexStoreKey, IndexEntryValue> {
822        #[cfg(test)]
823        record_journaled_snapshot_call();
824
825        let IndexStoreBackend::Journaled {
826            canonical,
827            live,
828            tombstones,
829            ..
830        } = backend
831        else {
832            return HeapBTreeMap::new();
833        };
834
835        let mut entries = HeapBTreeMap::new();
836        for entry in canonical.iter() {
837            let key = entry.key().clone();
838            if !tombstones.contains(&key) {
839                entries.insert(key, entry.value());
840            }
841        }
842        for (key, value) in live {
843            if !tombstones.contains(key) {
844                entries.insert(key.clone(), value.clone());
845            }
846        }
847
848        entries
849    }
850
851    pub(super) fn visit_journaled_entries_in_range<E>(
852        &self,
853        bounds: (&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>),
854        direction: Direction,
855        mut visit: impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<bool, E>,
856    ) -> Result<(), E> {
857        let IndexStoreBackend::Journaled {
858            canonical,
859            live,
860            tombstones,
861            ..
862        } = &self.backend
863        else {
864            return Ok(());
865        };
866
867        let lower = bounds.0.clone();
868        let upper = bounds.1.clone();
869        match direction {
870            Direction::Asc if canonical.is_empty() => {
871                for (key, value) in live.range((lower, upper)) {
872                    if visit_index_store_entry(key, value, &mut visit)? {
873                        return Ok(());
874                    }
875                }
876            }
877            Direction::Desc if canonical.is_empty() => {
878                for (key, value) in live.range((lower, upper)).rev() {
879                    if visit_index_store_entry(key, value, &mut visit)? {
880                        return Ok(());
881                    }
882                }
883            }
884            Direction::Asc if live.is_empty() && tombstones.is_empty() => {
885                for entry in canonical.range((lower, upper)) {
886                    if visit_index_store_entry(entry.key(), &entry.value(), &mut visit)? {
887                        return Ok(());
888                    }
889                }
890            }
891            Direction::Desc if live.is_empty() && tombstones.is_empty() => {
892                for entry in canonical.range((lower, upper)).rev() {
893                    if visit_index_store_entry(entry.key(), &entry.value(), &mut visit)? {
894                        return Ok(());
895                    }
896                }
897            }
898            Direction::Asc => {
899                visit_ordered_overlay(
900                    canonical.range((lower.clone(), upper.clone())),
901                    live.range((lower, upper)),
902                    direction,
903                    |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
904                    |canonical_entry| !tombstones.contains(canonical_entry.key()),
905                    |live_entry| !tombstones.contains(live_entry.0),
906                    |entry| {
907                        let should_stop = match entry {
908                            OrderedOverlayEntry::Canonical(canonical_entry) => {
909                                visit_index_store_entry(
910                                    canonical_entry.key(),
911                                    &canonical_entry.value(),
912                                    &mut visit,
913                                )?
914                            }
915                            OrderedOverlayEntry::Live((key, value)) => {
916                                visit_index_store_entry(key, value, &mut visit)?
917                            }
918                        };
919                        Ok(if should_stop {
920                            OrderedOverlayVisit::Stop
921                        } else {
922                            OrderedOverlayVisit::Continue
923                        })
924                    },
925                )?;
926            }
927            Direction::Desc => {
928                visit_ordered_overlay(
929                    canonical.range((lower.clone(), upper.clone())).rev(),
930                    live.range((lower, upper)).rev(),
931                    direction,
932                    |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
933                    |canonical_entry| !tombstones.contains(canonical_entry.key()),
934                    |live_entry| !tombstones.contains(live_entry.0),
935                    |entry| {
936                        let should_stop = match entry {
937                            OrderedOverlayEntry::Canonical(canonical_entry) => {
938                                visit_index_store_entry(
939                                    canonical_entry.key(),
940                                    &canonical_entry.value(),
941                                    &mut visit,
942                                )?
943                            }
944                            OrderedOverlayEntry::Live((key, value)) => {
945                                visit_index_store_entry(key, value, &mut visit)?
946                            }
947                        };
948                        Ok(if should_stop {
949                            OrderedOverlayVisit::Stop
950                        } else {
951                            OrderedOverlayVisit::Continue
952                        })
953                    },
954                )?;
955            }
956        }
957
958        Ok(())
959    }
960}
961
962#[cfg(test)]
963mod tests {
964    use super::*;
965    use crate::{
966        db::{
967            direction::Direction,
968            index::{IndexId, IndexKey, IndexKeyKind},
969            journal::JournalSequence,
970            key_taxonomy::{PrimaryKeyComponent, PrimaryKeyValue},
971            positioned_overlay::{JournalOverlayPosition, PositionedOverlayRetirement},
972            registry::StoreAllocationIdentity,
973        },
974        testing::test_memory,
975        types::EntityTag,
976    };
977    use ic_stable_structures::Storable;
978    use std::{borrow::Cow, convert::Infallible};
979
980    fn raw_key(value: u8) -> RawIndexStoreKey {
981        <RawIndexStoreKey as Storable>::from_bytes(Cow::Owned(vec![value]))
982    }
983
984    fn overlay_position(sequence: u64) -> JournalOverlayPosition {
985        JournalOverlayPosition::new(
986            StoreAllocationIdentity::new(231, "test::index"),
987            JournalSequence::new(sequence),
988        )
989    }
990
991    fn indexed_raw_key(
992        index_id: &IndexId,
993        components: Vec<Vec<u8>>,
994        primary_key: u64,
995    ) -> RawIndexStoreKey {
996        indexed_raw_key_with_kind(index_id, IndexKeyKind::User, components, primary_key)
997    }
998
999    fn indexed_raw_key_with_kind(
1000        index_id: &IndexId,
1001        key_kind: IndexKeyKind,
1002        components: Vec<Vec<u8>>,
1003        primary_key: u64,
1004    ) -> RawIndexStoreKey {
1005        IndexKey::new_from_components_with_primary_key_value(
1006            index_id,
1007            key_kind,
1008            components.as_slice(),
1009            &PrimaryKeyValue::from(PrimaryKeyComponent::Nat64(primary_key)),
1010        )
1011        .expect("test index key should build")
1012        .to_raw()
1013        .expect("test index key should encode")
1014    }
1015
1016    fn malformed_index_entry_value() -> IndexEntryValue {
1017        <IndexEntryValue as Storable>::from_bytes(Cow::Owned(vec![0xFF]))
1018    }
1019
1020    fn missing_index_entry_value() -> IndexEntryValue {
1021        <IndexEntryValue as Storable>::from_bytes(Cow::Owned(vec![1]))
1022    }
1023
1024    #[test]
1025    fn index_prefix_cardinality_requires_explicit_data_generation_sync() {
1026        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1027        let collection = b"collection-a".to_vec();
1028        let draft = b"Draft".to_vec();
1029        let review = b"Review".to_vec();
1030        let mut store = IndexStore::init_heap();
1031
1032        store.insert(
1033            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
1034            IndexEntryValue::presence(),
1035        );
1036        store.insert(
1037            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2),
1038            IndexEntryValue::presence(),
1039        );
1040        store.insert(
1041            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 3),
1042            IndexEntryValue::presence(),
1043        );
1044
1045        assert_eq!(
1046            store.exact_prefix_cardinality(
1047                0,
1048                IndexKeyKind::User,
1049                index_id,
1050                std::slice::from_ref(&collection),
1051            ),
1052            None,
1053            "raw index mutations must not be trusted until row generation sync is stamped",
1054        );
1055
1056        store.mark_prefix_cardinality_data_generation(7);
1057
1058        assert_eq!(
1059            store.exact_prefix_cardinality(
1060                7,
1061                IndexKeyKind::User,
1062                index_id,
1063                std::slice::from_ref(&collection),
1064            ),
1065            Some(3),
1066        );
1067        assert_eq!(
1068            store.exact_prefix_cardinality(
1069                7,
1070                IndexKeyKind::User,
1071                index_id,
1072                &[collection.clone(), draft],
1073            ),
1074            Some(2),
1075        );
1076        assert_eq!(
1077            store.exact_prefix_cardinality(8, IndexKeyKind::User, index_id, &[collection, review],),
1078            None,
1079            "row generation drift should force the caller to use the existing-row fallback",
1080        );
1081    }
1082
1083    #[test]
1084    fn index_prefix_cardinality_enumerates_bounded_child_prefixes() {
1085        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1086        let collection = b"collection-a".to_vec();
1087        let other_collection = b"collection-b".to_vec();
1088        let draft = b"Draft".to_vec();
1089        let review = b"Review".to_vec();
1090        let published = b"Published".to_vec();
1091        let mut store = IndexStore::init_heap();
1092
1093        store.insert(
1094            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
1095            IndexEntryValue::presence(),
1096        );
1097        store.insert(
1098            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2),
1099            IndexEntryValue::presence(),
1100        );
1101        store.insert(
1102            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 3),
1103            IndexEntryValue::presence(),
1104        );
1105        store.insert(
1106            indexed_raw_key(
1107                &index_id,
1108                vec![other_collection.clone(), published.clone()],
1109                4,
1110            ),
1111            IndexEntryValue::presence(),
1112        );
1113        store.mark_prefix_cardinality_data_generation(7);
1114
1115        assert_eq!(
1116            store.exact_child_prefixes_for_parent_set(
1117                7,
1118                IndexKeyKind::User,
1119                index_id,
1120                [std::slice::from_ref(&collection)],
1121                4,
1122            ),
1123            Some(vec![
1124                vec![collection.clone(), draft],
1125                vec![collection.clone(), review],
1126            ]),
1127            "child-prefix enumeration should return deterministic unique children under the requested parent",
1128        );
1129        assert_eq!(
1130            store.exact_child_prefixes_for_parent_set(
1131                7,
1132                IndexKeyKind::User,
1133                index_id,
1134                [std::slice::from_ref(&other_collection)],
1135                4,
1136            ),
1137            Some(vec![vec![other_collection, published]]),
1138            "child-prefix enumeration must stay scoped to the requested parent prefix",
1139        );
1140        assert_eq!(
1141            store.exact_child_prefixes_for_parent_set(
1142                8,
1143                IndexKeyKind::User,
1144                index_id,
1145                [std::slice::from_ref(&collection)],
1146                4,
1147            ),
1148            None,
1149            "row generation drift should keep child-prefix expansion fail-closed",
1150        );
1151        assert_eq!(
1152            store.exact_child_prefixes_for_parent_set(
1153                7,
1154                IndexKeyKind::User,
1155                index_id,
1156                [std::slice::from_ref(&collection)],
1157                1,
1158            ),
1159            None,
1160            "over-cap child-prefix expansion should fall back to the existing route",
1161        );
1162    }
1163
1164    #[test]
1165    fn index_prefix_cardinality_batches_sparse_child_prefixes() {
1166        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1167        let collection = b"collection-a".to_vec();
1168        let other_collection = b"collection-b".to_vec();
1169        let missing_a = b"missing-a".to_vec();
1170        let missing_b = b"missing-b".to_vec();
1171        let draft = b"Draft".to_vec();
1172        let review = b"Review".to_vec();
1173        let published = b"Published".to_vec();
1174        let mut store = IndexStore::init_heap();
1175
1176        store.insert(
1177            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
1178            IndexEntryValue::presence(),
1179        );
1180        store.insert(
1181            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 2),
1182            IndexEntryValue::presence(),
1183        );
1184        store.insert(
1185            indexed_raw_key(
1186                &index_id,
1187                vec![other_collection.clone(), published.clone()],
1188                3,
1189            ),
1190            IndexEntryValue::presence(),
1191        );
1192        store.mark_prefix_cardinality_data_generation(7);
1193
1194        let parents = [
1195            std::slice::from_ref(&missing_a),
1196            std::slice::from_ref(&collection),
1197            std::slice::from_ref(&missing_b),
1198            std::slice::from_ref(&other_collection),
1199        ];
1200        assert_eq!(
1201            store.exact_child_prefixes_for_parent_set(7, IndexKeyKind::User, index_id, parents, 4,),
1202            Some(vec![
1203                vec![collection.clone(), draft],
1204                vec![collection.clone(), review],
1205                vec![other_collection.clone(), published],
1206            ]),
1207            "batched child-prefix enumeration should skip missing sparse parents and return deterministic real children",
1208        );
1209        assert_eq!(
1210            store.exact_child_prefixes_for_parent_set(
1211                7,
1212                IndexKeyKind::User,
1213                index_id,
1214                [
1215                    std::slice::from_ref(&missing_a),
1216                    std::slice::from_ref(&missing_b)
1217                ],
1218                4,
1219            ),
1220            Some(Vec::new()),
1221            "missing-only sparse parent sets should be proven empty when cardinality is synchronized",
1222        );
1223        assert_eq!(
1224            store.exact_child_prefixes_for_parent_set(
1225                7,
1226                IndexKeyKind::User,
1227                index_id,
1228                [
1229                    std::slice::from_ref(&collection),
1230                    std::slice::from_ref(&other_collection)
1231                ],
1232                2,
1233            ),
1234            None,
1235            "over-cap sparse parent-set expansion should fail closed",
1236        );
1237        assert_eq!(
1238            store.exact_child_prefixes_for_parent_set(
1239                8,
1240                IndexKeyKind::User,
1241                index_id,
1242                [std::slice::from_ref(&collection)],
1243                4,
1244            ),
1245            None,
1246            "generation drift should keep batched child-prefix expansion fail-closed",
1247        );
1248    }
1249
1250    #[test]
1251    fn index_prefix_cardinality_ignores_system_index_mutations() {
1252        let user_index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1253        let system_index_id = IndexId::new(EntityTag::new(0xCA7D), 2);
1254        let collection = b"collection-a".to_vec();
1255        let draft = b"Draft".to_vec();
1256        let system_component = b"reverse-edge".to_vec();
1257        let mut store = IndexStore::init_heap();
1258
1259        store.insert(
1260            indexed_raw_key(&user_index_id, vec![collection.clone(), draft.clone()], 1),
1261            IndexEntryValue::presence(),
1262        );
1263        store.mark_prefix_cardinality_data_generation(7);
1264
1265        assert_eq!(
1266            store.exact_prefix_cardinality(
1267                7,
1268                IndexKeyKind::User,
1269                user_index_id,
1270                &[collection.clone(), draft.clone()],
1271            ),
1272            Some(1),
1273        );
1274
1275        let system_key = indexed_raw_key_with_kind(
1276            &system_index_id,
1277            IndexKeyKind::System,
1278            vec![system_component],
1279            1,
1280        );
1281        store.insert(system_key.clone(), IndexEntryValue::presence());
1282        assert_eq!(
1283            store.exact_prefix_cardinality(
1284                7,
1285                IndexKeyKind::User,
1286                user_index_id,
1287                &[collection.clone(), draft.clone()],
1288            ),
1289            Some(1),
1290            "system index writes must not invalidate synchronized user-prefix cardinality",
1291        );
1292
1293        store.remove(&system_key);
1294        assert_eq!(
1295            store.exact_prefix_cardinality(
1296                7,
1297                IndexKeyKind::User,
1298                user_index_id,
1299                &[collection.clone(), draft.clone()],
1300            ),
1301            Some(1),
1302            "system index removals must not invalidate synchronized user-prefix cardinality",
1303        );
1304
1305        let malformed_system_key = indexed_raw_key_with_kind(
1306            &system_index_id,
1307            IndexKeyKind::System,
1308            vec![b"malformed-reverse-edge".to_vec()],
1309            2,
1310        );
1311        store.insert(malformed_system_key.clone(), malformed_index_entry_value());
1312        assert_eq!(
1313            store.exact_prefix_cardinality(
1314                7,
1315                IndexKeyKind::User,
1316                user_index_id,
1317                &[collection.clone(), draft.clone()],
1318            ),
1319            Some(1),
1320            "malformed system index payloads must not invalidate user-prefix cardinality",
1321        );
1322
1323        store.remove(&malformed_system_key);
1324        assert_eq!(
1325            store.exact_prefix_cardinality(
1326                7,
1327                IndexKeyKind::User,
1328                user_index_id,
1329                &[collection.clone(), draft],
1330            ),
1331            Some(1),
1332            "malformed system index removals must not invalidate user-prefix cardinality",
1333        );
1334
1335        let review = b"Review".to_vec();
1336        store.insert(
1337            indexed_raw_key(&user_index_id, vec![collection.clone(), review.clone()], 2),
1338            IndexEntryValue::presence(),
1339        );
1340        assert_eq!(
1341            store.exact_prefix_cardinality(
1342                7,
1343                IndexKeyKind::User,
1344                user_index_id,
1345                &[collection, review]
1346            ),
1347            None,
1348            "user-prefix count changes must still require a fresh row-generation stamp",
1349        );
1350    }
1351
1352    #[test]
1353    fn index_prefix_cardinality_ignores_missing_user_index_mutations() {
1354        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1355        let collection = b"collection-a".to_vec();
1356        let draft = b"Draft".to_vec();
1357        let mut store = IndexStore::init_heap();
1358
1359        store.insert(
1360            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
1361            IndexEntryValue::presence(),
1362        );
1363        store.mark_prefix_cardinality_data_generation(7);
1364
1365        let stale_key = indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2);
1366        store.insert(stale_key.clone(), missing_index_entry_value());
1367        assert_eq!(
1368            store.exact_prefix_cardinality(
1369                7,
1370                IndexKeyKind::User,
1371                index_id,
1372                &[collection.clone(), draft.clone()],
1373            ),
1374            Some(1),
1375            "missing user index entries must not affect synchronized prefix cardinality",
1376        );
1377
1378        store.remove(&stale_key);
1379        assert_eq!(
1380            store.exact_prefix_cardinality(7, IndexKeyKind::User, index_id, &[collection, draft],),
1381            Some(1),
1382            "missing user index removals must not affect synchronized prefix cardinality",
1383        );
1384    }
1385
1386    #[cfg(all(feature = "sql", feature = "diagnostics"))]
1387    #[test]
1388    fn index_store_diagnostic_counters_record_gets_range_scans_and_entry_reads() {
1389        let mut store = IndexStore::init_heap();
1390        store.insert(raw_key(7), IndexEntryValue::presence());
1391        store.insert(raw_key(9), IndexEntryValue::presence());
1392
1393        let gets_before = IndexStore::current_get_call_count();
1394        assert_eq!(store.get(&raw_key(7)), Some(IndexEntryValue::presence()));
1395        assert_eq!(store.get(&raw_key(8)), None);
1396
1397        assert_eq!(
1398            IndexStore::current_get_call_count().saturating_sub(gets_before),
1399            2,
1400            "diagnostic index-store get counter should count both hit and miss reads",
1401        );
1402
1403        let range_scans_before = IndexStore::current_range_scan_call_count();
1404        let lower = Bound::Included(raw_key(7));
1405        let upper = Bound::Included(raw_key(9));
1406        store
1407            .visit_raw_entries_in_range((&lower, &upper), Direction::Asc, |_key, _entry| Ok(false))
1408            .expect("raw index range visit should succeed");
1409
1410        assert_eq!(
1411            IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before),
1412            1,
1413            "diagnostic index-store range-scan counter should count one range traversal probe",
1414        );
1415
1416        let entries_before = IndexStore::current_entry_read_count();
1417        store
1418            .visit_entries(|_key, _entry| Ok::<_, Infallible>(IndexStoreVisit::Continue))
1419            .expect("index entry visit should succeed");
1420
1421        assert_eq!(
1422            IndexStore::current_entry_read_count().saturating_sub(entries_before),
1423            2,
1424            "diagnostic index-store entry counter should count yielded traversal entries",
1425        );
1426    }
1427
1428    #[test]
1429    fn journaled_mixed_index_range_traversal_streams_without_snapshot() {
1430        let mut store = IndexStore::init_journaled(test_memory(93));
1431        for value in [1_u8, 3, 5] {
1432            store.insert(raw_key(value), IndexEntryValue::presence());
1433        }
1434        store
1435            .fold_journaled_materialized_view()
1436            .expect("canonical index seed should fold");
1437
1438        store.insert(raw_key(0), IndexEntryValue::presence());
1439        store.insert(raw_key(4), IndexEntryValue::presence());
1440        store.insert(raw_key(5), IndexEntryValue::presence());
1441        store.remove(&raw_key(1));
1442
1443        let lower = Bound::Included(raw_key(0));
1444        let upper = Bound::Included(raw_key(5));
1445
1446        reset_journaled_snapshot_call_count_for_tests();
1447        let mut asc = Vec::new();
1448        store
1449            .visit_journaled_entries_in_range((&lower, &upper), Direction::Asc, |key, _value| {
1450                asc.push(key.as_bytes()[0]);
1451                Ok::<_, Infallible>(asc.len() == 2)
1452            })
1453            .expect("asc journaled index range traversal should succeed");
1454        assert_eq!(asc, vec![0, 3]);
1455        assert_eq!(
1456            journaled_snapshot_call_count_for_tests(),
1457            0,
1458            "mixed journaled index range traversal should preserve early stop without materializing a snapshot",
1459        );
1460
1461        reset_journaled_snapshot_call_count_for_tests();
1462        let mut desc = Vec::new();
1463        store
1464            .visit_journaled_entries_in_range((&lower, &upper), Direction::Desc, |key, _value| {
1465                desc.push(key.as_bytes()[0]);
1466                Ok::<_, Infallible>(desc.len() == 2)
1467            })
1468            .expect("desc journaled index range traversal should succeed");
1469        assert_eq!(desc, vec![5, 4]);
1470        assert_eq!(
1471            journaled_snapshot_call_count_for_tests(),
1472            0,
1473            "mixed reverse journaled index range traversal should preserve early stop without materializing a snapshot",
1474        );
1475    }
1476
1477    #[test]
1478    fn journaled_index_store_reopens_without_materializing_prefix_cardinality() {
1479        let memory = test_memory(94);
1480        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1481        let collection = b"collection-a".to_vec();
1482        let mut store = IndexStore::init_journaled(memory.clone());
1483        let key = indexed_raw_key(&index_id, vec![collection.clone()], 1);
1484        store.insert(key.clone(), IndexEntryValue::presence());
1485        store
1486            .fold_journaled_materialized_view()
1487            .expect("canonical index seed should fold");
1488        drop(store);
1489
1490        let reopened = IndexStore::init_journaled(memory);
1491
1492        assert_eq!(reopened.get(&key), Some(IndexEntryValue::presence()));
1493        assert_eq!(
1494            reopened.exact_prefix_cardinality(
1495                0,
1496                IndexKeyKind::User,
1497                index_id,
1498                std::slice::from_ref(&collection),
1499            ),
1500            None,
1501            "startup must leave optional prefix cardinality unavailable without scanning stable entries",
1502        );
1503    }
1504
1505    #[test]
1506    fn empty_journaled_index_store_retains_exact_prefix_cardinality_without_scanning() {
1507        let memory = test_memory(95);
1508        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1509        let collection = b"collection-a".to_vec();
1510        let mut store = IndexStore::init_journaled(memory.clone());
1511
1512        assert_eq!(
1513            store.exact_prefix_cardinality(
1514                0,
1515                IndexKeyKind::User,
1516                index_id,
1517                std::slice::from_ref(&collection),
1518            ),
1519            Some(0),
1520        );
1521        store
1522            .reset_journaled_live_projection(7)
1523            .expect("empty projection reset should succeed");
1524        assert_eq!(
1525            store.exact_prefix_cardinality(
1526                7,
1527                IndexKeyKind::User,
1528                index_id,
1529                std::slice::from_ref(&collection),
1530            ),
1531            Some(0),
1532        );
1533        drop(store);
1534
1535        let reopened = IndexStore::init_journaled(memory);
1536        assert_eq!(
1537            reopened.exact_prefix_cardinality(
1538                0,
1539                IndexKeyKind::User,
1540                index_id,
1541                std::slice::from_ref(&collection),
1542            ),
1543            Some(0),
1544        );
1545    }
1546
1547    #[test]
1548    fn recovered_index_fold_maintains_available_prefix_cardinality() {
1549        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1550        let collection = b"collection-a".to_vec();
1551        let key = indexed_raw_key(&index_id, vec![collection.clone()], 1);
1552        let mut store = IndexStore::init_journaled(test_memory(96));
1553
1554        store
1555            .fold_recovered_journal_entry(key.clone(), Some(IndexEntryValue::presence()))
1556            .expect("recovered index put should fold");
1557        store.mark_prefix_cardinality_data_generation(1);
1558        assert_eq!(
1559            store.exact_prefix_cardinality(
1560                1,
1561                IndexKeyKind::User,
1562                index_id,
1563                std::slice::from_ref(&collection),
1564            ),
1565            Some(1),
1566        );
1567
1568        store
1569            .fold_recovered_journal_entry(key, None)
1570            .expect("recovered index delete should fold");
1571        store.mark_prefix_cardinality_data_generation(2);
1572        assert_eq!(
1573            store.exact_prefix_cardinality(
1574                2,
1575                IndexKeyKind::User,
1576                index_id,
1577                std::slice::from_ref(&collection),
1578            ),
1579            Some(0),
1580        );
1581    }
1582
1583    #[test]
1584    fn positioned_index_overlay_preserves_later_membership_until_exact_retirement() {
1585        let key = raw_key(7);
1586        let mut store = IndexStore::init_journaled(test_memory(97));
1587        store
1588            .fold_recovered_journal_entry(key.clone(), Some(IndexEntryValue::presence()))
1589            .expect("canonical membership should seed");
1590
1591        store
1592            .publish_positioned_journal_entry(key.clone(), None, overlay_position(1))
1593            .expect("positioned tombstone should publish");
1594        store
1595            .publish_positioned_journal_entry(
1596                key.clone(),
1597                Some(IndexEntryValue::presence()),
1598                overlay_position(2),
1599            )
1600            .expect("later membership should supersede the tombstone");
1601        store
1602            .fold_recovered_journal_entry(key.clone(), None)
1603            .expect("tombstone batch should become canonical");
1604        assert_eq!(
1605            store
1606                .retire_positioned_journal_effect(&key, overlay_position(1))
1607                .expect("older retirement should preserve later membership"),
1608            PositionedOverlayRetirement::Superseded,
1609        );
1610        assert_eq!(store.get(&key), Some(IndexEntryValue::presence()));
1611
1612        store
1613            .fold_recovered_journal_entry(key.clone(), Some(IndexEntryValue::presence()))
1614            .expect("membership batch should become canonical");
1615        assert_eq!(
1616            store
1617                .retire_positioned_journal_effect(&key, overlay_position(2))
1618                .expect("latest retirement should be exact"),
1619            PositionedOverlayRetirement::Exact,
1620        );
1621        assert_eq!(store.get(&key), Some(IndexEntryValue::presence()));
1622
1623        let mut visible = Vec::new();
1624        store
1625            .visit_entries(|visited_key, value| {
1626                visible.push((visited_key.clone(), value.clone()));
1627                Ok::<_, Infallible>(IndexStoreVisit::Continue)
1628            })
1629            .expect("positioned index should remain range-visible");
1630        assert_eq!(visible, vec![(key, IndexEntryValue::presence())]);
1631    }
1632
1633    #[test]
1634    fn positioned_index_overlay_coalesces_repeated_same_batch_target() {
1635        let key = raw_key(8);
1636        let position = overlay_position(3);
1637        let mut store = IndexStore::init_journaled(test_memory(98));
1638
1639        store
1640            .publish_positioned_journal_entry(
1641                key.clone(),
1642                Some(IndexEntryValue::presence()),
1643                position,
1644            )
1645            .expect("first same-batch effect should publish");
1646        store
1647            .publish_positioned_journal_entry(key.clone(), None, position)
1648            .expect("final same-batch effect should coalesce by logical target");
1649        store
1650            .fold_recovered_journal_entry(key.clone(), None)
1651            .expect("coalesced final effect should become canonical");
1652        assert_eq!(
1653            store
1654                .retire_positioned_journal_effect(&key, position)
1655                .expect("coalesced target should retire once"),
1656            PositionedOverlayRetirement::Exact,
1657        );
1658        assert!(store.get(&key).is_none());
1659    }
1660}
icydb_core/db/index/store.rs

icydb_core/db/index/
store.rs
