oxirs_tdb/distributed/

integration.rs

//! Distributed Features Integration Layer
//!
//! This module provides a high-level integration layer that connects distributed
//! transaction features (2PC, 3PC, Paxos, Saga) with the TdbStore, making it easy
//! to use distributed capabilities without managing low-level details.
//!
//! # Features
//!
//! - **Unified API**: Single interface for all distributed transaction protocols
//! - **Automatic Protocol Selection**: Choose optimal protocol based on requirements
//! - **Built-in Deadlock Prevention**: Integrated deadlock detection and resolution
//! - **Replication Integration**: Coordinate transactions with replication
//! - **Monitoring**: Comprehensive metrics and health tracking
//!
//! # Example
//!
//! ```rust,ignore
//! use oxirs_tdb::distributed::integration::{DistributedTdbStore, DistributedConfig};
//! use oxirs_tdb::consensus::transport::LoopbackSimulationTransport;
//!
//! // Create distributed store. A real multi-node deployment must supply a
//! // transport backed by real network I/O; LoopbackSimulationTransport is
//! // only for single-process tests/local development, and `new()` (no
//! // transport) will make every commit fail loudly instead of fabricating
//! // success.
//! let config = DistributedConfig::default();
//! let mut store = DistributedTdbStore::with_transport(
//!     "node1".to_string(),
//!     config,
//!     LoopbackSimulationTransport::arc(),
//! );
//!
//! // Register nodes
//! store.register_node("node1", "http://node1:8080").await?;
//! store.register_node("node2", "http://node2:8080").await?;
//!
//! // Execute distributed transaction
//! let txn_id = store.begin_distributed_transaction().await?;
//! // ... perform operations ...
//! store.commit_distributed_transaction(&txn_id).await?;
//! ```

use crate::consensus::transport::NetworkTransport;
use crate::distributed::coordinator::{CommitProtocol, CoordinatorConfig, TransactionCoordinator};
use crate::distributed::deadlock::{DeadlockDetectorConfig, DistributedDeadlockDetector};
use crate::distributed::replication::{ReplicationConfig, ReplicationManager};
use crate::distributed::saga::{SagaConfig, SagaOrchestrator};
use crate::error::{Result, TdbError};
use chrono::Utc;
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;

/// Distributed store configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedConfig {
    /// Transaction coordinator configuration
    pub coordinator_config: CoordinatorConfig,
    /// Deadlock detector configuration
    pub deadlock_config: DeadlockDetectorConfig,
    /// Replication configuration
    pub replication_config: ReplicationConfig,
    /// Saga configuration
    pub saga_config: SagaConfig,
    /// Default commit protocol
    pub default_protocol: CommitProtocol,
    /// Enable automatic deadlock detection
    pub enable_deadlock_detection: bool,
    /// Enable replication
    pub enable_replication: bool,
}

impl Default for DistributedConfig {
    fn default() -> Self {
        Self {
            coordinator_config: CoordinatorConfig::default(),
            deadlock_config: DeadlockDetectorConfig::default(),
            replication_config: ReplicationConfig::default(),
            saga_config: SagaConfig::default(),
            default_protocol: CommitProtocol::TwoPhase,
            enable_deadlock_detection: true,
            enable_replication: true,
        }
    }
}

/// Distributed transaction handle
#[derive(Debug, Clone)]
pub struct DistributedTransaction {
    /// Transaction ID
    pub txn_id: String,
    /// Protocol used
    pub protocol: CommitProtocol,
    /// Participating nodes
    pub nodes: Vec<String>,
}

/// Distributed TDB Store
///
/// High-level interface for distributed RDF storage with integrated
/// transaction coordination, deadlock detection, and replication.
pub struct DistributedTdbStore {
    /// Node ID
    node_id: String,
    /// Configuration
    config: DistributedConfig,
    /// Transaction coordinator
    coordinator: Arc<Mutex<TransactionCoordinator>>,
    /// Deadlock detector (optional)
    deadlock_detector: Option<Arc<Mutex<DistributedDeadlockDetector>>>,
    /// Replication manager (optional)
    replication_manager: Option<Arc<Mutex<ReplicationManager>>>,
    /// Active distributed transactions
    active_transactions: Arc<RwLock<HashMap<String, DistributedTransaction>>>,
    /// Statistics
    stats: Arc<Mutex<DistributedStoreStats>>,
}

/// Distributed store statistics
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct DistributedStoreStats {
    /// Total distributed transactions
    pub total_distributed_txns: u64,
    /// Successful distributed transactions
    pub successful_distributed_txns: u64,
    /// Failed distributed transactions
    pub failed_distributed_txns: u64,
    /// Deadlocks detected and resolved
    pub deadlocks_resolved: u64,
    /// Replications performed
    pub replications_performed: u64,
    /// Average transaction latency (milliseconds)
    pub avg_txn_latency_ms: f64,
    /// Total latency
    total_latency_ms: f64,
}

impl DistributedTdbStore {
    /// Create a new Distributed TDB Store with no network transport
    /// configured.
    ///
    /// **Note:** because no [`NetworkTransport`] is supplied, the underlying
    /// [`TransactionCoordinator`] and [`ReplicationManager`] cannot honestly
    /// reach any other node: `commit_distributed_transaction()` will fail
    /// with [`TdbError::DistributedTransportNotConfigured`] for every
    /// protocol rather than fabricating success. Use
    /// [`DistributedTdbStore::with_transport`] (with a real network-backed
    /// transport, or
    /// [`crate::consensus::transport::LoopbackSimulationTransport`] for
    /// single-node tests) to actually drive distributed commits.
    pub fn new(node_id: String, config: DistributedConfig) -> Self {
        let coordinator = Arc::new(Mutex::new(TransactionCoordinator::new(
            node_id.clone(),
            config.coordinator_config.clone(),
        )));

        let deadlock_detector = if config.enable_deadlock_detection {
            Some(Arc::new(Mutex::new(DistributedDeadlockDetector::new(
                format!("{}-deadlock-detector", node_id),
                config.deadlock_config.clone(),
            ))))
        } else {
            None
        };

        let replication_manager = if config.enable_replication {
            Some(Arc::new(Mutex::new(ReplicationManager::new(
                node_id.clone(),
                config.replication_config.clone(),
            ))))
        } else {
            None
        };

        Self {
            node_id,
            config,
            coordinator,
            deadlock_detector,
            replication_manager,
            active_transactions: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(Mutex::new(DistributedStoreStats::default())),
        }
    }

    /// Create a new Distributed TDB Store backed by a real (or
    /// explicitly-simulated) [`NetworkTransport`], shared by both the
    /// transaction coordinator and the replication manager. This is the
    /// constructor production code should use to actually drive distributed
    /// commits and replication.
    pub fn with_transport(
        node_id: String,
        config: DistributedConfig,
        transport: Arc<dyn NetworkTransport>,
    ) -> Self {
        let coordinator = Arc::new(Mutex::new(TransactionCoordinator::with_transport(
            node_id.clone(),
            config.coordinator_config.clone(),
            transport.clone(),
        )));

        let deadlock_detector = if config.enable_deadlock_detection {
            Some(Arc::new(Mutex::new(DistributedDeadlockDetector::new(
                format!("{}-deadlock-detector", node_id),
                config.deadlock_config.clone(),
            ))))
        } else {
            None
        };

        let replication_manager = if config.enable_replication {
            Some(Arc::new(Mutex::new(ReplicationManager::with_transport(
                node_id.clone(),
                config.replication_config.clone(),
                transport,
            ))))
        } else {
            None
        };

        Self {
            node_id,
            config,
            coordinator,
            deadlock_detector,
            replication_manager,
            active_transactions: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(Mutex::new(DistributedStoreStats::default())),
        }
    }

    /// Register a node in the distributed system
    #[allow(clippy::await_holding_lock)]
    pub async fn register_node(&mut self, node_id: &str, endpoint: &str) -> Result<()> {
        // Register with coordinator
        {
            self.coordinator
                .lock()
                .register_participant(node_id.to_string(), endpoint.to_string())
                .await?;
        }

        // Register with deadlock detector
        if let Some(ref detector) = self.deadlock_detector {
            detector.lock().register_node(node_id.to_string()).await?;
        }

        // Register with replication manager
        if let Some(ref manager) = self.replication_manager {
            manager
                .lock()
                .add_replica(node_id.to_string(), endpoint.to_string())
                .await?;
        }

        Ok(())
    }

    /// Begin a distributed transaction
    pub async fn begin_distributed_transaction(&mut self) -> Result<String> {
        self.begin_distributed_transaction_with_protocol(self.config.default_protocol)
            .await
    }

    /// Begin a distributed transaction with specific protocol
    #[allow(clippy::await_holding_lock)]
    pub async fn begin_distributed_transaction_with_protocol(
        &mut self,
        protocol: CommitProtocol,
    ) -> Result<String> {
        let txn_id = self.coordinator.lock().begin_transaction(protocol).await?;

        let metadata = {
            let coordinator = self.coordinator.lock();
            coordinator
                .get_transaction(&txn_id)
                .ok_or_else(|| TdbError::Other("Transaction not found".to_string()))?
        };

        let txn = DistributedTransaction {
            txn_id: txn_id.clone(),
            protocol,
            nodes: metadata.participants.clone(),
        };

        self.active_transactions.write().insert(txn_id.clone(), txn);

        let mut stats = self.stats.lock();
        stats.total_distributed_txns += 1;

        Ok(txn_id)
    }

    /// Commit a distributed transaction
    #[allow(clippy::await_holding_lock)]
    pub async fn commit_distributed_transaction(&mut self, txn_id: &str) -> Result<bool> {
        let start = std::time::Instant::now();

        // Check for deadlocks before commit
        if let Some(ref detector) = self.deadlock_detector {
            let deadlocks = detector.lock().detect_deadlocks().await?;

            if !deadlocks.is_empty() {
                // Abort victims
                for deadlock in &deadlocks {
                    if let Some(ref victim) = deadlock.victim {
                        detector.lock().abort_victim(victim).await?;

                        let mut stats = self.stats.lock();
                        stats.deadlocks_resolved += 1;
                    }
                }
            }
        }

        // Execute commit
        let result = self.coordinator.lock().commit_transaction(txn_id).await?;

        // Replicate if enabled and successful
        if result {
            if let Some(ref manager) = self.replication_manager {
                // NOTE: Transport is in-process simulation.
                // Record the committed transaction changes in the replication log.
                manager
                    .lock()
                    .record_change(txn_id.to_string(), txn_id.as_bytes().to_vec())
                    .await?;
                let mut stats = self.stats.lock();
                stats.replications_performed += 1;
            }
        }

        // Remove from active transactions
        self.active_transactions.write().remove(txn_id);

        // Update statistics
        let latency = start.elapsed().as_millis() as f64;
        let mut stats = self.stats.lock();

        if result {
            stats.successful_distributed_txns += 1;
        } else {
            stats.failed_distributed_txns += 1;
        }

        stats.total_latency_ms += latency;
        stats.avg_txn_latency_ms = stats.total_latency_ms / stats.total_distributed_txns as f64;

        Ok(result)
    }

    /// Abort a distributed transaction
    #[allow(clippy::await_holding_lock)]
    pub async fn abort_distributed_transaction(&mut self, txn_id: &str) -> Result<()> {
        self.coordinator.lock().abort_transaction(txn_id).await?;

        self.active_transactions.write().remove(txn_id);

        let mut stats = self.stats.lock();
        stats.failed_distributed_txns += 1;

        Ok(())
    }

    /// Create and execute a saga
    #[allow(clippy::await_holding_lock)]
    pub async fn execute_saga(&mut self, mut saga: SagaOrchestrator) -> Result<bool> {
        // NOTE: Transport is in-process simulation.
        // Execute the saga; replicate committed steps, and the compensation path if rolled back.
        let result = saga.execute().await?;

        // Replicate saga outcome to replication log
        if let Some(ref manager) = self.replication_manager {
            let saga_key = Utc::now().timestamp_millis().to_string();
            let change_data = if result {
                format!("saga:committed:{}", saga_key).into_bytes()
            } else {
                format!("saga:compensated:{}", saga_key).into_bytes()
            };
            manager.lock().record_change(saga_key, change_data).await?;
        }

        // Update coordinator stats
        {
            let mut stats = self.stats.lock();
            if result {
                stats.successful_distributed_txns += 1;
            } else {
                stats.failed_distributed_txns += 1;
            }
            stats.total_distributed_txns += 1;
        }

        Ok(result)
    }

    /// Get node ID
    pub fn node_id(&self) -> &str {
        &self.node_id
    }

    /// Get active transaction count
    pub fn active_transaction_count(&self) -> usize {
        self.active_transactions.read().len()
    }

    /// Get statistics
    pub fn stats(&self) -> DistributedStoreStats {
        self.stats.lock().clone()
    }

    /// Get coordinator statistics
    pub fn coordinator_stats(&self) -> crate::distributed::coordinator::CoordinatorStats {
        self.coordinator.lock().stats()
    }

    /// Get replication statistics (if enabled)
    pub fn replication_stats(&self) -> Option<crate::distributed::replication::ReplicationStats> {
        self.replication_manager.as_ref().map(|m| m.lock().stats())
    }

    /// Check system health
    pub async fn check_health(&self) -> Result<HealthStatus> {
        let coordinator_health = self.coordinator.lock().healthy_participant_count() > 0;

        let replication_health = if let Some(ref manager) = self.replication_manager {
            manager.lock().healthy_replica_count() > 0
        } else {
            true
        };

        let overall_health = coordinator_health && replication_health;

        Ok(HealthStatus {
            healthy: overall_health,
            coordinator_nodes: self.coordinator.lock().participant_count(),
            healthy_coordinators: self.coordinator.lock().healthy_participant_count(),
            replica_count: self
                .replication_manager
                .as_ref()
                .map(|m| m.lock().replica_count())
                .unwrap_or(0),
            healthy_replicas: self
                .replication_manager
                .as_ref()
                .map(|m| m.lock().healthy_replica_count())
                .unwrap_or(0),
            active_transactions: self.active_transaction_count(),
        })
    }
}

/// System health status
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthStatus {
    /// Overall health
    pub healthy: bool,
    /// Total coordinator nodes
    pub coordinator_nodes: usize,
    /// Healthy coordinator nodes
    pub healthy_coordinators: usize,
    /// Total replicas
    pub replica_count: usize,
    /// Healthy replicas
    pub healthy_replicas: usize,
    /// Active transactions
    pub active_transactions: usize,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_distributed_store_creation() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        assert_eq!(store.node_id(), "node1");
        assert_eq!(store.active_transaction_count(), 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_register_node() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        // Node should be registered with coordinator
        assert!(store.coordinator.lock().participant_count() > 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_begin_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        assert!(!txn_id.is_empty());
        assert_eq!(store.active_transaction_count(), 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_commit_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        let result = store.commit_distributed_transaction(&txn_id).await.unwrap();

        assert!(result);
        assert_eq!(store.active_transaction_count(), 0);

        let stats = store.stats();
        assert_eq!(stats.successful_distributed_txns, 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_abort_distributed_transaction() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();

        store.abort_distributed_transaction(&txn_id).await.unwrap();

        assert_eq!(store.active_transaction_count(), 0);

        let stats = store.stats();
        assert_eq!(stats.failed_distributed_txns, 1);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_health_check() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        let health = store.check_health().await.unwrap();

        assert_eq!(health.active_transactions, 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_statistics() {
        let config = DistributedConfig::default();
        let store = DistributedTdbStore::new("node1".to_string(), config);

        let stats = store.stats();
        assert_eq!(stats.total_distributed_txns, 0);

        let coord_stats = store.coordinator_stats();
        assert_eq!(coord_stats.total_transactions, 0);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_protocol_selection() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        // Test with 3PC
        let txn_id = store
            .begin_distributed_transaction_with_protocol(CommitProtocol::ThreePhase)
            .await
            .unwrap();

        let txn = store.active_transactions.read().get(&txn_id).cloned();
        assert!(txn.is_some());
        assert_eq!(txn.unwrap().protocol, CommitProtocol::ThreePhase);
    }

    #[tokio::test]
    #[ignore = "Distributed tests hang - needs network mock or investigation"]
    async fn test_multiple_transactions() {
        let config = DistributedConfig::default();
        let mut store = DistributedTdbStore::new("node1".to_string(), config);

        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn1 = store.begin_distributed_transaction().await.unwrap();
        let txn2 = store.begin_distributed_transaction().await.unwrap();

        assert_eq!(store.active_transaction_count(), 2);

        store.commit_distributed_transaction(&txn1).await.unwrap();
        assert_eq!(store.active_transaction_count(), 1);

        store.abort_distributed_transaction(&txn2).await.unwrap();
        assert_eq!(store.active_transaction_count(), 0);
    }

    #[tokio::test]
    async fn test_commit_triggers_replication() {
        use crate::consensus::transport::LoopbackSimulationTransport;

        let config = DistributedConfig {
            enable_deadlock_detection: false,
            ..Default::default()
        };
        let mut store = DistributedTdbStore::with_transport(
            "node1".to_string(),
            config,
            LoopbackSimulationTransport::arc(),
        );
        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();
        let result = store.commit_distributed_transaction(&txn_id).await.unwrap();
        assert!(result);

        let stats = store.stats();
        assert_eq!(stats.replications_performed, 1);
    }

    /// Regression test: a `DistributedTdbStore` built via the transport-less
    /// `new()` constructor must fail loudly on commit instead of fabricating
    /// success, since there is no way to honestly reach any participant.
    #[tokio::test]
    async fn test_commit_without_transport_fails_loudly() {
        let config = DistributedConfig {
            enable_deadlock_detection: false,
            ..Default::default()
        };
        let mut store = DistributedTdbStore::new("node1".to_string(), config);
        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let txn_id = store.begin_distributed_transaction().await.unwrap();
        let err = store
            .commit_distributed_transaction(&txn_id)
            .await
            .expect_err("commit without a configured transport must fail, not fabricate success");
        assert!(
            matches!(err, TdbError::DistributedTransportNotConfigured { .. }),
            "expected DistributedTransportNotConfigured, got {err:?}"
        );
    }

    #[tokio::test]
    async fn test_execute_saga_success_replicates() {
        use crate::distributed::saga::{SagaConfig, SagaOrchestrator, SagaStep};
        use std::sync::atomic::{AtomicU32, Ordering};
        use std::sync::Arc as StdArc;

        let config = DistributedConfig {
            enable_deadlock_detection: false,
            ..Default::default()
        };
        let mut store = DistributedTdbStore::new("node1".to_string(), config);
        store
            .register_node("node2", "http://node2:8080")
            .await
            .unwrap();

        let mut saga = SagaOrchestrator::new("test-saga".to_string(), SagaConfig::default());
        let counter = StdArc::new(AtomicU32::new(0));
        let counter_clone = counter.clone();
        saga.add_step(SagaStep {
            name: "step-1".to_string(),
            ..Default::default()
        });
        saga.register_step_callbacks(
            "step-1",
            move || {
                counter_clone.fetch_add(1, Ordering::SeqCst);
                Ok(())
            },
            || Ok(()),
        );

        let result = store.execute_saga(saga).await.unwrap();
        assert!(result);
        assert_eq!(counter.load(Ordering::SeqCst), 1);

        // Check replication manager received the saga record
        let rep_stats = store.replication_stats();
        assert!(rep_stats.is_some());
        let rs = rep_stats.unwrap();
        assert_eq!(rs.total_changes, 1);
    }
}