safemesh_crdt/crdt/
orset.rs

1// SafeMesh — delta-state CRDT convergence, built on crdt-lean.
2// Copyright (C) 2026 Ben Cassie
3// SPDX-License-Identifier: Apache-2.0
4
5use crate::{Crdt, MergeError, Mergeable, RecordId, WireError};
6use alloc::{collections::BTreeSet, vec::Vec};
7
8/// Delta for an observed-remove set.
9#[derive(Clone, Debug)]
10pub enum OrSetDelta<T, K> {
11    Add { element: T, token: K },
12    Remove { tokens: Vec<K> },
13}
14
15// Remove denotes a Finset in the Lean model. Compare that set for record
16// admission, while retaining the received vector for exact wire round trips.
17impl<T: PartialEq, K: Ord> PartialEq for OrSetDelta<T, K> {
18    fn eq(&self, other: &Self) -> bool {
19        match (self, other) {
20            (
21                Self::Add {
22                    element: a,
23                    token: x,
24                },
25                Self::Add {
26                    element: b,
27                    token: y,
28                },
29            ) => a == b && x == y,
30            (Self::Remove { tokens: a }, Self::Remove { tokens: b }) => {
31                a.iter().collect::<BTreeSet<_>>() == b.iter().collect::<BTreeSet<_>>()
32            }
33            _ => false,
34        }
35    }
36}
37
38impl<T: Eq, K: Ord> Eq for OrSetDelta<T, K> {}
39
40/// Observed-remove set with add-wins semantics.
41///
42/// Lean model: `Crdt.ORSet.State α τ = Finset (α × τ) × Finset τ`.
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub struct OrSet<T: Ord, K: Ord> {
45    pub(super) adds: BTreeSet<(T, K)>,
46    pub(super) tombstones: BTreeSet<K>,
47}
48
49impl<T: Ord, K: Ord> OrSet<T, K> {
50    pub fn new() -> Self {
51        OrSet {
52            adds: BTreeSet::new(),
53            tombstones: BTreeSet::new(),
54        }
55    }
56
57    pub fn add(&mut self, element: T, token: K) {
58        self.adds.insert((element, token));
59    }
60
61    pub fn apply_remove<I>(&mut self, tokens: I)
62    where
63        I: IntoIterator<Item = K>,
64    {
65        self.tombstones.extend(tokens);
66    }
67
68    pub fn merge(&mut self, other: &Self)
69    where
70        T: Clone,
71        K: Clone,
72    {
73        self.adds.extend(other.adds.iter().cloned());
74        self.tombstones.extend(other.tombstones.iter().cloned());
75    }
76
77    pub fn adds(&self) -> &BTreeSet<(T, K)> {
78        &self.adds
79    }
80
81    pub fn tombstones(&self) -> &BTreeSet<K> {
82        &self.tombstones
83    }
84}
85
86impl<T: Ord + Clone, K: Ord + Clone> OrSet<T, K> {
87    /// Return the live add tokens for `element`, excluding tombstoned tokens.
88    pub fn observed_tokens(&self, element: &T) -> BTreeSet<K> {
89        self.adds
90            .iter()
91            .filter_map(|(candidate, token)| {
92                if candidate == element && !self.tombstones.contains(token) {
93                    Some(token.clone())
94                } else {
95                    None
96                }
97            })
98            .collect()
99    }
100
101    pub fn contains(&self, element: &T) -> bool {
102        self.adds
103            .iter()
104            .any(|(candidate, token)| candidate == element && !self.tombstones.contains(token))
105    }
106
107    pub fn elements(&self) -> BTreeSet<T> {
108        self.adds
109            .iter()
110            .filter_map(|(element, token)| {
111                if self.tombstones.contains(token) {
112                    None
113                } else {
114                    Some(element.clone())
115                }
116            })
117            .collect()
118    }
119}
120
121impl<T: Ord, K: Ord> Default for OrSet<T, K> {
122    fn default() -> Self {
123        Self::new()
124    }
125}
126
127impl<T: Ord + Clone, K: Ord + Clone> Mergeable for OrSet<T, K> {
128    fn merge(&mut self, other: &Self) -> Result<(), MergeError> {
129        OrSet::merge(self, other);
130        Ok(())
131    }
132}
133
134impl<T: Ord + Clone, K: Ord + Clone> Crdt for OrSet<T, K> {
135    type Delta = OrSetDelta<T, K>;
136
137    /// Refuses sequence-0 adds and removes with [`WireError::ZeroSequenceAdd`]
138    /// and [`WireError::ZeroSequenceRemove`]. `RecordKernel.permitted` requires
139    /// a positive sequence for every payload. Adds also mint a token at their
140    /// record ID; `allocateToken` requires a positive sequence. These checks
141    /// need no writer count, so the raw carrier applies them too. The author
142    /// bound and add token value stay with the checked writers.
143    ///
144    /// Every positive-sequence record is accepted. An add joins the add set, even
145    /// when its token is already tombstoned (that is the observed-remove rule,
146    /// and a fresh set applies it). A remove tombstones every token it names
147    /// whether or not this replica observed them (`RecordKernel.payloadOwned
148    /// .remove`), so a fresh set applies it too. A remove naming no tokens is the
149    /// lattice bottom: the local writer emits it for an absent element, and
150    /// joining `⊥` on any carrier is the correct application of that record,
151    /// not a dropped one.
152    fn validate_record(&self, id: RecordId, delta: &Self::Delta) -> Result<(), WireError> {
153        match delta {
154            OrSetDelta::Add { .. } if id.sequence == 0 => Err(WireError::ZeroSequenceAdd {
155                replica: id.replica,
156            }),
157            OrSetDelta::Remove { .. } if id.sequence == 0 => Err(WireError::ZeroSequenceRemove {
158                replica: id.replica,
159            }),
160            _ => Ok(()),
161        }
162    }
163
164    fn apply_delta(&mut self, delta: Self::Delta) {
165        match delta {
166            OrSetDelta::Add { element, token } => self.add(element, token),
167            OrSetDelta::Remove { tokens } => self.apply_remove(tokens),
168        }
169    }
170}
safemesh_crdt/crdt/orset.rs

safemesh_crdt/crdt/
orset.rs
