indigo_kernel/
merge.rs

1use crate::IndigoBuildHasher;
2use std::{
3    collections::{BTreeSet, HashSet},
4    hash::{BuildHasher, Hash},
5    ops::Deref,
6    sync::atomic::{AtomicUsize, Ordering},
7};
8
9/// State-based merge
10///
11/// # Laws
12///
13/// - Associativity: _x_ ∨ (_y_ ∨ _z_) = (_x_ ∨ _y_) ∨ _z_
14/// - Commutativity: _x_ ∨ _y_ = _y_ ∨ _x_
15/// - Idempotence: _x_ ∨ _x_ = _x_
16/// - Identity: If `Self: Default`, then _x_ ∨ `Self::default()` = _x_
17///
18/// where _x_ ∨ _y_ means `x.join(y)`
19///
20/// # References
21///
22/// - Join-semilattices on Wikipedia: <https://en.wikipedia.org/wiki/Semilattice>
23/// - Rob Rix's `semilattices` Haskell package: <https://hackage.haskell.org/package/semilattices>
24/// - State-based CRDTs: <https://en.wikipedia.org/wiki/Conflict-free_replicated_data_type#State-based_CRDTs>
25pub trait Join {
26    fn join(&mut self, other: Self);
27}
28
29/// Operation-based merge
30///
31/// # Laws
32///
33/// - Atomicity: If `x.apply(d)` fails, `x` is unchanged
34/// - Coherence with `Join`: If `Self: Join` and `Delta = Self`, then `x.apply(y)` succeeds and is
35///   the same as `x.join(y)`
36///
37/// # References
38///
39/// - Operation-based CRDTs: <https://en.wikipedia.org/wiki/Conflict-free_replicated_data_type#Operation-based_CRDTs>
40pub trait Apply<Delta> {
41    type Error;
42
43    fn apply(&mut self, delta: Delta) -> Result<(), Self::Error>;
44}
45
46static CLOCK: AtomicUsize = AtomicUsize::new(0);
47
48#[derive(Clone)]
49pub struct LastWriteWins<T: ?Sized> {
50    tick: usize,
51    value: T,
52}
53
54impl<T> LastWriteWins<T> {
55    #[must_use]
56    pub fn new(value: T) -> Self {
57        Self {
58            tick: CLOCK.fetch_add(1, Ordering::Relaxed) + 1,
59            value,
60        }
61    }
62
63    #[must_use]
64    pub fn read(&self) -> &T {
65        &self.value
66    }
67
68    pub fn write(&mut self, value: T) {
69        self.tick = CLOCK.fetch_add(1, Ordering::Relaxed) + 1;
70        self.value = value;
71    }
72}
73
74impl<T> Deref for LastWriteWins<T> {
75    type Target = T;
76    fn deref(&self) -> &Self::Target {
77        self.read()
78    }
79}
80
81impl<T> Join for LastWriteWins<T> {
82    fn join(&mut self, other: Self) {
83        if self.tick < other.tick {
84            self.tick = other.tick;
85            self.value = other.value;
86        }
87    }
88}
89
90/* TODO: Rethink implementation so it satisfies `Join`'s invariant.
91pub struct GrowOnlyCounter<T>(Saturating<T>);
92
93impl<T> Deref for GrowOnlyCounter<T> {
94    type Target = T;
95    fn deref(&self) -> &Self::Target {
96        &self.0.0
97    }
98}
99
100impl<T, Rhs> AddAssign<Rhs> for GrowOnlyCounter<T>
101where
102    Saturating<T>: AddAssign<Rhs>,
103{
104    fn add_assign(&mut self, rhs: Rhs) {
105        self.0 += rhs;
106    }
107}
108*/
109
110pub struct GrowOnlyHashSet<T, S = IndigoBuildHasher> {
111    set: HashSet<T, S>,
112}
113
114impl<T, S> GrowOnlyHashSet<T, S> {
115    #[must_use]
116    pub fn new() -> Self
117    where
118        S: Default,
119    {
120        Self::default()
121    }
122
123    pub fn insert(&mut self, value: T) -> bool
124    where
125        T: Eq + Hash,
126        S: BuildHasher,
127    {
128        self.set.insert(value)
129    }
130}
131
132impl<T, S> Default for GrowOnlyHashSet<T, S>
133where
134    S: Default,
135{
136    fn default() -> Self {
137        Self {
138            set: HashSet::default(),
139        }
140    }
141}
142
143impl<T, S> Deref for GrowOnlyHashSet<T, S> {
144    type Target = HashSet<T, S>;
145    fn deref(&self) -> &Self::Target {
146        &self.set
147    }
148}
149
150impl<T, S> Join for GrowOnlyHashSet<T, S>
151where
152    T: Eq + Hash,
153    S: BuildHasher,
154{
155    fn join(&mut self, mut other: Self) {
156        self.set.extend(other.set.drain());
157    }
158}
159
160pub struct GrowOnlyBTreeSet<T> {
161    set: BTreeSet<T>,
162}
163
164impl<T> GrowOnlyBTreeSet<T> {
165    #[must_use]
166    pub fn new() -> Self {
167        Self::default()
168    }
169
170    pub fn insert(&mut self, value: T) -> bool
171    where
172        T: Ord,
173    {
174        self.set.insert(value)
175    }
176
177    pub fn append(&mut self, other: &mut Self)
178    where
179        T: Ord,
180    {
181        self.set.append(&mut other.set);
182    }
183}
184
185impl<T> Default for GrowOnlyBTreeSet<T> {
186    fn default() -> Self {
187        Self {
188            set: BTreeSet::default(),
189        }
190    }
191}
192
193impl<T> Deref for GrowOnlyBTreeSet<T> {
194    type Target = BTreeSet<T>;
195    fn deref(&self) -> &Self::Target {
196        &self.set
197    }
198}
199
200impl<T> Join for GrowOnlyBTreeSet<T>
201where
202    T: Ord,
203{
204    fn join(&mut self, mut other: Self) {
205        self.set.append(&mut other.set);
206    }
207}
208
209#[cfg(test)]
210mod tests {
211    use super::*;
212    use hegel::{TestCase, generators as gs};
213    use std::{cmp::max, iter::zip};
214
215    #[hegel::test(test_cases = 1_000)]
216    fn test_last_write_wins(tc: TestCase) {
217        #[derive(Default)]
218        struct StateMachine {
219            registers: Vec<LastWriteWins<i64>>,
220            model: Vec<(usize, i64)>,
221            next_sequence_number: usize,
222        }
223        #[expect(clippy::needless_pass_by_value)]
224        #[hegel::state_machine]
225        impl StateMachine {
226            fn draw_index(&self, tc: &TestCase) -> usize {
227                tc.assume(!self.registers.is_empty());
228                tc.draw(gs::integers::<usize>().max_value(self.registers.len() - 1))
229            }
230            fn next_write(&mut self, value: i64) -> (usize, i64) {
231                let sequence_number = self.next_sequence_number;
232                self.next_sequence_number += 1;
233                (sequence_number, value)
234            }
235            #[rule]
236            fn new_register(&mut self, tc: TestCase) {
237                let value = tc.draw(gs::integers::<i64>());
238                self.registers.push(LastWriteWins::new(value));
239                let write = self.next_write(value);
240                self.model.push(write);
241            }
242            #[rule]
243            fn write(&mut self, tc: TestCase) {
244                let index = self.draw_index(&tc);
245                let value = tc.draw(gs::integers::<i64>());
246                self.registers[index].write(value);
247                self.model[index] = self.next_write(value);
248            }
249            #[rule]
250            fn join(&mut self, tc: TestCase) {
251                let left = self.draw_index(&tc);
252                let right = self.draw_index(&tc);
253                let right_register = self.registers[right].clone();
254                self.registers[left].join(right_register);
255                self.model[left] = max(self.model[left], self.model[right]);
256            }
257            // TODO: Use `#[invariant(always_run)]`
258            // https://github.com/hegeldev/hegel-rust/releases/tag/v0.37.0
259            #[invariant]
260            fn reads_agree_with_model(&self, _: TestCase) {
261                for (register, (_, value)) in zip(&self.registers, &self.model) {
262                    assert_eq!(register.read(), value);
263                }
264            }
265        }
266        let mut state_machine = StateMachine::default();
267        let value = tc.draw(gs::integers::<i64>());
268        state_machine.registers.push(LastWriteWins::new(value));
269        let write = state_machine.next_write(value);
270        state_machine.model.push(write);
271        hegel::stateful::machine(state_machine).run(tc);
272    }
273}
indigo_kernel/merge.rs

indigo_kernel/
merge.rs
