reth_bench_compare/

comparison.rs

//! Results comparison and report generation.

use crate::cli::Args;
use chrono::{DateTime, Utc};
use csv::Reader;
use eyre::{eyre, Result, WrapErr};
use serde::{Deserialize, Serialize};
use std::{
    cmp::Ordering,
    collections::HashMap,
    fs,
    path::{Path, PathBuf},
};
use tracing::{info, warn};

/// Manages comparison between baseline and feature reference results
pub(crate) struct ComparisonGenerator {
    output_dir: PathBuf,
    timestamp: String,
    baseline_ref_name: String,
    feature_ref_name: String,
    baseline_results: Option<BenchmarkResults>,
    feature_results: Option<BenchmarkResults>,
    baseline_command: Option<String>,
    feature_command: Option<String>,
}

/// Represents the results from a single benchmark run
#[derive(Debug, Clone)]
pub(crate) struct BenchmarkResults {
    pub ref_name: String,
    pub combined_latency_data: Vec<CombinedLatencyRow>,
    pub summary: BenchmarkSummary,
    pub start_timestamp: Option<DateTime<Utc>>,
    pub end_timestamp: Option<DateTime<Utc>>,
}

/// Combined latency CSV row structure
#[derive(Debug, Clone, Deserialize, Serialize)]
pub(crate) struct CombinedLatencyRow {
    pub block_number: u64,
    pub transaction_count: u64,
    pub gas_used: u64,
    pub new_payload_latency: u128,
}

/// Total gas CSV row structure
#[derive(Debug, Clone, Deserialize, Serialize)]
pub(crate) struct TotalGasRow {
    pub block_number: u64,
    pub transaction_count: u64,
    pub gas_used: u64,
    pub time: u128,
}

/// Summary statistics for a benchmark run.
///
/// Latencies are derived from per-block `engine_newPayload` timings (converted from µs to ms):
/// - `mean_new_payload_latency_ms`: arithmetic mean latency across blocks.
/// - `median_new_payload_latency_ms`: p50 latency across blocks.
/// - `p90_new_payload_latency_ms` / `p99_new_payload_latency_ms`: tail latencies across blocks.
#[derive(Debug, Clone, Serialize)]
pub(crate) struct BenchmarkSummary {
    pub total_blocks: u64,
    pub total_gas_used: u64,
    pub total_duration_ms: u128,
    pub mean_new_payload_latency_ms: f64,
    pub median_new_payload_latency_ms: f64,
    pub p90_new_payload_latency_ms: f64,
    pub p99_new_payload_latency_ms: f64,
    pub gas_per_second: f64,
    pub blocks_per_second: f64,
    pub min_block_number: u64,
    pub max_block_number: u64,
}

/// Comparison report between two benchmark runs
#[derive(Debug, Serialize)]
pub(crate) struct ComparisonReport {
    pub timestamp: String,
    pub baseline: RefInfo,
    pub feature: RefInfo,
    pub comparison_summary: ComparisonSummary,
    pub per_block_comparisons: Vec<BlockComparison>,
}

/// Information about a reference in the comparison
#[derive(Debug, Serialize)]
pub(crate) struct RefInfo {
    pub ref_name: String,
    pub summary: BenchmarkSummary,
    pub start_timestamp: Option<DateTime<Utc>>,
    pub end_timestamp: Option<DateTime<Utc>>,
    pub reth_command: Option<String>,
}

/// Summary of the comparison between references.
///
/// Percent deltas are `(feature - baseline) / baseline * 100`:
/// - `new_payload_latency_p50_change_percent` / p90 / p99: percent changes of the respective
///   per-block percentiles.
/// - `per_block_latency_change_mean_percent` / `per_block_latency_change_median_percent` are the
///   mean and median of per-block percent deltas (feature vs baseline), capturing block-level
///   drift.
/// - `per_block_latency_change_std_dev_percent`: standard deviation of per-block percent changes,
///   measuring consistency of performance changes across blocks.
/// - `new_payload_total_latency_change_percent` is the percent change of the total newPayload time
///   across the run.
///
/// Positive means slower/higher; negative means faster/lower.
#[derive(Debug, Serialize)]
pub(crate) struct ComparisonSummary {
    pub per_block_latency_change_mean_percent: f64,
    pub per_block_latency_change_median_percent: f64,
    pub per_block_latency_change_std_dev_percent: f64,
    pub new_payload_total_latency_change_percent: f64,
    pub new_payload_latency_p50_change_percent: f64,
    pub new_payload_latency_p90_change_percent: f64,
    pub new_payload_latency_p99_change_percent: f64,
    pub gas_per_second_change_percent: f64,
    pub blocks_per_second_change_percent: f64,
}

/// Per-block comparison data
#[derive(Debug, Serialize)]
pub(crate) struct BlockComparison {
    pub block_number: u64,
    pub transaction_count: u64,
    pub gas_used: u64,
    pub baseline_new_payload_latency: u128,
    pub feature_new_payload_latency: u128,
    pub new_payload_latency_change_percent: f64,
}

impl ComparisonGenerator {
    /// Create a new comparison generator
    pub(crate) fn new(args: &Args) -> Self {
        let now: DateTime<Utc> = Utc::now();
        let timestamp = now.format("%Y%m%d_%H%M%S").to_string();

        Self {
            output_dir: args.output_dir_path(),
            timestamp,
            baseline_ref_name: args.baseline_ref.clone(),
            feature_ref_name: args.feature_ref.clone(),
            baseline_results: None,
            feature_results: None,
            baseline_command: None,
            feature_command: None,
        }
    }

    /// Get the output directory for a specific reference
    pub(crate) fn get_ref_output_dir(&self, ref_type: &str) -> PathBuf {
        self.output_dir.join("results").join(&self.timestamp).join(ref_type)
    }

    /// Get the main output directory for this comparison run
    pub(crate) fn get_output_dir(&self) -> PathBuf {
        self.output_dir.join("results").join(&self.timestamp)
    }

    /// Add benchmark results for a reference
    pub(crate) fn add_ref_results(&mut self, ref_type: &str, output_path: &Path) -> Result<()> {
        let ref_name = match ref_type {
            "baseline" => &self.baseline_ref_name,
            "feature" => &self.feature_ref_name,
            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
        };

        let results = self.load_benchmark_results(ref_name, output_path)?;

        match ref_type {
            "baseline" => self.baseline_results = Some(results),
            "feature" => self.feature_results = Some(results),
            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
        }

        info!("Loaded benchmark results for {} reference", ref_type);

        Ok(())
    }

    /// Set the benchmark run timestamps for a reference
    pub(crate) fn set_ref_timestamps(
        &mut self,
        ref_type: &str,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<()> {
        match ref_type {
            "baseline" => {
                if let Some(ref mut results) = self.baseline_results {
                    results.start_timestamp = Some(start);
                    results.end_timestamp = Some(end);
                } else {
                    return Err(eyre!("Baseline results not loaded yet"));
                }
            }
            "feature" => {
                if let Some(ref mut results) = self.feature_results {
                    results.start_timestamp = Some(start);
                    results.end_timestamp = Some(end);
                } else {
                    return Err(eyre!("Feature results not loaded yet"));
                }
            }
            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
        }

        Ok(())
    }

    /// Set the reth command for a reference
    pub(crate) fn set_ref_command(&mut self, ref_type: &str, command: String) -> Result<()> {
        match ref_type {
            "baseline" => {
                self.baseline_command = Some(command);
            }
            "feature" => {
                self.feature_command = Some(command);
            }
            _ => return Err(eyre!("Unknown reference type: {}", ref_type)),
        }

        Ok(())
    }

    /// Generate the final comparison report
    pub(crate) async fn generate_comparison_report(&self) -> Result<()> {
        info!("Generating comparison report...");

        let baseline =
            self.baseline_results.as_ref().ok_or_else(|| eyre!("Baseline results not loaded"))?;

        let feature =
            self.feature_results.as_ref().ok_or_else(|| eyre!("Feature results not loaded"))?;

        let per_block_comparisons = self.calculate_per_block_comparisons(baseline, feature)?;
        let comparison_summary = self.calculate_comparison_summary(
            &baseline.summary,
            &feature.summary,
            &per_block_comparisons,
        )?;

        let report = ComparisonReport {
            timestamp: self.timestamp.clone(),
            baseline: RefInfo {
                ref_name: baseline.ref_name.clone(),
                summary: baseline.summary.clone(),
                start_timestamp: baseline.start_timestamp,
                end_timestamp: baseline.end_timestamp,
                reth_command: self.baseline_command.clone(),
            },
            feature: RefInfo {
                ref_name: feature.ref_name.clone(),
                summary: feature.summary.clone(),
                start_timestamp: feature.start_timestamp,
                end_timestamp: feature.end_timestamp,
                reth_command: self.feature_command.clone(),
            },
            comparison_summary,
            per_block_comparisons,
        };

        // Write reports
        self.write_comparison_reports(&report).await?;

        // Print summary to console
        self.print_comparison_summary(&report);

        Ok(())
    }

    /// Load benchmark results from CSV files
    fn load_benchmark_results(
        &self,
        ref_name: &str,
        output_path: &Path,
    ) -> Result<BenchmarkResults> {
        let combined_latency_path = output_path.join("combined_latency.csv");
        let total_gas_path = output_path.join("total_gas.csv");

        let combined_latency_data = self.load_combined_latency_csv(&combined_latency_path)?;
        let total_gas_data = self.load_total_gas_csv(&total_gas_path)?;

        let summary = self.calculate_summary(&combined_latency_data, &total_gas_data)?;

        Ok(BenchmarkResults {
            ref_name: ref_name.to_string(),
            combined_latency_data,
            summary,
            start_timestamp: None,
            end_timestamp: None,
        })
    }

    /// Load combined latency CSV data
    fn load_combined_latency_csv(&self, path: &Path) -> Result<Vec<CombinedLatencyRow>> {
        let mut reader = Reader::from_path(path)
            .wrap_err_with(|| format!("Failed to open combined latency CSV: {path:?}"))?;

        let mut rows = Vec::new();
        for result in reader.deserialize() {
            let row: CombinedLatencyRow = result
                .wrap_err_with(|| format!("Failed to parse combined latency row in {path:?}"))?;
            rows.push(row);
        }

        if rows.is_empty() {
            return Err(eyre!("No data found in combined latency CSV: {:?}", path));
        }

        Ok(rows)
    }

    /// Load total gas CSV data
    fn load_total_gas_csv(&self, path: &Path) -> Result<Vec<TotalGasRow>> {
        let mut reader = Reader::from_path(path)
            .wrap_err_with(|| format!("Failed to open total gas CSV: {path:?}"))?;

        let mut rows = Vec::new();
        for result in reader.deserialize() {
            let row: TotalGasRow =
                result.wrap_err_with(|| format!("Failed to parse total gas row in {path:?}"))?;
            rows.push(row);
        }

        if rows.is_empty() {
            return Err(eyre!("No data found in total gas CSV: {:?}", path));
        }

        Ok(rows)
    }

    /// Calculate summary statistics for a benchmark run.
    ///
    /// Computes latency statistics from per-block `new_payload_latency` values in `combined_data`
    /// (converting from µs to ms), and throughput metrics using the total run duration from
    /// `total_gas_data`. Percentiles (p50/p90/p99) use linear interpolation on sorted latencies.
    fn calculate_summary(
        &self,
        combined_data: &[CombinedLatencyRow],
        total_gas_data: &[TotalGasRow],
    ) -> Result<BenchmarkSummary> {
        if combined_data.is_empty() || total_gas_data.is_empty() {
            return Err(eyre!("Cannot calculate summary for empty data"));
        }

        let total_blocks = combined_data.len() as u64;
        let total_gas_used: u64 = combined_data.iter().map(|r| r.gas_used).sum();

        let total_duration_ms = total_gas_data.last().unwrap().time / 1000; // Convert microseconds to milliseconds

        let latencies_ms: Vec<f64> =
            combined_data.iter().map(|r| r.new_payload_latency as f64 / 1000.0).collect();
        let mean_new_payload_latency_ms: f64 =
            latencies_ms.iter().sum::<f64>() / total_blocks as f64;

        let mut sorted_latencies_ms = latencies_ms;
        sorted_latencies_ms.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal));
        let median_new_payload_latency_ms = percentile(&sorted_latencies_ms, 0.5);
        let p90_new_payload_latency_ms = percentile(&sorted_latencies_ms, 0.9);
        let p99_new_payload_latency_ms = percentile(&sorted_latencies_ms, 0.99);

        let total_duration_seconds = total_duration_ms as f64 / 1000.0;
        let gas_per_second = if total_duration_seconds > f64::EPSILON {
            total_gas_used as f64 / total_duration_seconds
        } else {
            0.0
        };

        let blocks_per_second = if total_duration_seconds > f64::EPSILON {
            total_blocks as f64 / total_duration_seconds
        } else {
            0.0
        };

        let min_block_number = combined_data.first().unwrap().block_number;
        let max_block_number = combined_data.last().unwrap().block_number;

        Ok(BenchmarkSummary {
            total_blocks,
            total_gas_used,
            total_duration_ms,
            mean_new_payload_latency_ms,
            median_new_payload_latency_ms,
            p90_new_payload_latency_ms,
            p99_new_payload_latency_ms,
            gas_per_second,
            blocks_per_second,
            min_block_number,
            max_block_number,
        })
    }

    /// Calculate comparison summary between baseline and feature
    fn calculate_comparison_summary(
        &self,
        baseline: &BenchmarkSummary,
        feature: &BenchmarkSummary,
        per_block_comparisons: &[BlockComparison],
    ) -> Result<ComparisonSummary> {
        let calc_percent_change = |baseline: f64, feature: f64| -> f64 {
            if baseline.abs() > f64::EPSILON {
                ((feature - baseline) / baseline) * 100.0
            } else {
                0.0
            }
        };

        // Calculate per-block statistics. "Per-block" means: for each block, compute the percent
        // change (feature - baseline) / baseline * 100, then calculate statistics across those
        // per-block percent changes. This captures how consistently the feature performs relative
        // to baseline across all blocks.
        let per_block_percent_changes: Vec<f64> =
            per_block_comparisons.iter().map(|c| c.new_payload_latency_change_percent).collect();
        let per_block_latency_change_mean_percent = if per_block_percent_changes.is_empty() {
            0.0
        } else {
            per_block_percent_changes.iter().sum::<f64>() / per_block_percent_changes.len() as f64
        };
        let per_block_latency_change_median_percent = if per_block_percent_changes.is_empty() {
            0.0
        } else {
            let mut sorted = per_block_percent_changes.clone();
            sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal));
            percentile(&sorted, 0.5)
        };
        let per_block_latency_change_std_dev_percent =
            calculate_std_dev(&per_block_percent_changes, per_block_latency_change_mean_percent);

        let baseline_total_latency_ms =
            baseline.mean_new_payload_latency_ms * baseline.total_blocks as f64;
        let feature_total_latency_ms =
            feature.mean_new_payload_latency_ms * feature.total_blocks as f64;
        let new_payload_total_latency_change_percent =
            calc_percent_change(baseline_total_latency_ms, feature_total_latency_ms);

        Ok(ComparisonSummary {
            per_block_latency_change_mean_percent,
            per_block_latency_change_median_percent,
            per_block_latency_change_std_dev_percent,
            new_payload_total_latency_change_percent,
            new_payload_latency_p50_change_percent: calc_percent_change(
                baseline.median_new_payload_latency_ms,
                feature.median_new_payload_latency_ms,
            ),
            new_payload_latency_p90_change_percent: calc_percent_change(
                baseline.p90_new_payload_latency_ms,
                feature.p90_new_payload_latency_ms,
            ),
            new_payload_latency_p99_change_percent: calc_percent_change(
                baseline.p99_new_payload_latency_ms,
                feature.p99_new_payload_latency_ms,
            ),
            gas_per_second_change_percent: calc_percent_change(
                baseline.gas_per_second,
                feature.gas_per_second,
            ),
            blocks_per_second_change_percent: calc_percent_change(
                baseline.blocks_per_second,
                feature.blocks_per_second,
            ),
        })
    }

    /// Calculate per-block comparisons
    fn calculate_per_block_comparisons(
        &self,
        baseline: &BenchmarkResults,
        feature: &BenchmarkResults,
    ) -> Result<Vec<BlockComparison>> {
        let mut baseline_map: HashMap<u64, &CombinedLatencyRow> = HashMap::new();
        for row in &baseline.combined_latency_data {
            baseline_map.insert(row.block_number, row);
        }

        let mut comparisons = Vec::new();
        for feature_row in &feature.combined_latency_data {
            if let Some(baseline_row) = baseline_map.get(&feature_row.block_number) {
                let calc_percent_change = |baseline: u128, feature: u128| -> f64 {
                    if baseline > 0 {
                        ((feature as f64 - baseline as f64) / baseline as f64) * 100.0
                    } else {
                        0.0
                    }
                };

                let comparison = BlockComparison {
                    block_number: feature_row.block_number,
                    transaction_count: feature_row.transaction_count,
                    gas_used: feature_row.gas_used,
                    baseline_new_payload_latency: baseline_row.new_payload_latency,
                    feature_new_payload_latency: feature_row.new_payload_latency,
                    new_payload_latency_change_percent: calc_percent_change(
                        baseline_row.new_payload_latency,
                        feature_row.new_payload_latency,
                    ),
                };
                comparisons.push(comparison);
            } else {
                warn!("Block {} not found in baseline data", feature_row.block_number);
            }
        }

        Ok(comparisons)
    }

    /// Write comparison reports to files
    async fn write_comparison_reports(&self, report: &ComparisonReport) -> Result<()> {
        let report_dir = self.output_dir.join("results").join(&self.timestamp);
        fs::create_dir_all(&report_dir)
            .wrap_err_with(|| format!("Failed to create report directory: {report_dir:?}"))?;

        // Write JSON report
        let json_path = report_dir.join("comparison_report.json");
        let json_content = serde_json::to_string_pretty(report)
            .wrap_err("Failed to serialize comparison report to JSON")?;
        fs::write(&json_path, json_content)
            .wrap_err_with(|| format!("Failed to write JSON report: {json_path:?}"))?;

        // Write CSV report for per-block comparisons
        let csv_path = report_dir.join("per_block_comparison.csv");
        let mut writer = csv::Writer::from_path(&csv_path)
            .wrap_err_with(|| format!("Failed to create CSV writer: {csv_path:?}"))?;

        for comparison in &report.per_block_comparisons {
            writer.serialize(comparison).wrap_err("Failed to write comparison row to CSV")?;
        }
        writer.flush().wrap_err("Failed to flush CSV writer")?;

        info!("Comparison reports written to: {:?}", report_dir);
        Ok(())
    }

    /// Print comparison summary to console
    fn print_comparison_summary(&self, report: &ComparisonReport) {
        // Parse and format timestamp nicely
        let formatted_timestamp = if let Ok(dt) = chrono::DateTime::parse_from_str(
            &format!("{} +0000", report.timestamp.replace('_', " ")),
            "%Y%m%d %H%M%S %z",
        ) {
            dt.format("%Y-%m-%d %H:%M:%S UTC").to_string()
        } else {
            // Fallback to original if parsing fails
            report.timestamp.clone()
        };

        println!("\n=== BENCHMARK COMPARISON SUMMARY ===");
        println!("Timestamp: {formatted_timestamp}");
        println!("Baseline: {}", report.baseline.ref_name);
        println!("Feature:  {}", report.feature.ref_name);
        println!();

        let summary = &report.comparison_summary;

        println!("Performance Changes:");
        println!(
            "  NewPayload Latency per-block mean change:   {:+.2}%",
            summary.per_block_latency_change_mean_percent
        );
        println!(
            "  NewPayload Latency per-block median change: {:+.2}%",
            summary.per_block_latency_change_median_percent
        );
        println!(
            "  NewPayload Latency per-block std dev:       {:.2}%",
            summary.per_block_latency_change_std_dev_percent
        );
        println!(
            "  Total newPayload time change:                {:+.2}%",
            summary.new_payload_total_latency_change_percent
        );
        println!(
            "  NewPayload Latency p50:           {:+.2}%",
            summary.new_payload_latency_p50_change_percent
        );
        println!(
            "  NewPayload Latency p90:           {:+.2}%",
            summary.new_payload_latency_p90_change_percent
        );
        println!(
            "  NewPayload Latency p99:           {:+.2}%",
            summary.new_payload_latency_p99_change_percent
        );
        println!(
            "  Gas/Second:                      {:+.2}%",
            summary.gas_per_second_change_percent
        );
        println!(
            "  Blocks/Second:                   {:+.2}%",
            summary.blocks_per_second_change_percent
        );
        println!();

        println!("Baseline Summary:");
        let baseline = &report.baseline.summary;
        println!(
            "  Blocks: {} (blocks {} to {}), Gas: {}, Duration: {:.2}s",
            baseline.total_blocks,
            baseline.min_block_number,
            baseline.max_block_number,
            baseline.total_gas_used,
            baseline.total_duration_ms as f64 / 1000.0
        );
        println!("  NewPayload latency (ms):");
        println!(
            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}",
            baseline.mean_new_payload_latency_ms,
            baseline.median_new_payload_latency_ms,
            baseline.p90_new_payload_latency_ms,
            baseline.p99_new_payload_latency_ms
        );
        if let (Some(start), Some(end)) =
            (&report.baseline.start_timestamp, &report.baseline.end_timestamp)
        {
            println!(
                "  Started: {}, Ended: {}",
                start.format("%Y-%m-%d %H:%M:%S UTC"),
                end.format("%Y-%m-%d %H:%M:%S UTC")
            );
        }
        if let Some(ref cmd) = report.baseline.reth_command {
            println!("  Command: {}", cmd);
        }
        println!();

        println!("Feature Summary:");
        let feature = &report.feature.summary;
        println!(
            "  Blocks: {} (blocks {} to {}), Gas: {}, Duration: {:.2}s",
            feature.total_blocks,
            feature.min_block_number,
            feature.max_block_number,
            feature.total_gas_used,
            feature.total_duration_ms as f64 / 1000.0
        );
        println!("  NewPayload latency (ms):");
        println!(
            "    mean: {:.2}, p50: {:.2}, p90: {:.2}, p99: {:.2}",
            feature.mean_new_payload_latency_ms,
            feature.median_new_payload_latency_ms,
            feature.p90_new_payload_latency_ms,
            feature.p99_new_payload_latency_ms
        );
        if let (Some(start), Some(end)) =
            (&report.feature.start_timestamp, &report.feature.end_timestamp)
        {
            println!(
                "  Started: {}, Ended: {}",
                start.format("%Y-%m-%d %H:%M:%S UTC"),
                end.format("%Y-%m-%d %H:%M:%S UTC")
            );
        }
        if let Some(ref cmd) = report.feature.reth_command {
            println!("  Command: {}", cmd);
        }
        println!();
    }
}

/// Calculate standard deviation from a set of values and their mean.
///
/// Computes the population standard deviation using the formula:
/// `sqrt(sum((x - mean)²) / n)`
///
/// Returns 0.0 for empty input.
fn calculate_std_dev(values: &[f64], mean: f64) -> f64 {
    if values.is_empty() {
        return 0.0;
    }

    let variance = values
        .iter()
        .map(|x| {
            let diff = x - mean;
            diff * diff
        })
        .sum::<f64>() /
        values.len() as f64;

    variance.sqrt()
}

/// Calculate percentile using linear interpolation on a sorted slice.
///
/// Computes `rank = percentile × (n - 1)` where n is the array length. If the rank falls
/// between two indices, linearly interpolates between those values. For example, with 100 values,
/// p90 computes rank = 0.9 × 99 = 89.1, then returns `values[89] × 0.9 + values[90] × 0.1`.
///
/// Returns 0.0 for empty input.
fn percentile(sorted_values: &[f64], percentile: f64) -> f64 {
    if sorted_values.is_empty() {
        return 0.0;
    }

    let clamped = percentile.clamp(0.0, 1.0);
    let max_index = sorted_values.len() - 1;
    let rank = clamped * max_index as f64;
    let lower = rank.floor() as usize;
    let upper = rank.ceil() as usize;

    if lower == upper {
        sorted_values[lower]
    } else {
        let weight = rank - lower as f64;
        sorted_values[lower].mul_add(1.0 - weight, sorted_values[upper] * weight)
    }
}