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, OrderedOverlayVisit, visit_ordered_overlay},
11        positioned_overlay::{
12            JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
13        },
14    },
15    types::EntityTag,
16};
17use ic_stable_structures::{
18    BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
19};
20#[cfg(all(feature = "sql", feature = "diagnostics"))]
21use std::cell::Cell;
22use std::collections::{BTreeMap as HeapBTreeMap, BTreeSet};
23use std::convert::Infallible;
24use std::ops::{Bound, RangeBounds};
25
26#[cfg(all(feature = "sql", feature = "diagnostics"))]
27thread_local! {
28    static DATA_STORE_GET_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
29}
30
31#[cfg(all(feature = "sql", feature = "diagnostics"))]
32fn record_data_store_get_call() {
33    DATA_STORE_GET_CALL_COUNT.with(|count| {
34        count.set(count.get().saturating_add(1));
35    });
36}
37
38///
39/// DataStore
40///
41/// Thin persistence wrapper over one journaled or heap BTreeMap.
42///
43/// Invariant: callers provide already-validated `RawDataStoreKey` and canonical row bytes.
44/// This type intentionally does not enforce commit-phase ordering.
45///
46
47pub struct DataStore {
48    backend: DataStoreBackend,
49    generation: u64,
50    entity_cardinality: EntityCardinality,
51}
52
53enum DataStoreBackend {
54    Heap(HeapBTreeMap<RawDataStoreKey, RawRow>),
55    Journaled {
56        canonical: StableBTreeMap<RawDataStoreKey, RawRow, VirtualMemory<DefaultMemoryImpl>>,
57        live: HeapBTreeMap<RawDataStoreKey, RawRow>,
58        tombstones: BTreeSet<RawDataStoreKey>,
59        positions: PositionedOverlayMetadata<RawDataStoreKey>,
60    },
61}
62
63/// Preflighted provenance publication for direct journal record families.
64#[cfg(any(test, 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    #[must_use]
138    pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
139        let canonical = StableBTreeMap::init(memory);
140        let entity_cardinality = if canonical.is_empty() {
141            EntityCardinality::empty()
142        } else {
143            EntityCardinality::unavailable()
144        };
145        Self {
146            backend: DataStoreBackend::Journaled {
147                canonical,
148                live: HeapBTreeMap::new(),
149                tombstones: BTreeSet::new(),
150                positions: PositionedOverlayMetadata::new(),
151            },
152            generation: 0,
153            // Stable rows remain authoritative after reinitialization. Exact
154            // zero cardinality is still known for an empty canonical map;
155            // populated maps remain unavailable without a startup scan.
156            entity_cardinality,
157        }
158    }
159
160    /// Insert or replace one row by raw key.
161    pub(in crate::db) fn insert(
162        &mut self,
163        key: RawDataStoreKey,
164        row: CanonicalRow,
165    ) -> Option<RawRow> {
166        let row = row.into_raw_row();
167        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
168            self.get(&key)
169        } else {
170            None
171        };
172        let cardinality_key = key.clone();
173        let previous = match &mut self.backend {
174            DataStoreBackend::Heap(map) => map.insert(key, row),
175            DataStoreBackend::Journaled {
176                live, tombstones, ..
177            } => {
178                tombstones.remove(&key);
179                live.insert(key, row);
180                previous_journaled
181            }
182        };
183        self.entity_cardinality
184            .apply_insert(&cardinality_key, previous.as_ref());
185        self.bump_generation();
186        previous
187    }
188
189    /// Insert one raw row directly for corruption-focused test setup only.
190    #[cfg(test)]
191    pub(in crate::db) fn insert_raw_for_test(
192        &mut self,
193        key: RawDataStoreKey,
194        row: RawRow,
195    ) -> Option<RawRow> {
196        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
197            self.get(&key)
198        } else {
199            None
200        };
201        let cardinality_key = key.clone();
202        let previous = match &mut self.backend {
203            DataStoreBackend::Heap(map) => map.insert(key, row),
204            DataStoreBackend::Journaled {
205                live, tombstones, ..
206            } => {
207                tombstones.remove(&key);
208                live.insert(key, row);
209                previous_journaled
210            }
211        };
212        self.entity_cardinality
213            .apply_insert(&cardinality_key, previous.as_ref());
214        self.bump_generation();
215        previous
216    }
217
218    /// Remove one row by raw key.
219    pub(in crate::db) fn remove(&mut self, key: &RawDataStoreKey) -> Option<RawRow> {
220        let previous_journaled = if matches!(self.backend, DataStoreBackend::Journaled { .. }) {
221            self.get(key)
222        } else {
223            None
224        };
225        let previous = match &mut self.backend {
226            DataStoreBackend::Heap(map) => map.remove(key),
227            DataStoreBackend::Journaled {
228                live, tombstones, ..
229            } => {
230                live.remove(key);
231                tombstones.insert(key.clone());
232                previous_journaled
233            }
234        };
235        self.entity_cardinality.apply_remove(key, previous.as_ref());
236        self.bump_generation();
237        previous
238    }
239
240    /// Reset the volatile projection for journaled recovery without mutating
241    /// the canonical stable base.
242    pub(in crate::db) fn reset_journaled_live_projection(
243        &mut self,
244    ) -> Result<(), crate::error::InternalError> {
245        let DataStoreBackend::Journaled {
246            canonical,
247            live,
248            tombstones,
249            positions,
250        } = &mut self.backend
251        else {
252            return Err(crate::error::InternalError::store_invariant());
253        };
254
255        live.clear();
256        tombstones.clear();
257        positions.clear();
258        self.entity_cardinality = if canonical.is_empty() {
259            EntityCardinality::empty()
260        } else {
261            EntityCardinality::unavailable()
262        };
263        self.bump_generation();
264
265        Ok(())
266    }
267
268    /// Apply one recovered journal row put into the volatile projection.
269    pub(in crate::db) fn apply_recovered_journal_put(
270        &mut self,
271        key: RawDataStoreKey,
272        row: RawRow,
273    ) -> Result<Option<RawRow>, crate::error::InternalError> {
274        let DataStoreBackend::Journaled {
275            canonical,
276            live,
277            tombstones,
278            ..
279        } = &mut self.backend
280        else {
281            return Err(crate::error::InternalError::store_invariant());
282        };
283
284        let previous = if tombstones.contains(&key) {
285            None
286        } else {
287            live.get(&key).cloned().or_else(|| canonical.get(&key))
288        };
289        tombstones.remove(&key);
290        let cardinality_key = key.clone();
291        live.insert(key, row);
292        self.entity_cardinality
293            .apply_insert(&cardinality_key, previous.as_ref());
294        self.bump_generation();
295
296        Ok(previous)
297    }
298
299    /// Apply one recovered journal row delete into the volatile projection.
300    pub(in crate::db) fn apply_recovered_journal_delete(
301        &mut self,
302        key: &RawDataStoreKey,
303    ) -> Result<Option<RawRow>, crate::error::InternalError> {
304        let DataStoreBackend::Journaled {
305            canonical,
306            live,
307            tombstones,
308            ..
309        } = &mut self.backend
310        else {
311            return Err(crate::error::InternalError::store_invariant());
312        };
313
314        let previous = if tombstones.contains(key) {
315            None
316        } else {
317            live.get(key).cloned().or_else(|| canonical.get(key))
318        };
319        live.remove(key);
320        tombstones.insert(key.clone());
321        self.entity_cardinality.apply_remove(key, previous.as_ref());
322        self.bump_generation();
323
324        Ok(previous)
325    }
326
327    /// Publish one preflighted positioned row value or tombstone.
328    pub(in crate::db) fn publish_preflighted_journal_entry(
329        &mut self,
330        key: RawDataStoreKey,
331        row: Option<RawRow>,
332        position: JournalOverlayPosition,
333    ) -> Result<Option<RawRow>, crate::error::InternalError> {
334        let DataStoreBackend::Journaled {
335            canonical,
336            live,
337            tombstones,
338            positions,
339        } = &mut self.backend
340        else {
341            return Err(crate::error::InternalError::store_invariant());
342        };
343        let previous = if tombstones.contains(&key) {
344            None
345        } else {
346            live.get(&key).cloned().or_else(|| canonical.get(&key))
347        };
348        let cardinality_key = key.clone();
349        if let Some(row) = row {
350            tombstones.remove(&key);
351            live.insert(key.clone(), row);
352            self.entity_cardinality
353                .apply_insert(&cardinality_key, previous.as_ref());
354        } else {
355            live.remove(&key);
356            tombstones.insert(key.clone());
357            self.entity_cardinality
358                .apply_remove(&cardinality_key, previous.as_ref());
359        }
360        positions.publish_preflighted(key, position);
361        self.bump_generation();
362
363        Ok(previous)
364    }
365
366    /// Validate and publish one positioned row for direct store tests.
367    #[cfg(test)]
368    pub(in crate::db) fn publish_positioned_journal_entry(
369        &mut self,
370        key: RawDataStoreKey,
371        row: Option<RawRow>,
372        position: JournalOverlayPosition,
373    ) -> Result<Option<RawRow>, crate::error::InternalError> {
374        self.preflight_positioned_journal_entry(&key, position)?;
375        self.publish_preflighted_journal_entry(key, row, position)
376    }
377
378    /// Preflight row provenance before marker publication.
379    pub(in crate::db) fn preflight_positioned_journal_entry(
380        &self,
381        key: &RawDataStoreKey,
382        position: JournalOverlayPosition,
383    ) -> Result<(), crate::error::InternalError> {
384        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
385            return Err(crate::error::InternalError::store_invariant());
386        };
387        positions.preflight_publish(key, position)
388    }
389
390    /// Preflight direct row provenance before marker publication.
391    #[cfg(any(test, feature = "migration"))]
392    pub(in crate::db) fn prepare_position_publication(
393        &self,
394        keys: impl IntoIterator<Item = RawDataStoreKey>,
395        position: JournalOverlayPosition,
396    ) -> Result<PreparedDataPositionPublication, crate::error::InternalError> {
397        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
398            return Err(crate::error::InternalError::store_invariant());
399        };
400        let keys = keys.into_iter().collect::<BTreeSet<_>>();
401        for key in &keys {
402            positions.preflight_publish(key, position)?;
403        }
404        Ok(PreparedDataPositionPublication {
405            keys: keys.into_iter().collect(),
406            position,
407        })
408    }
409
410    /// Publish direct row provenance after its values have been applied.
411    #[cfg(any(test, feature = "migration"))]
412    pub(in crate::db) fn publish_prepared_positions(
413        &mut self,
414        prepared: PreparedDataPositionPublication,
415    ) {
416        let DataStoreBackend::Journaled { positions, .. } = &mut self.backend else {
417            debug_assert!(false, "preflighted row positions require a journaled store");
418            return;
419        };
420        for key in prepared.keys {
421            positions.publish_preflighted(key, prepared.position);
422        }
423    }
424
425    /// Preflight exact row-overlay retirement before canonical mutation.
426    pub(in crate::db) fn prepare_position_retirement(
427        &self,
428        keys: impl IntoIterator<Item = RawDataStoreKey>,
429        position: JournalOverlayPosition,
430    ) -> Result<PreparedDataPositionRetirement, crate::error::InternalError> {
431        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
432            return Err(crate::error::InternalError::store_invariant());
433        };
434        let entries = keys
435            .into_iter()
436            .collect::<BTreeSet<_>>()
437            .into_iter()
438            .map(|key| {
439                positions
440                    .preflight_retirement(&key, position)
441                    .map(|retirement| (key, retirement))
442            })
443            .collect::<Result<Vec<_>, _>>()?;
444        Ok(PreparedDataPositionRetirement { entries })
445    }
446
447    /// Retire only exact row overlays after canonical mutation succeeds.
448    pub(in crate::db) fn apply_prepared_position_retirement(
449        &mut self,
450        prepared: PreparedDataPositionRetirement,
451    ) {
452        let DataStoreBackend::Journaled {
453            live,
454            tombstones,
455            positions,
456            ..
457        } = &mut self.backend
458        else {
459            debug_assert!(
460                false,
461                "preflighted row retirement requires a journaled store"
462            );
463            return;
464        };
465        for (key, retirement) in prepared.entries {
466            if retirement == PositionedOverlayRetirement::Exact {
467                live.remove(&key);
468                tombstones.remove(&key);
469                positions.retire_preflighted(&key, retirement);
470            }
471        }
472    }
473
474    #[cfg(test)]
475    fn retire_positioned_journal_effect(
476        &mut self,
477        key: &RawDataStoreKey,
478        position: JournalOverlayPosition,
479    ) -> Result<PositionedOverlayRetirement, crate::error::InternalError> {
480        let DataStoreBackend::Journaled { positions, .. } = &self.backend else {
481            return Err(crate::error::InternalError::store_invariant());
482        };
483        let retirement = positions.preflight_retirement(key, position)?;
484        let prepared = PreparedDataPositionRetirement {
485            entries: vec![(key.clone(), retirement)],
486        };
487        self.apply_prepared_position_retirement(prepared);
488        Ok(retirement)
489    }
490
491    /// Apply one folded journal row put into the canonical stable base.
492    pub(in crate::db) fn fold_recovered_journal_put(
493        &mut self,
494        key: RawDataStoreKey,
495        row: RawRow,
496    ) -> Result<Option<RawRow>, crate::error::InternalError> {
497        let DataStoreBackend::Journaled {
498            canonical,
499            live,
500            tombstones,
501            ..
502        } = &mut self.backend
503        else {
504            return Err(crate::error::InternalError::store_invariant());
505        };
506
507        let visible = !live.contains_key(&key) && !tombstones.contains(&key);
508        let cardinality_key = key.clone();
509        let previous = canonical.insert(key, row);
510        if visible {
511            self.entity_cardinality
512                .apply_insert(&cardinality_key, previous.as_ref());
513        }
514        self.bump_generation();
515
516        Ok(previous)
517    }
518
519    /// Apply one folded journal row delete into the canonical stable base.
520    pub(in crate::db) fn fold_recovered_journal_delete(
521        &mut self,
522        key: &RawDataStoreKey,
523    ) -> Result<Option<RawRow>, crate::error::InternalError> {
524        let DataStoreBackend::Journaled {
525            canonical,
526            live,
527            tombstones,
528            ..
529        } = &mut self.backend
530        else {
531            return Err(crate::error::InternalError::store_invariant());
532        };
533
534        let visible = !live.contains_key(key) && !tombstones.contains(key);
535        let previous = canonical.remove(key);
536        if visible {
537            self.entity_cardinality.apply_remove(key, previous.as_ref());
538        }
539        self.bump_generation();
540
541        Ok(previous)
542    }
543
544    /// Prove that recovered journal rows can be folded into canonical storage.
545    pub(in crate::db) fn preflight_fold_recovered_journal(
546        &self,
547    ) -> Result<(), crate::error::InternalError> {
548        match self.backend {
549            DataStoreBackend::Journaled { .. } => Ok(()),
550            DataStoreBackend::Heap(_) => Err(crate::error::InternalError::store_invariant()),
551        }
552    }
553
554    /// Load one row by raw key.
555    pub(in crate::db) fn get(&self, key: &RawDataStoreKey) -> Option<RawRow> {
556        self.read(key).into_owned()
557    }
558
559    /// Load one row from the canonical predecessor view.
560    ///
561    /// Online journal folding uses this view so a newer positioned live effect
562    /// cannot replace the predecessor evidence for the older batch being
563    /// canonicalized.
564    pub(in crate::db) fn get_canonical(&self, key: &RawDataStoreKey) -> Option<RawRow> {
565        match &self.backend {
566            DataStoreBackend::Heap(map) => map.get(key).cloned(),
567            DataStoreBackend::Journaled { canonical, .. } => canonical.get(key),
568        }
569    }
570
571    /// Read one visible row without cloning heap/live payloads.
572    pub(in crate::db) fn read<'a>(&'a self, key: &RawDataStoreKey) -> StoredRowRead<'a> {
573        #[cfg(all(feature = "sql", feature = "diagnostics"))]
574        record_data_store_get_call();
575
576        match &self.backend {
577            DataStoreBackend::Heap(map) => map
578                .get(key)
579                .map_or(StoredRowRead::Missing, StoredRowRead::Borrowed),
580            DataStoreBackend::Journaled {
581                canonical,
582                live,
583                tombstones,
584                ..
585            } => {
586                if tombstones.contains(key) {
587                    StoredRowRead::Missing
588                } else if let Some(row) = live.get(key) {
589                    StoredRowRead::Borrowed(row)
590                } else {
591                    canonical
592                        .get(key)
593                        .map_or(StoredRowRead::Missing, StoredRowRead::Owned)
594                }
595            }
596        }
597    }
598
599    /// Return whether one raw key exists without cloning the row payload.
600    #[must_use]
601    pub(in crate::db) fn contains(&self, key: &RawDataStoreKey) -> bool {
602        match &self.backend {
603            DataStoreBackend::Heap(map) => map.contains_key(key),
604            DataStoreBackend::Journaled {
605                canonical,
606                live,
607                tombstones,
608                ..
609            } => {
610                !tombstones.contains(key)
611                    && (live.contains_key(key) || canonical.get(key).is_some())
612            }
613        }
614    }
615
616    /// Return the number of stored rows without exposing the backing map.
617    #[must_use]
618    pub(in crate::db) fn len(&self) -> u64 {
619        match &self.backend {
620            DataStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
621            DataStoreBackend::Journaled { .. } => {
622                let mut count = 0_u64;
623                let _: Result<(), Infallible> = self.visit_entries(|_key, _row| {
624                    count = count.saturating_add(1);
625                    Ok(StoreVisit::Continue)
626                });
627                count
628            }
629        }
630    }
631
632    /// Return the row-store generation used to prove index metadata freshness.
633    #[must_use]
634    pub(in crate::db) const fn generation(&self) -> u64 {
635        self.generation
636    }
637
638    /// Return an exact current row count for one entity when store metadata is valid.
639    #[must_use]
640    pub(in crate::db) fn exact_entity_count(&self, entity: EntityTag) -> Option<u64> {
641        self.entity_cardinality.exact_count(entity)
642    }
643
644    /// Visit raw row entries in canonical storage order.
645    pub(in crate::db) fn visit_entries<E>(
646        &self,
647        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
648    ) -> Result<(), E> {
649        match &self.backend {
650            DataStoreBackend::Heap(map) => {
651                for (key, row) in map {
652                    if visitor(key, row)?.should_stop() {
653                        break;
654                    }
655                }
656            }
657            DataStoreBackend::Journaled { .. } => Self::visit_journaled_entries_in_bounds(
658                &self.backend,
659                (Bound::Unbounded, Bound::Unbounded),
660                visitor,
661            )?,
662        }
663
664        Ok(())
665    }
666
667    /// Visit raw row entries whose keys belong to the provided storage range.
668    pub(in crate::db) fn visit_range<E>(
669        &self,
670        key_range: impl RangeBounds<RawDataStoreKey>,
671        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
672    ) -> Result<(), E> {
673        let bounds = Self::owned_range_bounds(&key_range);
674        match &self.backend {
675            DataStoreBackend::Heap(map) => {
676                for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
677                    if visitor(key, row)?.should_stop() {
678                        break;
679                    }
680                }
681            }
682            DataStoreBackend::Journaled { .. } => {
683                Self::visit_journaled_entries_in_bounds(&self.backend, bounds, visitor)?;
684            }
685        }
686
687        Ok(())
688    }
689
690    /// Visit one ascending row range while allowing the caller to stop after
691    /// seeing the key but before a stable row payload is materialized.
692    ///
693    /// Mixed journal overlays retain the ordinary key-then-point-read path;
694    /// one physical backing can keep the range iterator open for the complete
695    /// scan and avoid a second tree lookup per row.
696    pub(in crate::db) fn try_visit_range_with_row_preflight<E>(
697        &self,
698        key_range: impl RangeBounds<RawDataStoreKey>,
699        mut preflight: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
700        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
701    ) -> Result<Option<bool>, E> {
702        let bounds = Self::owned_range_bounds(&key_range);
703        let mut stopped = false;
704        match &self.backend {
705            DataStoreBackend::Heap(map) => {
706                for (key, row) in map.range((bounds.0.clone(), bounds.1)) {
707                    if preflight(key)?.should_stop() {
708                        stopped = true;
709                        break;
710                    }
711                    if visitor(key, row)?.should_stop() {
712                        stopped = true;
713                        break;
714                    }
715                }
716            }
717            DataStoreBackend::Journaled {
718                canonical,
719                live,
720                tombstones,
721                ..
722            } if canonical.is_empty() => {
723                for (key, row) in live.range((bounds.0.clone(), bounds.1)) {
724                    if tombstones.contains(key) {
725                        continue;
726                    }
727                    if preflight(key)?.should_stop() {
728                        stopped = true;
729                        break;
730                    }
731                    if visitor(key, row)?.should_stop() {
732                        stopped = true;
733                        break;
734                    }
735                }
736            }
737            DataStoreBackend::Journaled {
738                canonical,
739                live,
740                tombstones,
741                ..
742            } if live.is_empty() && tombstones.is_empty() => {
743                for entry in canonical.range((bounds.0.clone(), bounds.1)) {
744                    if preflight(entry.key())?.should_stop() {
745                        stopped = true;
746                        break;
747                    }
748                    if visitor(entry.key(), &entry.value())?.should_stop() {
749                        stopped = true;
750                        break;
751                    }
752                }
753            }
754            DataStoreBackend::Journaled { .. } => return Ok(None),
755        }
756
757        Ok(Some(!stopped))
758    }
759
760    /// Visit only raw keys in storage order without fetching row payloads.
761    ///
762    /// Primary-key access streams use this boundary to discover candidate
763    /// identities before the terminal row runtime decides whether the payload
764    /// is needed. Journaled traversal merges canonical and live keys while
765    /// preserving live overrides and tombstones without reading stable values.
766    pub(in crate::db) fn visit_key_range<E>(
767        &self,
768        key_range: impl RangeBounds<RawDataStoreKey>,
769        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
770    ) -> Result<(), E> {
771        self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), false, visitor)
772    }
773
774    /// Visit only raw keys in reverse storage order without fetching row payloads.
775    pub(in crate::db) fn visit_key_range_rev<E>(
776        &self,
777        key_range: impl RangeBounds<RawDataStoreKey>,
778        visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
779    ) -> Result<(), E> {
780        self.visit_keys_in_bounds(Self::owned_range_bounds(&key_range), true, visitor)
781    }
782
783    /// Sum of bytes used by all stored rows.
784    pub(in crate::db) fn memory_bytes(&self) -> u64 {
785        // Report map footprint as key bytes + row bytes per entry.
786        let mut bytes = 0u64;
787        let _: Result<(), Infallible> = self.visit_entries(|key, row| {
788            bytes = bytes.saturating_add(key.as_bytes().len() as u64 + row.len() as u64);
789            Ok(StoreVisit::Continue)
790        });
791        bytes
792    }
793
794    const fn bump_generation(&mut self) {
795        self.generation = self.generation.saturating_add(1);
796    }
797
798    #[cfg(test)]
799    fn rebuild_entity_cardinality_from_entries(&mut self) {
800        let mut cardinality = EntityCardinality::empty();
801        let _: Result<(), Infallible> = self.visit_entries(|key, _row| {
802            cardinality.apply_present_key(key);
803            Ok(StoreVisit::Continue)
804        });
805        self.entity_cardinality = cardinality;
806    }
807
808    /// Return the monotonic perf-only count of stable row fetches seen by this process.
809    #[cfg(all(feature = "sql", feature = "diagnostics"))]
810    pub(in crate::db) fn current_get_call_count() -> u64 {
811        DATA_STORE_GET_CALL_COUNT.with(Cell::get)
812    }
813
814    fn owned_range_bounds(
815        key_range: &impl RangeBounds<RawDataStoreKey>,
816    ) -> (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>) {
817        let lower = match key_range.start_bound() {
818            Bound::Included(key) => Bound::Included(key.clone()),
819            Bound::Excluded(key) => Bound::Excluded(key.clone()),
820            Bound::Unbounded => Bound::Unbounded,
821        };
822        let upper = match key_range.end_bound() {
823            Bound::Included(key) => Bound::Included(key.clone()),
824            Bound::Excluded(key) => Bound::Excluded(key.clone()),
825            Bound::Unbounded => Bound::Unbounded,
826        };
827
828        (lower, upper)
829    }
830
831    fn visit_keys_in_bounds<E>(
832        &self,
833        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
834        reverse: bool,
835        mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
836    ) -> Result<(), E> {
837        match &self.backend {
838            DataStoreBackend::Heap(map) => {
839                if reverse {
840                    for (key, _row) in map.range(bounds).rev() {
841                        if visitor(key)?.should_stop() {
842                            break;
843                        }
844                    }
845                } else {
846                    for (key, _row) in map.range(bounds) {
847                        if visitor(key)?.should_stop() {
848                            break;
849                        }
850                    }
851                }
852            }
853            DataStoreBackend::Journaled { .. } => {
854                Self::visit_journaled_keys_in_bounds(&self.backend, bounds, reverse, visitor)?;
855            }
856        }
857
858        Ok(())
859    }
860
861    fn visit_journaled_keys_in_bounds<E>(
862        backend: &DataStoreBackend,
863        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
864        reverse: bool,
865        mut visitor: impl FnMut(&RawDataStoreKey) -> Result<StoreVisit, E>,
866    ) -> Result<(), E> {
867        let DataStoreBackend::Journaled {
868            canonical,
869            live,
870            tombstones,
871            ..
872        } = backend
873        else {
874            return Ok(());
875        };
876
877        if canonical.is_empty() {
878            if reverse {
879                for (key, _row) in live.range(bounds).rev() {
880                    if visitor(key)?.should_stop() {
881                        return Ok(());
882                    }
883                }
884            } else {
885                for (key, _row) in live.range(bounds) {
886                    if visitor(key)?.should_stop() {
887                        return Ok(());
888                    }
889                }
890            }
891            return Ok(());
892        }
893
894        if live.is_empty() && tombstones.is_empty() {
895            if reverse {
896                for entry in canonical.range(bounds).rev() {
897                    if visitor(entry.key())?.should_stop() {
898                        return Ok(());
899                    }
900                }
901            } else {
902                for entry in canonical.range(bounds) {
903                    if visitor(entry.key())?.should_stop() {
904                        return Ok(());
905                    }
906                }
907            }
908            return Ok(());
909        }
910
911        let direction = if reverse {
912            Direction::Desc
913        } else {
914            Direction::Asc
915        };
916        match direction {
917            Direction::Asc => visit_ordered_overlay(
918                canonical.range((bounds.0.clone(), bounds.1.clone())),
919                live.range((bounds.0, bounds.1)),
920                direction,
921                |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
922                |canonical_entry| !tombstones.contains(canonical_entry.key()),
923                |live_entry| !tombstones.contains(live_entry.0),
924                |entry| {
925                    let visit = match entry {
926                        OrderedOverlayEntry::Canonical(canonical_entry) => {
927                            visitor(canonical_entry.key())?
928                        }
929                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
930                    };
931                    Ok(if visit.should_stop() {
932                        OrderedOverlayVisit::Stop
933                    } else {
934                        OrderedOverlayVisit::Continue
935                    })
936                },
937            ),
938            Direction::Desc => visit_ordered_overlay(
939                canonical.range((bounds.0.clone(), bounds.1.clone())).rev(),
940                live.range((bounds.0, bounds.1)).rev(),
941                direction,
942                |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
943                |canonical_entry| !tombstones.contains(canonical_entry.key()),
944                |live_entry| !tombstones.contains(live_entry.0),
945                |entry| {
946                    let visit = match entry {
947                        OrderedOverlayEntry::Canonical(canonical_entry) => {
948                            visitor(canonical_entry.key())?
949                        }
950                        OrderedOverlayEntry::Live((key, _row)) => visitor(key)?,
951                    };
952                    Ok(if visit.should_stop() {
953                        OrderedOverlayVisit::Stop
954                    } else {
955                        OrderedOverlayVisit::Continue
956                    })
957                },
958            ),
959        }
960    }
961
962    fn visit_journaled_entries_in_bounds<E>(
963        backend: &DataStoreBackend,
964        bounds: (Bound<RawDataStoreKey>, Bound<RawDataStoreKey>),
965        mut visitor: impl FnMut(&RawDataStoreKey, &RawRow) -> Result<StoreVisit, E>,
966    ) -> Result<(), E> {
967        let DataStoreBackend::Journaled {
968            canonical,
969            live,
970            tombstones,
971            ..
972        } = backend
973        else {
974            return Ok(());
975        };
976
977        if canonical.is_empty() {
978            for (key, row) in live.range(bounds) {
979                if visitor(key, row)?.should_stop() {
980                    return Ok(());
981                }
982            }
983            return Ok(());
984        }
985
986        if live.is_empty() && tombstones.is_empty() {
987            for entry in canonical.range(bounds) {
988                if visitor(entry.key(), &entry.value())?.should_stop() {
989                    return Ok(());
990                }
991            }
992            return Ok(());
993        }
994
995        visit_ordered_overlay(
996            canonical.range((bounds.0.clone(), bounds.1.clone())),
997            live.range((bounds.0, bounds.1)),
998            Direction::Asc,
999            |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1000            |canonical_entry| !tombstones.contains(canonical_entry.key()),
1001            |live_entry| !tombstones.contains(live_entry.0),
1002            |entry| {
1003                let visit = match entry {
1004                    OrderedOverlayEntry::Canonical(canonical_entry) => {
1005                        visitor(canonical_entry.key(), &canonical_entry.value())?
1006                    }
1007                    OrderedOverlayEntry::Live((key, row)) => visitor(key, row)?,
1008                };
1009                Ok(if visit.should_stop() {
1010                    OrderedOverlayVisit::Stop
1011                } else {
1012                    OrderedOverlayVisit::Continue
1013                })
1014            },
1015        )
1016    }
1017}
1018
1019#[derive(Clone, Debug)]
1020struct EntityCardinality {
1021    counts: HeapBTreeMap<EntityTag, u64>,
1022    decodable: bool,
1023}
1024
1025impl EntityCardinality {
1026    const fn empty() -> Self {
1027        Self {
1028            counts: HeapBTreeMap::new(),
1029            decodable: true,
1030        }
1031    }
1032
1033    const fn unavailable() -> Self {
1034        Self {
1035            counts: HeapBTreeMap::new(),
1036            decodable: false,
1037        }
1038    }
1039
1040    fn exact_count(&self, entity: EntityTag) -> Option<u64> {
1041        self.decodable
1042            .then(|| self.counts.get(&entity).copied().unwrap_or(0))
1043    }
1044
1045    fn apply_insert(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
1046        if previous.is_some() {
1047            return;
1048        }
1049        self.apply_present_key(key);
1050    }
1051
1052    fn apply_remove(&mut self, key: &RawDataStoreKey, previous: Option<&RawRow>) {
1053        if previous.is_none() {
1054            return;
1055        }
1056        self.apply_removed_key(key);
1057    }
1058
1059    fn apply_present_key(&mut self, key: &RawDataStoreKey) {
1060        if !self.decodable {
1061            return;
1062        }
1063        let Some(entity) = key.entity_tag_prefix() else {
1064            self.invalidate();
1065            return;
1066        };
1067
1068        let count = self.counts.entry(entity).or_insert(0);
1069        *count = count.saturating_add(1);
1070    }
1071
1072    fn apply_removed_key(&mut self, key: &RawDataStoreKey) {
1073        if !self.decodable {
1074            return;
1075        }
1076        let Some(entity) = key.entity_tag_prefix() else {
1077            self.invalidate();
1078            return;
1079        };
1080
1081        if let Some(count) = self.counts.get_mut(&entity) {
1082            *count = count.saturating_sub(1);
1083            if *count == 0 {
1084                self.counts.remove(&entity);
1085            }
1086        }
1087    }
1088
1089    fn invalidate(&mut self) {
1090        self.counts.clear();
1091        self.decodable = false;
1092    }
1093}
1094
1095#[cfg(test)]
1096mod tests;