safemesh_crdt/crdt/
rga.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 RGA-family ordered sequence.
9#[derive(Clone, Debug, PartialEq, Eq)]
10pub enum RgaDelta<P, V> {
11    Insert { position: P, value: V },
12    Delete { position: P },
13}
14
15/// RGA-family sequence state: positioned values plus tombstoned positions.
16///
17/// Lean model: `Crdt.RGA.State ι α = Finset (ι × α) × Finset ι`. The Lean
18/// read is the sorted list of live position identifiers; values are carried
19/// by lookup through the live positioned set.
20#[derive(Clone, Debug, PartialEq, Eq)]
21pub struct Rga<P: Ord, V: Ord> {
22    pub(super) placed: BTreeSet<(P, V)>,
23    pub(super) tombstones: BTreeSet<P>,
24}
25
26impl<P: Ord, V: Ord> Rga<P, V> {
27    pub fn new() -> Self {
28        Rga {
29            placed: BTreeSet::new(),
30            tombstones: BTreeSet::new(),
31        }
32    }
33
34    pub fn insert(&mut self, position: P, value: V) {
35        self.placed.insert((position, value));
36    }
37
38    pub fn delete(&mut self, position: P) {
39        self.tombstones.insert(position);
40    }
41
42    pub fn merge(&mut self, other: &Self)
43    where
44        P: Clone,
45        V: Clone,
46    {
47        self.placed.extend(other.placed.iter().cloned());
48        self.tombstones.extend(other.tombstones.iter().cloned());
49    }
50
51    pub fn placed(&self) -> &BTreeSet<(P, V)> {
52        &self.placed
53    }
54
55    pub fn tombstones(&self) -> &BTreeSet<P> {
56        &self.tombstones
57    }
58}
59
60impl<P: Ord + Clone, V: Ord + Clone> Rga<P, V> {
61    /// Every live positioned value, in total `(position, value)` order.
62    pub fn live_entries(&self) -> Vec<(P, V)> {
63        self.placed
64            .iter()
65            .filter_map(|(position, value)| {
66                if self.tombstones.contains(position) {
67                    None
68                } else {
69                    Some((position.clone(), value.clone()))
70                }
71            })
72            .collect()
73    }
74
75    /// Live positions sorted ascending, with no duplicates, matching Lean `read`.
76    /// Distinct values at a shared position remain available through `live_entries`.
77    pub fn read_positions(&self) -> Vec<P> {
78        let mut positions: Vec<P> = self
79            .live_entries()
80            .into_iter()
81            .map(|(position, _)| position)
82            .collect();
83        // live_entries is lexicographically sorted, so equal positions are adjacent.
84        positions.dedup();
85        positions
86    }
87}
88
89impl<P: Ord, V: Ord> Default for Rga<P, V> {
90    fn default() -> Self {
91        Self::new()
92    }
93}
94
95impl<P: Ord + Clone, V: Ord + Clone> Mergeable for Rga<P, V> {
96    fn merge(&mut self, other: &Self) -> Result<(), MergeError> {
97        Rga::merge(self, other);
98        Ok(())
99    }
100}
101
102impl<P: Ord + Clone, V: Ord + Clone> Crdt for Rga<P, V> {
103    type Delta = RgaDelta<P, V>;
104
105    /// Accepts every positive-sequence record: an insert joins the placed set even at a
106    /// tombstoned position, and a delete tombstones its position whether or not
107    /// it was observed; a fresh sequence applies either. The records replay
108    /// ignores (a duplicate insert, a repeated delete) are join idempotence.
109    fn validate_record(&self, id: RecordId, _delta: &Self::Delta) -> Result<(), WireError> {
110        id.require_positive_sequence("RGA")
111    }
112
113    fn apply_delta(&mut self, delta: Self::Delta) {
114        match delta {
115            RgaDelta::Insert { position, value } => self.insert(position, value),
116            RgaDelta::Delete { position } => self.delete(position),
117        }
118    }
119}
safemesh_crdt/crdt/rga.rs

safemesh_crdt/crdt/
rga.rs
