cbtop/federated_metrics/
crdt.rs

1//! CRDT (Conflict-free Replicated Data Types) for federated metrics.
2
3use std::collections::{HashMap, HashSet};
4
5/// G-Counter CRDT for monotonic counters
6#[derive(Debug, Clone, Default)]
7pub struct GCounter {
8    /// Per-host counts
9    counts: HashMap<String, u64>,
10}
11
12impl GCounter {
13    /// Create a new G-Counter
14    pub fn new() -> Self {
15        Self::default()
16    }
17
18    /// Increment counter for a host
19    pub fn increment(&mut self, host_id: &str, amount: u64) {
20        *self.counts.entry(host_id.to_string()).or_insert(0) += amount;
21    }
22
23    /// Get total count across all hosts
24    pub fn value(&self) -> u64 {
25        self.counts.values().sum()
26    }
27
28    /// Merge with another G-Counter (take max per host)
29    pub fn merge(&mut self, other: &GCounter) {
30        for (host, count) in &other.counts {
31            let entry = self.counts.entry(host.clone()).or_insert(0);
32            *entry = (*entry).max(*count);
33        }
34    }
35
36    /// Get count for a specific host
37    pub fn host_count(&self, host_id: &str) -> u64 {
38        self.counts.get(host_id).copied().unwrap_or(0)
39    }
40}
41
42/// LWW-Register CRDT for last-writer-wins values
43#[derive(Debug, Clone)]
44pub struct LwwRegister<T: Clone> {
45    /// Current value
46    value: T,
47    /// Timestamp of last write
48    timestamp: u64,
49    /// Host that performed last write
50    writer: String,
51}
52
53impl<T: Clone + Default> Default for LwwRegister<T> {
54    fn default() -> Self {
55        Self {
56            value: T::default(),
57            timestamp: 0,
58            writer: String::new(),
59        }
60    }
61}
62
63impl<T: Clone> LwwRegister<T> {
64    /// Create a new register with initial value
65    pub fn new(value: T, timestamp: u64, writer: impl Into<String>) -> Self {
66        Self {
67            value,
68            timestamp,
69            writer: writer.into(),
70        }
71    }
72
73    /// Update value if timestamp is newer
74    pub fn update(&mut self, value: T, timestamp: u64, writer: impl Into<String>) {
75        if timestamp > self.timestamp {
76            self.value = value;
77            self.timestamp = timestamp;
78            self.writer = writer.into();
79        }
80    }
81
82    /// Get current value
83    pub fn value(&self) -> &T {
84        &self.value
85    }
86
87    /// Get timestamp
88    pub fn timestamp(&self) -> u64 {
89        self.timestamp
90    }
91
92    /// Merge with another register (keep newer)
93    pub fn merge(&mut self, other: &LwwRegister<T>) {
94        if other.timestamp > self.timestamp {
95            self.value = other.value.clone();
96            self.timestamp = other.timestamp;
97            self.writer = other.writer.clone();
98        }
99    }
100}
101
102/// OR-Set CRDT for add/remove sets
103#[derive(Debug, Clone)]
104pub struct OrSet<T: Clone + Eq + std::hash::Hash> {
105    /// Elements with their unique tags
106    elements: HashMap<T, HashSet<String>>,
107    /// Tombstones for removed elements
108    tombstones: HashMap<T, HashSet<String>>,
109}
110
111impl<T: Clone + Eq + std::hash::Hash> Default for OrSet<T> {
112    fn default() -> Self {
113        Self {
114            elements: HashMap::new(),
115            tombstones: HashMap::new(),
116        }
117    }
118}
119
120impl<T: Clone + Eq + std::hash::Hash> OrSet<T> {
121    /// Create a new OR-Set
122    pub fn new() -> Self {
123        Self::default()
124    }
125
126    /// Add an element with a unique tag
127    pub fn add(&mut self, element: T, tag: String) {
128        self.elements.entry(element).or_default().insert(tag);
129    }
130
131    /// Remove an element (tombstone all tags)
132    pub fn remove(&mut self, element: &T) {
133        if let Some(tags) = self.elements.get(element) {
134            let tombstone_entry = self.tombstones.entry(element.clone()).or_default();
135            for tag in tags {
136                tombstone_entry.insert(tag.clone());
137            }
138        }
139    }
140
141    /// Check if element is in set
142    pub fn contains(&self, element: &T) -> bool {
143        if let Some(tags) = self.elements.get(element) {
144            let tombstones = self.tombstones.get(element);
145            tags.iter()
146                .any(|tag| tombstones.map_or(true, |ts| !ts.contains(tag)))
147        } else {
148            false
149        }
150    }
151
152    /// Get all active elements
153    pub fn elements(&self) -> Vec<&T> {
154        self.elements.keys().filter(|e| self.contains(e)).collect()
155    }
156
157    /// Merge with another OR-Set
158    pub fn merge(&mut self, other: &OrSet<T>) {
159        // Merge elements
160        for (elem, tags) in &other.elements {
161            let entry = self.elements.entry(elem.clone()).or_default();
162            entry.extend(tags.iter().cloned());
163        }
164        // Merge tombstones
165        for (elem, tags) in &other.tombstones {
166            let entry = self.tombstones.entry(elem.clone()).or_default();
167            entry.extend(tags.iter().cloned());
168        }
169    }
170}
cbtop/federated_metrics/crdt.rs

cbtop/federated_metrics/
crdt.rs
