lobe_core/engine/

burst.rs

//! Latency-under-contention probe: hold N requests in flight while pushing a
//! fixed total through, each measured with the normal phase decomposition.
//! Deliberately not a throughput benchmark — the output is the latency
//! distribution per concurrency level, not requests/second.

use std::sync::Arc;
use std::time::Instant;

use tokio::sync::Semaphore;

use crate::engine::bimodal::{detect_bimodal, BimodalDetection};
use crate::engine::proxy::SimpleProbe;
use crate::error::Result;
use crate::models::{NewProbeResult, ProbeStatus};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PhasePercentiles {
    pub p50_ms: u64,
    pub p95_ms: u64,
}

#[derive(Debug, Clone)]
pub struct BurstSummary {
    pub concurrency: usize,
    pub total_requests: usize,
    pub succeeded: usize,
    pub failed: usize,
    pub dns: PhasePercentiles,
    pub tcp: PhasePercentiles,
    pub tls: PhasePercentiles,
    pub ttfb: PhasePercentiles,
    pub total: PhasePercentiles,
    /// Fast/slow cohort split on TTFB across successful requests, when the
    /// distribution is bimodal (e.g. connection-pool exhaustion).
    pub bimodal_ttfb: Option<BimodalDetection>,
    /// First error message observed, for surfacing failures without dumping
    /// every result.
    pub first_error: Option<String>,
    pub elapsed_ms: u64,
}

/// Run `total_requests` probes against `target`, holding at most
/// `concurrency` in flight, and aggregate the phase distributions.
pub async fn run_burst(
    probe: &SimpleProbe,
    target: &str,
    method: &str,
    concurrency: usize,
    total_requests: usize,
) -> Result<BurstSummary> {
    let semaphore = Arc::new(Semaphore::new(concurrency.max(1)));
    let started = Instant::now();

    let mut handles = Vec::with_capacity(total_requests);
    for _ in 0..total_requests {
        let semaphore = Arc::clone(&semaphore);
        let probe = probe.clone();
        let target = target.to_string();
        let method = method.to_string();

        handles.push(tokio::spawn(async move {
            let _permit = semaphore.acquire_owned().await;
            probe.probe_with_method(&target, &method).await
        }));
    }

    let mut results = Vec::with_capacity(total_requests);
    for handle in handles {
        match handle.await {
            Ok(result) => results.push(result?),
            Err(join_error) => {
                return Err(crate::error::TloxError::Io(std::io::Error::other(
                    join_error.to_string(),
                )))
            }
        }
    }

    Ok(summarize_burst(
        concurrency,
        &results,
        started.elapsed().as_millis() as u64,
    ))
}

fn summarize_burst(
    concurrency: usize,
    results: &[NewProbeResult],
    elapsed_ms: u64,
) -> BurstSummary {
    let succeeded: Vec<&NewProbeResult> = results
        .iter()
        .filter(|result| result.status == ProbeStatus::Succeeded)
        .collect();
    let failed = results.len() - succeeded.len();
    let first_error = results
        .iter()
        .find_map(|result| result.error_message.clone());

    let phase = |selector: fn(&NewProbeResult) -> u64| {
        let values: Vec<u64> = succeeded.iter().map(|result| selector(result)).collect();
        PhasePercentiles {
            p50_ms: percentile_u64(&values, 0.50),
            p95_ms: percentile_u64(&values, 0.95),
        }
    };

    let ttfb_values: Vec<u64> = succeeded
        .iter()
        .map(|result| result.report.ttfb_ms)
        .collect();

    BurstSummary {
        concurrency,
        total_requests: results.len(),
        succeeded: succeeded.len(),
        failed,
        dns: phase(|result| result.report.dns_ms),
        tcp: phase(|result| result.report.tcp_ms),
        tls: phase(|result| result.report.tls_ms),
        ttfb: phase(|result| result.report.ttfb_ms),
        total: phase(|result| result.report.total_ms),
        bimodal_ttfb: detect_bimodal(&ttfb_values),
        first_error,
        elapsed_ms,
    }
}

fn percentile_u64(values: &[u64], percentile: f64) -> u64 {
    if values.is_empty() {
        return 0;
    }

    let mut sorted = values.to_vec();
    sorted.sort_unstable();
    let index = ((sorted.len() - 1) as f64 * percentile).round() as usize;
    sorted[index]
}

#[cfg(test)]
mod tests {
    use crate::models::{NewProbeResult, ProbeStatus, TimingReport};

    use super::{percentile_u64, summarize_burst};

    fn result_with_ttfb(ttfb_ms: u64, status: ProbeStatus) -> NewProbeResult {
        NewProbeResult {
            target: "http://example.test".to_string(),
            request_path: "/".to_string(),
            report: TimingReport {
                ttfb_ms,
                total_ms: ttfb_ms + 5,
                ..TimingReport::default()
            },
            status,
            response_status_code: (status == ProbeStatus::Succeeded).then_some(200),
            response_status_text: None,
            response_bytes: None,
            error_message: (status == ProbeStatus::Failed)
                .then(|| "connection refused".to_string()),
            tls: None,
        }
    }

    #[test]
    fn percentile_u64_handles_empty_and_single_values() {
        assert_eq!(percentile_u64(&[], 0.95), 0);
        assert_eq!(percentile_u64(&[42], 0.50), 42);
        assert_eq!(percentile_u64(&[42], 0.95), 42);
    }

    #[test]
    fn summarize_burst_excludes_failures_from_percentiles() {
        let mut results: Vec<NewProbeResult> = (0..10)
            .map(|i| result_with_ttfb(40 + i, ProbeStatus::Succeeded))
            .collect();
        results.push(result_with_ttfb(0, ProbeStatus::Failed));

        let summary = summarize_burst(5, &results, 1000);

        assert_eq!(summary.total_requests, 11);
        assert_eq!(summary.succeeded, 10);
        assert_eq!(summary.failed, 1);
        assert!(summary.ttfb.p50_ms >= 40);
        assert_eq!(summary.first_error.as_deref(), Some("connection refused"));
    }

    #[test]
    fn summarize_burst_flags_bimodal_ttfb() {
        let mut results: Vec<NewProbeResult> = (0..20)
            .map(|i| result_with_ttfb(38 + (i % 6), ProbeStatus::Succeeded))
            .collect();
        results.extend((0..40).map(|i| result_with_ttfb(590 + (i % 30), ProbeStatus::Succeeded)));

        let summary = summarize_burst(50, &results, 5000);

        let detection = summary.bimodal_ttfb.expect("split should be detected");
        assert!(detection.fast.median_ms < 50);
        assert!(detection.slow.median_ms > 500);
    }
}