#[cfg(test)]
mod tests {
use crate::convergence::VectorClock;
use crate::distributed_transactions::{
DistributedTransactionExecutor, ReplicationMessage, TransactionEnvelope,
};
use crate::metrics::ClusterMetrics;
use crate::state_hash::StateHash;
use crate::transactions::{
ConsistencyContract, Operation, OperationCommand, OperationId, StateVersion,
};
use serde_json::json;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant, SystemTime};
struct FailureInjectionConfig {
node_crash_interval_secs: u64,
partition_interval_secs: u64,
restart_interval_secs: u64,
compaction_interval_secs: u64,
}
impl FailureInjectionConfig {
fn default() -> Self {
Self {
node_crash_interval_secs: 45, partition_interval_secs: 180, restart_interval_secs: 600, compaction_interval_secs: 900, }
}
fn test_mode() -> Self {
Self {
node_crash_interval_secs: 5, partition_interval_secs: 15, restart_interval_secs: 30, compaction_interval_secs: 60, }
}
}
struct SoakNode {
node_id: String,
executor: Arc<Mutex<DistributedTransactionExecutor>>,
sequence: Arc<Mutex<u64>>,
is_crashed: Arc<AtomicBool>,
is_partitioned: Arc<AtomicBool>,
crash_count: Arc<Mutex<u64>>,
recovery_times_ms: Arc<Mutex<Vec<u64>>>,
}
impl SoakNode {
fn new(node_id: String) -> Self {
let initial_hash = StateHash::from_hex("0".repeat(64));
let executor = DistributedTransactionExecutor::new(node_id.clone(), initial_hash);
Self {
node_id,
executor: Arc::new(Mutex::new(executor)),
sequence: Arc::new(Mutex::new(0)),
is_crashed: Arc::new(AtomicBool::new(false)),
is_partitioned: Arc::new(AtomicBool::new(false)),
crash_count: Arc::new(Mutex::new(0)),
recovery_times_ms: Arc::new(Mutex::new(Vec::new())),
}
}
fn submit_operation(&self, op_idx: u64) -> Result<TransactionEnvelope, String> {
if self.is_crashed.load(Ordering::Relaxed) {
return Err("Node crashed".to_string());
}
let mut executor = self.executor.lock().map_err(|_| "lock failed")?;
let mut seq = self.sequence.lock().map_err(|_| "lock failed")?;
*seq += 1;
let op_id = OperationId::new(self.node_id.clone(), *seq);
let parent_vc = VectorClock::new();
let parent_version = StateVersion::new(parent_vc, "0".repeat(64));
let mut fields = HashMap::new();
fields.insert("value".to_string(), json!(op_idx));
let operation = Operation::new(
op_id,
parent_version.clone(),
format!("op_{}", op_idx),
OperationCommand {
op_type: "set".to_string(),
collection: "items".to_string(),
record_id: format!("item_{}", op_idx),
fields,
},
self.node_id.clone(),
);
executor.execute_local_transaction(
*seq,
format!("txn_{}", op_idx),
parent_version,
vec![operation],
ConsistencyContract::causal(),
)
}
fn apply_replicated_operation(&self, message: ReplicationMessage) -> Result<(), String> {
if self.is_crashed.load(Ordering::Relaxed)
|| self.is_partitioned.load(Ordering::Relaxed)
{
return Err("Node unavailable".to_string());
}
let mut executor = self.executor.lock().map_err(|_| "lock failed")?;
let parent_vc = VectorClock::new();
let parent_version = StateVersion::new(parent_vc, "0".repeat(64));
executor
.receive_replicated_transaction(message, parent_version)
.map(|_| ())
.map_err(|e| e)
}
fn crash(&self) {
self.is_crashed.store(true, Ordering::Relaxed);
let mut count = self.crash_count.lock().unwrap();
*count += 1;
}
fn recover(&self, recovery_time_ms: u64) {
self.is_crashed.store(false, Ordering::Relaxed);
let mut times = self.recovery_times_ms.lock().unwrap();
times.push(recovery_time_ms);
}
fn partition(&self) {
self.is_partitioned.store(true, Ordering::Relaxed);
}
fn heal_partition(&self) {
self.is_partitioned.store(false, Ordering::Relaxed);
}
fn state_hash(&self) -> StateHash {
let executor = self.executor.lock().unwrap();
executor
.get_replica_state(&self.node_id)
.map(|s| s.state_hash)
.unwrap_or_else(|| StateHash::from_hex("0".repeat(64)))
}
fn operations_applied(&self) -> u64 {
let executor = self.executor.lock().unwrap();
executor
.get_replica_state(&self.node_id)
.map(|s| s.operations_applied)
.unwrap_or(0)
}
}
struct SoakTest {
cluster_metrics: Arc<Mutex<ClusterMetrics>>,
leader: Arc<SoakNode>,
followers: Vec<Arc<SoakNode>>,
failure_config: FailureInjectionConfig,
start_time: Instant,
}
impl SoakTest {
fn new(node_count: usize, failure_config: FailureInjectionConfig) -> Self {
let node_ids: Vec<_> = (0..node_count)
.map(|i| {
if i == 0 {
"leader".to_string()
} else {
format!("follower-{}", i - 1)
}
})
.collect();
let cluster_metrics = Arc::new(Mutex::new(ClusterMetrics::new(node_ids)));
let leader = Arc::new(SoakNode::new("leader".to_string()));
let mut followers = Vec::new();
for i in 0..node_count - 1 {
followers.push(Arc::new(SoakNode::new(format!("follower-{}", i))));
}
{
let mut executor = leader.executor.lock().unwrap();
for follower in &followers {
executor.register_replica(
follower.node_id.clone(),
StateHash::from_hex("0".repeat(64)),
);
}
}
Self {
cluster_metrics,
leader,
followers,
failure_config,
start_time: Instant::now(),
}
}
fn run_soak_test(&self, duration_secs: u64, ops_per_batch: u64) -> SoakTestResult {
let deadline = Instant::now() + Duration::from_secs(duration_secs);
let mut operation_count = 0u64;
let mut failures_injected = 0u64;
let mut divergences = 0u64;
let mut convergence_checks = 0u64;
let mut last_crash_time = Instant::now();
let mut last_partition_time = Instant::now();
let mut last_restart_time = Instant::now();
let mut last_recovery_time = Instant::now();
while Instant::now() < deadline {
for i in 0..ops_per_batch {
if let Ok(envelope) = self.leader.submit_operation(operation_count + i) {
let message = ReplicationMessage::new(
envelope,
self.leader.node_id.clone(),
"followers".to_string(),
operation_count + i + 1,
);
for follower in &self.followers {
let _ = follower.apply_replicated_operation(message.clone());
}
operation_count += 1;
}
}
if last_crash_time.elapsed().as_secs()
>= self.failure_config.node_crash_interval_secs
{
if let Some(follower) = self.followers.first() {
follower.crash();
failures_injected += 1;
std::thread::sleep(Duration::from_millis(100));
let recovery_time = 100;
follower.recover(recovery_time);
}
last_crash_time = Instant::now();
}
if last_partition_time.elapsed().as_secs()
>= self.failure_config.partition_interval_secs
{
if let Some(follower) = self.followers.get(1) {
follower.partition();
failures_injected += 1;
std::thread::sleep(Duration::from_millis(50));
follower.heal_partition();
}
last_partition_time = Instant::now();
}
if last_recovery_time.elapsed().as_secs() >= 10 {
let leader_hash = self.leader.state_hash();
for follower in &self.followers {
if !follower.is_crashed.load(Ordering::Relaxed)
&& follower.state_hash() != leader_hash
{
divergences += 1;
if let Ok(mut metrics) = self.cluster_metrics.lock() {
metrics.record_divergence();
}
}
}
convergence_checks += 1;
last_recovery_time = Instant::now();
}
std::thread::sleep(Duration::from_millis(10));
}
let elapsed = self.start_time.elapsed();
SoakTestResult {
duration_secs: elapsed.as_secs(),
total_operations: operation_count,
failures_injected,
divergences,
convergence_checks,
all_stable: divergences == 0,
leader_ops: self.leader.operations_applied(),
final_hash_match: self.verify_final_convergence(),
}
}
fn verify_final_convergence(&self) -> bool {
let leader_hash = self.leader.state_hash();
self.followers
.iter()
.filter(|f| !f.is_crashed.load(Ordering::Relaxed))
.all(|f| f.state_hash() == leader_hash)
}
}
#[derive(Debug)]
struct SoakTestResult {
duration_secs: u64,
total_operations: u64,
failures_injected: u64,
divergences: u64,
convergence_checks: u64,
all_stable: bool,
leader_ops: u64,
final_hash_match: bool,
}
#[test]
fn soak_short_adversarial() {
let config = FailureInjectionConfig::test_mode();
let soak = SoakTest::new(3, config);
let result = soak.run_soak_test(30, 50);
println!("Short soak result: {:?}", result);
assert!(result.total_operations > 0, "Should process operations");
assert!(result.failures_injected > 0, "Should inject failures");
assert_eq!(result.divergences, 0, "Zero divergences under failures");
assert!(result.final_hash_match, "Final convergence achieved");
println!(
"✓ Short soak: {} ops, {} failures, {} divergences",
result.total_operations, result.failures_injected, result.divergences
);
}
#[test]
fn soak_medium_adversarial() {
let config = FailureInjectionConfig::test_mode();
let soak = SoakTest::new(3, config);
let result = soak.run_soak_test(60, 50);
println!("Medium soak result: {:?}", result);
assert!(
result.total_operations > 1000,
"Should process 1000+ operations"
);
assert!(
result.failures_injected > 5,
"Should inject multiple failures"
);
assert_eq!(result.divergences, 0, "Zero divergences");
assert!(result.all_stable, "System remains stable");
println!(
"✓ Medium soak: {} ops, {} failures injected",
result.total_operations, result.failures_injected
);
}
#[test]
fn soak_recovery_consistency() {
let config = FailureInjectionConfig::test_mode();
let soak = SoakTest::new(3, config);
let result = soak.run_soak_test(45, 40);
assert!(result.convergence_checks > 0, "Should check convergence");
assert_eq!(result.divergences, 0, "Recovery is consistent");
println!(
"✓ Recovery consistency over {} checks",
result.convergence_checks
);
}
#[test]
fn soak_long_stable() {
let config = FailureInjectionConfig::test_mode();
let soak = SoakTest::new(3, config);
let result = soak.run_soak_test(90, 60);
println!("Long soak result: {:?}", result);
assert!(
result.total_operations > 2000,
"Should process 2000+ operations"
);
assert!(
result.all_stable,
"System remains stable under sustained load"
);
assert_eq!(result.divergences, 0, "No divergences over extended period");
assert!(result.final_hash_match, "Final state converges");
println!(
"✓ Long soak: {} ops under {} failures",
result.total_operations, result.failures_injected
);
}
#[test]
fn soak_metrics_collection() {
let config = FailureInjectionConfig::test_mode();
let soak = SoakTest::new(3, config);
let result = soak.run_soak_test(30, 50);
let metrics = soak.cluster_metrics.lock().unwrap();
let snapshot = metrics.get_cluster_snapshot();
assert!(snapshot.divergence_events >= 0, "Metrics track divergences");
assert_eq!(snapshot.data_loss_events, 0, "No data loss detected");
println!(
"✓ Metrics: {} divergence events, {} data loss events",
snapshot.divergence_events, snapshot.data_loss_events
);
}
}