icydb_core/db/
read_set.rs

1//! Module: db::read_set
2//! Responsibility: bounded canonical source-revision proof vocabulary.
3//! Does not own: query planning, cursor traversal, or application job state.
4//! Boundary: registered physical stores + accepted runtime root -> public proof.
5
6use crate::{
7    db::{
8        QueryError,
9        codec::{finalize_hash_sha256, new_hash_sha256_prefixed, write_hash_str_u32},
10        integrity::DatabaseIncarnationId,
11        schema::AcceptedSchemaRuntimeRootIdentity,
12    },
13    error::InternalError,
14};
15
16use serde::Deserialize;
17use std::{error::Error as StdError, fmt};
18
19const READ_SET_STORE_IDENTITY_DOMAIN: &[u8] = b"icydb.read-set.store.v1";
20const READ_SET_PROOF_FIXED_BYTES: usize = 16 + 8 + 1 + 32 + 4;
21const READ_SET_STORE_ENTRY_BYTES: usize = 32 + 8 + 8;
22
23/// Maximum physical stores admitted by one exhaustive source proof.
24pub const MAX_READ_SET_PROOF_STORES: usize = 64;
25const MAX_READ_SET_PROOF_STORES_U32: u32 = 64;
26/// Maximum canonical binary bytes admitted by one exhaustive source proof.
27pub const MAX_READ_SET_PROOF_BYTES: usize = 8 * 1024;
28const MAX_READ_SET_PROOF_BYTES_U32: u32 = 8 * 1024;
29
30/// Opaque canonical identity of one registered physical store.
31#[derive(
32    serde::Serialize, Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd,
33)]
34pub struct ReadSetStoreIdentity([u8; 32]);
35
36impl ReadSetStoreIdentity {
37    pub(in crate::db) fn for_store_path(store_path: &str) -> Self {
38        let mut hasher = new_hash_sha256_prefixed(READ_SET_STORE_IDENTITY_DOMAIN);
39        write_hash_str_u32(&mut hasher, store_path);
40        Self(finalize_hash_sha256(hasher))
41    }
42
43    pub(in crate::db) const fn from_bytes(bytes: [u8; 32]) -> Self {
44        Self(bytes)
45    }
46
47    /// Return the opaque canonical bytes.
48    #[must_use]
49    pub const fn to_bytes(self) -> [u8; 32] {
50        self.0
51    }
52}
53
54/// One physical store's row and access-state revisions.
55#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
56pub struct ReadSetStoreRevision {
57    store: ReadSetStoreIdentity,
58    data_revision: u64,
59    access_state_revision: u64,
60}
61
62impl ReadSetStoreRevision {
63    pub(in crate::db) const fn new(
64        store: ReadSetStoreIdentity,
65        data_revision: u64,
66        access_state_revision: u64,
67    ) -> Self {
68        Self {
69            store,
70            data_revision,
71            access_state_revision,
72        }
73    }
74
75    /// Return the physical store identity.
76    #[must_use]
77    pub const fn store(&self) -> ReadSetStoreIdentity {
78        self.store
79    }
80
81    /// Return the logical row-mutation revision.
82    #[must_use]
83    pub const fn data_revision(&self) -> u64 {
84        self.data_revision
85    }
86
87    /// Return the physical access-readiness revision.
88    #[must_use]
89    pub const fn access_state_revision(&self) -> u64 {
90        self.access_state_revision
91    }
92}
93
94/// Canonical bounded proof for every physical source store in one exhaustive job.
95#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
96pub struct ReadSetRevisionProof {
97    database_incarnation: [u8; 16],
98    accepted_root_revision: u64,
99    accepted_root_fingerprint_method: u8,
100    accepted_root_fingerprint: [u8; 32],
101    stores: Vec<ReadSetStoreRevision>,
102}
103
104impl ReadSetRevisionProof {
105    pub(in crate::db) fn new(
106        root: AcceptedSchemaRuntimeRootIdentity,
107        stores: Vec<ReadSetStoreRevision>,
108    ) -> Result<Self, ReadSetRevisionError> {
109        let (accepted_root_fingerprint_method, accepted_root_fingerprint) = root.fingerprint();
110        let proof = Self {
111            database_incarnation: root.database_incarnation().to_bytes(),
112            accepted_root_revision: root.accepted_root_revision().get(),
113            accepted_root_fingerprint_method,
114            accepted_root_fingerprint,
115            stores,
116        };
117        proof.validate()?;
118        Ok(proof)
119    }
120
121    pub(in crate::db) fn from_parts(
122        database_incarnation: [u8; 16],
123        accepted_root_revision: u64,
124        accepted_root_fingerprint_method: u8,
125        accepted_root_fingerprint: [u8; 32],
126        stores: Vec<ReadSetStoreRevision>,
127    ) -> Result<Self, ReadSetRevisionError> {
128        let proof = Self {
129            database_incarnation,
130            accepted_root_revision,
131            accepted_root_fingerprint_method,
132            accepted_root_fingerprint,
133            stores,
134        };
135        proof.validate()?;
136        Ok(proof)
137    }
138
139    /// Return the durable database lifecycle identity.
140    #[must_use]
141    pub const fn database_incarnation(&self) -> [u8; 16] {
142        self.database_incarnation
143    }
144
145    /// Return the accepted runtime-root revision.
146    #[must_use]
147    pub const fn accepted_root_revision(&self) -> u64 {
148        self.accepted_root_revision
149    }
150
151    /// Return the accepted runtime-root fingerprint method.
152    #[must_use]
153    pub const fn accepted_root_fingerprint_method(&self) -> u8 {
154        self.accepted_root_fingerprint_method
155    }
156
157    /// Return the accepted runtime-root fingerprint.
158    #[must_use]
159    pub const fn accepted_root_fingerprint(&self) -> [u8; 32] {
160        self.accepted_root_fingerprint
161    }
162
163    /// Borrow canonically sorted participating stores.
164    #[must_use]
165    pub const fn stores(&self) -> &[ReadSetStoreRevision] {
166        self.stores.as_slice()
167    }
168
169    /// Return the exact current canonical binary size.
170    #[must_use]
171    pub const fn encoded_len(&self) -> usize {
172        READ_SET_PROOF_FIXED_BYTES
173            .saturating_add(self.stores.len().saturating_mul(READ_SET_STORE_ENTRY_BYTES))
174    }
175
176    /// Validate bounds, nonzero revisions, and canonical store ordering.
177    pub fn validate(&self) -> Result<(), ReadSetRevisionError> {
178        if self.stores.is_empty() {
179            return Err(ReadSetRevisionError::Empty);
180        }
181        if self.stores.len() > MAX_READ_SET_PROOF_STORES {
182            return Err(ReadSetRevisionError::TooManyStores {
183                limit: MAX_READ_SET_PROOF_STORES_U32,
184                actual: u32::try_from(self.stores.len()).unwrap_or(u32::MAX),
185            });
186        }
187        let encoded_len = self.encoded_len();
188        if encoded_len > MAX_READ_SET_PROOF_BYTES {
189            return Err(ReadSetRevisionError::EncodedBytesExceeded {
190                limit: MAX_READ_SET_PROOF_BYTES_U32,
191                actual: u32::try_from(encoded_len).unwrap_or(u32::MAX),
192            });
193        }
194        if self.database_incarnation == [0; 16]
195            || self.accepted_root_revision == 0
196            || self.accepted_root_fingerprint_method == 0
197            || self.accepted_root_fingerprint == [0; 32]
198            || self.stores.iter().any(|store| {
199                store.store.to_bytes() == [0; 32]
200                    || store.data_revision == 0
201                    || store.access_state_revision == 0
202            })
203            || self
204                .stores
205                .windows(2)
206                .any(|pair| pair[0].store >= pair[1].store)
207        {
208            return Err(ReadSetRevisionError::NonCanonical);
209        }
210        Ok(())
211    }
212
213    pub(in crate::db) fn contains_store(&self, store: ReadSetStoreIdentity) -> bool {
214        self.stores
215            .binary_search_by_key(&store, ReadSetStoreRevision::store)
216            .is_ok()
217    }
218
219    pub(in crate::db) fn signature_bytes(&self) -> Vec<u8> {
220        let mut bytes = Vec::with_capacity(self.encoded_len());
221        bytes.extend_from_slice(&self.database_incarnation);
222        bytes.extend_from_slice(&self.accepted_root_revision.to_be_bytes());
223        bytes.push(self.accepted_root_fingerprint_method);
224        bytes.extend_from_slice(&self.accepted_root_fingerprint);
225        let store_count = u32::try_from(self.stores.len()).unwrap_or(u32::MAX);
226        bytes.extend_from_slice(&store_count.to_be_bytes());
227        for store in &self.stores {
228            bytes.extend_from_slice(&store.store.to_bytes());
229            bytes.extend_from_slice(&store.data_revision.to_be_bytes());
230            bytes.extend_from_slice(&store.access_state_revision.to_be_bytes());
231        }
232        bytes
233    }
234
235    pub(in crate::db) fn root_matches(
236        &self,
237        incarnation: DatabaseIncarnationId,
238        root: AcceptedSchemaRuntimeRootIdentity,
239    ) -> bool {
240        let (method, fingerprint) = root.fingerprint();
241        self.database_incarnation == incarnation.to_bytes()
242            && self.accepted_root_revision == root.accepted_root_revision().get()
243            && self.accepted_root_fingerprint_method == method
244            && self.accepted_root_fingerprint == fingerprint
245    }
246}
247
248/// Typed failure while capturing or validating an exhaustive source proof.
249#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
250pub enum ReadSetRevisionError {
251    /// No participating physical store was supplied.
252    Empty,
253    /// The proof exceeds the bounded participating-store count.
254    TooManyStores { limit: u32, actual: u32 },
255    /// The canonical proof encoding exceeds its byte ceiling.
256    EncodedBytesExceeded { limit: u32, actual: u32 },
257    /// Proof authority, revisions, or store ordering are not canonical.
258    NonCanonical,
259    /// One requested entity is absent from accepted runtime authority.
260    UnknownEntity,
261    /// The page's physical source was not declared in the initial proof.
262    StoreMissingFromProof { store: ReadSetStoreIdentity },
263    /// A continuation was supplied without its associated source proof.
264    ResumeProofRequired,
265    /// The database was recreated after the proof was captured.
266    DatabaseIncarnationChanged,
267    /// Accepted runtime authority changed after the proof was captured.
268    AcceptedRootChanged,
269    /// Rows in one participating physical store changed.
270    StoreDataChanged { store: ReadSetStoreIdentity },
271    /// Physical access readiness in one participating store changed.
272    StoreAccessChanged { store: ReadSetStoreIdentity },
273    /// A cross-call job attempted to use a volatile source store.
274    DurableStoreRequired { store: ReadSetStoreIdentity },
275}
276
277impl fmt::Display for ReadSetRevisionError {
278    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
279        formatter.write_str("exhaustive read source proof is invalid")
280    }
281}
282
283impl StdError for ReadSetRevisionError {}
284
285impl ReadSetRevisionError {
286    pub(in crate::db) const fn is_source_change(&self) -> bool {
287        matches!(
288            self,
289            Self::DatabaseIncarnationChanged
290                | Self::AcceptedRootChanged
291                | Self::StoreDataChanged { .. }
292                | Self::StoreAccessChanged { .. }
293        )
294    }
295}
296
297/// Query/runtime or typed source-proof failure from an exhaustive operation.
298#[derive(Debug)]
299pub enum ExhaustiveReadError {
300    /// Query planning, admission, execution, or database authority failed.
301    Query(QueryError),
302    /// The bounded source proof was invalid or changed.
303    Revision(ReadSetRevisionError),
304}
305
306impl fmt::Display for ExhaustiveReadError {
307    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
308        match self {
309            Self::Query(error) => error.fmt(formatter),
310            Self::Revision(error) => error.fmt(formatter),
311        }
312    }
313}
314
315impl StdError for ExhaustiveReadError {}
316
317impl From<QueryError> for ExhaustiveReadError {
318    fn from(error: QueryError) -> Self {
319        Self::Query(error)
320    }
321}
322
323impl From<InternalError> for ExhaustiveReadError {
324    fn from(error: InternalError) -> Self {
325        Self::Query(QueryError::execute(error))
326    }
327}
328
329impl From<ReadSetRevisionError> for ExhaustiveReadError {
330    fn from(error: ReadSetRevisionError) -> Self {
331        Self::Revision(error)
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use super::*;
338
339    fn store(byte: u8, data_revision: u64) -> ReadSetStoreRevision {
340        ReadSetStoreRevision::new(
341            ReadSetStoreIdentity::from_bytes([byte; 32]),
342            data_revision,
343            1,
344        )
345    }
346
347    fn proof(
348        stores: Vec<ReadSetStoreRevision>,
349    ) -> Result<ReadSetRevisionProof, ReadSetRevisionError> {
350        ReadSetRevisionProof::from_parts([1; 16], 1, 1, [2; 32], stores)
351    }
352
353    #[test]
354    fn read_set_proof_requires_bounded_canonical_distinct_store_order() {
355        assert_eq!(proof(Vec::new()), Err(ReadSetRevisionError::Empty));
356        assert_eq!(
357            proof(vec![store(2, 1), store(1, 1)]),
358            Err(ReadSetRevisionError::NonCanonical),
359        );
360        assert_eq!(
361            proof(vec![store(1, 1), store(1, 2)]),
362            Err(ReadSetRevisionError::NonCanonical),
363        );
364
365        let too_many = (1..=MAX_READ_SET_PROOF_STORES + 1)
366            .map(|index| {
367                let mut identity = [0; 32];
368                identity[24..].copy_from_slice(
369                    &u64::try_from(index)
370                        .expect("bounded test store count should fit u64")
371                        .to_be_bytes(),
372                );
373                ReadSetStoreRevision::new(ReadSetStoreIdentity::from_bytes(identity), 1, 1)
374            })
375            .collect();
376        assert_eq!(
377            proof(too_many),
378            Err(ReadSetRevisionError::TooManyStores {
379                limit: MAX_READ_SET_PROOF_STORES_U32,
380                actual: u32::try_from(MAX_READ_SET_PROOF_STORES + 1)
381                    .expect("bounded test store count should fit u32"),
382            }),
383        );
384    }
385
386    #[test]
387    fn read_set_proof_rejects_zero_authority_or_revision_components() {
388        let valid = proof(vec![store(1, 1)]).expect("nonzero canonical proof should admit");
389        assert_eq!(
390            valid.encoded_len(),
391            READ_SET_PROOF_FIXED_BYTES + READ_SET_STORE_ENTRY_BYTES
392        );
393
394        assert_eq!(
395            ReadSetRevisionProof::from_parts([1; 16], 0, 1, [2; 32], vec![store(1, 1)]),
396            Err(ReadSetRevisionError::NonCanonical),
397        );
398        assert_eq!(
399            proof(vec![store(1, 0)]),
400            Err(ReadSetRevisionError::NonCanonical),
401        );
402    }
403}
icydb_core/db/read_set.rs

icydb_core/db/
read_set.rs
