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};
12
13use candid::CandidType;
14use ic_stable_structures::{
15    BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
16};
17use serde::Deserialize;
18#[cfg(any(test, all(feature = "sql", feature = "diagnostics")))]
19use std::cell::Cell;
20use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
21use std::ops::Bound;
22
23#[cfg(test)]
24thread_local! {
25    static JOURNALED_SNAPSHOT_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
26}
27
28#[cfg(all(feature = "sql", feature = "diagnostics"))]
29thread_local! {
30    static INDEX_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
31    static INDEX_STORE_RANGE_SCAN_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
32    static INDEX_STORE_ENTRY_READ_COUNT: Cell<u64> = const { Cell::new(0) };
33}
34
35#[cfg(all(feature = "sql", feature = "diagnostics"))]
36fn record_index_store_get_call() {
37    INDEX_STORE_GET_CALL_COUNT.with(|count| {
38        count.set(count.get().saturating_add(1));
39    });
40}
41
42#[cfg(all(feature = "sql", feature = "diagnostics"))]
43fn record_index_store_range_scan_call() {
44    INDEX_STORE_RANGE_SCAN_CALL_COUNT.with(|count| {
45        count.set(count.get().saturating_add(1));
46    });
47}
48
49#[cfg(all(feature = "sql", feature = "diagnostics"))]
50fn record_index_store_entry_read() {
51    INDEX_STORE_ENTRY_READ_COUNT.with(|count| {
52        count.set(count.get().saturating_add(1));
53    });
54}
55
56fn visit_index_store_entry<E>(
57    key: &RawIndexStoreKey,
58    value: &IndexEntryValue,
59    visit: &mut impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<bool, E>,
60) -> Result<bool, E> {
61    #[cfg(all(feature = "sql", feature = "diagnostics"))]
62    record_index_store_entry_read();
63
64    visit(key, value)
65}
66
67#[cfg(test)]
68fn record_journaled_snapshot_call() {
69    JOURNALED_SNAPSHOT_CALL_COUNT.with(|count| {
70        count.set(count.get().saturating_add(1));
71    });
72}
73
74#[cfg(test)]
75fn reset_journaled_snapshot_call_count_for_tests() {
76    JOURNALED_SNAPSHOT_CALL_COUNT.with(|count| count.set(0));
77}
78
79#[cfg(test)]
80fn journaled_snapshot_call_count_for_tests() -> u64 {
81    JOURNALED_SNAPSHOT_CALL_COUNT.with(Cell::get)
82}
83
84//
85// IndexState
86//
87// Explicit lifecycle visibility state for one index store.
88// Visibility matters because planner-visible indexes must already be complete:
89// the index contents are fully built and query-visible for reads.
90//
91#[derive(CandidType, Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
92pub enum IndexState {
93    Building,
94    #[default]
95    Ready,
96}
97
98impl IndexState {
99    /// Return the stable lowercase text label for this lifecycle state.
100    #[must_use]
101    pub const fn as_str(self) -> &'static str {
102        match self {
103            Self::Building => "building",
104            Self::Ready => "ready",
105        }
106    }
107}
108
109///
110/// IndexStore
111///
112/// Thin persistence wrapper over one journaled or heap BTreeMap.
113///
114/// Invariant: callers provide already-validated `RawIndexStoreKey`/`IndexEntryValue`.
115///
116
117pub struct IndexStore {
118    pub(super) backend: IndexStoreBackend,
119    generation: u64,
120    state: IndexState,
121    access_state_revision: u64,
122    prefix_cardinality: IndexPrefixCardinality,
123}
124
125pub(super) enum IndexStoreBackend {
126    Heap(HeapBTreeMap<RawIndexStoreKey, IndexEntryValue>),
127    Journaled {
128        canonical:
129            StableBTreeMap<RawIndexStoreKey, IndexEntryValue, VirtualMemory<DefaultMemoryImpl>>,
130        live: HeapBTreeMap<RawIndexStoreKey, IndexEntryValue>,
131        tombstones: BTreeSet<RawIndexStoreKey>,
132    },
133}
134
135/// Control-flow result for index-store traversal visitors.
136#[derive(Clone, Copy, Debug, Eq, PartialEq)]
137pub(in crate::db) enum IndexStoreVisit {
138    Continue,
139    Stop,
140}
141
142impl IndexStoreVisit {
143    const fn should_stop(self) -> bool {
144        matches!(self, Self::Stop)
145    }
146}
147
148impl IndexStore {
149    /// Initialize a volatile heap-backed index store.
150    #[must_use]
151    pub const fn init_heap() -> Self {
152        Self {
153            backend: IndexStoreBackend::Heap(HeapBTreeMap::new()),
154            generation: 0,
155            state: IndexState::Ready,
156            access_state_revision: 1,
157            prefix_cardinality: IndexPrefixCardinality::synchronized_empty(),
158        }
159    }
160
161    /// Initialize a journaled cached-stable index store.
162    ///
163    /// Normal writes update only the live materialized projection. The
164    /// canonical stable index is updated by future fold/rebuild paths.
165    #[must_use]
166    pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
167        let mut store = Self {
168            backend: IndexStoreBackend::Journaled {
169                canonical: StableBTreeMap::init(memory),
170                live: HeapBTreeMap::new(),
171                tombstones: BTreeSet::new(),
172            },
173            generation: 0,
174            state: IndexState::Ready,
175            access_state_revision: 1,
176            prefix_cardinality: IndexPrefixCardinality::synchronized_empty(),
177        };
178        store.rebuild_prefix_cardinality_from_entries(Some(0));
179        store
180    }
181
182    /// Visit all index entries in canonical store order without exposing the
183    /// backing stable-map iterator.
184    pub(in crate::db) fn visit_entries<E>(
185        &self,
186        mut visitor: impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<IndexStoreVisit, E>,
187    ) -> Result<(), E> {
188        match &self.backend {
189            IndexStoreBackend::Heap(map) => {
190                for (key, value) in map {
191                    #[cfg(all(feature = "sql", feature = "diagnostics"))]
192                    record_index_store_entry_read();
193
194                    if visitor(key, value)?.should_stop() {
195                        return Ok(());
196                    }
197                }
198            }
199            IndexStoreBackend::Journaled {
200                canonical: _,
201                live: _,
202                tombstones: _,
203            } => self.visit_journaled_entries_in_range(
204                (&Bound::Unbounded, &Bound::Unbounded),
205                Direction::Asc,
206                |key, value| visitor(key, value).map(IndexStoreVisit::should_stop),
207            )?,
208        }
209
210        Ok(())
211    }
212
213    pub(in crate::db) fn get(&self, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
214        #[cfg(all(feature = "sql", feature = "diagnostics"))]
215        record_index_store_get_call();
216
217        match &self.backend {
218            IndexStoreBackend::Heap(map) => map.get(key).cloned(),
219            IndexStoreBackend::Journaled { .. } => Self::journaled_get(&self.backend, key),
220        }
221    }
222
223    pub fn len(&self) -> u64 {
224        match &self.backend {
225            IndexStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
226            IndexStoreBackend::Journaled { .. } => {
227                let mut count = 0_u64;
228                let _: Result<(), std::convert::Infallible> = self.visit_entries(|_key, _value| {
229                    count = count.saturating_add(1);
230                    Ok(IndexStoreVisit::Continue)
231                });
232                count
233            }
234        }
235    }
236
237    pub fn is_empty(&self) -> bool {
238        match &self.backend {
239            IndexStoreBackend::Heap(map) => map.is_empty(),
240            IndexStoreBackend::Journaled { .. } => {
241                let mut empty = true;
242                let _: Result<(), std::convert::Infallible> = self.visit_entries(|_key, _value| {
243                    empty = false;
244                    Ok(IndexStoreVisit::Stop)
245                });
246                empty
247            }
248        }
249    }
250
251    #[must_use]
252    pub(in crate::db) const fn generation(&self) -> u64 {
253        self.generation
254    }
255
256    /// Return the explicit lifecycle state for this index store.
257    #[must_use]
258    pub(in crate::db) const fn state(&self) -> IndexState {
259        self.state
260    }
261
262    /// Return the current physical access-readiness revision.
263    #[must_use]
264    pub(in crate::db) const fn access_state_revision(&self) -> u64 {
265        self.access_state_revision
266    }
267
268    /// Return an exact user-index prefix count when the index metadata is
269    /// synchronized with the caller's authoritative row-store generation.
270    #[must_use]
271    pub(in crate::db) fn exact_prefix_cardinality(
272        &self,
273        data_generation: u64,
274        key_kind: IndexKeyKind,
275        index_id: IndexId,
276        components: &[Vec<u8>],
277    ) -> Option<u64> {
278        self.prefix_cardinality
279            .exact_count(data_generation, key_kind, index_id, components)
280    }
281
282    /// Return the sum of exact prefix counts for prefixes on the same index
283    /// when synchronized metadata can prove all requested counts.
284    #[must_use]
285    pub(in crate::db) fn exact_prefix_cardinality_sum<'a>(
286        &self,
287        data_generation: u64,
288        key_kind: IndexKeyKind,
289        index_id: IndexId,
290        component_prefixes: impl IntoIterator<Item = &'a [Vec<u8>]>,
291        stop_after: Option<u64>,
292    ) -> Option<u64> {
293        self.prefix_cardinality.exact_count_sum(
294            data_generation,
295            key_kind,
296            index_id,
297            component_prefixes,
298            stop_after,
299        )
300    }
301
302    /// Return non-empty exact child prefixes under a sparse set of already-encoded
303    /// parent prefixes when synchronized metadata can prove the bounded child set.
304    #[must_use]
305    pub(in crate::db) fn exact_child_prefixes_for_parent_set<'a>(
306        &self,
307        data_generation: u64,
308        key_kind: IndexKeyKind,
309        index_id: IndexId,
310        parent_component_prefixes: impl IntoIterator<Item = &'a [Vec<u8>]>,
311        max_children: usize,
312    ) -> Option<Vec<Vec<Vec<u8>>>> {
313        self.prefix_cardinality.exact_child_prefixes_for_parent_set(
314            data_generation,
315            key_kind,
316            index_id,
317            parent_component_prefixes,
318            max_children,
319        )
320    }
321
322    /// Mark prefix-cardinality metadata synchronized with the authoritative
323    /// row-store generation after a committed row/index transition.
324    pub(in crate::db) const fn mark_prefix_cardinality_data_generation(&mut self, generation: u64) {
325        self.prefix_cardinality.mark_synchronized(generation);
326    }
327
328    /// Mark this index store as in-progress and therefore ineligible for
329    /// planner visibility until a full authoritative rebuild ends.
330    pub(in crate::db) const fn set_access_state(&mut self, state: IndexState, revision: u64) {
331        self.state = state;
332        self.access_state_revision = revision;
333    }
334
335    pub(crate) fn insert(
336        &mut self,
337        key: RawIndexStoreKey,
338        entry: IndexEntryValue,
339    ) -> Option<IndexEntryValue> {
340        let previous_journaled = if matches!(self.backend, IndexStoreBackend::Journaled { .. }) {
341            self.get(&key)
342        } else {
343            None
344        };
345        let cardinality_key = key.clone();
346        let previous = match &mut self.backend {
347            IndexStoreBackend::Heap(map) => map.insert(key, entry.clone()),
348            IndexStoreBackend::Journaled {
349                live, tombstones, ..
350            } => {
351                tombstones.remove(&key);
352                live.insert(key, entry.clone());
353                previous_journaled
354            }
355        };
356        self.prefix_cardinality
357            .apply_insert(&cardinality_key, previous.as_ref(), &entry);
358        self.bump_generation();
359        previous
360    }
361
362    pub(crate) fn remove(&mut self, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
363        let previous_journaled = if matches!(self.backend, IndexStoreBackend::Journaled { .. }) {
364            self.get(key)
365        } else {
366            None
367        };
368        let previous = match &mut self.backend {
369            IndexStoreBackend::Heap(map) => map.remove(key),
370            IndexStoreBackend::Journaled {
371                live, tombstones, ..
372            } => {
373                live.remove(key);
374                tombstones.insert(key.clone());
375                previous_journaled
376            }
377        };
378        self.prefix_cardinality.apply_remove(key, previous.as_ref());
379        self.bump_generation();
380        previous
381    }
382
383    pub fn clear(&mut self) {
384        match &mut self.backend {
385            IndexStoreBackend::Heap(map) => map.clear(),
386            IndexStoreBackend::Journaled {
387                canonical,
388                live,
389                tombstones,
390            } => {
391                live.clear();
392                tombstones.clear();
393                for entry in canonical.iter() {
394                    tombstones.insert(entry.key().clone());
395                }
396            }
397        }
398        self.prefix_cardinality.clear_unsynchronized();
399        self.bump_generation();
400    }
401
402    /// Fold the current journaled materialized index view into the canonical
403    /// stable base and clear volatile projection state.
404    pub(in crate::db) fn fold_journaled_materialized_view(
405        &mut self,
406    ) -> Result<(), crate::error::InternalError> {
407        let entries = Self::journaled_entries_snapshot_for_fold(&self.backend);
408        let IndexStoreBackend::Journaled {
409            canonical,
410            live,
411            tombstones,
412        } = &mut self.backend
413        else {
414            return Err(crate::error::InternalError::store_invariant());
415        };
416
417        canonical.clear_new();
418        for (key, value) in entries {
419            canonical.insert(key, value);
420        }
421        live.clear();
422        tombstones.clear();
423        let data_generation = self.prefix_cardinality.synchronized_generation();
424        self.rebuild_prefix_cardinality_from_entries(data_generation);
425        self.bump_generation();
426
427        Ok(())
428    }
429
430    /// Sum of bytes used by all stored index entries.
431    pub fn memory_bytes(&self) -> u64 {
432        let mut bytes = 0u64;
433        let _: Result<(), std::convert::Infallible> = self.visit_entries(|key, value| {
434            bytes = bytes.saturating_add(key.as_bytes().len() as u64 + value.len() as u64);
435            Ok(IndexStoreVisit::Continue)
436        });
437        bytes
438    }
439
440    /// Return the monotonic perf-only count of index-entry fetches seen by this process.
441    #[cfg(all(feature = "sql", feature = "diagnostics"))]
442    pub(in crate::db) fn current_get_call_count() -> u64 {
443        INDEX_STORE_GET_CALL_COUNT.with(Cell::get)
444    }
445
446    /// Return the monotonic perf-only count of index range traversal probes seen by this process.
447    #[cfg(all(feature = "sql", feature = "diagnostics"))]
448    pub(in crate::db) fn current_range_scan_call_count() -> u64 {
449        INDEX_STORE_RANGE_SCAN_CALL_COUNT.with(Cell::get)
450    }
451
452    /// Return the monotonic perf-only count of index entries yielded by traversal.
453    #[cfg(all(feature = "sql", feature = "diagnostics"))]
454    pub(in crate::db) fn current_entry_read_count() -> u64 {
455        INDEX_STORE_ENTRY_READ_COUNT.with(Cell::get)
456    }
457
458    #[cfg(all(feature = "sql", feature = "diagnostics"))]
459    pub(in crate::db::index) fn record_range_scan_call() {
460        record_index_store_range_scan_call();
461    }
462
463    #[cfg(all(feature = "sql", feature = "diagnostics"))]
464    pub(in crate::db::index) fn record_merged_entry_reads(count: u64) {
465        INDEX_STORE_ENTRY_READ_COUNT.with(|total| {
466            total.set(total.get().saturating_add(count));
467        });
468    }
469
470    const fn bump_generation(&mut self) {
471        self.generation = self.generation.saturating_add(1);
472    }
473
474    fn rebuild_prefix_cardinality_from_entries(&mut self, data_generation: Option<u64>) {
475        self.prefix_cardinality.clear_unsynchronized();
476        let entries = Self::entries_snapshot_for_cardinality(&self.backend);
477        for (key, value) in &entries {
478            self.prefix_cardinality.apply_insert(key, None, value);
479        }
480        if let Some(data_generation) = data_generation {
481            self.prefix_cardinality.mark_synchronized(data_generation);
482        }
483    }
484
485    fn entries_snapshot_for_cardinality(
486        backend: &IndexStoreBackend,
487    ) -> HeapBTreeMap<RawIndexStoreKey, IndexEntryValue> {
488        match backend {
489            IndexStoreBackend::Heap(map) => map.clone(),
490            IndexStoreBackend::Journaled { .. } => {
491                Self::journaled_entries_snapshot_for_fold(backend)
492            }
493        }
494    }
495
496    fn journaled_get(
497        backend: &IndexStoreBackend,
498        key: &RawIndexStoreKey,
499    ) -> Option<IndexEntryValue> {
500        let IndexStoreBackend::Journaled {
501            canonical,
502            live,
503            tombstones,
504        } = backend
505        else {
506            return None;
507        };
508
509        if tombstones.contains(key) {
510            return None;
511        }
512        live.get(key).cloned().or_else(|| canonical.get(key))
513    }
514
515    pub(super) fn journaled_entries_snapshot_for_fold(
516        backend: &IndexStoreBackend,
517    ) -> HeapBTreeMap<RawIndexStoreKey, IndexEntryValue> {
518        #[cfg(test)]
519        record_journaled_snapshot_call();
520
521        let IndexStoreBackend::Journaled {
522            canonical,
523            live,
524            tombstones,
525        } = backend
526        else {
527            return HeapBTreeMap::new();
528        };
529
530        let mut entries = HeapBTreeMap::new();
531        for entry in canonical.iter() {
532            let key = entry.key().clone();
533            if !tombstones.contains(&key) {
534                entries.insert(key, entry.value());
535            }
536        }
537        for (key, value) in live {
538            if !tombstones.contains(key) {
539                entries.insert(key.clone(), value.clone());
540            }
541        }
542
543        entries
544    }
545
546    pub(super) fn visit_journaled_entries_in_range<E>(
547        &self,
548        bounds: (&Bound<RawIndexStoreKey>, &Bound<RawIndexStoreKey>),
549        direction: Direction,
550        mut visit: impl FnMut(&RawIndexStoreKey, &IndexEntryValue) -> Result<bool, E>,
551    ) -> Result<(), E> {
552        let IndexStoreBackend::Journaled {
553            canonical,
554            live,
555            tombstones,
556        } = &self.backend
557        else {
558            return Ok(());
559        };
560
561        let lower = bounds.0.clone();
562        let upper = bounds.1.clone();
563        match direction {
564            Direction::Asc if canonical.is_empty() => {
565                for (key, value) in live.range((lower, upper)) {
566                    if visit_index_store_entry(key, value, &mut visit)? {
567                        return Ok(());
568                    }
569                }
570            }
571            Direction::Desc if canonical.is_empty() => {
572                for (key, value) in live.range((lower, upper)).rev() {
573                    if visit_index_store_entry(key, value, &mut visit)? {
574                        return Ok(());
575                    }
576                }
577            }
578            Direction::Asc if live.is_empty() && tombstones.is_empty() => {
579                for entry in canonical.range((lower, upper)) {
580                    if visit_index_store_entry(entry.key(), &entry.value(), &mut visit)? {
581                        return Ok(());
582                    }
583                }
584            }
585            Direction::Desc if live.is_empty() && tombstones.is_empty() => {
586                for entry in canonical.range((lower, upper)).rev() {
587                    if visit_index_store_entry(entry.key(), &entry.value(), &mut visit)? {
588                        return Ok(());
589                    }
590                }
591            }
592            Direction::Asc => {
593                visit_ordered_overlay(
594                    canonical.range((lower.clone(), upper.clone())),
595                    live.range((lower, upper)),
596                    direction,
597                    |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
598                    |canonical_entry| !tombstones.contains(canonical_entry.key()),
599                    |live_entry| !tombstones.contains(live_entry.0),
600                    |entry| {
601                        let should_stop = match entry {
602                            OrderedOverlayEntry::Canonical(canonical_entry) => {
603                                visit_index_store_entry(
604                                    canonical_entry.key(),
605                                    &canonical_entry.value(),
606                                    &mut visit,
607                                )?
608                            }
609                            OrderedOverlayEntry::Live((key, value)) => {
610                                visit_index_store_entry(key, value, &mut visit)?
611                            }
612                        };
613                        Ok(if should_stop {
614                            OrderedOverlayVisit::Stop
615                        } else {
616                            OrderedOverlayVisit::Continue
617                        })
618                    },
619                )?;
620            }
621            Direction::Desc => {
622                visit_ordered_overlay(
623                    canonical.range((lower.clone(), upper.clone())).rev(),
624                    live.range((lower, upper)).rev(),
625                    direction,
626                    |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
627                    |canonical_entry| !tombstones.contains(canonical_entry.key()),
628                    |live_entry| !tombstones.contains(live_entry.0),
629                    |entry| {
630                        let should_stop = match entry {
631                            OrderedOverlayEntry::Canonical(canonical_entry) => {
632                                visit_index_store_entry(
633                                    canonical_entry.key(),
634                                    &canonical_entry.value(),
635                                    &mut visit,
636                                )?
637                            }
638                            OrderedOverlayEntry::Live((key, value)) => {
639                                visit_index_store_entry(key, value, &mut visit)?
640                            }
641                        };
642                        Ok(if should_stop {
643                            OrderedOverlayVisit::Stop
644                        } else {
645                            OrderedOverlayVisit::Continue
646                        })
647                    },
648                )?;
649            }
650        }
651
652        Ok(())
653    }
654}
655
656#[cfg(test)]
657mod tests {
658    use super::*;
659    use crate::{
660        db::{
661            direction::Direction,
662            index::{IndexId, IndexKey, IndexKeyKind},
663            key_taxonomy::{PrimaryKeyComponent, PrimaryKeyValue},
664        },
665        testing::test_memory,
666        types::EntityTag,
667    };
668    use ic_stable_structures::Storable;
669    use std::{borrow::Cow, convert::Infallible};
670
671    fn raw_key(value: u8) -> RawIndexStoreKey {
672        <RawIndexStoreKey as Storable>::from_bytes(Cow::Owned(vec![value]))
673    }
674
675    fn indexed_raw_key(
676        index_id: &IndexId,
677        components: Vec<Vec<u8>>,
678        primary_key: u64,
679    ) -> RawIndexStoreKey {
680        indexed_raw_key_with_kind(index_id, IndexKeyKind::User, components, primary_key)
681    }
682
683    fn indexed_raw_key_with_kind(
684        index_id: &IndexId,
685        key_kind: IndexKeyKind,
686        components: Vec<Vec<u8>>,
687        primary_key: u64,
688    ) -> RawIndexStoreKey {
689        IndexKey::new_from_components_with_primary_key_value(
690            index_id,
691            key_kind,
692            components.as_slice(),
693            &PrimaryKeyValue::from(PrimaryKeyComponent::Nat64(primary_key)),
694        )
695        .expect("test index key should build")
696        .to_raw()
697        .expect("test index key should encode")
698    }
699
700    fn malformed_index_entry_value() -> IndexEntryValue {
701        <IndexEntryValue as Storable>::from_bytes(Cow::Owned(vec![0xFF]))
702    }
703
704    fn missing_index_entry_value() -> IndexEntryValue {
705        <IndexEntryValue as Storable>::from_bytes(Cow::Owned(vec![1]))
706    }
707
708    #[test]
709    fn index_prefix_cardinality_requires_explicit_data_generation_sync() {
710        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
711        let collection = b"collection-a".to_vec();
712        let draft = b"Draft".to_vec();
713        let review = b"Review".to_vec();
714        let mut store = IndexStore::init_heap();
715
716        store.insert(
717            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
718            IndexEntryValue::presence(),
719        );
720        store.insert(
721            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2),
722            IndexEntryValue::presence(),
723        );
724        store.insert(
725            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 3),
726            IndexEntryValue::presence(),
727        );
728
729        assert_eq!(
730            store.exact_prefix_cardinality(
731                0,
732                IndexKeyKind::User,
733                index_id,
734                std::slice::from_ref(&collection),
735            ),
736            None,
737            "raw index mutations must not be trusted until row generation sync is stamped",
738        );
739
740        store.mark_prefix_cardinality_data_generation(7);
741
742        assert_eq!(
743            store.exact_prefix_cardinality(
744                7,
745                IndexKeyKind::User,
746                index_id,
747                std::slice::from_ref(&collection),
748            ),
749            Some(3),
750        );
751        assert_eq!(
752            store.exact_prefix_cardinality(
753                7,
754                IndexKeyKind::User,
755                index_id,
756                &[collection.clone(), draft],
757            ),
758            Some(2),
759        );
760        assert_eq!(
761            store.exact_prefix_cardinality(8, IndexKeyKind::User, index_id, &[collection, review],),
762            None,
763            "row generation drift should force the caller to use the existing-row fallback",
764        );
765    }
766
767    #[test]
768    fn index_prefix_cardinality_enumerates_bounded_child_prefixes() {
769        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
770        let collection = b"collection-a".to_vec();
771        let other_collection = b"collection-b".to_vec();
772        let draft = b"Draft".to_vec();
773        let review = b"Review".to_vec();
774        let published = b"Published".to_vec();
775        let mut store = IndexStore::init_heap();
776
777        store.insert(
778            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
779            IndexEntryValue::presence(),
780        );
781        store.insert(
782            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2),
783            IndexEntryValue::presence(),
784        );
785        store.insert(
786            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 3),
787            IndexEntryValue::presence(),
788        );
789        store.insert(
790            indexed_raw_key(
791                &index_id,
792                vec![other_collection.clone(), published.clone()],
793                4,
794            ),
795            IndexEntryValue::presence(),
796        );
797        store.mark_prefix_cardinality_data_generation(7);
798
799        assert_eq!(
800            store.exact_child_prefixes_for_parent_set(
801                7,
802                IndexKeyKind::User,
803                index_id,
804                [std::slice::from_ref(&collection)],
805                4,
806            ),
807            Some(vec![
808                vec![collection.clone(), draft],
809                vec![collection.clone(), review],
810            ]),
811            "child-prefix enumeration should return deterministic unique children under the requested parent",
812        );
813        assert_eq!(
814            store.exact_child_prefixes_for_parent_set(
815                7,
816                IndexKeyKind::User,
817                index_id,
818                [std::slice::from_ref(&other_collection)],
819                4,
820            ),
821            Some(vec![vec![other_collection, published]]),
822            "child-prefix enumeration must stay scoped to the requested parent prefix",
823        );
824        assert_eq!(
825            store.exact_child_prefixes_for_parent_set(
826                8,
827                IndexKeyKind::User,
828                index_id,
829                [std::slice::from_ref(&collection)],
830                4,
831            ),
832            None,
833            "row generation drift should keep child-prefix expansion fail-closed",
834        );
835        assert_eq!(
836            store.exact_child_prefixes_for_parent_set(
837                7,
838                IndexKeyKind::User,
839                index_id,
840                [std::slice::from_ref(&collection)],
841                1,
842            ),
843            None,
844            "over-cap child-prefix expansion should fall back to the existing route",
845        );
846    }
847
848    #[test]
849    fn index_prefix_cardinality_batches_sparse_child_prefixes() {
850        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
851        let collection = b"collection-a".to_vec();
852        let other_collection = b"collection-b".to_vec();
853        let missing_a = b"missing-a".to_vec();
854        let missing_b = b"missing-b".to_vec();
855        let draft = b"Draft".to_vec();
856        let review = b"Review".to_vec();
857        let published = b"Published".to_vec();
858        let mut store = IndexStore::init_heap();
859
860        store.insert(
861            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
862            IndexEntryValue::presence(),
863        );
864        store.insert(
865            indexed_raw_key(&index_id, vec![collection.clone(), review.clone()], 2),
866            IndexEntryValue::presence(),
867        );
868        store.insert(
869            indexed_raw_key(
870                &index_id,
871                vec![other_collection.clone(), published.clone()],
872                3,
873            ),
874            IndexEntryValue::presence(),
875        );
876        store.mark_prefix_cardinality_data_generation(7);
877
878        let parents = [
879            std::slice::from_ref(&missing_a),
880            std::slice::from_ref(&collection),
881            std::slice::from_ref(&missing_b),
882            std::slice::from_ref(&other_collection),
883        ];
884        assert_eq!(
885            store.exact_child_prefixes_for_parent_set(7, IndexKeyKind::User, index_id, parents, 4,),
886            Some(vec![
887                vec![collection.clone(), draft],
888                vec![collection.clone(), review],
889                vec![other_collection.clone(), published],
890            ]),
891            "batched child-prefix enumeration should skip missing sparse parents and return deterministic real children",
892        );
893        assert_eq!(
894            store.exact_child_prefixes_for_parent_set(
895                7,
896                IndexKeyKind::User,
897                index_id,
898                [
899                    std::slice::from_ref(&missing_a),
900                    std::slice::from_ref(&missing_b)
901                ],
902                4,
903            ),
904            Some(Vec::new()),
905            "missing-only sparse parent sets should be proven empty when cardinality is synchronized",
906        );
907        assert_eq!(
908            store.exact_child_prefixes_for_parent_set(
909                7,
910                IndexKeyKind::User,
911                index_id,
912                [
913                    std::slice::from_ref(&collection),
914                    std::slice::from_ref(&other_collection)
915                ],
916                2,
917            ),
918            None,
919            "over-cap sparse parent-set expansion should fail closed",
920        );
921        assert_eq!(
922            store.exact_child_prefixes_for_parent_set(
923                8,
924                IndexKeyKind::User,
925                index_id,
926                [std::slice::from_ref(&collection)],
927                4,
928            ),
929            None,
930            "generation drift should keep batched child-prefix expansion fail-closed",
931        );
932    }
933
934    #[test]
935    fn index_prefix_cardinality_ignores_system_index_mutations() {
936        let user_index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
937        let system_index_id = IndexId::new(EntityTag::new(0xCA7D), 2);
938        let collection = b"collection-a".to_vec();
939        let draft = b"Draft".to_vec();
940        let system_component = b"reverse-edge".to_vec();
941        let mut store = IndexStore::init_heap();
942
943        store.insert(
944            indexed_raw_key(&user_index_id, vec![collection.clone(), draft.clone()], 1),
945            IndexEntryValue::presence(),
946        );
947        store.mark_prefix_cardinality_data_generation(7);
948
949        assert_eq!(
950            store.exact_prefix_cardinality(
951                7,
952                IndexKeyKind::User,
953                user_index_id,
954                &[collection.clone(), draft.clone()],
955            ),
956            Some(1),
957        );
958
959        let system_key = indexed_raw_key_with_kind(
960            &system_index_id,
961            IndexKeyKind::System,
962            vec![system_component],
963            1,
964        );
965        store.insert(system_key.clone(), IndexEntryValue::presence());
966        assert_eq!(
967            store.exact_prefix_cardinality(
968                7,
969                IndexKeyKind::User,
970                user_index_id,
971                &[collection.clone(), draft.clone()],
972            ),
973            Some(1),
974            "system index writes must not invalidate synchronized user-prefix cardinality",
975        );
976
977        store.remove(&system_key);
978        assert_eq!(
979            store.exact_prefix_cardinality(
980                7,
981                IndexKeyKind::User,
982                user_index_id,
983                &[collection.clone(), draft.clone()],
984            ),
985            Some(1),
986            "system index removals must not invalidate synchronized user-prefix cardinality",
987        );
988
989        let malformed_system_key = indexed_raw_key_with_kind(
990            &system_index_id,
991            IndexKeyKind::System,
992            vec![b"malformed-reverse-edge".to_vec()],
993            2,
994        );
995        store.insert(malformed_system_key.clone(), malformed_index_entry_value());
996        assert_eq!(
997            store.exact_prefix_cardinality(
998                7,
999                IndexKeyKind::User,
1000                user_index_id,
1001                &[collection.clone(), draft.clone()],
1002            ),
1003            Some(1),
1004            "malformed system index payloads must not invalidate user-prefix cardinality",
1005        );
1006
1007        store.remove(&malformed_system_key);
1008        assert_eq!(
1009            store.exact_prefix_cardinality(
1010                7,
1011                IndexKeyKind::User,
1012                user_index_id,
1013                &[collection.clone(), draft],
1014            ),
1015            Some(1),
1016            "malformed system index removals must not invalidate user-prefix cardinality",
1017        );
1018
1019        let review = b"Review".to_vec();
1020        store.insert(
1021            indexed_raw_key(&user_index_id, vec![collection.clone(), review.clone()], 2),
1022            IndexEntryValue::presence(),
1023        );
1024        assert_eq!(
1025            store.exact_prefix_cardinality(
1026                7,
1027                IndexKeyKind::User,
1028                user_index_id,
1029                &[collection, review]
1030            ),
1031            None,
1032            "user-prefix count changes must still require a fresh row-generation stamp",
1033        );
1034    }
1035
1036    #[test]
1037    fn index_prefix_cardinality_ignores_missing_user_index_mutations() {
1038        let index_id = IndexId::new(EntityTag::new(0xCA7D), 1);
1039        let collection = b"collection-a".to_vec();
1040        let draft = b"Draft".to_vec();
1041        let mut store = IndexStore::init_heap();
1042
1043        store.insert(
1044            indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 1),
1045            IndexEntryValue::presence(),
1046        );
1047        store.mark_prefix_cardinality_data_generation(7);
1048
1049        let stale_key = indexed_raw_key(&index_id, vec![collection.clone(), draft.clone()], 2);
1050        store.insert(stale_key.clone(), missing_index_entry_value());
1051        assert_eq!(
1052            store.exact_prefix_cardinality(
1053                7,
1054                IndexKeyKind::User,
1055                index_id,
1056                &[collection.clone(), draft.clone()],
1057            ),
1058            Some(1),
1059            "missing user index entries must not affect synchronized prefix cardinality",
1060        );
1061
1062        store.remove(&stale_key);
1063        assert_eq!(
1064            store.exact_prefix_cardinality(7, IndexKeyKind::User, index_id, &[collection, draft],),
1065            Some(1),
1066            "missing user index removals must not affect synchronized prefix cardinality",
1067        );
1068    }
1069
1070    #[cfg(all(feature = "sql", feature = "diagnostics"))]
1071    #[test]
1072    fn index_store_diagnostic_counters_record_gets_range_scans_and_entry_reads() {
1073        let mut store = IndexStore::init_heap();
1074        store.insert(raw_key(7), IndexEntryValue::presence());
1075        store.insert(raw_key(9), IndexEntryValue::presence());
1076
1077        let gets_before = IndexStore::current_get_call_count();
1078        assert_eq!(store.get(&raw_key(7)), Some(IndexEntryValue::presence()));
1079        assert_eq!(store.get(&raw_key(8)), None);
1080
1081        assert_eq!(
1082            IndexStore::current_get_call_count().saturating_sub(gets_before),
1083            2,
1084            "diagnostic index-store get counter should count both hit and miss reads",
1085        );
1086
1087        let range_scans_before = IndexStore::current_range_scan_call_count();
1088        let lower = Bound::Included(raw_key(7));
1089        let upper = Bound::Included(raw_key(9));
1090        store
1091            .visit_raw_entries_in_range((&lower, &upper), Direction::Asc, |_key, _entry| Ok(false))
1092            .expect("raw index range visit should succeed");
1093
1094        assert_eq!(
1095            IndexStore::current_range_scan_call_count().saturating_sub(range_scans_before),
1096            1,
1097            "diagnostic index-store range-scan counter should count one range traversal probe",
1098        );
1099
1100        let entries_before = IndexStore::current_entry_read_count();
1101        store
1102            .visit_entries(|_key, _entry| Ok::<_, Infallible>(IndexStoreVisit::Continue))
1103            .expect("index entry visit should succeed");
1104
1105        assert_eq!(
1106            IndexStore::current_entry_read_count().saturating_sub(entries_before),
1107            2,
1108            "diagnostic index-store entry counter should count yielded traversal entries",
1109        );
1110    }
1111
1112    #[test]
1113    fn journaled_mixed_index_range_traversal_streams_without_snapshot() {
1114        let mut store = IndexStore::init_journaled(test_memory(93));
1115        for value in [1_u8, 3, 5] {
1116            store.insert(raw_key(value), IndexEntryValue::presence());
1117        }
1118        store
1119            .fold_journaled_materialized_view()
1120            .expect("canonical index seed should fold");
1121
1122        store.insert(raw_key(0), IndexEntryValue::presence());
1123        store.insert(raw_key(4), IndexEntryValue::presence());
1124        store.insert(raw_key(5), IndexEntryValue::presence());
1125        store.remove(&raw_key(1));
1126
1127        let lower = Bound::Included(raw_key(0));
1128        let upper = Bound::Included(raw_key(5));
1129
1130        reset_journaled_snapshot_call_count_for_tests();
1131        let mut asc = Vec::new();
1132        store
1133            .visit_journaled_entries_in_range((&lower, &upper), Direction::Asc, |key, _value| {
1134                asc.push(key.as_bytes()[0]);
1135                Ok::<_, Infallible>(asc.len() == 2)
1136            })
1137            .expect("asc journaled index range traversal should succeed");
1138        assert_eq!(asc, vec![0, 3]);
1139        assert_eq!(
1140            journaled_snapshot_call_count_for_tests(),
1141            0,
1142            "mixed journaled index range traversal should preserve early stop without materializing a snapshot",
1143        );
1144
1145        reset_journaled_snapshot_call_count_for_tests();
1146        let mut desc = Vec::new();
1147        store
1148            .visit_journaled_entries_in_range((&lower, &upper), Direction::Desc, |key, _value| {
1149                desc.push(key.as_bytes()[0]);
1150                Ok::<_, Infallible>(desc.len() == 2)
1151            })
1152            .expect("desc journaled index range traversal should succeed");
1153        assert_eq!(desc, vec![5, 4]);
1154        assert_eq!(
1155            journaled_snapshot_call_count_for_tests(),
1156            0,
1157            "mixed reverse journaled index range traversal should preserve early stop without materializing a snapshot",
1158        );
1159    }
1160}
icydb_core/db/index/store.rs

icydb_core/db/index/
store.rs
