Skip to main content

icydb_core/db/data/
store.rs

1//! Module: data::store
2//! Responsibility: journaled-or-heap row storage behind the data-store boundary.
3//! Does not own: key/row validation policy beyond type boundaries.
4//! Boundary: commit/executor call into this layer after prevalidation.
5
6use crate::{
7    db::{
8        data::{CanonicalRow, RawDataStoreKey, RawRow},
9        direction::Direction,
10        ordered_overlay::{OrderedOverlayEntry, ordered_overlay_entries},
11        positioned_overlay::{
12            JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
13        },
14    },
15    types::EntityTag,
16};
17use ic_memory::{DefaultMemoryImpl, Memory, RuntimeMemory};
18use ic_storage::adapter::StableBTreeMap;
19#[cfg(test)]
20use std::cell::Cell;
21use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet, btree_map::Entry};
22use std::convert::Infallible;
23use std::ops::{Bound, RangeBounds};
24
25#[cfg(test)]
26thread_local! {
27    static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
28}
29
30#[cfg(test)]
31fn record_data_store_get_call() {
32    DATA_STORE_GET_CALL_COUNT.with(|count| {
33        count.set(count.get().saturating_add(1));
34    });
35}
36
37///
38/// DataStore
39///
40/// Thin persistence wrapper over one journaled or heap BTreeMap.
41///
42/// Invariant: callers provide already-validated `RawDataStoreKey` and canonical row bytes.
43/// This type intentionally does not enforce commit-phase ordering.
44///
45
46pub struct DataStore {
47    backend: DataStoreBackend,
48    generation: u64,
49    entity_cardinality: EntityCardinality,
50}
51
52enum DataStoreBackend {
53    Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
54    Journaled {
55        canonical: StableBTreeMap<RawDataStoreKey, RawRow, DefaultMemoryImpl>,
56        live: HeapBTreeMap<RawDataStoreKey, RawRow>,
57        tombstones: BTreeSet<RawDataStoreKey>,
58        positions: PositionedOverlayMetadata<RawDataStoreKey>,
59        entity_cardinality_delta: EntityCardinalityDelta,
60    },
61}
62
63/// Preflighted provenance publication for direct journal record families.
64#[cfg(feature = "migration")]
65pub(in crate::db) struct PreparedDataPositionPublication {
66    keys: Vec<RawDataStoreKey>,
67    position: JournalOverlayPosition,
68}
69
70/// Preflighted exact retirement for one complete journal batch.
71pub(in crate::db) struct PreparedDataPositionRetirement {
72    entries: Vec<(RawDataStoreKey, PositionedOverlayRetirement)>,
73}
74
75/// One visible row read that borrows heap/live state and owns stable state.
76///
77/// Callers that only need selected fields can evaluate a borrowed row while
78/// the store handle is active. Stable-structure reads remain owned because
79/// that backend cannot expose a value reference beyond its storage call.
80pub(in crate::db) enum StoredRowRead<'a> {
81    Missing,
82    Borrowed(&'a RawRow),
83    Owned(RawRow),
84}
85
86impl StoredRowRead<'_> {
87    /// Borrow the visible row regardless of its physical backing.
88    #[must_use]
89    pub(in crate::db) const fn as_row(&self) -> Option<&RawRow> {
90        match self {
91            Self::Missing => None,
92            Self::Borrowed(row) => Some(row),
93            Self::Owned(row) => Some(row),
94        }
95    }
96
97    /// Convert the visible row into the existing owned get contract.
98    #[must_use]
99    fn into_owned(self) -> Option<RawRow> {
100        match self {
101            Self::Missing => None,
102            Self::Borrowed(row) => Some(row.clone()),
103            Self::Owned(row) => Some(row),
104        }
105    }
106}
107
108/// Control-flow result for store traversal visitors.
109#[derive(Clone, Copy, Debug, Eq, PartialEq)]
110pub(in crate::db) enum StoreVisit {
111    Continue,
112    Stop,
113}
114
115impl StoreVisit {
116    const fn should_stop(self) -> bool {
117        matches!(self, Self::Stop)
118    }
119}
120
121impl DataStore {
122    /// Initialize a volatile heap-backed data store.
123    #[must_use]
124    pub const fn init_heap() -> Self {
125        Self {
126            backend: DataStoreBackend::Heap(HeapBTreeMap::new()),
127            generation: 0,
128            entity_cardinality: EntityCardinality::empty(),
129        }
130    }
131
132    /// Initialize a journaled cached-stable data store.
133    ///
134    /// Normal writes update only the live projection. The canonical stable map
135    /// is the future fold target and is not mutated by this wrapper's write
136    /// methods.
137    ///
138    /// # Panics
139    /// Panics if retained storage has an invalid header or unsupported layout,
140    /// or the backing storage fails. Invalid retained contents are not replaced.
141    #[must_use]
142    pub fn init_journaled(memory: RuntimeMemory<DefaultMemoryImpl>) -> Self {
143        // Retained maps must never be replaced when their header is corrupt.
144        let canonical = if memory.size() == 0 {
145            StableBTreeMap::init(memory)
146        } else {
147            StableBTreeMap::load(memory)
148        };
149        let entity_cardinality = if canonical.is_empty() {
150            EntityCardinality::empty()
151        } else {
152            EntityCardinality::unavailable()
153        };
154        Self {
155            backend: DataStoreBackend::Journaled {
156                canonical,
157                live: HeapBTreeMap::new(),
158                tombstones: BTreeSet::new(),
159                positions: PositionedOverlayMetadata::new(),
160                entity_cardinality_delta: EntityCardinalityDelta::empty(),
161            },
162            generation: 0,
163            // Stable rows remain authoritative after reinitialization. Exact
164            // zero cardinality is still known for an empty canonical map;
165            // populated maps remain unavailable without a startup scan.
166            entity_cardinality,
167        }
168    }
169
170    /// Insert or replace one row by raw key.
171    pub(in crate::db) fn insert(
172        &mut self,
173        key: RawDataStoreKey,
174        row: CanonicalRow,
175    ) -> Option<RawRow> {
176        self.insert_raw(key, row.into_raw_row())
177    }
178
179    // One insertion owner for admitted writes and corruption-fixture raw rows.
180    fn insert_raw(&mut self, key: RawDataStoreKey, row: RawRow) -> Option<RawRow> {
181        // Counts retain only entity identity, so copy that fact before the map
182        // takes the row key instead of retaining a second key buffer.
183        let entity = key.entity_tag_prefix();
184        let previous = match &mut self.backend {
185            DataStoreBackend::Heap(map) => map.insert(key, row),
186            DataStoreBackend::Journaled {
187                canonical,
188                live,
189                tombstones,
190                ..
191            } => {
192                let hidden = tombstones.remove(&key);
193                // Move a live predecessor; read stable bytes only when visible
194                // and not overridden. Neither path clones the live row payload.
195                match live.entry(key) {
196                    Entry::Occupied(mut entry) => {
197                        let previous = entry.insert(row);
198                        (!hidden).then_some(previous)
199                    }
200                    Entry::Vacant(entry) => {
201                        let previous = if hidden {
202                            None
203                        } else {
204                            canonical.get(entry.key())
205                        };
206                        entry.insert(row);
207                        previous
208                    }
209                }
210            }
211        };
212        self.entity_cardinality
213            .apply_insert(entity, previous.is_some());
214        self.apply_entity_overlay_delta(entity, previous.is_some(), true);
215        self.bump_generation();
216        previous
217    }
218
219    /// Insert one raw row directly for corruption-focused test setup only.
220    #[cfg(test)]
221    pub(in crate::db) fn insert_raw_for_test(
222        &mut self,
223        key: RawDataStoreKey,
224        row: RawRow,
225    ) -> Option<RawRow> {
226        self.insert_raw(key, row)
227    }
228
229    /// Remove one row by raw key.
230    pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
231        let previous = match &mut self.backend {
232            DataStoreBackend::Heap(map) => map.remove(key),
233            DataStoreBackend::Journaled {
234                canonical,
235                live,
236                tombstones,
237                ..
238            } => {
239                let previous = live.remove(key);
240                let previous = if tombstones.contains(key) {
241                    None
242                } else {
243                    previous.or_else(|| canonical.get(key))
244                };
245                tombstones.insert(key.clone());
246                previous
247            }
248        };
249        self.entity_cardinality
250            .apply_remove(key.entity_tag_prefix(), previous.is_some());
251        self.apply_entity_overlay_delta(key.entity_tag_prefix(), previous.is_some(), false);
252        self.bump_generation();
253        previous
254    }
255
256    /// Reset the volatile projection for journaled recovery without mutating
257    /// the canonical stable base.
258    pub(in crate::db) fn reset_journaled_live_projection(
259        &mut self,
260    ) -> Result<(), crate::error::InternalError> {
261        let DataStoreBackend::Journaled {
262            canonical,
263            live,
264            tombstones,
265            positions,
266            entity_cardinality_delta,
267        } = &mut self.backend
268        else {
269            return Err(crate::error::InternalError::store_invariant());
270        };
271
272        live.clear();
273        tombstones.clear();
274        positions.clear();
275        *entity_cardinality_delta = EntityCardinalityDelta::empty();
276        self.entity_cardinality = if canonical.is_empty() {
277            EntityCardinality::empty()
278        } else {
279            EntityCardinality::unavailable()
280        };
281        self.bump_generation();
282
283        Ok(())
284    }
285
286    /// Apply one recovered journal row put into the volatile projection.
287    #[cfg(any(test, feature = "migration"))]
288    pub(in crate::db) fn apply_recovered_journal_put(
289        &mut self,
290        key: RawDataStoreKey,
291        row: RawRow,
292    ) -> Result<(), crate::error::InternalError> {
293        // Replay callers discard predecessor bytes; cardinality needs presence only.
294        let previous_present = self.contains(&key);
295        let DataStoreBackend::Journaled {
296            live, tombstones, ..
297        } = &mut self.backend
298        else {
299            return Err(crate::error::InternalError::store_invariant());
300        };
301
302        tombstones.remove(&key);
303        let entity = key.entity_tag_prefix();
304        live.insert(key, row);
305        self.entity_cardinality
306            .apply_insert(entity, previous_present);
307        self.apply_entity_overlay_delta(entity, previous_present, true);
308        self.bump_generation();
309
310        Ok(())
311    }
312
313    /// Publish one preflighted positioned row value or tombstone.
314    pub(in crate::db) fn publish_preflighted_journal_entry(
315        &mut self,
316        key: RawDataStoreKey,
317        row: Option<RawRow>,
318        position: JournalOverlayPosition,
319    ) -> Result<(), crate::error::InternalError> {
320        // Cardinality needs presence only; never materialize a previous row.
321        let previous_present = self.contains(&key);
322        let DataStoreBackend::Journaled {
323            live,
324            tombstones,
325            positions,
326            entity_cardinality_delta,
327            ..
328        } = &mut self.backend
329        else {
330            return Err(crate::error::InternalError::store_invariant());
331        };
332        let next_present = row.is_some();
333        let entity = key.entity_tag_prefix();
334        if let Some(row) = row {
335            tombstones.remove(&key);
336            live.insert(key.clone(), row);
337            self.entity_cardinality
338                .apply_insert(entity, previous_present);
339        } else {
340            live.remove(&key);
341            tombstones.insert(key.clone());
342            self.entity_cardinality
343                .apply_remove(entity, previous_present);
344        }
345        entity_cardinality_delta.apply_presence_transition(entity, previous_present, next_present);
346        positions.publish_preflighted(key, position);
347        self.bump_generation();
348
349        Ok(())
350    }
351
352    /// Validate and publish one positioned row for direct store tests.
353    #[cfg(test)]
354    pub(in crate::db) fn publish_positioned_journal_entry(
355        &mut self,
356        key: RawDataStoreKey,
357        row: Option<RawRow>,
358        position: JournalOverlayPosition,
359    ) -> Result<(), crate::error::InternalError> {
360        self.preflight_positioned_journal_entry(&key, position)?;
361        self.publish_preflighted_journal_entry(key, row, position)
362    }
363
364    /// Preflight row provenance before marker publication.
365    pub(in crate::db) fn preflight_positioned_journal_entry(
366        &self,
367        key: &RawDataStoreKey,
368        position: JournalOverlayPosition,
369    ) -> Result<(), crate::error::InternalError> {
370        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
371            return Err(crate::error::InternalError::store_invariant());
372        };
373        positions.preflight_publish(key, position)
374    }
375
376    /// Preflight direct row provenance before marker publication.
377    #[cfg(feature = "migration")]
378    pub(in crate::db) fn prepare_position_publication(
379        &self,
380        keys: impl IntoIterator<Item = RawDataStoreKey>,
381        position: JournalOverlayPosition,
382    ) -> Result<PreparedDataPositionPublication, crate::error::InternalError> {
383        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
384            return Err(crate::error::InternalError::store_invariant());
385        };
386        let keys = keys.into_iter().collect::<BTreeSet<_>>();
387        for key in &keys {
388            positions.preflight_publish(key, position)?;
389        }
390        Ok(PreparedDataPositionPublication {
391            keys: keys.into_iter().collect(),
392            position,
393        })
394    }
395
396    /// Publish direct row provenance after its values have been applied.
397    #[cfg(feature = "migration")]
398    pub(in crate::db) fn publish_prepared_positions(
399        &mut self,
400        prepared: PreparedDataPositionPublication,
401    ) {
402        let DataStoreBackend::Journaled { positions, .. } = &mut self.backend else {
403            debug_assert!(false, "preflighted row positions require a journaled store");
404            return;
405        };
406        for key in prepared.keys {
407            positions.publish_preflighted(key, prepared.position);
408        }
409    }
410
411    /// Preflight exact row-overlay retirement before canonical mutation.
412    pub(in crate::db) fn prepare_position_retirement(
413        &self,
414        keys: impl IntoIterator<Item = RawDataStoreKey>,
415        position: JournalOverlayPosition,
416    ) -> Result<PreparedDataPositionRetirement, crate::error::InternalError> {
417        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
418            return Err(crate::error::InternalError::store_invariant());
419        };
420        let entries = keys
421            .into_iter()
422            .collect::<BTreeSet<_>>()
423            .into_iter()
424            .map(|key| {
425                positions
426                    .preflight_retirement(&key, position)
427                    .map(|retirement| (key, retirement))
428            })
429            .collect::<Result<Vec<_>, _>>()?;
430        Ok(PreparedDataPositionRetirement { entries })
431    }
432
433    /// Retire only exact row overlays after canonical mutation succeeds.
434    pub(in crate::db) fn apply_prepared_position_retirement(
435        &mut self,
436        prepared: PreparedDataPositionRetirement,
437    ) {
438        let DataStoreBackend::Journaled {
439            live,
440            tombstones,
441            positions,
442            ..
443        } = &mut self.backend
444        else {
445            debug_assert!(
446                false,
447                "preflighted row retirement requires a journaled store"
448            );
449            return;
450        };
451        positions.retire_overlays_preflighted(prepared.entries, live, tombstones);
452    }
453
454    /// Apply one folded journal row put into the canonical stable base.
455    pub(in crate::db) fn fold_recovered_journal_put(
456        &mut self,
457        key: RawDataStoreKey,
458        row: RawRow,
459    ) -> Result<Option<RawRow>, crate::error::InternalError> {
460        let DataStoreBackend::Journaled {
461            canonical,
462            live,
463            tombstones,
464            ..
465        } = &mut self.backend
466        else {
467            return Err(crate::error::InternalError::store_invariant());
468        };
469
470        let visible = !live.contains_key(&key) && !tombstones.contains(&key);
471        let entity = key.entity_tag_prefix();
472        let previous = canonical.insert(key, row);
473        if visible {
474            self.entity_cardinality
475                .apply_insert(entity, previous.is_some());
476        } else if previous.is_none() {
477            self.apply_entity_overlay_delta(entity, true, false);
478        }
479        self.bump_generation();
480
481        Ok(previous)
482    }
483
484    /// Apply one folded journal row delete into the canonical stable base.
485    pub(in crate::db) fn fold_recovered_journal_delete(
486        &mut self,
487        key: &RawDataStoreKey,
488    ) -> Result<Option<RawRow>, crate::error::InternalError> {
489        let DataStoreBackend::Journaled {
490            canonical,
491            live,
492            tombstones,
493            ..
494        } = &mut self.backend
495        else {
496            return Err(crate::error::InternalError::store_invariant());
497        };
498
499        let visible = !live.contains_key(key) && !tombstones.contains(key);
500        let previous = canonical.remove(key);
501        if visible {
502            self.entity_cardinality
503                .apply_remove(key.entity_tag_prefix(), previous.is_some());
504        } else if previous.is_some() {
505            self.apply_entity_overlay_delta(key.entity_tag_prefix(), false, true);
506        }
507        self.bump_generation();
508
509        Ok(previous)
510    }
511
512    /// Prove that recovered journal rows can be folded into canonical storage.
513    pub(in crate::db) fn preflight_fold_recovered_journal(
514        &self,
515    ) -> Result<(), crate::error::InternalError> {
516        match self.backend {
517            DataStoreBackend::Journaled { .. } => Ok(()),
518            DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
519        }
520    }
521
522    /// Load one row by raw key.
523    pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
524        self.read(key).into_owned()
525    }
526
527    /// Load one row from the canonical predecessor view.
528    ///
529    /// Online journal folding uses this view so a newer positioned live effect
530    /// cannot replace the predecessor evidence for the older batch being
531    /// canonicalized.
532    pub(in crate::db) fn get_canonical(&self, key: &RawDataStoreKey) -> Option<RawRow> {
533        match &self.backend {
534            DataStoreBackend::Heap(map) => map.get(key).cloned(),
535            DataStoreBackend::Journaled { canonical, .. } => canonical.get(key),
536        }
537    }
538
539    /// Test canonical predecessor presence without loading its row payload.
540    /// Positioned live updates and tombstones do not change this view.
541    #[must_use]
542    pub(in crate::db) fn contains_canonical(&self, key: &RawDataStoreKey) -> bool {
543        match &self.backend {
544            DataStoreBackend::Heap(map) => map.contains_key(key),
545            DataStoreBackend::Journaled { canonical, .. } => canonical.contains_key(key),
546        }
547    }
548
549    /// Read one visible row without cloning heap/live payloads.
550    pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
551        #[cfg(test)]
552        record_data_store_get_call();
553
554        match &self.backend {
555            DataStoreBackend::Heap(map) => map
556                .get(key)
557                .map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
558            DataStoreBackend::Journaled {
559                canonical,
560                live,
561                tombstones,
562                ..
563            } => {
564                if tombstones.contains(key) {
565                    StoredRowRead::Missing
566                } else if let Some(row) = live.get(key) {
567                    StoredRowRead::Borrowed(row)
568                } else {
569                    canonical
570                        .get(key)
571                        .map_or(StoredRowRead::Missing, StoredRowRead::Owned)
572                }
573            }
574        }
575    }
576
577    /// Return whether one raw key exists without cloning the row payload.
578    #[must_use]
579    pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
580        match &self.backend {
581            DataStoreBackend::Heap(map) => map.contains_key(key),
582            DataStoreBackend::Journaled {
583                canonical,
584                live,
585                tombstones,
586                ..
587            } => {
588                !tombstones.contains(key) && (live.contains_key(key) || canonical.contains_key(key))
589            }
590        }
591    }
592
593    /// Return whether the effective row domain is empty without reading payloads.
594    ///
595    /// Stop at the first visible key. Journal tombstones may hide canonical
596    /// keys, so physical canonical emptiness alone is not sufficient.
597    #[must_use]
598    pub(in crate::db) fn is_empty(&self) -> bool {
599        let mut empty = true;
600        let _: Result<(), Infallible> = self.visit_key_range(.., |_key| {
601            empty = false;
602            Ok(StoreVisit::Stop)
603        });
604        empty
605    }
606
607    /// Return an exact row count for tests without exposing the backing map.
608    #[cfg(test)]
609    #[must_use]
610    pub(in crate::db) fn len(&self) -> u64 {
611        match &self.backend {
612            DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
613            DataStoreBackend::Journaled { .. } => {
614                let mut count = 0_u64;
615                let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
616                    count = count.saturating_add(1);
617                    Ok(StoreVisit::Continue)
618                });
619                count
620            }
621        }
622    }
623
624    /// Return the row-store generation used to prove index metadata freshness.
625    #[must_use]
626    pub(in crate::db) const fn generation(&self) -> u64 {
627        self.generation
628    }
629
630    /// Return an exact current row count for one entity when store metadata is valid.
631    #[must_use]
632    pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
633        self.entity_cardinality.exact_count(entity)
634    }
635
636    /// Return the exact live-overlay delta from the canonical base for one entity.
637    #[must_use]
638    pub(in crate::db) fn exact_entity_cardinality_delta(&self, entity: EntityTag) -> Option<i64> {
639        match &self.backend {
640            DataStoreBackend::Heap(_) => Some(0),
641            DataStoreBackend::Journaled {
642                entity_cardinality_delta,
643                ..
644            } => entity_cardinality_delta.exact_delta(entity),
645        }
646    }
647
648    /// Visit raw row entries in canonical storage order.
649    pub(in crate::db) fn visit_entries<E>(
650        &self,
651        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
652    ) -> Result<(), E> {
653        match &self.backend {
654            DataStoreBackend::Heap(map) => {
655                for (key, row) in map {
656                    if visitor(key, row)?.should_stop() {
657                        break;
658                    }
659                }
660            }
661            DataStoreBackend::Journaled { .. } => Self::visit_journaled_entries_in_bounds(
662                &self.backend,
663                (Bound::Unbounded, Bound::Unbounded),
664                &mut visitor,
665            )?,
666        }
667
668        Ok(())
669    }
670
671    /// Visit canonical predecessor rows after one exclusive physical checkpoint.
672    ///
673    /// The visitor returns true before consuming the current row. On that stop,
674    /// return the preceding consumed key; exhaustion returns no checkpoint.
675    /// Retain the previous entry until the next admission decision so only the
676    /// returned checkpoint copies key bytes, including for stable iterators.
677    /// Optional derived-evidence builders use this canonical view so newer live
678    /// overlays cannot contaminate a generation bound to its fold watermark.
679    pub(in crate::db) fn visit_canonical_entries_after(
680        &self,
681        checkpoint: Option<&RawDataStoreKey>,
682        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<bool, crate::error::InternalError>,
683    ) -> Result<Option<RawDataStoreKey>, crate::error::InternalError> {
684        let lower = checkpoint.map_or(Bound::Unbounded, Bound::Excluded);
685        match &self.backend {
686            DataStoreBackend::Heap(map) => {
687                let mut previous: Option<&RawDataStoreKey> = None;
688                for (key, row) in map.range((lower, Bound::Unbounded)) {
689                    if visitor(key, row)? {
690                        return Ok(previous.cloned());
691                    }
692                    previous = Some(key);
693                }
694                Ok(None)
695            }
696            DataStoreBackend::Journaled { canonical, .. } => {
697                let mut previous = None;
698                let mut stopped = false;
699                for entry in canonical.range((lower, Bound::Unbounded)) {
700                    if visitor(entry.key(), &entry.value())? {
701                        stopped = true;
702                        break;
703                    }
704                    previous = Some(entry);
705                }
706                Ok(if stopped {
707                    previous.map(|entry| entry.key().clone())
708                } else {
709                    None
710                })
711            }
712        }
713    }
714
715    /// Return whether the canonical predecessor domain is physically empty.
716    ///
717    /// This bounded root observation never walks live overlays or materializes
718    /// row payloads.
719    pub(in crate::db) fn canonical_is_empty(&self) -> Result<bool, crate::error::InternalError> {
720        match &self.backend {
721            DataStoreBackend::Journaled { canonical, .. } => Ok(canonical.is_empty()),
722            DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
723        }
724    }
725
726    /// Visit raw row entries whose keys belong to the provided storage range.
727    pub(in crate::db) fn visit_range<E>(
728        &self,
729        key_range: impl RangeBounds<RawDataStoreKey>,
730        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
731    ) -> Result<(), E> {
732        let bounds = (key_range.start_bound(), key_range.end_bound());
733        match &self.backend {
734            DataStoreBackend::Heap(map) => {
735                for (key, row) in map.range(bounds) {
736                    if visitor(key, row)?.should_stop() {
737                        break;
738                    }
739                }
740            }
741            DataStoreBackend::Journaled { .. } => {
742                Self::visit_journaled_entries_in_bounds(&self.backend, bounds, &mut visitor)?;
743            }
744        }
745
746        Ok(())
747    }
748
749    /// Visit one directed row range while allowing the caller to stop after
750    /// seeing the key but before a stable row payload is materialized.
751    ///
752    /// Mixed journal overlays retain the ordinary key-then-point-read path;
753    /// one physical backing can keep the range iterator open for the complete
754    /// scan and avoid a second tree lookup per row.
755    pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
756        &self,
757        key_range: impl RangeBounds<RawDataStoreKey>,
758        direction: Direction,
759        mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
760        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
761    ) -> Result<Option<bool>, E> {
762        let bounds = (key_range.start_bound(), key_range.end_bound());
763        let mut stopped = false;
764        match &self.backend {
765            DataStoreBackend::Heap(map) => {
766                let mut entries = map.range(bounds);
767                while let Some((key, row)) = match direction {
768                    Direction::Asc => entries.next(),
769                    Direction::Desc => entries.next_back(),
770                } {
771                    if preflight(key)?.should_stop() {
772                        stopped = true;
773                        break;
774                    }
775                    if visitor(key, row)?.should_stop() {
776                        stopped = true;
777                        break;
778                    }
779                }
780            }
781            DataStoreBackend::Journaled {
782                canonical,
783                live,
784                tombstones,
785                ..
786            } if canonical.is_empty() => {
787                let mut entries = live.range(bounds);
788                while let Some((key, row)) = match direction {
789                    Direction::Asc => entries.next(),
790                    Direction::Desc => entries.next_back(),
791                } {
792                    if tombstones.contains(key) {
793                        continue;
794                    }
795                    if preflight(key)?.should_stop() {
796                        stopped = true;
797                        break;
798                    }
799                    if visitor(key, row)?.should_stop() {
800                        stopped = true;
801                        break;
802                    }
803                }
804            }
805            DataStoreBackend::Journaled {
806                canonical,
807                live,
808                tombstones,
809                ..
810            } if live.is_empty() && tombstones.is_empty() => {
811                let mut entries = canonical.range(bounds);
812                while let Some(entry) = match direction {
813                    Direction::Asc => entries.next(),
814                    Direction::Desc => entries.next_back(),
815                } {
816                    if preflight(entry.key())?.should_stop() {
817                        stopped = true;
818                        break;
819                    }
820                    if visitor(entry.key(), &entry.value())?.should_stop() {
821                        stopped = true;
822                        break;
823                    }
824                }
825            }
826            DataStoreBackend::Journaled { .. } => return Ok(None),
827        }
828
829        Ok(Some(!stopped))
830    }
831
832    /// Visit only raw keys in storage order without fetching row payloads.
833    ///
834    /// Primary-key access streams use this boundary to discover candidate
835    /// identities before the terminal row runtime decides whether the payload
836    /// is needed. Journaled traversal merges canonical and live keys while
837    /// preserving live overrides and tombstones without reading stable values.
838    pub(in crate::db) fn visit_key_range<E>(
839        &self,
840        key_range: impl RangeBounds<RawDataStoreKey>,
841        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
842    ) -> Result<(), E> {
843        self.visit_keys_in_bounds(
844            (key_range.start_bound(), key_range.end_bound()),
845            false,
846            visitor,
847        )
848    }
849
850    /// Visit only raw keys in reverse storage order without fetching row payloads.
851    pub(in crate::db) fn visit_key_range_rev<E>(
852        &self,
853        key_range: impl RangeBounds<RawDataStoreKey>,
854        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
855    ) -> Result<(), E> {
856        self.visit_keys_in_bounds(
857            (key_range.start_bound(), key_range.end_bound()),
858            true,
859            visitor,
860        )
861    }
862
863    /// Independent physical byte baseline for storage-report tests.
864    #[cfg(test)]
865    pub(in crate::db) fn memory_bytes(&self) -> u64 {
866        // Report map footprint as key bytes + row bytes per entry.
867        let mut bytes = 0u64;
868        let _: Result<(), Infallible> = self.visit_entries(|key, row| {
869            bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
870            Ok(StoreVisit::Continue)
871        });
872        bytes
873    }
874
875    const fn bump_generation(&mut self) {
876        self.generation = self.generation.saturating_add(1);
877    }
878
879    #[cfg(test)]
880    fn rebuild_entity_cardinality_from_entries(&mut self) {
881        let mut cardinality = EntityCardinality::empty();
882        let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
883            cardinality.apply_present_entity(key.entity_tag_prefix());
884            Ok(StoreVisit::Continue)
885        });
886        self.entity_cardinality = cardinality;
887    }
888
889    fn apply_entity_overlay_delta(
890        &mut self,
891        entity: Option<EntityTag>,
892        previous_present: bool,
893        next_present: bool,
894    ) {
895        let DataStoreBackend::Journaled {
896            entity_cardinality_delta,
897            ..
898        } = &mut self.backend
899        else {
900            return;
901        };
902        entity_cardinality_delta.apply_presence_transition(entity, previous_present, next_present);
903    }
904
905    /// Return the monotonic perf-only count of stable row fetches seen by this process.
906    #[cfg(test)]
907    pub(in crate::db) fn current_get_call_count() -> u64 {
908        DATA_STORE_GET_CALL_COUNT.with(Cell::get)
909    }
910
911    fn visit_keys_in_bounds<E>(
912        &self,
913        bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
914        reverse: bool,
915        mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
916    ) -> Result<(), E> {
917        match &self.backend {
918            DataStoreBackend::Heap(map) => {
919                if reverse {
920                    for (key, _row) in map.range(bounds).rev() {
921                        if visitor(key)?.should_stop() {
922                            break;
923                        }
924                    }
925                } else {
926                    for (key, _row) in map.range(bounds) {
927                        if visitor(key)?.should_stop() {
928                            break;
929                        }
930                    }
931                }
932            }
933            DataStoreBackend::Journaled { .. } => {
934                Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, &mut visitor)?;
935            }
936        }
937
938        Ok(())
939    }
940
941    // Share journaled traversal across callback types while retaining typed
942    // caller errors. Key-only traversal must never materialize row payloads.
943    fn visit_journaled_keys_in_bounds<E>(
944        backend: &DataStoreBackend,
945        bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
946        reverse: bool,
947        visitor: &mut dyn FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
948    ) -> Result<(), E> {
949        let DataStoreBackend::Journaled {
950            canonical,
951            live,
952            tombstones,
953            ..
954        } = backend
955        else {
956            return Ok(());
957        };
958
959        if canonical.is_empty() {
960            if reverse {
961                for (key, _row) in live.range(bounds).rev() {
962                    if visitor(key)?.should_stop() {
963                        return Ok(());
964                    }
965                }
966            } else {
967                for (key, _row) in live.range(bounds) {
968                    if visitor(key)?.should_stop() {
969                        return Ok(());
970                    }
971                }
972            }
973            return Ok(());
974        }
975
976        if live.is_empty() && tombstones.is_empty() {
977            if reverse {
978                for entry in canonical.range(bounds).rev() {
979                    if visitor(entry.key())?.should_stop() {
980                        return Ok(());
981                    }
982                }
983            } else {
984                for entry in canonical.range(bounds) {
985                    if visitor(entry.key())?.should_stop() {
986                        return Ok(());
987                    }
988                }
989            }
990            return Ok(());
991        }
992
993        let direction = if reverse {
994            Direction::Desc
995        } else {
996            Direction::Asc
997        };
998        match direction {
999            Direction::Asc => {
1000                for entry in ordered_overlay_entries(
1001                    canonical.range(bounds),
1002                    live.range(bounds),
1003                    direction,
1004                    |entry| entry.key(),
1005                    |entry| entry.0,
1006                    tombstones,
1007                ) {
1008                    let visit = match entry {
1009                        OrderedOverlayEntry::Canonical(canonical_entry) => {
1010                            visitor(canonical_entry.key())?
1011                        }
1012                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
1013                    };
1014                    if visit.should_stop() {
1015                        return Ok(());
1016                    }
1017                }
1018            }
1019            Direction::Desc => {
1020                for entry in ordered_overlay_entries(
1021                    canonical.range(bounds).rev(),
1022                    live.range(bounds).rev(),
1023                    direction,
1024                    |entry| entry.key(),
1025                    |entry| entry.0,
1026                    tombstones,
1027                ) {
1028                    let visit = match entry {
1029                        OrderedOverlayEntry::Canonical(canonical_entry) => {
1030                            visitor(canonical_entry.key())?
1031                        }
1032                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
1033                    };
1034                    if visit.should_stop() {
1035                        return Ok(());
1036                    }
1037                }
1038            }
1039        }
1040
1041        Ok(())
1042    }
1043
1044    // Keep one journaled row traversal body per error type, independently of
1045    // each caller's closure. The ordered overlay remains the merge authority.
1046    fn visit_journaled_entries_in_bounds<E>(
1047        backend: &DataStoreBackend,
1048        bounds: (Bound<&RawDataStoreKey>, Bound<&RawDataStoreKey>),
1049        visitor: &mut dyn FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
1050    ) -> Result<(), E> {
1051        let DataStoreBackend::Journaled {
1052            canonical,
1053            live,
1054            tombstones,
1055            ..
1056        } = backend
1057        else {
1058            return Ok(());
1059        };
1060
1061        if canonical.is_empty() {
1062            for (key, row) in live.range(bounds) {
1063                if visitor(key, row)?.should_stop() {
1064                    return Ok(());
1065                }
1066            }
1067            return Ok(());
1068        }
1069
1070        if live.is_empty() && tombstones.is_empty() {
1071            for entry in canonical.range(bounds) {
1072                if visitor(entry.key(), &entry.value())?.should_stop() {
1073                    return Ok(());
1074                }
1075            }
1076            return Ok(());
1077        }
1078
1079        for entry in ordered_overlay_entries(
1080            canonical.range(bounds),
1081            live.range(bounds),
1082            Direction::Asc,
1083            |entry| entry.key(),
1084            |entry| entry.0,
1085            tombstones,
1086        ) {
1087            let visit = match entry {
1088                OrderedOverlayEntry::Canonical(canonical_entry) => {
1089                    visitor(canonical_entry.key(), &canonical_entry.value())?
1090                }
1091                OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
1092            };
1093            if visit.should_stop() {
1094                return Ok(());
1095            }
1096        }
1097
1098        Ok(())
1099    }
1100}
1101
1102#[derive(Clone, Debug)]
1103struct EntityCardinality {
1104    counts: HeapBTreeMap<EntityTag, u64>,
1105    decodable: bool,
1106}
1107
1108#[derive(Clone, Debug)]
1109struct EntityCardinalityDelta {
1110    counts: HeapBTreeMap<EntityTag, i64>,
1111    decodable: bool,
1112}
1113
1114impl EntityCardinalityDelta {
1115    const fn empty() -> Self {
1116        Self {
1117            counts: HeapBTreeMap::new(),
1118            decodable: true,
1119        }
1120    }
1121
1122    fn exact_delta(&self, entity: EntityTag) -> Option<i64> {
1123        self.decodable
1124            .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1125    }
1126
1127    fn apply_presence_transition(
1128        &mut self,
1129        entity: Option<EntityTag>,
1130        previous_present: bool,
1131        next_present: bool,
1132    ) {
1133        if !self.decodable || previous_present == next_present {
1134            return;
1135        }
1136        let Some(entity) = entity else {
1137            self.counts.clear();
1138            self.decodable = false;
1139            return;
1140        };
1141        let delta = if next_present { 1_i64 } else { -1_i64 };
1142        let count = self.counts.entry(entity).or_insert(0);
1143        let Some(next) = count.checked_add(delta) else {
1144            self.counts.clear();
1145            self.decodable = false;
1146            return;
1147        };
1148        *count = next;
1149        if *count == 0 {
1150            self.counts.remove(&entity);
1151        }
1152    }
1153}
1154
1155impl EntityCardinality {
1156    const fn empty() -> Self {
1157        Self {
1158            counts: HeapBTreeMap::new(),
1159            decodable: true,
1160        }
1161    }
1162
1163    const fn unavailable() -> Self {
1164        Self {
1165            counts: HeapBTreeMap::new(),
1166            decodable: false,
1167        }
1168    }
1169
1170    fn exact_count(&self, entity: EntityTag) -> Option<u64> {
1171        self.decodable
1172            .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1173    }
1174
1175    fn apply_insert(&mut self, entity: Option<EntityTag>, previous_present: bool) {
1176        if previous_present {
1177            return;
1178        }
1179        self.apply_present_entity(entity);
1180    }
1181
1182    fn apply_remove(&mut self, entity: Option<EntityTag>, previous_present: bool) {
1183        if !previous_present {
1184            return;
1185        }
1186        self.apply_removed_entity(entity);
1187    }
1188
1189    fn apply_present_entity(&mut self, entity: Option<EntityTag>) {
1190        if !self.decodable {
1191            return;
1192        }
1193        let Some(entity) = entity else {
1194            self.invalidate();
1195            return;
1196        };
1197
1198        let count = self.counts.entry(entity).or_insert(0);
1199        *count = count.saturating_add(1);
1200    }
1201
1202    fn apply_removed_entity(&mut self, entity: Option<EntityTag>) {
1203        if !self.decodable {
1204            return;
1205        }
1206        let Some(entity) = entity else {
1207            self.invalidate();
1208            return;
1209        };
1210
1211        if let Some(count) = self.counts.get_mut(&entity) {
1212            *count = count.saturating_sub(1);
1213            if *count == 0 {
1214                self.counts.remove(&entity);
1215            }
1216        }
1217    }
1218
1219    fn invalidate(&mut self) {
1220        self.counts.clear();
1221        self.decodable = false;
1222    }
1223}
1224
1225#[cfg(test)]
1226mod tests;