feltdb/
distributed_indexing.rs

1/**
2 * Distributed Indexing - Multi-node index replication and synchronization
3 *
4 * Extends the local indexing layer with distributed capabilities:
5 * - Index replication across peers
6 * - Consistency protocols based on vector clocks
7 * - Remote index state tracking
8 * - Query routing and distribution
9 * - Index merging strategies
10 */
11use serde::{Deserialize, Serialize};
12use std::collections::{BTreeMap, HashMap, HashSet};
13
14#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
15pub struct PeerId(pub String);
16
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub struct IndexVectorClock {
19    pub clock: BTreeMap<String, u64>,
20}
21
22impl IndexVectorClock {
23    pub fn new() -> Self {
24        IndexVectorClock {
25            clock: BTreeMap::new(),
26        }
27    }
28
29    pub fn increment(&mut self, node_id: &str) {
30        *self.clock.entry(node_id.to_string()).or_insert(0) += 1;
31    }
32
33    pub fn merge(&mut self, other: &IndexVectorClock) {
34        for (node, timestamp) in &other.clock {
35            let current = self.clock.entry(node.clone()).or_insert(0);
36            *current = (*current).max(*timestamp);
37        }
38    }
39
40    pub fn happens_before(&self, other: &IndexVectorClock) -> bool {
41        let mut less = false;
42        for (node, timestamp) in &self.clock {
43            let other_ts = other.clock.get(node).copied().unwrap_or(0);
44            if timestamp > &other_ts {
45                return false;
46            }
47            if timestamp < &other_ts {
48                less = true;
49            }
50        }
51        less
52    }
53
54    pub fn concurrent_with(&self, other: &IndexVectorClock) -> bool {
55        !self.happens_before(other) && !other.happens_before(self)
56    }
57}
58
59#[derive(Debug, Clone, Serialize, Deserialize)]
60pub struct IndexReplicationState {
61    pub index_name: String,
62    pub local_version: IndexVectorClock,
63    pub peer_versions: HashMap<String, IndexVectorClock>,
64    pub replicated_peers: HashSet<String>,
65    pub pending_sync: Vec<IndexOperation>,
66}
67
68#[derive(Debug, Clone, Serialize, Deserialize)]
69pub enum IndexOperation {
70    CreateIndex {
71        name: String,
72        field: String,
73        index_type: String,
74    },
75    UpdateIndex {
76        name: String,
77        record_id: String,
78        value: String,
79    },
80    DeleteIndex {
81        name: String,
82    },
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct IndexSyncMessage {
87    pub source_peer: String,
88    pub target_peer: String,
89    pub index_name: String,
90    pub version: IndexVectorClock,
91    pub operations: Vec<IndexOperation>,
92    pub timestamp: u64,
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize)]
96pub struct RemoteIndexState {
97    pub peer_id: String,
98    pub indexes: HashMap<String, RemoteIndexMetadata>,
99    pub last_sync: u64,
100    pub reachable: bool,
101}
102
103#[derive(Debug, Clone, Serialize, Deserialize)]
104pub struct RemoteIndexMetadata {
105    pub name: String,
106    pub version: IndexVectorClock,
107    pub entry_count: usize,
108    pub avg_query_time_ms: f64,
109    pub last_updated: u64,
110}
111
112/// Distributed index manager - coordinates indexes across peers
113pub struct DistributedIndexManager {
114    local_peer_id: String,
115    replication_states: HashMap<String, IndexReplicationState>,
116    peer_states: HashMap<String, RemoteIndexState>,
117    pending_operations: Vec<IndexSyncMessage>,
118    consistency_protocol: ConsistencyProtocol,
119}
120
121#[derive(Debug, Clone, Copy, PartialEq)]
122pub enum ConsistencyProtocol {
123    EventualConsistency,
124    StrongConsistency,
125    CausalConsistency,
126}
127
128impl DistributedIndexManager {
129    pub fn new(local_peer_id: String) -> Self {
130        DistributedIndexManager {
131            local_peer_id,
132            replication_states: HashMap::new(),
133            peer_states: HashMap::new(),
134            pending_operations: Vec::new(),
135            consistency_protocol: ConsistencyProtocol::CausalConsistency,
136        }
137    }
138
139    /// Create a local index and prepare for replication
140    pub fn create_replicated_index(
141        &mut self,
142        name: String,
143        field: String,
144        index_type: String,
145    ) -> Result<(), String> {
146        let mut version = IndexVectorClock::new();
147        version.increment(&self.local_peer_id);
148
149        let replication_state = IndexReplicationState {
150            index_name: name.clone(),
151            local_version: version,
152            peer_versions: HashMap::new(),
153            replicated_peers: HashSet::new(),
154            pending_sync: vec![IndexOperation::CreateIndex {
155                name: name.clone(),
156                field,
157                index_type,
158            }],
159        };
160
161        self.replication_states.insert(name, replication_state);
162        Ok(())
163    }
164
165    /// Track a remote peer's index state
166    pub fn register_peer(&mut self, peer_id: String) {
167        self.peer_states.insert(
168            peer_id.clone(),
169            RemoteIndexState {
170                peer_id,
171                indexes: HashMap::new(),
172                last_sync: current_time_ms(),
173                reachable: true,
174            },
175        );
176    }
177
178    /// Update remote peer's index metadata after sync
179    pub fn update_peer_index_metadata(
180        &mut self,
181        peer_id: &str,
182        index_name: String,
183        metadata: RemoteIndexMetadata,
184    ) -> Result<(), String> {
185        if let Some(peer_state) = self.peer_states.get_mut(peer_id) {
186            peer_state.indexes.insert(index_name.clone(), metadata);
187            peer_state.last_sync = current_time_ms();
188            Ok(())
189        } else {
190            Err(format!("Peer {} not found", peer_id))
191        }
192    }
193
194    /// Enqueue a sync message to a peer
195    pub fn enqueue_sync(&mut self, message: IndexSyncMessage) {
196        self.pending_operations.push(message);
197    }
198
199    /// Get pending operations for a peer
200    pub fn get_pending_for_peer(&self, peer_id: &str) -> Vec<IndexSyncMessage> {
201        self.pending_operations
202            .iter()
203            .filter(|msg| msg.target_peer == peer_id)
204            .cloned()
205            .collect()
206    }
207
208    /// Acknowledge sync from peer
209    pub fn acknowledge_sync(&mut self, peer_id: &str, index_name: &str) -> Result<(), String> {
210        if let Some(replication_state) = self.replication_states.get_mut(index_name) {
211            replication_state
212                .replicated_peers
213                .insert(peer_id.to_string());
214            Ok(())
215        } else {
216            Err(format!("Index {} not found", index_name))
217        }
218    }
219
220    /// Check replication status
221    pub fn get_replication_status(&self, index_name: &str) -> Option<ReplicationStatus> {
222        self.replication_states.get(index_name).map(|state| {
223            let replicated_count = state.replicated_peers.len();
224            let total_peers = self.peer_states.len();
225            let reachable_peers = self.peer_states.values().filter(|p| p.reachable).count();
226
227            ReplicationStatus {
228                index_name: index_name.to_string(),
229                replicated_to: replicated_count,
230                total_peers,
231                reachable_peers,
232                pending_syncs: state.pending_sync.len(),
233                consistency_level: self.calculate_consistency_level(index_name),
234            }
235        })
236    }
237
238    /// Calculate current consistency level for an index
239    fn calculate_consistency_level(&self, index_name: &str) -> ConsistencyLevel {
240        if let Some(replication_state) = self.replication_states.get(index_name) {
241            let total = self.peer_states.len();
242            let replicated = replication_state.replicated_peers.len();
243
244            if replicated == total {
245                ConsistencyLevel::Strong
246            } else if replicated > total / 2 {
247                ConsistencyLevel::Causal
248            } else {
249                ConsistencyLevel::Eventual
250            }
251        } else {
252            ConsistencyLevel::Local
253        }
254    }
255
256    /// Get list of peers where index needs replication
257    pub fn get_replication_targets(&self, index_name: &str) -> Vec<String> {
258        if let Some(replication_state) = self.replication_states.get(index_name) {
259            self.peer_states
260                .keys()
261                .filter(|peer_id| {
262                    peer_id != &&self.local_peer_id
263                        && !replication_state.replicated_peers.contains(*peer_id)
264                })
265                .cloned()
266                .collect()
267        } else {
268            Vec::new()
269        }
270    }
271
272    /// Detect conflicts between peer versions
273    pub fn detect_conflicts(&self, index_name: &str) -> Vec<VersionConflict> {
274        let mut conflicts = Vec::new();
275
276        if let Some(replication_state) = self.replication_states.get(index_name) {
277            for (peer_id, peer_version) in &replication_state.peer_versions {
278                if replication_state
279                    .local_version
280                    .concurrent_with(peer_version)
281                {
282                    conflicts.push(VersionConflict {
283                        index_name: index_name.to_string(),
284                        local_version: replication_state.local_version.clone(),
285                        remote_peer: peer_id.clone(),
286                        remote_version: peer_version.clone(),
287                    });
288                }
289            }
290        }
291
292        conflicts
293    }
294
295    /// Merge concurrent versions using CRDT strategy
296    pub fn merge_versions(&mut self, index_name: &str) -> Result<(), String> {
297        if let Some(replication_state) = self.replication_states.get_mut(index_name) {
298            for peer_version in replication_state.peer_versions.values() {
299                replication_state.local_version.merge(peer_version);
300            }
301            Ok(())
302        } else {
303            Err(format!("Index {} not found", index_name))
304        }
305    }
306
307    /// Set peer reachability
308    pub fn set_peer_reachable(&mut self, peer_id: &str, reachable: bool) {
309        if let Some(peer_state) = self.peer_states.get_mut(peer_id) {
310            peer_state.reachable = reachable;
311        }
312    }
313
314    /// Get synchronization status across all indexes
315    pub fn get_sync_status(&self) -> DistributedSyncStatus {
316        let total_indexes = self.replication_states.len();
317        let fully_replicated = self
318            .replication_states
319            .values()
320            .filter(|state| state.replicated_peers.len() == self.peer_states.len())
321            .count();
322
323        let pending_total: usize = self
324            .replication_states
325            .values()
326            .map(|state| state.pending_sync.len())
327            .sum();
328
329        DistributedSyncStatus {
330            total_indexes,
331            fully_replicated,
332            partially_replicated: total_indexes - fully_replicated,
333            pending_operations: pending_total,
334            unreachable_peers: self.peer_states.values().filter(|p| !p.reachable).count(),
335        }
336    }
337}
338
339#[derive(Debug, Clone, Serialize, Deserialize)]
340pub enum ConsistencyLevel {
341    Local,
342    Eventual,
343    Causal,
344    Strong,
345}
346
347#[derive(Debug, Clone, Serialize, Deserialize)]
348pub struct ReplicationStatus {
349    pub index_name: String,
350    pub replicated_to: usize,
351    pub total_peers: usize,
352    pub reachable_peers: usize,
353    pub pending_syncs: usize,
354    pub consistency_level: ConsistencyLevel,
355}
356
357#[derive(Debug, Clone, Serialize, Deserialize)]
358pub struct VersionConflict {
359    pub index_name: String,
360    pub local_version: IndexVectorClock,
361    pub remote_peer: String,
362    pub remote_version: IndexVectorClock,
363}
364
365#[derive(Debug, Clone, Serialize, Deserialize)]
366pub struct DistributedSyncStatus {
367    pub total_indexes: usize,
368    pub fully_replicated: usize,
369    pub partially_replicated: usize,
370    pub pending_operations: usize,
371    pub unreachable_peers: usize,
372}
373
374fn current_time_ms() -> u64 {
375    std::time::SystemTime::now()
376        .duration_since(std::time::UNIX_EPOCH)
377        .map(|d| d.as_millis() as u64)
378        .unwrap_or(0)
379}
380
381#[cfg(test)]
382mod tests {
383    use super::*;
384
385    #[test]
386    fn test_vector_clock_increment() {
387        let mut clock = IndexVectorClock::new();
388        clock.increment("node1");
389        assert_eq!(clock.clock.get("node1"), Some(&1));
390    }
391
392    #[test]
393    fn test_vector_clock_merge() {
394        let mut clock1 = IndexVectorClock::new();
395        clock1.increment("node1");
396        clock1.increment("node1");
397
398        let mut clock2 = IndexVectorClock::new();
399        clock2.increment("node2");
400        clock2.increment("node2");
401
402        clock1.merge(&clock2);
403        assert_eq!(clock1.clock.get("node1"), Some(&2));
404        assert_eq!(clock1.clock.get("node2"), Some(&2));
405    }
406
407    #[test]
408    fn test_vector_clock_happens_before() {
409        let mut clock1 = IndexVectorClock::new();
410        clock1.increment("node1");
411
412        let mut clock2 = IndexVectorClock::new();
413        clock2.increment("node1");
414        clock2.increment("node1");
415
416        assert!(clock1.happens_before(&clock2));
417        assert!(!clock2.happens_before(&clock1));
418    }
419
420    #[test]
421    fn test_concurrent_detection() {
422        let mut clock1 = IndexVectorClock::new();
423        clock1.increment("node1");
424
425        let mut clock2 = IndexVectorClock::new();
426        clock2.increment("node2");
427
428        assert!(clock1.concurrent_with(&clock2));
429    }
430
431    #[test]
432    fn test_distributed_manager_creation() {
433        let mut manager = DistributedIndexManager::new("node1".to_string());
434        manager.register_peer("node2".to_string());
435
436        assert!(manager
437            .create_replicated_index(
438                "status_idx".to_string(),
439                "status".to_string(),
440                "hash".to_string()
441            )
442            .is_ok());
443
444        let status = manager.get_replication_status("status_idx");
445        assert!(status.is_some());
446    }
447
448    #[test]
449    fn test_replication_targets() {
450        let mut manager = DistributedIndexManager::new("node1".to_string());
451        manager.register_peer("node2".to_string());
452        manager.register_peer("node3".to_string());
453
454        manager
455            .create_replicated_index("idx1".to_string(), "field".to_string(), "hash".to_string())
456            .ok();
457
458        let targets = manager.get_replication_targets("idx1");
459        assert_eq!(targets.len(), 2);
460    }
461
462    #[test]
463    fn test_sync_status() {
464        let mut manager = DistributedIndexManager::new("node1".to_string());
465        manager.register_peer("node2".to_string());
466
467        manager
468            .create_replicated_index("idx1".to_string(), "field".to_string(), "hash".to_string())
469            .ok();
470        manager
471            .create_replicated_index("idx2".to_string(), "field".to_string(), "hash".to_string())
472            .ok();
473
474        let status = manager.get_sync_status();
475        assert_eq!(status.total_indexes, 2);
476    }
477}
feltdb/distributed_indexing.rs

feltdb/
distributed_indexing.rs
