icydb_core/db/session/
read_set.rs

1//! Module: session::read_set
2//! Responsibility: capture and recompare bounded physical source proofs.
3//! Does not own: cursor execution or durable application job state.
4//! Boundary: accepted entity names -> registered store revisions.
5
6use crate::{
7    db::{
8        DbSession, ExhaustiveReadError, ReadSetRevisionError, ReadSetRevisionProof,
9        ReadSetStoreIdentity, ReadSetStoreRevision, StoreHandle, StoreRuntimeStorageMode,
10        commit::database_incarnation_id, executor::budget::HardExecutionContext,
11    },
12    traits::CanisterKind,
13};
14use icydb_diagnostic_code::{
15    DiagnosticExecutionBudgetResource, DiagnosticExecutionBudgetScope, DiagnosticExecutionLane,
16};
17
18const READ_SET_CAPTURE_SHAPE: u64 = 0x7265_6164_7365_7401;
19
20impl<C: CanisterKind> DbSession<C> {
21    /// Capture one canonical proof covering every physical store behind the
22    /// named accepted entities.
23    pub fn capture_read_set_revision_proof(
24        &self,
25        entity_names: &[&str],
26    ) -> Result<ReadSetRevisionProof, ExhaustiveReadError> {
27        self.db
28            .request_execution_scope()
29            .charge(
30                HardExecutionContext::new(
31                    DiagnosticExecutionBudgetScope::Execution,
32                    DiagnosticExecutionLane::TrustedRead,
33                    READ_SET_CAPTURE_SHAPE,
34                ),
35                DiagnosticExecutionBudgetResource::QueryExecutions,
36                1,
37            )
38            .map_err(crate::error::InternalError::from)?;
39        if entity_names.is_empty() {
40            return Err(ReadSetRevisionError::Empty.into());
41        }
42        let root = self.current_accepted_runtime_root_identity()?;
43        let mut store_paths = Vec::with_capacity(entity_names.len());
44        for entity in entity_names {
45            let catalog = self
46                .find_accepted_schema_catalog_context_for_entity_name(entity)?
47                .ok_or(ReadSetRevisionError::UnknownEntity)?;
48            store_paths.push(catalog.identity().store_path());
49        }
50        store_paths.sort_unstable();
51        store_paths.dedup();
52        let stores = store_paths
53            .into_iter()
54            .map(|store_path| {
55                let handle = self.db.recovered_store(store_path)?;
56                Self::capture_store_revision(store_path, handle)
57            })
58            .collect::<Result<Vec<_>, ExhaustiveReadError>>()?;
59        ReadSetRevisionProof::new(root, stores).map_err(Into::into)
60    }
61
62    pub(in crate::db) fn capture_entity_read_set_revision_proof(
63        &self,
64        store_path: &'static str,
65    ) -> Result<ReadSetRevisionProof, ExhaustiveReadError> {
66        let root = self.current_accepted_runtime_root_identity()?;
67        let handle = self.db.recovered_store(store_path)?;
68        ReadSetRevisionProof::new(
69            root,
70            vec![Self::capture_store_revision(store_path, handle)?],
71        )
72        .map_err(Into::into)
73    }
74
75    pub(in crate::db) fn verify_read_set_revision_proof(
76        &self,
77        proof: &ReadSetRevisionProof,
78    ) -> Result<(), ExhaustiveReadError> {
79        proof.validate()?;
80        let incarnation = database_incarnation_id()?;
81        let root = self.current_accepted_runtime_root_identity()?;
82        if proof.database_incarnation() != incarnation.to_bytes() {
83            return Err(ReadSetRevisionError::DatabaseIncarnationChanged.into());
84        }
85        if !proof.root_matches(incarnation, root) {
86            return Err(ReadSetRevisionError::AcceptedRootChanged.into());
87        }
88        for expected in proof.stores() {
89            let (store_path, handle) = self
90                .store_for_read_set_identity(expected.store())
91                .ok_or(ReadSetRevisionError::NonCanonical)?;
92            let current = Self::capture_store_revision(store_path, handle)?;
93            if current.data_revision() != expected.data_revision() {
94                return Err(ReadSetRevisionError::StoreDataChanged {
95                    store: expected.store(),
96                }
97                .into());
98            }
99            if current.access_state_revision() != expected.access_state_revision() {
100                return Err(ReadSetRevisionError::StoreAccessChanged {
101                    store: expected.store(),
102                }
103                .into());
104            }
105        }
106        Ok(())
107    }
108
109    pub(in crate::db) fn verify_durable_read_set_revision_proof(
110        &self,
111        proof: &ReadSetRevisionProof,
112    ) -> Result<(), ExhaustiveReadError> {
113        self.verify_read_set_revision_proof(proof)?;
114        for store in proof.stores() {
115            let (_, handle) = self
116                .store_for_read_set_identity(store.store())
117                .ok_or(ReadSetRevisionError::NonCanonical)?;
118            if handle.storage_capabilities().storage_mode() != StoreRuntimeStorageMode::Journaled {
119                return Err(ReadSetRevisionError::DurableStoreRequired {
120                    store: store.store(),
121                }
122                .into());
123            }
124        }
125        Ok(())
126    }
127
128    pub(in crate::db) fn ensure_read_set_contains_store(
129        proof: &ReadSetRevisionProof,
130        store_path: &str,
131    ) -> Result<(), ExhaustiveReadError> {
132        let store = ReadSetStoreIdentity::for_store_path(store_path);
133        if !proof.contains_store(store) {
134            return Err(ReadSetRevisionError::StoreMissingFromProof { store }.into());
135        }
136        Ok(())
137    }
138
139    fn capture_store_revision(
140        store_path: &str,
141        handle: StoreHandle,
142    ) -> Result<ReadSetStoreRevision, ExhaustiveReadError> {
143        let data_revision = handle.journal_tail_store().map_or_else(
144            || {
145                handle
146                    .with_data(crate::db::data::DataStore::generation)
147                    .checked_add(1)
148                    .ok_or_else(crate::error::InternalError::store_invariant)
149            },
150            |journal| {
151                journal.with_borrow(crate::db::journal::JournalTailStore::data_mutation_revision)
152            },
153        )?;
154        let access_state_revision = handle.access_state_revision()?;
155        Ok(ReadSetStoreRevision::new(
156            ReadSetStoreIdentity::for_store_path(store_path),
157            data_revision,
158            access_state_revision,
159        ))
160    }
161
162    fn store_for_read_set_identity(
163        &self,
164        identity: ReadSetStoreIdentity,
165    ) -> Option<(&'static str, StoreHandle)> {
166        self.db.with_store_registry(|registry| {
167            registry.iter().find(|(store_path, _)| {
168                ReadSetStoreIdentity::for_store_path(store_path) == identity
169            })
170        })
171    }
172}
icydb_core/db/session/read_set.rs

icydb_core/db/session/
read_set.rs
