oxirs_tdb/distributed/
replication.rs

1//! Database Replication System
2//!
3//! This module implements comprehensive database replication for high availability
4//! and disaster recovery. Supports both master-slave and master-master topologies.
5//!
6//! # Replication Modes
7//!
8//! - **Master-Slave**: One master handles writes, multiple slaves replicate
9//! - **Master-Master**: Multiple masters, all handle writes with conflict resolution
10//! - **Async Replication**: Fire-and-forget, low latency
11//! - **Sync Replication**: Wait for acknowledgment, data consistency
12//! - **Semi-Sync**: Wait for at least one replica acknowledgment
13//!
14//! # Features
15//!
16//! - **Automatic Failover**: Promote slave to master on failure
17//! - **Conflict Resolution**: Last-write-wins, vector clocks, custom strategies
18//! - **Lag Monitoring**: Track replication lag across replicas
19//! - **Split-Brain Prevention**: Quorum-based writes in master-master
20//! - **Incremental Sync**: Only replicate changes, not full snapshots
21//!
22//! # Architecture
23//!
24//! ```text
25//! Master-Slave:
26//! ┌─────────┐     writes      ┌────────────┐
27//! │ Clients │──────────────────▶│   Master   │
28//! └─────────┘                  └──────┬─────┘
29//!                                     │
30//!                       ┌─────────────┼────────────┐
31//!                       │             │            │
32//!                       ▼             ▼            ▼
33//!                  ┌────────┐    ┌────────┐  ┌────────┐
34//!                  │ Slave1 │    │ Slave2 │  │ Slave3 │
35//!                  └────────┘    └────────┘  └────────┘
36//!
37//! Master-Master:
38//! ┌─────────┐    writes     ┌──────────┐
39//! │Clients A│───────────────▶│ Master A │
40//! └─────────┘                └─────┬────┘
41//!                                  │
42//!                              sync│
43//!                                  │
44//! ┌─────────┐    writes     ┌─────▼────┐
45//! │Clients B│───────────────▶│ Master B │
46//! └─────────┘                └──────────┘
47//! ```
48//!
49//! # Example
50//!
51//! ```rust,no_run
52//! use oxirs_tdb::distributed::replication::{ReplicationManager, ReplicationConfig, ReplicationMode};
53//! use oxirs_tdb::consensus::transport::LoopbackSimulationTransport;
54//!
55//! # async fn example() -> anyhow::Result<()> {
56//! // Create replication manager. A real multi-node deployment must supply a
57//! // transport backed by real network I/O; LoopbackSimulationTransport is
58//! // only for single-process tests/local development.
59//! let config = ReplicationConfig {
60//!     mode: ReplicationMode::MasterSlave,
61//!     ..Default::default()
62//! };
63//!
64//! let mut manager =
65//!     ReplicationManager::with_transport("node1".to_string(), config, LoopbackSimulationTransport::arc());
66//!
67//! // Add replicas
68//! manager.add_replica("replica1".to_string(), "http://replica1:8080".to_string()).await?;
69//!
70//! // Replicate changes
71//! let changes = vec![/* ... */];
72//! manager.replicate_changes(changes).await?;
73//! # Ok(())
74//! # }
75//! ```
76
77use crate::consensus::transport::NetworkTransport;
78use crate::error::{Result, TdbError};
79use anyhow::Context;
80use chrono::{DateTime, Duration, Utc};
81use parking_lot::{Mutex, RwLock};
82use serde::{Deserialize, Serialize};
83use std::collections::{HashMap, VecDeque};
84use std::sync::Arc;
85
86/// Replication mode
87#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
88pub enum ReplicationMode {
89    /// Master-slave replication (one master, multiple slaves)
90    MasterSlave,
91    /// Master-master replication (multiple masters with conflict resolution)
92    MasterMaster,
93}
94
95/// Replication synchronization mode
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
97pub enum SyncMode {
98    /// Asynchronous replication (fire-and-forget)
99    Async,
100    /// Synchronous replication (wait for all replicas)
101    Sync,
102    /// Semi-synchronous (wait for at least one replica)
103    SemiSync,
104}
105
106/// Replica role
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108pub enum ReplicaRole {
109    /// Master node (handles writes)
110    Master,
111    /// Slave node (read-only, replicates from master)
112    Slave,
113}
114
115/// Replica node status
116#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
117pub enum ReplicationStatus {
118    /// Replica is healthy and up-to-date
119    Healthy,
120    /// Replica is lagging behind
121    Lagging,
122    /// Replica is unreachable
123    Unreachable,
124    /// Replica is in error state
125    Error,
126}
127
128/// Replica node information
129#[derive(Debug, Clone, Serialize, Deserialize)]
130pub struct ReplicaNode {
131    /// Node ID
132    pub node_id: String,
133    /// Network endpoint
134    pub endpoint: String,
135    /// Replica role
136    pub role: ReplicaRole,
137    /// Current status
138    pub status: ReplicationStatus,
139    /// Last successful replication timestamp
140    pub last_sync: DateTime<Utc>,
141    /// Replication lag (milliseconds)
142    pub lag_ms: i64,
143    /// Last sequence number (LSN) replicated
144    pub last_lsn: u64,
145}
146
147/// Replication change event
148#[derive(Debug, Clone, Serialize, Deserialize)]
149pub struct ReplicationChange {
150    /// Log sequence number
151    pub lsn: u64,
152    /// Transaction ID
153    pub txn_id: String,
154    /// Change timestamp
155    pub timestamp: DateTime<Utc>,
156    /// Change data (serialized)
157    pub data: Vec<u8>,
158    /// Node that originated the change
159    pub source_node: String,
160}
161
162/// Conflict resolution strategy for master-master
163#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
164pub enum ConflictResolution {
165    /// Last write wins (based on timestamp)
166    LastWriteWins,
167    /// First write wins
168    FirstWriteWins,
169    /// Manual resolution required
170    Manual,
171    /// Custom strategy (application-defined)
172    Custom,
173}
174
175/// Replication configuration
176#[derive(Debug, Clone, Serialize, Deserialize)]
177pub struct ReplicationConfig {
178    /// Replication mode
179    pub mode: ReplicationMode,
180    /// Synchronization mode
181    pub sync_mode: SyncMode,
182    /// Conflict resolution strategy
183    pub conflict_resolution: ConflictResolution,
184    /// Maximum replication lag before marking replica as unhealthy (milliseconds)
185    pub max_lag_ms: i64,
186    /// Heartbeat interval
187    pub heartbeat_interval: Duration,
188    /// Automatic failover enabled
189    pub auto_failover: bool,
190    /// Minimum replicas for quorum (master-master)
191    pub quorum_size: usize,
192    /// Change buffer size
193    pub change_buffer_size: usize,
194}
195
196impl Default for ReplicationConfig {
197    fn default() -> Self {
198        Self {
199            mode: ReplicationMode::MasterSlave,
200            sync_mode: SyncMode::Async,
201            conflict_resolution: ConflictResolution::LastWriteWins,
202            max_lag_ms: 1000,
203            heartbeat_interval: Duration::seconds(5),
204            auto_failover: true,
205            quorum_size: 2,
206            change_buffer_size: 10000,
207        }
208    }
209}
210
211/// Replication Manager
212///
213/// Manages database replication across multiple nodes.
214pub struct ReplicationManager {
215    /// Node ID
216    node_id: String,
217    /// Configuration
218    config: ReplicationConfig,
219    /// Local role
220    role: Arc<RwLock<ReplicaRole>>,
221    /// Registered replicas
222    replicas: Arc<RwLock<HashMap<String, ReplicaNode>>>,
223    /// Change buffer for replication
224    change_buffer: Arc<Mutex<VecDeque<ReplicationChange>>>,
225    /// Current LSN
226    current_lsn: Arc<Mutex<u64>>,
227    /// Statistics
228    stats: Arc<Mutex<ReplicationStats>>,
229    /// Real (or explicitly-simulated) network transport used to ship changes
230    /// to replicas. `None` means no transport has been configured, in which
231    /// case `replicate_changes()` fails loudly with
232    /// `TdbError::DistributedTransportNotConfigured` instead of counting
233    /// fabricated acknowledgements as successful replication.
234    transport: Option<Arc<dyn NetworkTransport>>,
235}
236
237/// Replication statistics
238#[derive(Debug, Clone, Default, Serialize, Deserialize)]
239pub struct ReplicationStats {
240    /// Total changes replicated
241    pub total_changes: u64,
242    /// Total bytes replicated
243    pub total_bytes_replicated: u64,
244    /// Successful replications
245    pub successful_replications: u64,
246    /// Failed replications
247    pub failed_replications: u64,
248    /// Conflicts detected (master-master)
249    pub conflicts_detected: u64,
250    /// Conflicts resolved
251    pub conflicts_resolved: u64,
252    /// Failovers performed
253    pub failovers: u64,
254    /// Average replication latency (milliseconds)
255    pub avg_replication_latency_ms: f64,
256    /// Total replication latency (for calculating average)
257    total_latency_ms: f64,
258}
259
260impl ReplicationManager {
261    /// Create a new Replication Manager with no transport configured.
262    ///
263    /// **Note:** `replicate_changes()` on a manager built this way always
264    /// fails with [`TdbError::DistributedTransportNotConfigured`] — there is
265    /// no way to honestly ship changes to replicas without a transport. Use
266    /// [`ReplicationManager::with_transport`] (with a real network-backed
267    /// transport, or
268    /// [`crate::consensus::transport::LoopbackSimulationTransport`] for
269    /// single-node tests) to actually replicate.
270    pub fn new(node_id: String, config: ReplicationConfig) -> Self {
271        let initial_role = match config.mode {
272            ReplicationMode::MasterSlave => ReplicaRole::Master,
273            ReplicationMode::MasterMaster => ReplicaRole::Master,
274        };
275
276        Self {
277            node_id,
278            config,
279            role: Arc::new(RwLock::new(initial_role)),
280            replicas: Arc::new(RwLock::new(HashMap::new())),
281            change_buffer: Arc::new(Mutex::new(VecDeque::new())),
282            current_lsn: Arc::new(Mutex::new(0)),
283            stats: Arc::new(Mutex::new(ReplicationStats::default())),
284            transport: None,
285        }
286    }
287
288    /// Create a new Replication Manager backed by a real (or
289    /// explicitly-simulated) [`NetworkTransport`]. This is the constructor
290    /// production code should use.
291    pub fn with_transport(
292        node_id: String,
293        config: ReplicationConfig,
294        transport: Arc<dyn NetworkTransport>,
295    ) -> Self {
296        let mut manager = Self::new(node_id, config);
297        manager.transport = Some(transport);
298        manager
299    }
300
301    /// Add a replica node
302    pub async fn add_replica(&mut self, node_id: String, endpoint: String) -> Result<()> {
303        let replica = ReplicaNode {
304            node_id: node_id.clone(),
305            endpoint,
306            role: ReplicaRole::Slave,
307            status: ReplicationStatus::Healthy,
308            last_sync: Utc::now(),
309            lag_ms: 0,
310            last_lsn: 0,
311        };
312
313        self.replicas.write().insert(node_id, replica);
314        Ok(())
315    }
316
317    /// Remove a replica node
318    pub async fn remove_replica(&mut self, node_id: &str) -> Result<()> {
319        self.replicas.write().remove(node_id);
320        Ok(())
321    }
322
323    /// Promote a slave to master (failover)
324    pub async fn promote_to_master(&mut self, node_id: &str) -> Result<()> {
325        let mut replicas = self.replicas.write();
326        if let Some(replica) = replicas.get_mut(node_id) {
327            replica.role = ReplicaRole::Master;
328            replica.status = ReplicationStatus::Healthy;
329
330            let mut stats = self.stats.lock();
331            stats.failovers += 1;
332        }
333
334        Ok(())
335    }
336
337    /// Record a change for replication
338    pub async fn record_change(&mut self, txn_id: String, data: Vec<u8>) -> Result<u64> {
339        let lsn = {
340            let mut current_lsn = self.current_lsn.lock();
341            *current_lsn += 1;
342            *current_lsn
343        };
344
345        let change = ReplicationChange {
346            lsn,
347            txn_id,
348            timestamp: Utc::now(),
349            data: data.clone(),
350            source_node: self.node_id.clone(),
351        };
352
353        let mut buffer = self.change_buffer.lock();
354        buffer.push_back(change);
355
356        // Trim buffer if too large
357        while buffer.len() > self.config.change_buffer_size {
358            buffer.pop_front();
359        }
360
361        let mut stats = self.stats.lock();
362        stats.total_changes += 1;
363        stats.total_bytes_replicated += data.len() as u64;
364
365        Ok(lsn)
366    }
367
368    /// Replicate changes to all replicas
369    pub async fn replicate_changes(&mut self, changes: Vec<ReplicationChange>) -> Result<()> {
370        if self.transport.is_none() {
371            return Err(TdbError::DistributedTransportNotConfigured {
372                node_id: self.node_id.clone(),
373                protocol: "Replication".to_string(),
374            });
375        }
376
377        let start_time = Utc::now();
378
379        match self.config.sync_mode {
380            SyncMode::Async => self.replicate_async(changes).await?,
381            SyncMode::Sync => self.replicate_sync(changes).await?,
382            SyncMode::SemiSync => self.replicate_semi_sync(changes).await?,
383        }
384
385        let latency = (Utc::now() - start_time).num_milliseconds() as f64;
386        let mut stats = self.stats.lock();
387        stats.total_latency_ms += latency;
388        stats.avg_replication_latency_ms = stats.total_latency_ms / stats.total_changes as f64;
389
390        Ok(())
391    }
392
393    /// Asynchronous replication (fire-and-forget)
394    async fn replicate_async(&self, changes: Vec<ReplicationChange>) -> Result<()> {
395        let replicas = self.replicas.read().clone();
396
397        for node_id in replicas.keys() {
398            // Simulate async replication
399            self.send_changes_to_replica(node_id, &changes).await?;
400        }
401
402        let mut stats = self.stats.lock();
403        stats.successful_replications += replicas.len() as u64;
404
405        Ok(())
406    }
407
408    /// Synchronous replication (wait for all replicas)
409    async fn replicate_sync(&self, changes: Vec<ReplicationChange>) -> Result<()> {
410        let replicas = self.replicas.read().clone();
411        let mut success_count = 0;
412
413        for node_id in replicas.keys() {
414            match self.send_changes_to_replica(node_id, &changes).await {
415                Ok(_) => success_count += 1,
416                Err(_) => {
417                    let mut stats = self.stats.lock();
418                    stats.failed_replications += 1;
419                }
420            }
421        }
422
423        // All replicas must succeed
424        if success_count != replicas.len() {
425            return Err(TdbError::Other(format!(
426                "Synchronous replication failed: {}/{} replicas acknowledged",
427                success_count,
428                replicas.len()
429            )));
430        }
431
432        let mut stats = self.stats.lock();
433        stats.successful_replications += success_count as u64;
434
435        Ok(())
436    }
437
438    /// Semi-synchronous replication (wait for at least one replica)
439    async fn replicate_semi_sync(&self, changes: Vec<ReplicationChange>) -> Result<()> {
440        let replicas = self.replicas.read().clone();
441        let mut success_count = 0;
442
443        for node_id in replicas.keys() {
444            match self.send_changes_to_replica(node_id, &changes).await {
445                Ok(_) => success_count += 1,
446                Err(_) => {
447                    let mut stats = self.stats.lock();
448                    stats.failed_replications += 1;
449                }
450            }
451        }
452
453        // At least one replica must succeed
454        if success_count == 0 {
455            return Err(TdbError::Other(
456                "Semi-sync replication failed: no replicas acknowledged".to_string(),
457            ));
458        }
459
460        let mut stats = self.stats.lock();
461        stats.successful_replications += success_count as u64;
462
463        Ok(())
464    }
465
466    /// Send changes to a specific replica over the configured
467    /// `NetworkTransport`.
468    async fn send_changes_to_replica(
469        &self,
470        node_id: &str,
471        changes: &[ReplicationChange],
472    ) -> Result<()> {
473        let transport =
474            self.transport
475                .as_ref()
476                .ok_or_else(|| TdbError::DistributedTransportNotConfigured {
477                    node_id: self.node_id.clone(),
478                    protocol: "Replication".to_string(),
479                })?;
480        let payload = oxicode::serde::encode_to_vec(&changes.to_vec(), oxicode::config::standard())
481            .map_err(|e| TdbError::Serialization(e.to_string()))?;
482        transport.send_replication_changes(node_id, &payload).await
483    }
484
485    /// Check replica health and update status
486    pub async fn check_replica_health(&self) -> Result<()> {
487        let now = Utc::now();
488        let max_lag = self.config.max_lag_ms;
489
490        let mut replicas = self.replicas.write();
491        for replica in replicas.values_mut() {
492            let lag = (now - replica.last_sync).num_milliseconds();
493            replica.lag_ms = lag;
494
495            if lag > max_lag {
496                replica.status = ReplicationStatus::Lagging;
497            } else {
498                replica.status = ReplicationStatus::Healthy;
499            }
500        }
501
502        Ok(())
503    }
504
505    /// Handle conflict in master-master replication
506    pub async fn resolve_conflict(
507        &mut self,
508        change1: &ReplicationChange,
509        change2: &ReplicationChange,
510    ) -> Result<ReplicationChange> {
511        let mut stats = self.stats.lock();
512        stats.conflicts_detected += 1;
513
514        let resolved = match self.config.conflict_resolution {
515            ConflictResolution::LastWriteWins => {
516                // Choose change with latest timestamp
517                if change1.timestamp > change2.timestamp {
518                    change1.clone()
519                } else {
520                    change2.clone()
521                }
522            }
523            ConflictResolution::FirstWriteWins => {
524                // Choose change with earliest timestamp
525                if change1.timestamp < change2.timestamp {
526                    change1.clone()
527                } else {
528                    change2.clone()
529                }
530            }
531            ConflictResolution::Manual | ConflictResolution::Custom => {
532                // Future enhancement: Implement manual resolution queue with admin UI.
533                // For v0.1.0: Falls back to last-write-wins for automatic resolution.
534                // Manual resolution would require persistent queue and admin interface.
535                if change1.timestamp > change2.timestamp {
536                    change1.clone()
537                } else {
538                    change2.clone()
539                }
540            }
541        };
542
543        stats.conflicts_resolved += 1;
544        Ok(resolved)
545    }
546
547    /// Perform automatic failover if master is down
548    pub async fn check_and_failover(&mut self) -> Result<bool> {
549        if !self.config.auto_failover {
550            return Ok(false);
551        }
552
553        // Find unhealthy master (if any)
554        let replicas = self.replicas.read().clone();
555        let unhealthy_masters: Vec<_> = replicas
556            .values()
557            .filter(|r| r.role == ReplicaRole::Master && r.status != ReplicationStatus::Healthy)
558            .collect();
559
560        if unhealthy_masters.is_empty() {
561            return Ok(false);
562        }
563
564        // Find best slave to promote
565        let best_slave = replicas
566            .values()
567            .filter(|r| r.role == ReplicaRole::Slave && r.status == ReplicationStatus::Healthy)
568            .max_by_key(|r| r.last_lsn);
569
570        if let Some(slave) = best_slave {
571            self.promote_to_master(&slave.node_id).await?;
572            return Ok(true);
573        }
574
575        Ok(false)
576    }
577
578    /// Get node ID
579    pub fn node_id(&self) -> &str {
580        &self.node_id
581    }
582
583    /// Get current role
584    pub fn role(&self) -> ReplicaRole {
585        *self.role.read()
586    }
587
588    /// Get statistics
589    pub fn stats(&self) -> ReplicationStats {
590        self.stats.lock().clone()
591    }
592
593    /// Get replica count
594    pub fn replica_count(&self) -> usize {
595        self.replicas.read().len()
596    }
597
598    /// Get healthy replica count
599    pub fn healthy_replica_count(&self) -> usize {
600        self.replicas
601            .read()
602            .values()
603            .filter(|r| r.status == ReplicationStatus::Healthy)
604            .count()
605    }
606
607    /// Get current LSN
608    pub fn current_lsn(&self) -> u64 {
609        *self.current_lsn.lock()
610    }
611
612    /// Get change buffer size
613    pub fn change_buffer_size(&self) -> usize {
614        self.change_buffer.lock().len()
615    }
616}
617
618#[cfg(test)]
619mod tests {
620    use super::*;
621
622    #[tokio::test]
623    async fn test_replication_manager_creation() {
624        let config = ReplicationConfig::default();
625        let manager = ReplicationManager::with_transport(
626            "node1".to_string(),
627            config,
628            crate::consensus::transport::LoopbackSimulationTransport::arc(),
629        );
630
631        assert_eq!(manager.node_id(), "node1");
632        assert_eq!(manager.role(), ReplicaRole::Master);
633        assert_eq!(manager.replica_count(), 0);
634    }
635
636    #[tokio::test]
637    async fn test_replicate_changes_without_transport_fails_loudly() {
638        // Regression test for the fabricated-replication-ack bug: without a
639        // NetworkTransport, replicate_changes() must never report
640        // successful_replications for acks that never happened.
641        let config = ReplicationConfig::default();
642        let mut manager = ReplicationManager::new("node1".to_string(), config);
643        manager
644            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
645            .await
646            .unwrap();
647
648        let changes = vec![ReplicationChange {
649            lsn: 1,
650            txn_id: "txn-1".to_string(),
651            timestamp: Utc::now(),
652            data: vec![1, 2, 3],
653            source_node: "node1".to_string(),
654        }];
655
656        let err = manager.replicate_changes(changes).await.unwrap_err();
657        assert!(matches!(
658            err,
659            TdbError::DistributedTransportNotConfigured { .. }
660        ));
661        assert_eq!(manager.stats().successful_replications, 0);
662    }
663
664    #[tokio::test]
665    async fn test_add_replicas() {
666        let config = ReplicationConfig::default();
667        let mut manager = ReplicationManager::with_transport(
668            "node1".to_string(),
669            config,
670            crate::consensus::transport::LoopbackSimulationTransport::arc(),
671        );
672
673        manager
674            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
675            .await
676            .unwrap();
677
678        manager
679            .add_replica("replica2".to_string(), "http://replica2:8080".to_string())
680            .await
681            .unwrap();
682
683        assert_eq!(manager.replica_count(), 2);
684    }
685
686    #[tokio::test]
687    async fn test_record_change() {
688        let config = ReplicationConfig::default();
689        let mut manager = ReplicationManager::with_transport(
690            "node1".to_string(),
691            config,
692            crate::consensus::transport::LoopbackSimulationTransport::arc(),
693        );
694
695        let lsn = manager
696            .record_change("txn-001".to_string(), vec![1, 2, 3, 4])
697            .await
698            .unwrap();
699
700        assert_eq!(lsn, 1);
701        assert_eq!(manager.current_lsn(), 1);
702
703        let stats = manager.stats();
704        assert_eq!(stats.total_changes, 1);
705        assert_eq!(stats.total_bytes_replicated, 4);
706    }
707
708    #[tokio::test]
709    async fn test_async_replication() {
710        let config = ReplicationConfig {
711            sync_mode: SyncMode::Async,
712            ..Default::default()
713        };
714
715        let mut manager = ReplicationManager::with_transport(
716            "node1".to_string(),
717            config,
718            crate::consensus::transport::LoopbackSimulationTransport::arc(),
719        );
720
721        manager
722            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
723            .await
724            .unwrap();
725
726        let changes = vec![ReplicationChange {
727            lsn: 1,
728            txn_id: "txn-001".to_string(),
729            timestamp: Utc::now(),
730            data: vec![1, 2, 3],
731            source_node: "node1".to_string(),
732        }];
733
734        manager.replicate_changes(changes).await.unwrap();
735
736        let stats = manager.stats();
737        assert_eq!(stats.successful_replications, 1);
738    }
739
740    #[tokio::test]
741    async fn test_sync_replication() {
742        let config = ReplicationConfig {
743            sync_mode: SyncMode::Sync,
744            ..Default::default()
745        };
746
747        let mut manager = ReplicationManager::with_transport(
748            "node1".to_string(),
749            config,
750            crate::consensus::transport::LoopbackSimulationTransport::arc(),
751        );
752
753        manager
754            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
755            .await
756            .unwrap();
757
758        manager
759            .add_replica("replica2".to_string(), "http://replica2:8080".to_string())
760            .await
761            .unwrap();
762
763        let changes = vec![ReplicationChange {
764            lsn: 1,
765            txn_id: "txn-001".to_string(),
766            timestamp: Utc::now(),
767            data: vec![1, 2, 3],
768            source_node: "node1".to_string(),
769        }];
770
771        manager.replicate_changes(changes).await.unwrap();
772
773        let stats = manager.stats();
774        assert_eq!(stats.successful_replications, 2);
775    }
776
777    #[tokio::test]
778    async fn test_semi_sync_replication() {
779        let config = ReplicationConfig {
780            sync_mode: SyncMode::SemiSync,
781            ..Default::default()
782        };
783
784        let mut manager = ReplicationManager::with_transport(
785            "node1".to_string(),
786            config,
787            crate::consensus::transport::LoopbackSimulationTransport::arc(),
788        );
789
790        manager
791            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
792            .await
793            .unwrap();
794
795        let changes = vec![ReplicationChange {
796            lsn: 1,
797            txn_id: "txn-001".to_string(),
798            timestamp: Utc::now(),
799            data: vec![1, 2, 3],
800            source_node: "node1".to_string(),
801        }];
802
803        manager.replicate_changes(changes).await.unwrap();
804
805        let stats = manager.stats();
806        assert!(stats.successful_replications >= 1);
807    }
808
809    #[tokio::test]
810    async fn test_promote_to_master() {
811        let config = ReplicationConfig::default();
812        let mut manager = ReplicationManager::with_transport(
813            "node1".to_string(),
814            config,
815            crate::consensus::transport::LoopbackSimulationTransport::arc(),
816        );
817
818        manager
819            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
820            .await
821            .unwrap();
822
823        manager.promote_to_master("replica1").await.unwrap();
824
825        let stats = manager.stats();
826        assert_eq!(stats.failovers, 1);
827    }
828
829    #[tokio::test]
830    async fn test_conflict_resolution_last_write_wins() {
831        let config = ReplicationConfig {
832            conflict_resolution: ConflictResolution::LastWriteWins,
833            mode: ReplicationMode::MasterMaster,
834            ..Default::default()
835        };
836
837        let mut manager = ReplicationManager::with_transport(
838            "node1".to_string(),
839            config,
840            crate::consensus::transport::LoopbackSimulationTransport::arc(),
841        );
842
843        let change1 = ReplicationChange {
844            lsn: 1,
845            txn_id: "txn-001".to_string(),
846            timestamp: Utc::now(),
847            data: vec![1, 2, 3],
848            source_node: "node1".to_string(),
849        };
850
851        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
852
853        let change2 = ReplicationChange {
854            lsn: 2,
855            txn_id: "txn-001".to_string(),
856            timestamp: Utc::now(),
857            data: vec![4, 5, 6],
858            source_node: "node2".to_string(),
859        };
860
861        let resolved = manager.resolve_conflict(&change1, &change2).await.unwrap();
862
863        // change2 should win (latest timestamp)
864        assert_eq!(resolved.data, vec![4, 5, 6]);
865
866        let stats = manager.stats();
867        assert_eq!(stats.conflicts_detected, 1);
868        assert_eq!(stats.conflicts_resolved, 1);
869    }
870
871    #[tokio::test]
872    async fn test_conflict_resolution_first_write_wins() {
873        let config = ReplicationConfig {
874            conflict_resolution: ConflictResolution::FirstWriteWins,
875            mode: ReplicationMode::MasterMaster,
876            ..Default::default()
877        };
878
879        let mut manager = ReplicationManager::with_transport(
880            "node1".to_string(),
881            config,
882            crate::consensus::transport::LoopbackSimulationTransport::arc(),
883        );
884
885        let change1 = ReplicationChange {
886            lsn: 1,
887            txn_id: "txn-001".to_string(),
888            timestamp: Utc::now(),
889            data: vec![1, 2, 3],
890            source_node: "node1".to_string(),
891        };
892
893        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
894
895        let change2 = ReplicationChange {
896            lsn: 2,
897            txn_id: "txn-001".to_string(),
898            timestamp: Utc::now(),
899            data: vec![4, 5, 6],
900            source_node: "node2".to_string(),
901        };
902
903        let resolved = manager.resolve_conflict(&change1, &change2).await.unwrap();
904
905        // change1 should win (earliest timestamp)
906        assert_eq!(resolved.data, vec![1, 2, 3]);
907    }
908
909    #[tokio::test]
910    async fn test_replica_health_check() {
911        let config = ReplicationConfig {
912            max_lag_ms: 100,
913            ..Default::default()
914        };
915
916        let manager = ReplicationManager::with_transport(
917            "node1".to_string(),
918            config,
919            crate::consensus::transport::LoopbackSimulationTransport::arc(),
920        );
921
922        manager.replicas.write().insert(
923            "replica1".to_string(),
924            ReplicaNode {
925                node_id: "replica1".to_string(),
926                endpoint: "http://replica1:8080".to_string(),
927                role: ReplicaRole::Slave,
928                status: ReplicationStatus::Healthy,
929                last_sync: Utc::now() - Duration::seconds(1),
930                lag_ms: 0,
931                last_lsn: 10,
932            },
933        );
934
935        manager.check_replica_health().await.unwrap();
936
937        let replicas = manager.replicas.read();
938        let replica = replicas.get("replica1").unwrap();
939        assert!(replica.lag_ms > 100);
940        assert_eq!(replica.status, ReplicationStatus::Lagging);
941    }
942
943    #[tokio::test]
944    async fn test_change_buffer() {
945        let config = ReplicationConfig {
946            change_buffer_size: 3,
947            ..Default::default()
948        };
949
950        let mut manager = ReplicationManager::with_transport(
951            "node1".to_string(),
952            config,
953            crate::consensus::transport::LoopbackSimulationTransport::arc(),
954        );
955
956        // Add 5 changes (buffer size is 3)
957        for i in 0..5 {
958            manager
959                .record_change(format!("txn-{:03}", i), vec![i as u8])
960                .await
961                .unwrap();
962        }
963
964        // Buffer should contain only last 3 changes
965        assert_eq!(manager.change_buffer_size(), 3);
966    }
967
968    #[tokio::test]
969    async fn test_replication_stats() {
970        let config = ReplicationConfig::default();
971        let mut manager = ReplicationManager::with_transport(
972            "node1".to_string(),
973            config,
974            crate::consensus::transport::LoopbackSimulationTransport::arc(),
975        );
976
977        manager
978            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
979            .await
980            .unwrap();
981
982        // Record multiple changes
983        for i in 0..5 {
984            manager
985                .record_change(format!("txn-{:03}", i), vec![1, 2, 3])
986                .await
987                .unwrap();
988        }
989
990        let stats = manager.stats();
991        assert_eq!(stats.total_changes, 5);
992        assert_eq!(stats.total_bytes_replicated, 15);
993    }
994
995    #[tokio::test]
996    async fn test_remove_replica() {
997        let config = ReplicationConfig::default();
998        let mut manager = ReplicationManager::with_transport(
999            "node1".to_string(),
1000            config,
1001            crate::consensus::transport::LoopbackSimulationTransport::arc(),
1002        );
1003
1004        manager
1005            .add_replica("replica1".to_string(), "http://replica1:8080".to_string())
1006            .await
1007            .unwrap();
1008
1009        assert_eq!(manager.replica_count(), 1);
1010
1011        manager.remove_replica("replica1").await.unwrap();
1012        assert_eq!(manager.replica_count(), 0);
1013    }
1014}
oxirs_tdb/distributed/replication.rs

oxirs_tdb/distributed/
replication.rs
