#[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::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant, SystemTime};
struct BaselineWorkload {
cluster_metrics: Arc<Mutex<ClusterMetrics>>,
leader: BaselineNode,
followers: Vec<BaselineNode>,
start_time: Instant,
}
struct BaselineNode {
node_id: String,
executor: Arc<Mutex<DistributedTransactionExecutor>>,
sequence: Arc<Mutex<u64>>,
}
impl BaselineNode {
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)),
}
}
fn submit_operation(&self, op_idx: u64) -> Result<TransactionEnvelope, 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> {
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 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)
}
}
impl BaselineWorkload {
fn new(follower_count: usize) -> Self {
let node_ids: Vec<_> = (0..=follower_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.clone())));
let leader = BaselineNode::new("leader".to_string());
let mut followers = Vec::new();
for i in 0..follower_count {
followers.push(BaselineNode::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,
start_time: Instant::now(),
}
}
fn run_clean_workload(&self, duration_secs: u64, ops_per_batch: u64) -> BaselineResult {
let deadline = Instant::now() + Duration::from_secs(duration_secs);
let mut operation_count = 0u64;
let mut batches = 0u64;
let mut convergence_checks = 0u64;
let mut divergences = 0u64;
let mut last_metrics_time = Instant::now();
while Instant::now() < deadline {
for i in 0..ops_per_batch {
let op_idx = operation_count + i;
match self.leader.submit_operation(op_idx) {
Ok(envelope) => {
let message = ReplicationMessage::new(
envelope,
self.leader.node_id.clone(),
"followers".to_string(),
op_idx + 1,
);
for follower in &self.followers {
let _ = follower.apply_replicated_operation(message.clone());
}
operation_count += 1;
}
Err(_) => break,
}
}
batches += 1;
if operation_count % 100 == 0 {
let leader_hash = self.leader.state_hash();
for follower in &self.followers {
if follower.state_hash() != leader_hash {
divergences += 1;
if let Ok(mut metrics) = self.cluster_metrics.lock() {
metrics.record_divergence();
}
}
}
convergence_checks += 1;
}
if last_metrics_time.elapsed() >= Duration::from_secs(5) {
if let Ok(mut metrics) = self.cluster_metrics.lock() {
metrics.nodes[0].record_operation_applied(); }
last_metrics_time = Instant::now();
}
}
let elapsed = self.start_time.elapsed();
let throughput = (operation_count as f64) / (elapsed.as_secs_f64());
BaselineResult {
duration_secs: elapsed.as_secs(),
total_operations: operation_count,
total_batches: batches,
throughput_ops_sec: throughput as u64,
convergence_checks,
divergences,
all_converged: divergences == 0,
}
}
fn verify_convergence(&self) -> bool {
let leader_hash = self.leader.state_hash();
self.followers.iter().all(|f| f.state_hash() == leader_hash)
}
fn export_baseline_report(&self) -> String {
let metrics = self.cluster_metrics.lock().unwrap();
let snapshot = metrics.get_cluster_snapshot();
let mut report = String::new();
report.push_str("=== Baseline Workload Report ===\n\n");
report.push_str(&format!("Uptime: {} seconds\n", snapshot.uptime_secs));
report.push_str(&format!(
"Total Operations: {}\n",
snapshot.total_operations_applied
));
report.push_str(&format!("Total Crashes: {}\n", snapshot.total_crashes));
report.push_str(&format!(
"Divergence Events: {}\n",
snapshot.divergence_events
));
report.push_str(&format!(
"Data Loss Events: {}\n",
snapshot.data_loss_events
));
report.push_str("\nNode Metrics:\n");
for node in &snapshot.node_snapshots {
report.push_str(&format!(
" {}: {} ops applied, {} acked, {} committed\n",
node.node_id,
node.operations_applied,
node.operations_acknowledged,
node.operations_committed
));
}
report
}
}
#[derive(Debug)]
struct BaselineResult {
duration_secs: u64,
total_operations: u64,
total_batches: u64,
throughput_ops_sec: u64,
convergence_checks: u64,
divergences: u64,
all_converged: bool,
}
#[test]
fn baseline_short_clean_run() {
let workload = BaselineWorkload::new(2);
let result = workload.run_clean_workload(10, 10);
println!("Baseline result: {:?}", result);
println!("{}", workload.export_baseline_report());
assert!(
result.total_operations > 0,
"Should have processed operations"
);
assert!(result.all_converged, "All nodes should converge");
assert_eq!(result.divergences, 0, "Zero divergences during clean run");
println!("✓ Short baseline: {} ops/sec", result.throughput_ops_sec);
}
#[test]
fn baseline_medium_clean_run() {
let workload = BaselineWorkload::new(2);
let result = workload.run_clean_workload(30, 25);
println!("Medium baseline result: {:?}", result);
assert!(
result.total_operations > 100,
"Should process 100+ operations"
);
assert!(result.all_converged, "All nodes should converge");
assert_eq!(result.divergences, 0, "Zero divergences");
println!("✓ Medium baseline: {} ops/sec", result.throughput_ops_sec);
}
#[test]
fn baseline_convergence_consistency() {
let workload = BaselineWorkload::new(2);
let result = workload.run_clean_workload(15, 20);
assert!(workload.verify_convergence(), "Final convergence check");
assert_eq!(result.divergences, 0, "No divergences during baseline");
println!(
"✓ Convergence consistency verified over {} ops",
result.total_operations
);
}
#[test]
fn baseline_high_throughput() {
let workload = BaselineWorkload::new(2);
let result = workload.run_clean_workload(20, 50);
println!(
"High-throughput baseline: {} ops/sec",
result.throughput_ops_sec
);
assert!(
result.throughput_ops_sec > 0,
"Should measure positive throughput"
);
assert_eq!(result.divergences, 0, "Zero divergences");
println!("✓ High-throughput: {} ops/sec", result.throughput_ops_sec);
}
#[test]
fn baseline_metrics_export() {
let workload = BaselineWorkload::new(2);
let _result = workload.run_clean_workload(5, 10);
let metrics = workload.cluster_metrics.lock().unwrap();
let csv = metrics.export_csv();
let json = metrics.export_json();
assert!(csv.contains("leader"), "CSV should contain leader metrics");
assert!(
json.contains("total_operations_applied"),
"JSON should contain operations"
);
println!("✓ Metrics export validated");
}
}