iroh_db_core/
crdt.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use crate::{AuthorId, Dot, VersionVector};
4
5/// A deterministic CRDT operation or merge failure.
6#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
7pub enum CrdtError {
8    /// Distinct values claimed the same globally unique operation dot.
9    #[error("one operation dot identifies conflicting values")]
10    ConflictingDot,
11    /// A multi-value read requires explicit conflict handling.
12    #[error("value has {0} concurrent alternatives")]
13    Conflict(usize),
14    /// A write-once value was assigned again in its causal future.
15    #[error("immutable value is already set")]
16    ImmutableAlreadySet,
17    /// A list insertion referenced an element that is not present.
18    #[error("ordered-list predecessor is missing")]
19    MissingPredecessor,
20    /// A list removal referenced an element that is not present.
21    #[error("ordered-list element is missing")]
22    MissingElement,
23    /// A list removal did not causally observe the referenced element.
24    #[error("ordered-list removal does not observe the element")]
25    InvalidRemoval,
26    /// An arithmetic operation exceeded its durable integer range.
27    #[error("counter overflow")]
28    CounterOverflow,
29    /// A commit attempted to allocate more operation identities than fit on the wire.
30    #[error("commit operation index exhausted")]
31    OperationIndexExhausted,
32    /// An ordered list contains a parent cycle or unreachable element.
33    #[error("ordered-list parent graph contains a cycle")]
34    CyclicList,
35}
36
37/// Deterministically combines state from another replica.
38pub trait Merge {
39    /// Merges `other` into this value.
40    fn merge(&mut self, other: &Self) -> Result<(), CrdtError>;
41}
42
43#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
44#[cbor(array)]
45struct Assignment<T> {
46    #[n(0)]
47    value: T,
48    #[n(1)]
49    dot: Dot,
50    #[n(2)]
51    context: VersionVector,
52}
53
54/// A causal last-writer-wins register that never uses wall-clock time.
55#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
56#[cbor(array)]
57pub struct Lww<T> {
58    #[n(0)]
59    assignment: Option<Assignment<T>>,
60}
61
62impl<T> Lww<T> {
63    /// Constructs an empty register.
64    pub const fn new() -> Self {
65        Self { assignment: None }
66    }
67
68    /// Returns the current value, if assigned.
69    pub fn get(&self) -> Option<&T> {
70        self.assignment.as_ref().map(|entry| &entry.value)
71    }
72
73    /// Returns the database-issued identity of the visible assignment.
74    pub fn dot(&self) -> Option<Dot> {
75        self.assignment.as_ref().map(|entry| entry.dot)
76    }
77
78    /// Returns the dependency-derived context of the visible assignment.
79    pub fn context(&self) -> Option<&VersionVector> {
80        self.assignment.as_ref().map(|entry| &entry.context)
81    }
82}
83
84impl<T: Clone + Eq> Lww<T> {
85    /// Applies an assignment using causal order, then dot order for concurrency.
86    pub fn assign(&mut self, value: T, dot: Dot, context: VersionVector) -> Result<(), CrdtError> {
87        merge_lww_assignment(
88            &mut self.assignment,
89            Assignment {
90                value,
91                dot,
92                context,
93            },
94        )
95    }
96
97    /// Deterministically combines another replica's state.
98    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
99        Merge::merge(self, other)
100    }
101
102    /// Reissues causal metadata from an authenticated commit while preserving the value.
103    pub fn canonicalize_for_commit(
104        &self,
105        author: AuthorId,
106        sequence: u64,
107        context: &VersionVector,
108        next_operation: &mut u32,
109    ) -> Result<Self, CrdtError> {
110        let mut canonical = Self::new();
111        if let Some(value) = self.get() {
112            canonical.assign(
113                value.clone(),
114                allocate_dot(author, sequence, next_operation)?,
115                context.clone(),
116            )?;
117        }
118        Ok(canonical)
119    }
120}
121
122impl<T> Default for Lww<T> {
123    fn default() -> Self {
124        Self::new()
125    }
126}
127
128impl<T: Clone + Eq> Merge for Lww<T> {
129    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
130        if let Some(candidate) = &other.assignment {
131            merge_lww_assignment(&mut self.assignment, candidate.clone())?;
132        }
133        Ok(())
134    }
135}
136
137fn merge_lww_assignment<T: Eq>(
138    current: &mut Option<Assignment<T>>,
139    mut candidate: Assignment<T>,
140) -> Result<(), CrdtError> {
141    let Some(existing) = current else {
142        *current = Some(candidate);
143        return Ok(());
144    };
145
146    if candidate.dot == existing.dot {
147        if candidate.value != existing.value {
148            return Err(CrdtError::ConflictingDot);
149        }
150        existing.context.merge(&candidate.context);
151        return Ok(());
152    }
153
154    let candidate_is_later = candidate.context.covers(&existing.dot);
155    let existing_is_later = existing.context.covers(&candidate.dot);
156    if candidate_is_later || (!existing_is_later && candidate.dot > existing.dot) {
157        candidate.context.merge(&existing.context);
158        *existing = candidate;
159    }
160    Ok(())
161}
162
163/// An add-wins observed-remove set.
164#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
165#[cbor(array)]
166pub struct OrSet<T: Ord> {
167    #[n(0)]
168    adds: BTreeMap<T, BTreeSet<Dot>>,
169    #[n(1)]
170    removed: BTreeSet<Dot>,
171}
172
173impl<T: Ord> OrSet<T> {
174    /// Constructs an empty set.
175    pub const fn new() -> Self {
176        Self {
177            adds: BTreeMap::new(),
178            removed: BTreeSet::new(),
179        }
180    }
181
182    /// Returns whether a value has at least one live add dot.
183    pub fn contains(&self, value: &T) -> bool
184    where
185        T: Ord,
186    {
187        self.adds.get(value).is_some_and(|dots| !dots.is_empty())
188    }
189
190    /// Iterates over live values in canonical value order.
191    pub fn values(&self) -> impl Iterator<Item = &T> {
192        self.adds.keys()
193    }
194}
195
196impl<T: Clone + Ord> OrSet<T> {
197    /// Adds a value under a globally unique operation dot.
198    pub fn insert(&mut self, value: T, dot: Dot) -> Result<(), CrdtError> {
199        for (existing_value, dots) in &self.adds {
200            if dots.contains(&dot) && existing_value != &value {
201                return Err(CrdtError::ConflictingDot);
202            }
203        }
204        if !self.removed.contains(&dot) {
205            self.adds.entry(value).or_default().insert(dot);
206        }
207        Ok(())
208    }
209
210    /// Removes only add operations visible in `context`.
211    pub fn remove(&mut self, value: &T, context: &VersionVector) {
212        let Some(dots) = self.adds.get_mut(value) else {
213            return;
214        };
215        let observed: Vec<_> = dots
216            .iter()
217            .copied()
218            .filter(|dot| context.covers(dot))
219            .collect();
220        for dot in observed {
221            dots.remove(&dot);
222            self.removed.insert(dot);
223        }
224        if dots.is_empty() {
225            self.adds.remove(value);
226        }
227    }
228
229    /// Deterministically combines another replica's state.
230    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
231        Merge::merge(self, other)
232    }
233
234    /// Reissues live additions under commit-owned dots and drops superseded tombstones.
235    pub fn canonicalize_for_commit(
236        &self,
237        author: AuthorId,
238        sequence: u64,
239        _context: &VersionVector,
240        next_operation: &mut u32,
241    ) -> Result<Self, CrdtError> {
242        let mut canonical = Self::new();
243        for value in self.values() {
244            canonical.insert(
245                value.clone(),
246                allocate_dot(author, sequence, next_operation)?,
247            )?;
248        }
249        Ok(canonical)
250    }
251}
252
253impl<T: Ord> Default for OrSet<T> {
254    fn default() -> Self {
255        Self::new()
256    }
257}
258
259impl<T: Clone + Ord> Merge for OrSet<T> {
260    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
261        self.removed.extend(other.removed.iter().copied());
262        for (value, dots) in &other.adds {
263            for &dot in dots {
264                self.insert(value.clone(), dot)?;
265            }
266        }
267        for dots in self.adds.values_mut() {
268            dots.retain(|dot| !self.removed.contains(dot));
269        }
270        self.adds.retain(|_, dots| !dots.is_empty());
271        Ok(())
272    }
273}
274
275/// A set whose only operation is insertion.
276#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
277#[cbor(array)]
278pub struct GrowOnlySet<T: Ord> {
279    #[n(0)]
280    values: BTreeSet<T>,
281}
282
283impl<T: Ord> GrowOnlySet<T> {
284    /// Constructs an empty set.
285    pub const fn new() -> Self {
286        Self {
287            values: BTreeSet::new(),
288        }
289    }
290
291    /// Iterates over values in canonical order.
292    pub fn values(&self) -> impl Iterator<Item = &T> {
293        self.values.iter()
294    }
295}
296
297impl<T: Clone + Ord> GrowOnlySet<T> {
298    /// Copies value-only state; this CRDT carries no caller-controlled causal metadata.
299    pub fn canonicalize_for_commit(
300        &self,
301        _author: AuthorId,
302        _sequence: u64,
303        _context: &VersionVector,
304        _next_operation: &mut u32,
305    ) -> Result<Self, CrdtError> {
306        Ok(self.clone())
307    }
308}
309
310impl<T: Ord> GrowOnlySet<T> {
311    /// Inserts a value and returns whether it was new.
312    pub fn insert(&mut self, value: T) -> bool {
313        self.values.insert(value)
314    }
315
316    /// Deterministically combines another replica's state.
317    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError>
318    where
319        T: Clone,
320    {
321        Merge::merge(self, other)
322    }
323}
324
325impl<T: Ord> Default for GrowOnlySet<T> {
326    fn default() -> Self {
327        Self::new()
328    }
329}
330
331impl<T: Clone + Ord> Merge for GrowOnlySet<T> {
332    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
333        self.values.extend(other.values.iter().cloned());
334        Ok(())
335    }
336}
337
338/// A state-based positive/negative counter with per-author components.
339#[derive(Debug, Clone, Default, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
340#[cbor(array)]
341pub struct PnCounter {
342    #[n(0)]
343    positive: BTreeMap<AuthorId, u64>,
344    #[n(1)]
345    negative: BTreeMap<AuthorId, u64>,
346}
347
348impl PnCounter {
349    /// Constructs a zero counter.
350    pub const fn new() -> Self {
351        Self {
352            positive: BTreeMap::new(),
353            negative: BTreeMap::new(),
354        }
355    }
356
357    /// Adds to an author's monotonic positive component.
358    pub fn increment(&mut self, author: AuthorId, amount: u64) -> Result<(), CrdtError> {
359        add_component(&mut self.positive, author, amount)
360    }
361
362    /// Adds to an author's monotonic negative component.
363    pub fn decrement(&mut self, author: AuthorId, amount: u64) -> Result<(), CrdtError> {
364        add_component(&mut self.negative, author, amount)
365    }
366
367    /// Computes the signed counter value with overflow checking.
368    pub fn value(&self) -> Result<i128, CrdtError> {
369        let positive = sum_components(&self.positive)?;
370        let negative = sum_components(&self.negative)?;
371        positive
372            .checked_sub(negative)
373            .ok_or(CrdtError::CounterOverflow)
374    }
375
376    /// Deterministically combines another replica's state.
377    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
378        Merge::merge(self, other)
379    }
380
381    /// Attributes the visible signed value solely to the authenticated commit author.
382    pub fn canonicalize_for_commit(
383        &self,
384        author: AuthorId,
385        _sequence: u64,
386        _context: &VersionVector,
387        _next_operation: &mut u32,
388    ) -> Result<Self, CrdtError> {
389        let value = self.value()?;
390        let mut canonical = Self::new();
391        if value >= 0 {
392            canonical.increment(
393                author,
394                u64::try_from(value).map_err(|_| CrdtError::CounterOverflow)?,
395            )?;
396        } else {
397            let magnitude = value.checked_abs().ok_or(CrdtError::CounterOverflow)?;
398            canonical.decrement(
399                author,
400                u64::try_from(magnitude).map_err(|_| CrdtError::CounterOverflow)?,
401            )?;
402        }
403        Ok(canonical)
404    }
405}
406
407impl Merge for PnCounter {
408    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
409        merge_components(&mut self.positive, &other.positive);
410        merge_components(&mut self.negative, &other.negative);
411        self.value().map(|_| ())
412    }
413}
414
415fn add_component(
416    components: &mut BTreeMap<AuthorId, u64>,
417    author: AuthorId,
418    amount: u64,
419) -> Result<(), CrdtError> {
420    let value = components.entry(author).or_default();
421    *value = value
422        .checked_add(amount)
423        .ok_or(CrdtError::CounterOverflow)?;
424    Ok(())
425}
426
427fn merge_components(components: &mut BTreeMap<AuthorId, u64>, other: &BTreeMap<AuthorId, u64>) {
428    for (&author, &other_value) in other {
429        components
430            .entry(author)
431            .and_modify(|value| *value = (*value).max(other_value))
432            .or_insert(other_value);
433    }
434}
435
436fn sum_components(components: &BTreeMap<AuthorId, u64>) -> Result<i128, CrdtError> {
437    components.values().try_fold(0_i128, |total, value| {
438        total
439            .checked_add(i128::from(*value))
440            .ok_or(CrdtError::CounterOverflow)
441    })
442}
443
444/// A register retaining every causally maximal assignment.
445#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
446#[cbor(array)]
447pub struct MultiValue<T> {
448    #[n(0)]
449    entries: BTreeMap<Dot, Assignment<T>>,
450}
451
452impl<T> MultiValue<T> {
453    /// Constructs an empty register.
454    pub const fn new() -> Self {
455        Self {
456            entries: BTreeMap::new(),
457        }
458    }
459
460    /// Iterates over concurrent values in canonical dot order.
461    pub fn values(&self) -> impl Iterator<Item = &T> {
462        self.entries.values().map(|entry| &entry.value)
463    }
464
465    /// Reads zero or one value, returning a typed conflict for concurrency.
466    pub fn read_one(&self) -> Result<Option<&T>, CrdtError> {
467        match self.entries.len() {
468            0 => Ok(None),
469            1 => Ok(self.entries.values().next().map(|entry| &entry.value)),
470            count => Err(CrdtError::Conflict(count)),
471        }
472    }
473}
474
475impl<T: Clone + Eq> MultiValue<T> {
476    /// Applies an assignment and removes only values it causally observes.
477    pub fn assign(&mut self, value: T, dot: Dot, context: VersionVector) -> Result<(), CrdtError> {
478        if let Some(existing) = self.entries.get_mut(&dot) {
479            if existing.value != value {
480                return Err(CrdtError::ConflictingDot);
481            }
482            existing.context.merge(&context);
483            return Ok(());
484        }
485        if self
486            .entries
487            .values()
488            .any(|existing| existing.context.covers(&dot))
489        {
490            return Ok(());
491        }
492        self.entries
493            .retain(|_, existing| !context.covers(&existing.dot));
494        self.entries.insert(
495            dot,
496            Assignment {
497                value,
498                dot,
499                context,
500            },
501        );
502        Ok(())
503    }
504
505    /// Deterministically combines another replica's state.
506    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
507        Merge::merge(self, other)
508    }
509
510    /// Reissues every visible alternative under commit-owned dots.
511    pub fn canonicalize_for_commit(
512        &self,
513        author: AuthorId,
514        sequence: u64,
515        context: &VersionVector,
516        next_operation: &mut u32,
517    ) -> Result<Self, CrdtError> {
518        let mut canonical = Self::new();
519        for value in self.values() {
520            canonical.assign(
521                value.clone(),
522                allocate_dot(author, sequence, next_operation)?,
523                context.clone(),
524            )?;
525        }
526        Ok(canonical)
527    }
528}
529
530impl<T> Default for MultiValue<T> {
531    fn default() -> Self {
532        Self::new()
533    }
534}
535
536impl<T: Clone + Eq> Merge for MultiValue<T> {
537    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
538        for (&dot, entry) in &other.entries {
539            if let Some(existing) = self.entries.get_mut(&dot) {
540                if existing.value != entry.value {
541                    return Err(CrdtError::ConflictingDot);
542                }
543                existing.context.merge(&entry.context);
544            } else {
545                self.entries.insert(dot, entry.clone());
546            }
547        }
548
549        let entries: Vec<_> = self.entries.values().cloned().collect();
550        self.entries.retain(|dot, _| {
551            !entries
552                .iter()
553                .any(|other| other.dot != *dot && other.context.covers(dot))
554        });
555        Ok(())
556    }
557}
558
559/// A write-once register that retains unequal concurrent initial values.
560#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
561#[cbor(array)]
562pub struct Immutable<T> {
563    #[n(0)]
564    values: MultiValue<T>,
565}
566
567impl<T> Immutable<T> {
568    /// Constructs an unset immutable value.
569    pub const fn new() -> Self {
570        Self {
571            values: MultiValue::new(),
572        }
573    }
574
575    /// Reads the value, returning a typed error for concurrent initial values.
576    pub fn get(&self) -> Result<Option<&T>, CrdtError> {
577        self.values.read_one()
578    }
579
580    /// Iterates over all causally concurrent initial values in dot order.
581    pub fn values(&self) -> impl Iterator<Item = &T> {
582        self.values.values()
583    }
584}
585
586impl<T: Clone + Eq> Immutable<T> {
587    /// Sets an initial value, rejecting causal reassignment.
588    pub fn set(&mut self, value: T, dot: Dot, context: VersionVector) -> Result<(), CrdtError> {
589        if self
590            .values
591            .entries
592            .values()
593            .any(|existing| context.covers(&existing.dot))
594        {
595            return Err(CrdtError::ImmutableAlreadySet);
596        }
597        self.values.assign(value, dot, context)
598    }
599
600    /// Deterministically combines another replica's state.
601    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
602        Merge::merge(self, other)
603    }
604
605    /// Reissues every concurrent initial value under commit-owned dots.
606    pub fn canonicalize_for_commit(
607        &self,
608        author: AuthorId,
609        sequence: u64,
610        context: &VersionVector,
611        next_operation: &mut u32,
612    ) -> Result<Self, CrdtError> {
613        let mut canonical = Self::new();
614        for value in self.values() {
615            canonical.set(
616                value.clone(),
617                allocate_dot(author, sequence, next_operation)?,
618                context.clone(),
619            )?;
620        }
621        Ok(canonical)
622    }
623}
624
625impl<T> Default for Immutable<T> {
626    fn default() -> Self {
627        Self::new()
628    }
629}
630
631impl<T: Clone + Eq> Merge for Immutable<T> {
632    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
633        for left in self.values.entries.values() {
634            for right in other.values.entries.values() {
635                if left.dot != right.dot
636                    && (left.context.covers(&right.dot) || right.context.covers(&left.dot))
637                {
638                    return Err(CrdtError::ImmutableAlreadySet);
639                }
640            }
641        }
642        self.values.merge(&other.values)
643    }
644}
645
646#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
647#[cbor(array)]
648struct ListElement<T> {
649    #[n(0)]
650    parent: Option<Dot>,
651    #[n(1)]
652    value: T,
653    #[n(2)]
654    removed: bool,
655}
656
657/// A replicated growable ordered list with stable element dots.
658#[derive(Debug, Clone, PartialEq, Eq, minicbor::Encode, minicbor::Decode)]
659#[cbor(array)]
660pub struct OrderedList<T> {
661    #[n(0)]
662    elements: BTreeMap<Dot, ListElement<T>>,
663}
664
665impl<T> OrderedList<T> {
666    /// Constructs an empty list.
667    pub const fn new() -> Self {
668        Self {
669            elements: BTreeMap::new(),
670        }
671    }
672
673    /// Iterates over visible values in replicated list order.
674    pub fn values(&self) -> impl Iterator<Item = &T> {
675        let mut values = Vec::with_capacity(self.elements.len());
676        let mut pending: Vec<_> = self
677            .elements
678            .iter()
679            .rev()
680            .filter_map(|(&dot, element)| element.parent.is_none().then_some(dot))
681            .collect();
682        while let Some(dot) = pending.pop() {
683            let element = &self.elements[&dot];
684            if !element.removed {
685                values.push(&element.value);
686            }
687            pending.extend(
688                self.elements
689                    .iter()
690                    .rev()
691                    .filter_map(|(&child, element)| (element.parent == Some(dot)).then_some(child)),
692            );
693        }
694        values.into_iter()
695    }
696}
697
698impl<T: Clone + Eq> OrderedList<T> {
699    /// Inserts a stable element after `parent`, or at the replicated root.
700    pub fn insert_after(
701        &mut self,
702        parent: Option<Dot>,
703        dot: Dot,
704        value: T,
705    ) -> Result<(), CrdtError> {
706        if parent.is_some_and(|parent| !self.elements.contains_key(&parent)) {
707            return Err(CrdtError::MissingPredecessor);
708        }
709        if let Some(existing) = self.elements.get(&dot) {
710            if existing.parent == parent && existing.value == value {
711                return Ok(());
712            }
713            return Err(CrdtError::ConflictingDot);
714        }
715        self.elements.insert(
716            dot,
717            ListElement {
718                parent,
719                value,
720                removed: false,
721            },
722        );
723        Ok(())
724    }
725
726    /// Removes an element that is visible in `context`.
727    pub fn remove(&mut self, dot: Dot, context: &VersionVector) -> Result<(), CrdtError> {
728        let element = self
729            .elements
730            .get_mut(&dot)
731            .ok_or(CrdtError::MissingElement)?;
732        if !context.covers(&dot) {
733            return Err(CrdtError::InvalidRemoval);
734        }
735        element.removed = true;
736        Ok(())
737    }
738
739    /// Deterministically combines another replica's state.
740    pub fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
741        Merge::merge(self, other)
742    }
743
744    /// Rebuilds the visible sequence with commit-owned element identities.
745    pub fn canonicalize_for_commit(
746        &self,
747        author: AuthorId,
748        sequence: u64,
749        _context: &VersionVector,
750        next_operation: &mut u32,
751    ) -> Result<Self, CrdtError> {
752        self.validate_structure()?;
753        let mut canonical = Self::new();
754        let mut parent = None;
755        for value in self.values() {
756            let dot = allocate_dot(author, sequence, next_operation)?;
757            canonical.insert_after(parent, dot, value.clone())?;
758            parent = Some(dot);
759        }
760        Ok(canonical)
761    }
762
763    fn validate_structure(&self) -> Result<(), CrdtError> {
764        for &start in self.elements.keys() {
765            let mut path = BTreeSet::new();
766            let mut current = Some(start);
767            while let Some(dot) = current {
768                if !path.insert(dot) {
769                    return Err(CrdtError::CyclicList);
770                }
771                let element = self
772                    .elements
773                    .get(&dot)
774                    .ok_or(CrdtError::MissingPredecessor)?;
775                current = element.parent;
776            }
777        }
778        Ok(())
779    }
780}
781
782impl<T> Default for OrderedList<T> {
783    fn default() -> Self {
784        Self::new()
785    }
786}
787
788impl<T: Clone + Eq> Merge for OrderedList<T> {
789    fn merge(&mut self, other: &Self) -> Result<(), CrdtError> {
790        for (&dot, other_element) in &other.elements {
791            if let Some(element) = self.elements.get_mut(&dot) {
792                if element.parent != other_element.parent || element.value != other_element.value {
793                    return Err(CrdtError::ConflictingDot);
794                }
795                element.removed |= other_element.removed;
796            } else {
797                self.elements.insert(dot, other_element.clone());
798            }
799        }
800        if self.elements.values().any(|element| {
801            element
802                .parent
803                .is_some_and(|parent| !self.elements.contains_key(&parent))
804        }) {
805            return Err(CrdtError::MissingPredecessor);
806        }
807        self.validate_structure()?;
808        Ok(())
809    }
810}
811
812fn allocate_dot(
813    author: AuthorId,
814    sequence: u64,
815    next_operation: &mut u32,
816) -> Result<Dot, CrdtError> {
817    let operation = *next_operation;
818    *next_operation = next_operation
819        .checked_add(1)
820        .ok_or(CrdtError::OperationIndexExhausted)?;
821    Ok(Dot::new(author, sequence, operation))
822}
iroh_db_core/crdt.rs

iroh_db_core/
crdt.rs
