iroh_db_core/
record.rs

1use std::{
2    any::type_name, collections::BTreeMap, convert::Infallible, marker::PhantomData, sync::Arc,
3};
4
5use minicbor::{Decode, Decoder, Encode, Encoder};
6
7use crate::{
8    AuthorId, CrdtError, CrdtKind, GrowOnlySet, Immutable, Lww, Merge, MultiValue, OrSet,
9    OrderedList, PnCounter, SchemaDescriptor, SchemaError, VersionVector,
10};
11
12const MAX_RECORD_FIELDS: usize = 4_096;
13const MAX_FIELD_STATE: usize = 16 * 1024 * 1024;
14
15/// Compile-time metadata for one strongly typed query field.
16#[derive(Debug, PartialEq, Eq, Hash)]
17pub struct TypedField<Record, Value> {
18    field_id: u32,
19    name: &'static str,
20    indexed: bool,
21    marker: PhantomData<fn() -> (Record, Value)>,
22}
23
24/// A type-checked equality predicate over canonical visible field values.
25#[derive(Debug)]
26pub struct EqualityFilter<Record> {
27    field_id: u32,
28    encoded_value: Result<Vec<u8>, RecordError>,
29    marker: PhantomData<fn() -> Record>,
30}
31
32/// A type-checked ordered predicate over canonical visible field values.
33pub struct RangeFilter<Record> {
34    field_id: u32,
35    operator: RangeOperator,
36    encoded_bound: Result<Vec<u8>, RecordError>,
37    matches: Arc<RangeMatcher>,
38    marker: PhantomData<fn() -> Record>,
39}
40
41type RangeMatcher = dyn Fn(&[Vec<u8>]) -> Result<bool, RecordError> + Send + Sync;
42
43impl<Record> std::fmt::Debug for RangeFilter<Record> {
44    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        formatter
46            .debug_struct("RangeFilter")
47            .field("field_id", &self.field_id)
48            .field("operator", &self.operator)
49            .finish_non_exhaustive()
50    }
51}
52
53/// One supported ordered comparison.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
55pub enum RangeOperator {
56    /// Strictly less than the bound.
57    Less,
58    /// Less than or equal to the bound.
59    LessOrEqual,
60    /// Strictly greater than the bound.
61    Greater,
62    /// Greater than or equal to the bound.
63    GreaterOrEqual,
64}
65
66impl<Record> RangeFilter<Record> {
67    /// Returns the stable field targeted by this predicate.
68    pub const fn field_id(&self) -> u32 {
69        self.field_id
70    }
71
72    /// Returns the comparison operator.
73    pub const fn operator(&self) -> RangeOperator {
74        self.operator
75    }
76
77    /// Returns the canonical bound or its deferred codec failure.
78    pub fn encoded_bound(&self) -> Result<&[u8], &RecordError> {
79        self.encoded_bound.as_deref()
80    }
81
82    /// Applies this predicate to canonical visible values.
83    pub fn matches(&self, values: &[Vec<u8>]) -> Result<bool, RecordError> {
84        (self.matches)(values)
85    }
86}
87
88impl<Record> EqualityFilter<Record> {
89    /// Returns the stable field targeted by this predicate.
90    pub const fn field_id(&self) -> u32 {
91        self.field_id
92    }
93
94    /// Returns the canonical equality key or its deferred codec failure.
95    pub fn encoded_value(&self) -> Result<&[u8], &RecordError> {
96        self.encoded_value.as_deref()
97    }
98}
99
100impl<Record, Value> TypedField<Record, Value> {
101    /// Constructs generated field metadata.
102    pub const fn new(field_id: u32, name: &'static str, indexed: bool) -> Self {
103        Self {
104            field_id,
105            name,
106            indexed,
107            marker: PhantomData,
108        }
109    }
110
111    /// Returns the stable numeric field ID.
112    pub const fn field_id(self) -> u32 {
113        self.field_id
114    }
115
116    /// Returns the declared Rust field name.
117    pub const fn name(self) -> &'static str {
118        self.name
119    }
120
121    /// Returns whether the schema materializes a secondary index.
122    pub const fn is_indexed(self) -> bool {
123        self.indexed
124    }
125
126    /// Builds a type-checked equality predicate.
127    pub fn eq(self, value: impl Into<Value>) -> EqualityFilter<Record>
128    where
129        Value: RecordKey,
130    {
131        EqualityFilter {
132            field_id: self.field_id,
133            encoded_value: value.into().encode_key(),
134            marker: PhantomData,
135        }
136    }
137
138    /// Builds a type-checked less-than predicate.
139    pub fn lt(self, value: impl Into<Value>) -> RangeFilter<Record>
140    where
141        Value: RecordKey + Ord + Send + Sync + 'static,
142    {
143        self.range(value.into(), RangeOperator::Less)
144    }
145
146    /// Builds a type-checked less-than-or-equal predicate.
147    pub fn le(self, value: impl Into<Value>) -> RangeFilter<Record>
148    where
149        Value: RecordKey + Ord + Send + Sync + 'static,
150    {
151        self.range(value.into(), RangeOperator::LessOrEqual)
152    }
153
154    /// Builds a type-checked greater-than predicate.
155    pub fn gt(self, value: impl Into<Value>) -> RangeFilter<Record>
156    where
157        Value: RecordKey + Ord + Send + Sync + 'static,
158    {
159        self.range(value.into(), RangeOperator::Greater)
160    }
161
162    /// Builds a type-checked greater-than-or-equal predicate.
163    pub fn ge(self, value: impl Into<Value>) -> RangeFilter<Record>
164    where
165        Value: RecordKey + Ord + Send + Sync + 'static,
166    {
167        self.range(value.into(), RangeOperator::GreaterOrEqual)
168    }
169
170    #[allow(clippy::needless_pass_by_value)]
171    fn range(self, bound: Value, operator: RangeOperator) -> RangeFilter<Record>
172    where
173        Value: RecordKey + Ord + Send + Sync + 'static,
174    {
175        let encoded_bound = bound.encode_key();
176        let matches_bound = encoded_bound.clone();
177        RangeFilter {
178            field_id: self.field_id,
179            operator,
180            encoded_bound,
181            matches: Arc::new(move |values| {
182                let bound = Value::decode_key(matches_bound.as_deref().map_err(Clone::clone)?)?;
183                values.iter().try_fold(false, |matched, encoded| {
184                    let value = Value::decode_key(encoded)?;
185                    let current = match operator {
186                        RangeOperator::Less => value < bound,
187                        RangeOperator::LessOrEqual => value <= bound,
188                        RangeOperator::Greater => value > bound,
189                        RangeOperator::GreaterOrEqual => value >= bound,
190                    };
191                    Ok(matched || current)
192                })
193            }),
194            marker: PhantomData,
195        }
196    }
197}
198
199impl<Record, Value> Clone for TypedField<Record, Value> {
200    fn clone(&self) -> Self {
201        *self
202    }
203}
204
205impl<Record, Value> Copy for TypedField<Record, Value> {}
206
207/// A typed-record schema or canonical codec failure.
208#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
209pub enum RecordError {
210    /// The generated or hand-written schema is invalid.
211    #[error(transparent)]
212    Schema(#[from] SchemaError),
213    /// A value could not be encoded.
214    #[error("record encoding failed: {0}")]
215    Encode(String),
216    /// A value could not be decoded.
217    #[error("record decoding failed: {0}")]
218    Decode(String),
219    /// Bytes followed an otherwise complete value.
220    #[error("record contains trailing data")]
221    TrailingData,
222    /// The input was valid CBOR but not its unique canonical representation.
223    #[error("record encoding is not canonical")]
224    NonCanonicalEncoding,
225    /// The record omitted a field required by its schema.
226    #[error("record is missing field {0}")]
227    MissingField(u32),
228    /// A field ID appeared more than once.
229    #[error("record repeats field {0}")]
230    DuplicateField(u32),
231    /// The record exceeded the field-count safety limit.
232    #[error("record has too many fields: {0}")]
233    TooManyFields(usize),
234    /// One encoded field state exceeded the safety limit.
235    #[error("field {field_id} state is too large: {size}")]
236    FieldTooLarge {
237        /// Stable numeric field identifier.
238        field_id: u32,
239        /// Encoded byte length.
240        size: usize,
241    },
242    /// A replicated field rejected invalid or conflicting causal state.
243    #[error(transparent)]
244    Crdt(#[from] CrdtError),
245    /// A hand-written record did not provide a deterministic merge adapter.
246    #[error("record type does not provide a merge adapter")]
247    MergeUnavailable,
248    /// A hand-written record did not provide commit-owned causal normalization.
249    #[error("record type does not provide a causal normalization adapter")]
250    CausalNormalizationUnavailable,
251}
252
253/// Canonical application record identifier encoding.
254pub trait RecordKey: Sized {
255    /// Encodes this key into its unique durable representation.
256    fn encode_key(&self) -> Result<Vec<u8>, RecordError>;
257
258    /// Strictly decodes a key from its unique durable representation.
259    fn decode_key(bytes: &[u8]) -> Result<Self, RecordError>;
260}
261
262impl<T> RecordKey for T
263where
264    T: Encode<()> + for<'bytes> Decode<'bytes, ()>,
265{
266    fn encode_key(&self) -> Result<Vec<u8>, RecordError> {
267        encode_value(self)
268    }
269
270    fn decode_key(bytes: &[u8]) -> Result<Self, RecordError> {
271        decode_value(bytes)
272    }
273}
274
275/// Canonical state codec and schema metadata for a replicated field.
276pub trait FieldState: Sized {
277    /// The application value carried by this field strategy.
278    type Value;
279
280    /// The deterministic merge strategy.
281    const CRDT_KIND: CrdtKind;
282
283    /// Stable Rust logical value type used in schema compatibility checks.
284    fn value_type_name() -> &'static str {
285        type_name::<Self::Value>()
286    }
287
288    /// Encodes complete causal field state.
289    fn encode_state(&self) -> Result<Vec<u8>, RecordError>;
290
291    /// Strictly decodes complete causal field state.
292    fn decode_state(bytes: &[u8]) -> Result<Self, RecordError>;
293
294    /// Encodes each currently visible application value for filtering and indexes.
295    fn visible_values(&self) -> Result<Vec<Vec<u8>>, RecordError>;
296
297    /// Deterministically merges complete causal state from another replica.
298    fn merge_state(&mut self, other: &Self) -> Result<(), RecordError>;
299
300    /// Reissues all causal identities from one authenticated commit envelope.
301    fn canonicalize_for_commit(
302        &self,
303        author: AuthorId,
304        sequence: u64,
305        context: &VersionVector,
306        next_operation: &mut u32,
307    ) -> Result<Self, RecordError>;
308}
309
310macro_rules! field_state {
311    ($type:ident<$value:ident>, $kind:ident) => {
312        impl<$value> FieldState for $type<$value>
313        where
314            $value: Clone + Eq + Encode<()> + for<'bytes> Decode<'bytes, ()>,
315        {
316            type Value = $value;
317
318            const CRDT_KIND: CrdtKind = CrdtKind::$kind;
319
320            fn encode_state(&self) -> Result<Vec<u8>, RecordError> {
321                encode_value(self)
322            }
323
324            fn decode_state(bytes: &[u8]) -> Result<Self, RecordError> {
325                decode_value(bytes)
326            }
327
328            fn visible_values(&self) -> Result<Vec<Vec<u8>>, RecordError> {
329                self.values().map(encode_value).collect()
330            }
331
332            fn merge_state(&mut self, other: &Self) -> Result<(), RecordError> {
333                Merge::merge(self, other).map_err(RecordError::from)
334            }
335
336            fn canonicalize_for_commit(
337                &self,
338                author: AuthorId,
339                sequence: u64,
340                context: &VersionVector,
341                next_operation: &mut u32,
342            ) -> Result<Self, RecordError> {
343                self.canonicalize_for_commit(author, sequence, context, next_operation)
344                    .map_err(RecordError::from)
345            }
346        }
347    };
348}
349
350impl<T> FieldState for Lww<T>
351where
352    T: Clone + Eq + Encode<()> + for<'bytes> Decode<'bytes, ()>,
353{
354    type Value = T;
355
356    const CRDT_KIND: CrdtKind = CrdtKind::Lww;
357
358    fn encode_state(&self) -> Result<Vec<u8>, RecordError> {
359        encode_value(self)
360    }
361
362    fn decode_state(bytes: &[u8]) -> Result<Self, RecordError> {
363        decode_value(bytes)
364    }
365
366    fn visible_values(&self) -> Result<Vec<Vec<u8>>, RecordError> {
367        self.get().map(encode_value).into_iter().collect()
368    }
369
370    fn merge_state(&mut self, other: &Self) -> Result<(), RecordError> {
371        Merge::merge(self, other).map_err(RecordError::from)
372    }
373
374    fn canonicalize_for_commit(
375        &self,
376        author: AuthorId,
377        sequence: u64,
378        context: &VersionVector,
379        next_operation: &mut u32,
380    ) -> Result<Self, RecordError> {
381        self.canonicalize_for_commit(author, sequence, context, next_operation)
382            .map_err(RecordError::from)
383    }
384}
385
386field_state!(MultiValue<T>, MultiValue);
387field_state!(Immutable<T>, Immutable);
388field_state!(OrderedList<T>, OrderedList);
389
390macro_rules! ordered_field_state {
391    ($type:ident<$value:ident>, $kind:ident) => {
392        impl<$value> FieldState for $type<$value>
393        where
394            $value: Clone + Ord + Encode<()> + for<'bytes> Decode<'bytes, ()>,
395        {
396            type Value = $value;
397
398            const CRDT_KIND: CrdtKind = CrdtKind::$kind;
399
400            fn encode_state(&self) -> Result<Vec<u8>, RecordError> {
401                encode_value(self)
402            }
403
404            fn decode_state(bytes: &[u8]) -> Result<Self, RecordError> {
405                decode_value(bytes)
406            }
407
408            fn visible_values(&self) -> Result<Vec<Vec<u8>>, RecordError> {
409                self.values().map(encode_value).collect()
410            }
411
412            fn merge_state(&mut self, other: &Self) -> Result<(), RecordError> {
413                Merge::merge(self, other).map_err(RecordError::from)
414            }
415
416            fn canonicalize_for_commit(
417                &self,
418                author: AuthorId,
419                sequence: u64,
420                context: &VersionVector,
421                next_operation: &mut u32,
422            ) -> Result<Self, RecordError> {
423                self.canonicalize_for_commit(author, sequence, context, next_operation)
424                    .map_err(RecordError::from)
425            }
426        }
427    };
428}
429
430ordered_field_state!(OrSet<T>, OrSet);
431ordered_field_state!(GrowOnlySet<T>, GrowOnlySet);
432
433impl FieldState for PnCounter {
434    type Value = i128;
435
436    const CRDT_KIND: CrdtKind = CrdtKind::PnCounter;
437
438    fn encode_state(&self) -> Result<Vec<u8>, RecordError> {
439        encode_value(self)
440    }
441
442    fn decode_state(bytes: &[u8]) -> Result<Self, RecordError> {
443        decode_value(bytes)
444    }
445
446    fn visible_values(&self) -> Result<Vec<Vec<u8>>, RecordError> {
447        Ok(vec![
448            self.value()
449                .map_err(|error| RecordError::Encode(error.to_string()))?
450                .to_be_bytes()
451                .to_vec(),
452        ])
453    }
454
455    fn merge_state(&mut self, other: &Self) -> Result<(), RecordError> {
456        Merge::merge(self, other).map_err(RecordError::from)
457    }
458
459    fn canonicalize_for_commit(
460        &self,
461        author: AuthorId,
462        sequence: u64,
463        context: &VersionVector,
464        next_operation: &mut u32,
465    ) -> Result<Self, RecordError> {
466        self.canonicalize_for_commit(author, sequence, context, next_operation)
467            .map_err(RecordError::from)
468    }
469}
470
471/// Strongly typed record contract generated by `#[derive(IrohRecord)]`.
472pub trait IrohRecord: Sized {
473    /// Application-controlled primary-key type.
474    type Id: RecordKey;
475
476    /// Returns the validated durable schema descriptor.
477    fn schema() -> Result<SchemaDescriptor, RecordError>;
478
479    /// Borrows the primary key.
480    fn id(&self) -> &Self::Id;
481
482    /// Encodes all record fields in stable numeric-field order.
483    fn encode_record(&self) -> Result<Vec<u8>, RecordError>;
484
485    /// Strictly decodes a complete typed record.
486    fn decode_record(bytes: &[u8]) -> Result<Self, RecordError>;
487
488    /// Encodes the canonical primary key used by storage and indexes.
489    fn record_id(&self) -> Result<Vec<u8>, RecordError> {
490        self.id().encode_key()
491    }
492
493    /// Returns canonical visible values for one field, or an empty set if unknown.
494    fn field_values(&self, field_id: u32) -> Result<Vec<Vec<u8>>, RecordError>;
495
496    /// Merges every replicated field from another incarnation of the same record.
497    fn merge_record(&mut self, _other: &Self) -> Result<(), RecordError> {
498        Err(RecordError::MergeUnavailable)
499    }
500
501    /// Reissues every replicated field identity from one authenticated commit.
502    fn canonicalize_causality(
503        &mut self,
504        _author: AuthorId,
505        _sequence: u64,
506        _context: &VersionVector,
507        _next_operation: &mut u32,
508    ) -> Result<(), RecordError> {
509        Err(RecordError::CausalNormalizationUnavailable)
510    }
511}
512
513/// Strictly decoded stable field-state map.
514#[derive(Debug, Clone, PartialEq, Eq)]
515pub struct RecordFields {
516    fields: BTreeMap<u32, Vec<u8>>,
517}
518
519impl RecordFields {
520    /// Returns one field or a typed missing-field error.
521    pub fn required(&self, field_id: u32) -> Result<&[u8], RecordError> {
522        self.fields
523            .get(&field_id)
524            .map(Vec::as_slice)
525            .ok_or(RecordError::MissingField(field_id))
526    }
527}
528
529/// Encodes field states as a canonical CBOR map keyed by stable numeric IDs.
530pub fn encode_record_fields(
531    fields: impl IntoIterator<Item = (u32, Vec<u8>)>,
532) -> Result<Vec<u8>, RecordError> {
533    let mut canonical = BTreeMap::new();
534    for (field_id, state) in fields {
535        if state.len() > MAX_FIELD_STATE {
536            return Err(RecordError::FieldTooLarge {
537                field_id,
538                size: state.len(),
539            });
540        }
541        if canonical.insert(field_id, state).is_some() {
542            return Err(RecordError::DuplicateField(field_id));
543        }
544        if canonical.len() > MAX_RECORD_FIELDS {
545            return Err(RecordError::TooManyFields(canonical.len()));
546        }
547    }
548
549    let mut encoder = Encoder::new(Vec::new());
550    encoder
551        .map(
552            u64::try_from(canonical.len())
553                .map_err(|error| RecordError::Encode(error.to_string()))?,
554        )
555        .map_err(encode_error)?;
556    for (field_id, state) in canonical {
557        encoder.u32(field_id).map_err(encode_error)?;
558        encoder.bytes(&state).map_err(encode_error)?;
559    }
560    Ok(encoder.into_writer())
561}
562
563/// Strictly decodes a canonical stable field-state map.
564pub fn decode_record_fields(bytes: &[u8]) -> Result<RecordFields, RecordError> {
565    let mut decoder = Decoder::new(bytes);
566    let count = decoder
567        .map()
568        .map_err(decode_error)?
569        .ok_or_else(|| RecordError::Decode("indefinite record maps are forbidden".into()))?;
570    let count = usize::try_from(count).map_err(|error| RecordError::Decode(error.to_string()))?;
571    if count > MAX_RECORD_FIELDS {
572        return Err(RecordError::TooManyFields(count));
573    }
574
575    let mut fields = BTreeMap::new();
576    for _ in 0..count {
577        let field_id = decoder.u32().map_err(decode_error)?;
578        let state = decoder.bytes().map_err(decode_error)?;
579        if state.len() > MAX_FIELD_STATE {
580            return Err(RecordError::FieldTooLarge {
581                field_id,
582                size: state.len(),
583            });
584        }
585        if fields.insert(field_id, state.to_vec()).is_some() {
586            return Err(RecordError::DuplicateField(field_id));
587        }
588    }
589    if decoder.position() != bytes.len() {
590        return Err(RecordError::TrailingData);
591    }
592
593    let record_fields = RecordFields { fields };
594    let canonical = encode_record_fields(
595        record_fields
596            .fields
597            .iter()
598            .map(|(&field_id, state)| (field_id, state.clone())),
599    )?;
600    if canonical != bytes {
601        return Err(RecordError::NonCanonicalEncoding);
602    }
603    Ok(record_fields)
604}
605
606fn encode_value<T: Encode<()>>(value: &T) -> Result<Vec<u8>, RecordError> {
607    minicbor::to_vec(value).map_err(|error| RecordError::Encode(error.to_string()))
608}
609
610fn decode_value<T>(bytes: &[u8]) -> Result<T, RecordError>
611where
612    T: Encode<()> + for<'value> Decode<'value, ()>,
613{
614    let mut decoder = Decoder::new(bytes);
615    let value = decoder.decode::<T>().map_err(decode_error)?;
616    if decoder.position() != bytes.len() {
617        return Err(RecordError::TrailingData);
618    }
619    if encode_value(&value)? != bytes {
620        return Err(RecordError::NonCanonicalEncoding);
621    }
622    Ok(value)
623}
624
625#[allow(clippy::needless_pass_by_value)]
626fn encode_error(error: minicbor::encode::Error<Infallible>) -> RecordError {
627    RecordError::Encode(error.to_string())
628}
629
630#[allow(clippy::needless_pass_by_value)]
631fn decode_error(error: minicbor::decode::Error) -> RecordError {
632    RecordError::Decode(error.to_string())
633}
iroh_db_core/record.rs

iroh_db_core/
record.rs
