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

icydb_core/db/index/
store.rs
