foxhunt/services/load_tests/throughput_tests.rs

//! Comprehensive throughput validation tests for trading service
//!
//! This module implements load tests to validate the trading system's capacity:
//! - 10,000 orders/sec sustained load (60 seconds)
//! - 50,000 orders/sec peak burst (10 seconds)
//! - 1M concurrent market data updates (streaming)
//! - Connection pool saturation (1000 concurrent clients)
//!
//! Run with: cargo test -p load_tests --release -- --nocapture

use anyhow::{Context, Result};
use dashmap::DashMap;
use hdrhistogram::Histogram;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use sysinfo::{ProcessRefreshKind, RefreshKind, System};
use tokio::sync::{mpsc, Barrier};
use tokio::task::JoinSet;
use tonic::transport::Channel;

// Generated proto code
pub mod trading {
    tonic::include_proto!("trading");
}

use trading::{
    trading_service_client::TradingServiceClient, OrderSide, OrderType, StreamMarketDataRequest,
    SubmitOrderRequest,
};

/// Test configuration
const TRADING_SERVICE_URL: &str = "http://localhost:50052";
const TEST_ACCOUNT_ID: &str = "load-test-account";

/// Metrics collected during load tests
#[derive(Debug)]
pub struct LoadTestMetrics {
    /// Total requests sent
    pub total_requests: AtomicU64,
    /// Successful responses
    pub successful_requests: AtomicU64,
    /// Failed requests
    pub failed_requests: AtomicU64,
    /// Latency histogram (microseconds)
    pub latency_us: parking_lot::Mutex<Histogram<u64>>,
    /// Memory usage samples (bytes)
    pub memory_samples: parking_lot::Mutex<Vec<u64>>,
    /// Test start time
    pub start_time: Instant,
}

impl LoadTestMetrics {
    pub fn new() -> Self {
        Self {
            total_requests: AtomicU64::new(0),
            successful_requests: AtomicU64::new(0),
            failed_requests: AtomicU64::new(0),
            latency_us: parking_lot::Mutex::new(
                Histogram::<u64>::new_with_bounds(1, 60_000_000, 3).unwrap(),
            ),
            memory_samples: parking_lot::Mutex::new(Vec::new()),
            start_time: Instant::now(),
        }
    }

    pub fn record_request(&self, duration: Duration, success: bool) {
        self.total_requests.fetch_add(1, Ordering::Relaxed);
        if success {
            self.successful_requests.fetch_add(1, Ordering::Relaxed);
        } else {
            self.failed_requests.fetch_add(1, Ordering::Relaxed);
        }

        let mut hist = self.latency_us.lock();
        let _ = hist.record(duration.as_micros() as u64);
    }

    pub fn record_memory(&self, bytes: u64) {
        self.memory_samples.lock().push(bytes);
    }

    pub fn report(&self) -> LoadTestReport {
        let elapsed = self.start_time.elapsed();
        let total = self.total_requests.load(Ordering::Relaxed);
        let successful = self.successful_requests.load(Ordering::Relaxed);
        let failed = self.failed_requests.load(Ordering::Relaxed);

        let hist = self.latency_us.lock();
        let p50 = hist.value_at_quantile(0.50);
        let p95 = hist.value_at_quantile(0.95);
        let p99 = hist.value_at_quantile(0.99);
        let max = hist.max();

        let throughput = total as f64 / elapsed.as_secs_f64();
        let error_rate = if total > 0 {
            (failed as f64 / total as f64) * 100.0
        } else {
            0.0
        };

        let memory_samples = self.memory_samples.lock();
        let avg_memory_mb = if !memory_samples.is_empty() {
            memory_samples.iter().sum::<u64>() as f64 / memory_samples.len() as f64 / 1_048_576.0
        } else {
            0.0
        };

        LoadTestReport {
            test_duration: elapsed,
            total_requests: total,
            successful_requests: successful,
            failed_requests: failed,
            throughput_per_sec: throughput,
            error_rate_percent: error_rate,
            latency_p50_us: p50,
            latency_p95_us: p95,
            latency_p99_us: p99,
            latency_max_us: max,
            avg_memory_mb,
        }
    }
}

#[derive(Debug, Clone)]
pub struct LoadTestReport {
    pub test_duration: Duration,
    pub total_requests: u64,
    pub successful_requests: u64,
    pub failed_requests: u64,
    pub throughput_per_sec: f64,
    pub error_rate_percent: f64,
    pub latency_p50_us: u64,
    pub latency_p95_us: u64,
    pub latency_p99_us: u64,
    pub latency_max_us: u64,
    pub avg_memory_mb: f64,
}

impl LoadTestReport {
    pub fn print(&self, test_name: &str) {
        println!("\n{'='*80}");
        println!("Load Test Report: {}", test_name);
        println!("{'='*80}");
        println!("Duration: {:?}", self.test_duration);
        println!(
            "Total Requests: {} (Success: {}, Failed: {})",
            self.total_requests, self.successful_requests, self.failed_requests
        );
        println!("Throughput: {:.2} req/sec", self.throughput_per_sec);
        println!("Error Rate: {:.2}%", self.error_rate_percent);
        println!("\nLatency (microseconds):");
        println!("  P50: {} μs", self.latency_p50_us);
        println!("  P95: {} μs", self.latency_p95_us);
        println!("  P99: {} μs", self.latency_p99_us);
        println!("  Max: {} μs", self.latency_max_us);
        println!("\nMemory: {:.2} MB (avg)", self.avg_memory_mb);
        println!("{'='*80}\n");
    }
}

/// Create authenticated gRPC client
async fn create_client() -> Result<TradingServiceClient<Channel>> {
    let channel = Channel::from_static(TRADING_SERVICE_URL)
        .connect()
        .await
        .context("Failed to connect to trading service")?;

    Ok(TradingServiceClient::new(channel))
}

/// Submit a single order and measure latency
async fn submit_order(
    client: &mut TradingServiceClient<Channel>,
    symbol: String,
    order_id: u64,
) -> Result<Duration> {
    let start = Instant::now();

    let request = SubmitOrderRequest {
        symbol,
        side: OrderSide::Buy as i32,
        quantity: 100.0,
        order_type: OrderType::Market as i32,
        price: None,
        stop_price: None,
        account_id: TEST_ACCOUNT_ID.to_string(),
        metadata: std::collections::HashMap::new(),
    };

    let _ = client.submit_order(request).await?;
    Ok(start.elapsed())
}

/// Test 1: 10,000 orders/sec sustained load (60 seconds)
#[tokio::test]
#[ignore] // Run explicitly with --ignored
async fn test_sustained_load_10k_orders() -> Result<()> {
    const TARGET_RPS: usize = 10_000;
    const DURATION_SECS: u64 = 60;
    const CONCURRENT_CLIENTS: usize = 100;

    println!("\n🚀 Starting Sustained Load Test: 10,000 orders/sec for 60s");

    let metrics = Arc::new(LoadTestMetrics::new());
    let mut join_set = JoinSet::new();

    // Spawn concurrent clients
    for client_id in 0..CONCURRENT_CLIENTS {
        let metrics = Arc::clone(&metrics);

        join_set.spawn(async move {
            let mut client = create_client().await?;
            let start = Instant::now();
            let mut order_count = 0u64;

            while start.elapsed().as_secs() < DURATION_SECS {
                let symbol = format!("SYM{:04}", client_id % 100);
                match submit_order(&mut client, symbol, order_count).await {
                    Ok(duration) => metrics.record_request(duration, true),
                    Err(e) => {
                        tracing::warn!("Order failed: {:?}", e);
                        metrics.record_request(Duration::from_micros(0), false);
                    }
                }

                order_count += 1;

                // Rate limiting: each client sends ~100 orders/sec
                tokio::time::sleep(Duration::from_micros(10_000)).await;
            }

            Ok::<_, anyhow::Error>(())
        });
    }

    // Memory monitoring task
    let metrics_clone = Arc::clone(&metrics);
    let monitor_handle = tokio::spawn(async move {
        let mut sys = System::new_with_specifics(
            RefreshKind::new().with_processes(ProcessRefreshKind::everything()),
        );

        for _ in 0..60 {
            tokio::time::sleep(Duration::from_secs(1)).await;
            sys.refresh_all();

            if let Some(process) = sys.process(sysinfo::get_current_pid().unwrap()) {
                metrics_clone.record_memory(process.memory());
            }
        }
    });

    // Wait for all clients
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            tracing::error!("Client task failed: {:?}", e);
        }
    }

    monitor_handle.abort();

    // Generate report
    let report = metrics.report();
    report.print("Sustained Load: 10,000 orders/sec");

    // Assertions
    assert!(
        report.throughput_per_sec >= 9_000.0,
        "Throughput too low: {:.2} req/sec (expected >= 9,000)",
        report.throughput_per_sec
    );
    assert!(
        report.error_rate_percent < 1.0,
        "Error rate too high: {:.2}% (expected < 1%)",
        report.error_rate_percent
    );
    assert!(
        report.latency_p95_us < 10_000,
        "P95 latency too high: {} μs (expected < 10ms)",
        report.latency_p95_us
    );

    Ok(())
}

/// Test 2: 50,000 orders/sec peak burst (10 seconds)
#[tokio::test]
#[ignore]
async fn test_peak_burst_50k_orders() -> Result<()> {
    const TARGET_RPS: usize = 50_000;
    const DURATION_SECS: u64 = 10;
    const CONCURRENT_CLIENTS: usize = 500;

    println!("\n🚀 Starting Peak Burst Test: 50,000 orders/sec for 10s");

    let metrics = Arc::new(LoadTestMetrics::new());
    let barrier = Arc::new(Barrier::new(CONCURRENT_CLIENTS));
    let mut join_set = JoinSet::new();

    // Spawn all clients at once for burst
    for client_id in 0..CONCURRENT_CLIENTS {
        let metrics = Arc::clone(&metrics);
        let barrier = Arc::clone(&barrier);

        join_set.spawn(async move {
            let mut client = create_client().await?;

            // Wait for all clients to be ready
            barrier.wait().await;

            let start = Instant::now();
            let mut order_count = 0u64;

            while start.elapsed().as_secs() < DURATION_SECS {
                let symbol = format!("BURST{:04}", client_id % 100);
                match submit_order(&mut client, symbol, order_count).await {
                    Ok(duration) => metrics.record_request(duration, true),
                    Err(e) => {
                        tracing::warn!("Burst order failed: {:?}", e);
                        metrics.record_request(Duration::from_micros(0), false);
                    }
                }

                order_count += 1;

                // Each client sends ~100 orders/sec (500 clients * 100 = 50k/sec)
                tokio::time::sleep(Duration::from_micros(10_000)).await;
            }

            Ok::<_, anyhow::Error>(())
        });
    }

    // Wait for all clients
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            tracing::error!("Burst client failed: {:?}", e);
        }
    }

    let report = metrics.report();
    report.print("Peak Burst: 50,000 orders/sec");

    // Assertions (more lenient for burst)
    assert!(
        report.throughput_per_sec >= 40_000.0,
        "Burst throughput too low: {:.2} req/sec (expected >= 40,000)",
        report.throughput_per_sec
    );
    assert!(
        report.error_rate_percent < 5.0,
        "Burst error rate too high: {:.2}% (expected < 5%)",
        report.error_rate_percent
    );

    Ok(())
}

/// Test 3: 1M concurrent market data updates (streaming)
#[tokio::test]
#[ignore]
async fn test_1m_market_data_streaming() -> Result<()> {
    const NUM_STREAMS: usize = 1000;
    const UPDATES_PER_STREAM: usize = 1000;
    const DURATION_SECS: u64 = 30;

    println!("\n🚀 Starting Market Data Streaming: 1M updates across {} streams", NUM_STREAMS);

    let metrics = Arc::new(LoadTestMetrics::new());
    let update_count = Arc::new(AtomicUsize::new(0));
    let mut join_set = JoinSet::new();

    for stream_id in 0..NUM_STREAMS {
        let metrics = Arc::clone(&metrics);
        let update_count = Arc::clone(&update_count);

        join_set.spawn(async move {
            let mut client = create_client().await?;
            let symbols = vec![format!("STREAM{:04}", stream_id % 100)];

            let request = StreamMarketDataRequest { symbols };
            let start = Instant::now();

            match client.stream_market_data(request).await {
                Ok(response) => {
                    let mut stream = response.into_inner();
                    let mut count = 0;

                    while let Ok(Some(_event)) = stream.message().await {
                        count += 1;
                        update_count.fetch_add(1, Ordering::Relaxed);

                        if count >= UPDATES_PER_STREAM || start.elapsed().as_secs() >= DURATION_SECS
                        {
                            break;
                        }
                    }

                    metrics.record_request(start.elapsed(), true);
                }
                Err(e) => {
                    tracing::warn!("Stream {} failed: {:?}", stream_id, e);
                    metrics.record_request(start.elapsed(), false);
                }
            }

            Ok::<_, anyhow::Error>(())
        });
    }

    // Wait for all streams
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            tracing::error!("Stream task failed: {:?}", e);
        }
    }

    let total_updates = update_count.load(Ordering::Relaxed);
    let report = metrics.report();

    println!("\n📊 Market Data Streaming Results:");
    println!("Total Updates Received: {}", total_updates);
    println!("Active Streams: {}", NUM_STREAMS);
    report.print("Market Data Streaming");

    // Assertions
    assert!(
        total_updates >= 900_000,
        "Not enough updates received: {} (expected >= 900k)",
        total_updates
    );

    Ok(())
}

/// Test 4: Connection pool saturation (1000 concurrent clients)
#[tokio::test]
#[ignore]
async fn test_connection_pool_saturation() -> Result<()> {
    const NUM_CLIENTS: usize = 1000;
    const REQUESTS_PER_CLIENT: usize = 100;

    println!("\n🚀 Starting Connection Pool Saturation: {} concurrent clients", NUM_CLIENTS);

    let metrics = Arc::new(LoadTestMetrics::new());
    let barrier = Arc::new(Barrier::new(NUM_CLIENTS));
    let mut join_set = JoinSet::new();

    // Create all clients simultaneously
    for client_id in 0..NUM_CLIENTS {
        let metrics = Arc::clone(&metrics);
        let barrier = Arc::clone(&barrier);

        join_set.spawn(async move {
            // Connect client
            let connect_start = Instant::now();
            let mut client = match create_client().await {
                Ok(c) => {
                    metrics.record_request(connect_start.elapsed(), true);
                    c
                }
                Err(e) => {
                    tracing::error!("Client {} connection failed: {:?}", client_id, e);
                    metrics.record_request(connect_start.elapsed(), false);
                    return Ok::<_, anyhow::Error>(());
                }
            };

            // Wait for all clients to connect
            barrier.wait().await;

            // Submit requests
            for req_id in 0..REQUESTS_PER_CLIENT {
                let symbol = format!("POOL{:04}", client_id % 100);
                match submit_order(&mut client, symbol, req_id as u64).await {
                    Ok(duration) => metrics.record_request(duration, true),
                    Err(e) => {
                        tracing::warn!("Client {} request {} failed: {:?}", client_id, req_id, e);
                        metrics.record_request(Duration::from_micros(0), false);
                    }
                }

                tokio::time::sleep(Duration::from_millis(10)).await;
            }

            Ok::<_, anyhow::Error>(())
        });
    }

    // Wait for all clients
    while let Some(result) = join_set.join_next().await {
        if let Err(e) = result {
            tracing::error!("Connection pool client failed: {:?}", e);
        }
    }

    let report = metrics.report();
    report.print("Connection Pool Saturation");

    // Assertions
    assert!(
        report.error_rate_percent < 5.0,
        "Connection pool error rate too high: {:.2}% (expected < 5%)",
        report.error_rate_percent
    );
    assert!(
        report.latency_p99_us < 100_000,
        "P99 latency too high under pool saturation: {} μs (expected < 100ms)",
        report.latency_p99_us
    );

    Ok(())
}

/// Integration test: Run all throughput tests sequentially
#[tokio::test]
#[ignore]
async fn test_comprehensive_throughput_suite() -> Result<()> {
    println!("\n{'='*80}");
    println!("🎯 Comprehensive Throughput Validation Suite");
    println!("{'='*80}\n");

    // Test 1: Sustained load
    println!("Test 1/4: Sustained Load");
    test_sustained_load_10k_orders().await?;

    // Cool down
    tokio::time::sleep(Duration::from_secs(5)).await;

    // Test 2: Peak burst
    println!("\nTest 2/4: Peak Burst");
    test_peak_burst_50k_orders().await?;

    // Cool down
    tokio::time::sleep(Duration::from_secs(5)).await;

    // Test 3: Market data streaming
    println!("\nTest 3/4: Market Data Streaming");
    test_1m_market_data_streaming().await?;

    // Cool down
    tokio::time::sleep(Duration::from_secs(5)).await;

    // Test 4: Connection pool
    println!("\nTest 4/4: Connection Pool Saturation");
    test_connection_pool_saturation().await?;

    println!("\n{'='*80}");
    println!("✅ All throughput tests completed successfully!");
    println!("{'='*80}\n");

    Ok(())
}

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

    #[test]
    fn test_metrics_initialization() {
        let metrics = LoadTestMetrics::new();
        assert_eq!(metrics.total_requests.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.successful_requests.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.failed_requests.load(Ordering::Relaxed), 0);
    }

    #[test]
    fn test_metrics_recording() {
        let metrics = LoadTestMetrics::new();

        metrics.record_request(Duration::from_micros(100), true);
        metrics.record_request(Duration::from_micros(200), false);

        assert_eq!(metrics.total_requests.load(Ordering::Relaxed), 2);
        assert_eq!(metrics.successful_requests.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.failed_requests.load(Ordering::Relaxed), 1);

        let report = metrics.report();
        assert_eq!(report.total_requests, 2);
        assert_eq!(report.error_rate_percent, 50.0);
    }
}