iroh_db_store/
materialized.rs

1use iroh_db_core::SchemaDescriptor;
2
3const MAX_RECORD_ID: usize = 1_024;
4const MAX_RECORD_STATE: usize = 64 * 1024 * 1024;
5const MAX_INDEX_VALUE: usize = 1024 * 1024;
6
7/// One primary-key ordered materialized record returned by a collection scan.
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct StoredRecord {
10    pub(crate) record_id: Vec<u8>,
11    pub(crate) state: Vec<u8>,
12}
13
14impl StoredRecord {
15    /// Returns the canonical application record ID.
16    pub fn record_id(&self) -> &[u8] {
17        &self.record_id
18    }
19
20    /// Returns the canonical typed-record state.
21    pub fn state(&self) -> &[u8] {
22        &self.state
23    }
24}
25
26/// One canonical value emitted into a schema-declared secondary index.
27#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
28pub struct IndexValue {
29    pub(crate) field_id: u32,
30    pub(crate) value: Vec<u8>,
31}
32
33impl IndexValue {
34    /// Creates an index entry for a stable field and canonical value.
35    pub fn new(field_id: u32, value: Vec<u8>) -> Self {
36        Self { field_id, value }
37    }
38
39    /// Returns the indexed stable field ID.
40    pub const fn field_id(&self) -> u32 {
41        self.field_id
42    }
43
44    /// Returns the canonical indexed value.
45    pub fn value(&self) -> &[u8] {
46        &self.value
47    }
48}
49
50/// One record replacement or deletion applied in the commit metadata transaction.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct MaterializedRecord {
53    pub(crate) schema: SchemaDescriptor,
54    pub(crate) record_id: Vec<u8>,
55    pub(crate) state: Option<Vec<u8>>,
56    pub(crate) indexes: Vec<IndexValue>,
57}
58
59impl MaterializedRecord {
60    /// Validates a complete canonical record replacement and its index values.
61    pub fn upsert(
62        schema: SchemaDescriptor,
63        record_id: Vec<u8>,
64        state: Vec<u8>,
65        mut indexes: Vec<IndexValue>,
66    ) -> Result<Self, MaterializedError> {
67        validate_record_id(&record_id)?;
68        if state.len() > MAX_RECORD_STATE {
69            return Err(MaterializedError::RecordStateTooLarge(state.len()));
70        }
71        for index in &indexes {
72            if index.field_id == 0 {
73                return Err(MaterializedError::InvalidIndexField(0));
74            }
75            if index.value.len() > MAX_INDEX_VALUE {
76                return Err(MaterializedError::IndexValueTooLarge(index.value.len()));
77            }
78        }
79        indexes.sort_unstable();
80        indexes.dedup();
81        Ok(Self {
82            schema,
83            record_id,
84            state: Some(state),
85            indexes,
86        })
87    }
88
89    /// Validates a record deletion under its registered schema.
90    pub fn delete(schema: SchemaDescriptor, record_id: Vec<u8>) -> Result<Self, MaterializedError> {
91        validate_record_id(&record_id)?;
92        Ok(Self {
93            schema,
94            record_id,
95            state: None,
96            indexes: Vec::new(),
97        })
98    }
99
100    /// Returns the record's registered schema.
101    pub const fn schema(&self) -> &SchemaDescriptor {
102        &self.schema
103    }
104
105    /// Returns the canonical primary key.
106    pub fn record_id(&self) -> &[u8] {
107        &self.record_id
108    }
109
110    /// Returns complete canonical record state, or `None` for deletion.
111    pub fn state(&self) -> Option<&[u8]> {
112        self.state.as_deref()
113    }
114
115    /// Returns canonical secondary-index values.
116    pub fn indexes(&self) -> &[IndexValue] {
117        &self.indexes
118    }
119
120    /// Encodes the complete rebuildable materialization carried by a commit operation.
121    pub fn encode_canonical(&self) -> Result<Vec<u8>, MaterializedError> {
122        let wire = MaterializedWire {
123            schema: self.schema.encode_canonical(),
124            record_id: self.record_id.clone(),
125            state: self.state.clone(),
126            indexes: self
127                .indexes
128                .iter()
129                .map(|index| IndexWire {
130                    field_id: index.field_id,
131                    value: index.value.clone(),
132                })
133                .collect(),
134        };
135        minicbor::to_vec(wire).map_err(codec_error)
136    }
137
138    /// Strictly decodes and validates a commit-carried materialization.
139    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, MaterializedError> {
140        let mut decoder = minicbor::Decoder::new(bytes);
141        let wire: MaterializedWire = decoder.decode().map_err(codec_error)?;
142        if decoder.position() != bytes.len() {
143            return Err(MaterializedError::NonCanonicalEncoding);
144        }
145        let schema = SchemaDescriptor::decode_canonical(&wire.schema)
146            .map_err(|error| MaterializedError::Schema(error.to_string()))?;
147        let indexes = wire
148            .indexes
149            .into_iter()
150            .map(|index| IndexValue::new(index.field_id, index.value))
151            .collect();
152        let value = match wire.state {
153            Some(state) => Self::upsert(schema, wire.record_id, state, indexes)?,
154            None if indexes.is_empty() => Self::delete(schema, wire.record_id)?,
155            None => return Err(MaterializedError::DeleteHasIndexes),
156        };
157        if value.encode_canonical()? != bytes {
158            return Err(MaterializedError::NonCanonicalEncoding);
159        }
160        Ok(value)
161    }
162}
163
164#[derive(minicbor::Encode, minicbor::Decode)]
165#[cbor(array)]
166struct MaterializedWire {
167    #[n(0)]
168    schema: Vec<u8>,
169    #[n(1)]
170    record_id: Vec<u8>,
171    #[n(2)]
172    state: Option<Vec<u8>>,
173    #[n(3)]
174    indexes: Vec<IndexWire>,
175}
176
177#[derive(minicbor::Encode, minicbor::Decode)]
178#[cbor(array)]
179struct IndexWire {
180    #[n(0)]
181    field_id: u32,
182    #[n(1)]
183    value: Vec<u8>,
184}
185
186/// A rejected materialized-record input.
187#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
188pub enum MaterializedError {
189    /// A record key exceeded the protocol bound.
190    #[error("record ID is too large: {0}")]
191    RecordIdTooLarge(usize),
192    /// A materialized record exceeded the local safety bound.
193    #[error("record state is too large: {0}")]
194    RecordStateTooLarge(usize),
195    /// An index value exceeded the local safety bound.
196    #[error("index value is too large: {0}")]
197    IndexValueTooLarge(usize),
198    /// Field zero is reserved for record-level operations.
199    #[error("field {0} cannot be used as a secondary index")]
200    InvalidIndexField(u32),
201    /// A durable mutation could not be encoded or decoded.
202    #[error("materialized record codec failed: {0}")]
203    Codec(String),
204    /// A durable mutation contained an invalid schema.
205    #[error("materialized record schema failed: {0}")]
206    Schema(String),
207    /// A decoded representation was not uniquely canonical.
208    #[error("materialized record encoding is not canonical")]
209    NonCanonicalEncoding,
210    /// A deletion illegally carried secondary index values.
211    #[error("a materialized deletion cannot contain indexes")]
212    DeleteHasIndexes,
213}
214
215#[allow(clippy::needless_pass_by_value)]
216fn codec_error(error: impl std::fmt::Display) -> MaterializedError {
217    MaterializedError::Codec(error.to_string())
218}
219
220fn validate_record_id(record_id: &[u8]) -> Result<(), MaterializedError> {
221    if record_id.len() > MAX_RECORD_ID {
222        return Err(MaterializedError::RecordIdTooLarge(record_id.len()));
223    }
224    Ok(())
225}
iroh_db_store/materialized.rs

iroh_db_store/
materialized.rs
