dbx_core/replication/
vector_clock.rs

1//! 벡터 클록(Vector Clock) — 분산 이벤트 인과관계 추적
2//!
3//! ## 용도
4//! - LWW(Timestamp)는 clock skew 시 데이터 손실 위험이 있음
5//! - 벡터 클록은 각 노드의 논리 시계를 독립적으로 관리하여
6//!   **동시 발생 이벤트(concurrent)**를 감지할 수 있습니다.
7//!
8//! ## 비교 결과
9//! - `HappensBefore(A → B)`: A가 B 이전에 발생
10//! - `HappensAfter(B → A)`: B가 A 이전에 발생
11//! - `Concurrent`: 인과관계 없음 → 충돌 → 애플리케이션 레벨 처리 필요
12
13use std::collections::HashMap;
14
15/// 벡터 클록 비교 결과
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum VectorClockOrder {
18    /// self 가 other 이전 (self → other)
19    HappensBefore,
20    /// self 가 other 이후 (other → self)
21    HappensAfter,
22    /// 인과관계 없음 — 동시 발생 (충돌)
23    Concurrent,
24    /// 완전히 동일
25    Equal,
26}
27
28/// 벡터 클록
29///
30/// `node_id → logical_clock` 의 맵으로 구성됩니다.
31#[derive(Debug, Clone, Default, PartialEq)]
32pub struct VectorClock {
33    clocks: HashMap<u32, u64>,
34}
35
36impl VectorClock {
37    /// 새 빈 벡터 클록 생성
38    pub fn new() -> Self {
39        Self {
40            clocks: HashMap::new(),
41        }
42    }
43
44    /// 특정 노드의 논리 클록 값 조회
45    pub fn get(&self, node_id: u32) -> u64 {
46        *self.clocks.get(&node_id).unwrap_or(&0)
47    }
48
49    /// 특정 노드의 클록을 1 증가 (이벤트 발생 시 호출)
50    pub fn tick(&mut self, node_id: u32) {
51        let entry = self.clocks.entry(node_id).or_insert(0);
52        *entry += 1;
53    }
54
55    /// 메시지 수신 시 병합: `self[node] = max(self[node], other[node])` + tick
56    pub fn merge_and_tick(&mut self, other: &VectorClock, self_node_id: u32) {
57        for (&node_id, &clock) in &other.clocks {
58            let entry = self.clocks.entry(node_id).or_insert(0);
59            *entry = (*entry).max(clock);
60        }
61        self.tick(self_node_id);
62    }
63
64    /// 단순 병합 (tick 없이)
65    pub fn merge(&mut self, other: &VectorClock) {
66        for (&node_id, &clock) in &other.clocks {
67            let entry = self.clocks.entry(node_id).or_insert(0);
68            *entry = (*entry).max(clock);
69        }
70    }
71
72    /// 두 벡터 클록을 비교하여 인과관계 판단
73    pub fn compare(&self, other: &VectorClock) -> VectorClockOrder {
74        let mut self_gt = false; // self가 더 큰 항목 존재?
75        let mut other_gt = false; // other가 더 큰 항목 존재?
76
77        // 양쪽 노드 집합 합집합
78        let all_nodes: std::collections::HashSet<u32> = self
79            .clocks
80            .keys()
81            .chain(other.clocks.keys())
82            .copied()
83            .collect();
84
85        for node in all_nodes {
86            let lhs = self.get(node);
87            let rhs = other.get(node);
88            if lhs > rhs {
89                self_gt = true;
90            }
91            if rhs > lhs {
92                other_gt = true;
93            }
94        }
95
96        match (self_gt, other_gt) {
97            (false, false) => VectorClockOrder::Equal,
98            (true, false) => VectorClockOrder::HappensAfter,
99            (false, true) => VectorClockOrder::HappensBefore,
100            (true, true) => VectorClockOrder::Concurrent,
101        }
102    }
103
104    /// self가 other보다 먼저 발생했는지 여부
105    pub fn happens_before(&self, other: &VectorClock) -> bool {
106        self.compare(other) == VectorClockOrder::HappensBefore
107    }
108
109    /// 두 클록이 동시 발생(충돌)인지 여부
110    pub fn is_concurrent(&self, other: &VectorClock) -> bool {
111        self.compare(other) == VectorClockOrder::Concurrent
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[test]
120    fn test_tick_and_get() {
121        let mut vc = VectorClock::new();
122        vc.tick(1);
123        vc.tick(1);
124        vc.tick(2);
125        assert_eq!(vc.get(1), 2);
126        assert_eq!(vc.get(2), 1);
127        assert_eq!(vc.get(99), 0);
128    }
129
130    #[test]
131    fn test_happens_before() {
132        let mut a = VectorClock::new();
133        a.tick(1);
134        // b는 a를 받은 뒤 b 자신의 이벤트 발생
135        let mut b = VectorClock::new();
136        b.merge_and_tick(&a, 2);
137        // a → b
138        assert_eq!(a.compare(&b), VectorClockOrder::HappensBefore);
139        assert_eq!(b.compare(&a), VectorClockOrder::HappensAfter);
140    }
141
142    #[test]
143    fn test_concurrent() {
144        let mut a = VectorClock::new();
145        a.tick(1);
146        let mut b = VectorClock::new();
147        b.tick(2);
148        // 두 이벤트은 서로 독립적 → 동시 발생
149        assert_eq!(a.compare(&b), VectorClockOrder::Concurrent);
150        assert!(a.is_concurrent(&b));
151    }
152
153    #[test]
154    fn test_merge() {
155        let mut a = VectorClock::new();
156        a.tick(1);
157        a.tick(1); // [1:2]
158        let mut b = VectorClock::new();
159        b.tick(2); // [2:1]
160        b.merge(&a);
161        // b = [1:2, 2:1]
162        assert_eq!(b.get(1), 2);
163        assert_eq!(b.get(2), 1);
164    }
165
166    #[test]
167    fn test_equal_clocks() {
168        let mut a = VectorClock::new();
169        let mut b = VectorClock::new();
170        a.tick(1);
171        b.tick(1);
172        assert_eq!(a.compare(&b), VectorClockOrder::Equal);
173    }
174}
dbx_core/replication/vector_clock.rs

dbx_core/replication/
vector_clock.rs
