foxhunt/testing/api-load/src/scenarios/stress_test.rs

use anyhow::Result;
use tokio::sync::mpsc;
use tokio::task::JoinSet;

use crate::clients::{AuthenticatedClient, MixedWorkloadClient};
use crate::metrics::{collector::MetricsCollector, LoadTestReport, TestConfig};

pub async fn run(
    gateway_url: String,
    initial_clients: usize,
    increment: usize,
    increment_interval_secs: u64,
    max_p99_latency_ms: f64,
    max_error_rate_pct: f64,
) -> Result<LoadTestReport> {
    tracing::info!(
        "Starting STRESS test: initial {} clients, increment by {} every {}s",
        initial_clients,
        increment,
        increment_interval_secs
    );

    let (metrics_tx, metrics_rx) = mpsc::unbounded_channel();
    let mut collector = MetricsCollector::new(metrics_rx);

    // Spawn collector task
    let collector_handle = tokio::spawn(async move {
        collector.run(initial_clients).await;
        collector
    });

    let mut join_set = JoinSet::new();
    #[allow(unused_assignments)]
    let mut current_clients = 0usize;
    #[allow(unused_assignments)]
    let mut max_clients_reached = initial_clients;
    let increment_duration = std::time::Duration::from_secs(increment_interval_secs);
    let test_start = std::time::Instant::now();

    // Helper to spawn a batch of clients
    let spawn_clients = |join_set: &mut JoinSet<Result<()>>,
                         metrics_tx: &mpsc::UnboundedSender<crate::metrics::RequestMetric>,
                         gateway_url: &String,
                         start_id: usize,
                         count: usize| {
        for i in 0..count {
            let client_id = start_id + i;
            let gateway_url = gateway_url.clone();
            let metrics_tx = metrics_tx.clone();

            join_set.spawn(async move {
                let auth_client = AuthenticatedClient::new(
                    gateway_url,
                    "test-secret-key-for-load-testing",
                    &format!("user-{}", client_id),
                    &format!("loadtest-user-{}", client_id),
                )
                .await?;

                let mut mixed_client = MixedWorkloadClient::new(auth_client, client_id, metrics_tx);

                // Run for a long time (clients will be terminated when threshold is hit)
                mixed_client
                    .run_mixed_workload(std::time::Duration::from_secs(3600))
                    .await
            });
        }
    };

    // Spawn initial batch
    spawn_clients(
        &mut join_set,
        &metrics_tx,
        &gateway_url,
        0_usize,
        initial_clients,
    );
    current_clients = initial_clients;

    tracing::info!("Initial {} clients spawned", current_clients);

    // Incremental load increase loop
    loop {
        // Wait for increment interval
        tokio::time::sleep(increment_duration).await;

        // Check current performance metrics (simplified - in real scenario, query collector)
        // For now, we'll use a heuristic: if we've had any errors in recent metrics, consider stopping
        // In production, you'd pull latest metrics from collector via a channel

        // Spawn next batch
        spawn_clients(
            &mut join_set,
            &metrics_tx,
            &gateway_url,
            current_clients,
            increment,
        );
        current_clients += increment;
        max_clients_reached = current_clients;

        tracing::info!(
            "Increased load: {} active clients (elapsed: {}s)",
            current_clients,
            test_start.elapsed().as_secs()
        );

        // In a real implementation, we'd check live metrics here
        // For demonstration, we'll run until a maximum (e.g., 5000 clients or 10 minutes)
        if current_clients >= 5000 || test_start.elapsed().as_secs() >= 600 {
            tracing::info!(
                "Stress test limit reached: {} clients or 10 minutes elapsed",
                current_clients
            );
            break;
        }
    }

    // Let the current load run for one more interval to measure performance
    tracing::info!(
        "Sustaining max load ({} clients) for {}s to measure breaking point...",
        current_clients,
        increment_interval_secs
    );
    tokio::time::sleep(increment_duration).await;

    // Abort all client tasks
    join_set.shutdown().await;

    // Drop the sender to signal collector to finish
    drop(metrics_tx);

    // Wait for collector to finish and get the report
    let collector = collector_handle.await?;
    let mut report = collector.generate_report(
        "Stress Test".to_owned(),
        TestConfig {
            num_clients: max_clients_reached,
            duration_secs: test_start.elapsed().as_secs(),
            test_type: "stress_test".to_owned(),
        },
    );

    // Determine breaking point and capacity limits
    let breaking_point_identified = report.metrics.error_rate_pct >= max_error_rate_pct
        || report.metrics.latency_stats.p99_ms >= max_p99_latency_ms;

    let recommended_max_clients = if breaking_point_identified {
        // Recommend 80% of breaking point as safe limit
        usize::try_from(
            (f64::from(u32::try_from(max_clients_reached).unwrap_or(u32::MAX)) * 0.8) as u64,
        )
        .unwrap_or(usize::MAX)
    } else {
        max_clients_reached
    };

    let bottleneck = if report.metrics.error_rate_pct >= max_error_rate_pct {
        Some(format!(
            "Error rate threshold exceeded: {:.2}% (max: {:.2}%)",
            report.metrics.error_rate_pct, max_error_rate_pct
        ))
    } else if report.metrics.latency_stats.p99_ms >= max_p99_latency_ms {
        Some(format!(
            "Latency threshold exceeded: {:.2}ms P99 (max: {:.2}ms)",
            report.metrics.latency_stats.p99_ms, max_p99_latency_ms
        ))
    } else {
        Some("Test limit reached without failure".to_owned())
    };

    report.capacity_recommendation = Some(crate::metrics::CapacityRecommendation {
        max_sustainable_clients: recommended_max_clients,
        max_sustainable_rps: report.metrics.requests_per_second * 0.8,
        bottleneck_identified: bottleneck.clone(),
        recommendation: if breaking_point_identified {
            format!(
                "BREAKING POINT IDENTIFIED at {} clients. {}\n\
                 Recommended production capacity: {} concurrent clients ({} RPS).\n\
                 System degraded with {:.2}% error rate and {:.2}ms P99 latency.",
                max_clients_reached,
                bottleneck.unwrap_or_default(),
                recommended_max_clients,
                usize::try_from((report.metrics.requests_per_second * 0.8) as u64)
                    .unwrap_or(usize::MAX),
                report.metrics.error_rate_pct,
                report.metrics.latency_stats.p99_ms
            )
        } else {
            format!(
                "System sustained {} clients without failure (test limit reached).\n\
                 Error rate: {:.2}%, P99 latency: {:.2}ms.\n\
                 Breaking point not identified within test constraints. \
                 System capacity exceeds {} concurrent clients.",
                max_clients_reached,
                report.metrics.error_rate_pct,
                report.metrics.latency_stats.p99_ms,
                max_clients_reached
            )
        },
    });

    tracing::info!("Stress test completed: {:?}", report.metrics);

    Ok(report)
}