ruvector_replication/
conflict.rs

1//! Conflict resolution strategies for distributed replication
2//!
3//! Provides vector clocks for causality tracking and various
4//! conflict resolution strategies including Last-Write-Wins
5//! and custom merge functions.
6
7use crate::{ReplicationError, Result};
8use serde::{Deserialize, Serialize};
9use std::cmp::Ordering;
10use std::collections::HashMap;
11use std::fmt;
12
13/// Vector clock for tracking causality in distributed systems
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct VectorClock {
16    /// Map of replica ID to logical timestamp
17    clock: HashMap<String, u64>,
18}
19
20impl VectorClock {
21    /// Create a new vector clock
22    pub fn new() -> Self {
23        Self {
24            clock: HashMap::new(),
25        }
26    }
27
28    /// Increment the clock for a replica
29    pub fn increment(&mut self, replica_id: &str) {
30        let counter = self.clock.entry(replica_id.to_string()).or_insert(0);
31        *counter += 1;
32    }
33
34    /// Get the timestamp for a replica
35    pub fn get(&self, replica_id: &str) -> u64 {
36        self.clock.get(replica_id).copied().unwrap_or(0)
37    }
38
39    /// Update with another vector clock (taking max of each component)
40    pub fn merge(&mut self, other: &VectorClock) {
41        for (replica_id, &timestamp) in &other.clock {
42            let current = self.clock.entry(replica_id.clone()).or_insert(0);
43            *current = (*current).max(timestamp);
44        }
45    }
46
47    /// Check if this clock happens-before another clock
48    pub fn happens_before(&self, other: &VectorClock) -> bool {
49        let mut less = false;
50        let mut equal = true;
51
52        // Check all replicas in self
53        for (replica_id, &self_ts) in &self.clock {
54            let other_ts = other.get(replica_id);
55            if self_ts > other_ts {
56                return false;
57            }
58            if self_ts < other_ts {
59                less = true;
60                equal = false;
61            }
62        }
63
64        // Check replicas only in other
65        for (replica_id, &other_ts) in &other.clock {
66            if !self.clock.contains_key(replica_id) && other_ts > 0 {
67                less = true;
68                equal = false;
69            }
70        }
71
72        less || equal
73    }
74
75    /// Compare vector clocks for causality
76    pub fn compare(&self, other: &VectorClock) -> ClockOrdering {
77        if self == other {
78            return ClockOrdering::Equal;
79        }
80
81        if self.happens_before(other) {
82            return ClockOrdering::Before;
83        }
84
85        if other.happens_before(self) {
86            return ClockOrdering::After;
87        }
88
89        ClockOrdering::Concurrent
90    }
91
92    /// Check if two clocks are concurrent (conflicting)
93    pub fn is_concurrent(&self, other: &VectorClock) -> bool {
94        matches!(self.compare(other), ClockOrdering::Concurrent)
95    }
96}
97
98impl Default for VectorClock {
99    fn default() -> Self {
100        Self::new()
101    }
102}
103
104impl fmt::Display for VectorClock {
105    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
106        write!(f, "{{")?;
107        for (i, (replica, ts)) in self.clock.iter().enumerate() {
108            if i > 0 {
109                write!(f, ", ")?;
110            }
111            write!(f, "{}: {}", replica, ts)?;
112        }
113        write!(f, "}}")
114    }
115}
116
117/// Ordering relationship between vector clocks
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119pub enum ClockOrdering {
120    /// Clocks are equal
121    Equal,
122    /// First clock happens before second
123    Before,
124    /// First clock happens after second
125    After,
126    /// Clocks are concurrent (conflicting)
127    Concurrent,
128}
129
130/// A versioned value with vector clock
131#[derive(Debug, Clone, Serialize, Deserialize)]
132pub struct Versioned<T> {
133    /// The value
134    pub value: T,
135    /// Vector clock for this version
136    pub clock: VectorClock,
137    /// Replica that created this version
138    pub replica_id: String,
139}
140
141impl<T> Versioned<T> {
142    /// Create a new versioned value
143    pub fn new(value: T, replica_id: String) -> Self {
144        let mut clock = VectorClock::new();
145        clock.increment(&replica_id);
146        Self {
147            value,
148            clock,
149            replica_id,
150        }
151    }
152
153    /// Update the version with a new value
154    pub fn update(&mut self, value: T) {
155        self.value = value;
156        self.clock.increment(&self.replica_id);
157    }
158
159    /// Compare versions for causality
160    pub fn compare(&self, other: &Versioned<T>) -> ClockOrdering {
161        self.clock.compare(&other.clock)
162    }
163}
164
165/// Trait for conflict resolution strategies
166pub trait ConflictResolver<T: Clone>: Send + Sync {
167    /// Resolve a conflict between two versions
168    fn resolve(&self, v1: &Versioned<T>, v2: &Versioned<T>) -> Result<Versioned<T>>;
169
170    /// Resolve multiple conflicting versions
171    fn resolve_many(&self, versions: Vec<Versioned<T>>) -> Result<Versioned<T>> {
172        if versions.is_empty() {
173            return Err(ReplicationError::ConflictResolution(
174                "No versions to resolve".to_string(),
175            ));
176        }
177
178        if versions.len() == 1 {
179            return Ok(versions.into_iter().next().unwrap());
180        }
181
182        let mut result = versions[0].clone();
183        for version in versions.iter().skip(1) {
184            result = self.resolve(&result, version)?;
185        }
186        Ok(result)
187    }
188}
189
190/// Last-Write-Wins conflict resolution strategy
191pub struct LastWriteWins;
192
193impl<T: Clone> ConflictResolver<T> for LastWriteWins {
194    fn resolve(&self, v1: &Versioned<T>, v2: &Versioned<T>) -> Result<Versioned<T>> {
195        match v1.compare(v2) {
196            ClockOrdering::Before | ClockOrdering::Concurrent => Ok(v2.clone()),
197            ClockOrdering::After | ClockOrdering::Equal => Ok(v1.clone()),
198        }
199    }
200}
201
202/// Custom merge function for conflict resolution
203pub struct MergeFunction<T, F>
204where
205    F: Fn(&T, &T) -> T + Send + Sync,
206{
207    merge_fn: F,
208    _phantom: std::marker::PhantomData<T>,
209}
210
211impl<T, F> MergeFunction<T, F>
212where
213    F: Fn(&T, &T) -> T + Send + Sync,
214{
215    /// Create a new merge function resolver
216    pub fn new(merge_fn: F) -> Self {
217        Self {
218            merge_fn,
219            _phantom: std::marker::PhantomData,
220        }
221    }
222}
223
224impl<T: Clone + Send + Sync, F> ConflictResolver<T> for MergeFunction<T, F>
225where
226    F: Fn(&T, &T) -> T + Send + Sync,
227{
228    fn resolve(&self, v1: &Versioned<T>, v2: &Versioned<T>) -> Result<Versioned<T>> {
229        match v1.compare(v2) {
230            ClockOrdering::Equal | ClockOrdering::Before => Ok(v2.clone()),
231            ClockOrdering::After => Ok(v1.clone()),
232            ClockOrdering::Concurrent => {
233                let merged_value = (self.merge_fn)(&v1.value, &v2.value);
234                let mut merged_clock = v1.clock.clone();
235                merged_clock.merge(&v2.clock);
236
237                Ok(Versioned {
238                    value: merged_value,
239                    clock: merged_clock,
240                    replica_id: v1.replica_id.clone(),
241                })
242            }
243        }
244    }
245}
246
247/// CRDT-inspired merge for numeric values (takes max)
248pub struct MaxMerge;
249
250impl ConflictResolver<i64> for MaxMerge {
251    fn resolve(&self, v1: &Versioned<i64>, v2: &Versioned<i64>) -> Result<Versioned<i64>> {
252        match v1.compare(v2) {
253            ClockOrdering::Equal | ClockOrdering::Before => Ok(v2.clone()),
254            ClockOrdering::After => Ok(v1.clone()),
255            ClockOrdering::Concurrent => {
256                let merged_value = v1.value.max(v2.value);
257                let mut merged_clock = v1.clock.clone();
258                merged_clock.merge(&v2.clock);
259
260                Ok(Versioned {
261                    value: merged_value,
262                    clock: merged_clock,
263                    replica_id: v1.replica_id.clone(),
264                })
265            }
266        }
267    }
268}
269
270/// CRDT-inspired merge for sets (takes union)
271pub struct SetUnion;
272
273impl<T: Clone + Eq + std::hash::Hash> ConflictResolver<Vec<T>> for SetUnion {
274    fn resolve(&self, v1: &Versioned<Vec<T>>, v2: &Versioned<Vec<T>>) -> Result<Versioned<Vec<T>>> {
275        match v1.compare(v2) {
276            ClockOrdering::Equal | ClockOrdering::Before => Ok(v2.clone()),
277            ClockOrdering::After => Ok(v1.clone()),
278            ClockOrdering::Concurrent => {
279                let mut merged_value = v1.value.clone();
280                for item in &v2.value {
281                    if !merged_value.contains(item) {
282                        merged_value.push(item.clone());
283                    }
284                }
285
286                let mut merged_clock = v1.clock.clone();
287                merged_clock.merge(&v2.clock);
288
289                Ok(Versioned {
290                    value: merged_value,
291                    clock: merged_clock,
292                    replica_id: v1.replica_id.clone(),
293                })
294            }
295        }
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302
303    #[test]
304    fn test_vector_clock() {
305        let mut clock1 = VectorClock::new();
306        clock1.increment("r1");
307        clock1.increment("r1");
308
309        let mut clock2 = VectorClock::new();
310        clock2.increment("r1");
311
312        assert_eq!(clock1.compare(&clock2), ClockOrdering::After);
313        assert_eq!(clock2.compare(&clock1), ClockOrdering::Before);
314    }
315
316    #[test]
317    fn test_concurrent_clocks() {
318        let mut clock1 = VectorClock::new();
319        clock1.increment("r1");
320
321        let mut clock2 = VectorClock::new();
322        clock2.increment("r2");
323
324        assert_eq!(clock1.compare(&clock2), ClockOrdering::Concurrent);
325        assert!(clock1.is_concurrent(&clock2));
326    }
327
328    #[test]
329    fn test_clock_merge() {
330        let mut clock1 = VectorClock::new();
331        clock1.increment("r1");
332        clock1.increment("r1");
333
334        let mut clock2 = VectorClock::new();
335        clock2.increment("r2");
336        clock2.increment("r2");
337        clock2.increment("r2");
338
339        clock1.merge(&clock2);
340        assert_eq!(clock1.get("r1"), 2);
341        assert_eq!(clock1.get("r2"), 3);
342    }
343
344    #[test]
345    fn test_versioned() {
346        let mut v1 = Versioned::new(100, "r1".to_string());
347        v1.update(200);
348
349        assert_eq!(v1.value, 200);
350        assert_eq!(v1.clock.get("r1"), 2);
351    }
352
353    #[test]
354    fn test_last_write_wins() {
355        let v1 = Versioned::new(100, "r1".to_string());
356        let mut v2 = Versioned::new(200, "r1".to_string());
357        v2.clock.increment("r1");
358
359        let resolver = LastWriteWins;
360        let result = resolver.resolve(&v1, &v2).unwrap();
361        assert_eq!(result.value, 200);
362    }
363
364    #[test]
365    fn test_merge_function() {
366        let v1 = Versioned::new(100, "r1".to_string());
367        let v2 = Versioned::new(200, "r2".to_string());
368
369        let resolver = MergeFunction::new(|a, b| a + b);
370        let result = resolver.resolve(&v1, &v2).unwrap();
371        assert_eq!(result.value, 300);
372    }
373
374    #[test]
375    fn test_max_merge() {
376        let v1 = Versioned::new(100, "r1".to_string());
377        let v2 = Versioned::new(200, "r2".to_string());
378
379        let resolver = MaxMerge;
380        let result = resolver.resolve(&v1, &v2).unwrap();
381        assert_eq!(result.value, 200);
382    }
383
384    #[test]
385    fn test_set_union() {
386        let v1 = Versioned::new(vec![1, 2, 3], "r1".to_string());
387        let v2 = Versioned::new(vec![3, 4, 5], "r2".to_string());
388
389        let resolver = SetUnion;
390        let result = resolver.resolve(&v1, &v2).unwrap();
391        assert_eq!(result.value.len(), 5);
392        assert!(result.value.contains(&1));
393        assert!(result.value.contains(&4));
394    }
395}
ruvector_replication/conflict.rs

ruvector_replication/
conflict.rs
