ic_testkit/benchmark/

mod.rs

//! Parse, pair, aggregate, compare, and report compact benchmark markers.
//!
//! The normal pipeline is [`parse_benchmark_events`],
//! [`pair_benchmark_spans`], [`aggregate_benchmark_spans`], and optionally
//! [`compare_benchmark_aggregates`] plus [`write_benchmark_report_dir`]. Marker
//! producers can use [`format_marker`] on the host or
//! [`crate::performance::Performance`] in canister code.
//!
//! Totals and run counts remain exact integers. Means use IC Metrics' exact
//! scaled integer projection, rounded down. Callers choose the scale and handle
//! projection errors; reports use scale one in the original counter units.
//! Comparisons retain current and previous means instead of percentage estimates.

use std::{
    collections::{BTreeMap, btree_map::Entry},
    fmt::Write as _,
    fs, io,
    path::{Path, PathBuf},
};

use ic_metrics::{MeasurementRatioError, checked_scaled_ratio};
use serde_json::Value;

/// Default prefix for benchmark marker lines.
pub const DEFAULT_PREFIX: &str = "ICTK";
const ALL_SUITES_LABEL: &str = "ALL";

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
enum AggregateScope {
    Suite(String),
    All,
}

impl AggregateScope {
    fn label(&self) -> &str {
        match self {
            Self::Suite(suite) => suite,
            Self::All => ALL_SUITES_LABEL,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkParserConfig {
    pub prefixes: Vec<String>,
    pub suite_derivation: SuiteDerivation,
    /// When enabled, non-empty lines without a configured marker prefix are
    /// reported as malformed markers instead of ignored log noise.
    pub strict: bool,
}

impl Default for BenchmarkParserConfig {
    fn default() -> Self {
        Self {
            prefixes: vec![DEFAULT_PREFIX.to_string()],
            suite_derivation: SuiteDerivation::FirstPathSegment,
            strict: false,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SuiteDerivation {
    FirstPathSegment,
    Fixed(String),
}

impl SuiteDerivation {
    #[must_use]
    pub fn derive_suite(&self, span_label: &str) -> String {
        match self {
            Self::FirstPathSegment => span_label
                .split('/')
                .next()
                .filter(|part| !part.is_empty())
                .unwrap_or(span_label)
                .to_string(),
            Self::Fixed(suite) => suite.clone(),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BenchmarkEventKind {
    Start,
    End,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BenchmarkEventSource {
    Unknown,
    Stdout,
    Stderr,
    FetchedLog,
}

impl BenchmarkEventSource {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Unknown => "unknown",
            Self::Stdout => "stdout",
            Self::Stderr => "stderr",
            Self::FetchedLog => "fetched_log",
        }
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct BenchmarkCounters {
    pub instructions: u128,
    pub heap_bytes: u128,
    pub memory_bytes: u128,
    pub total_allocation: u128,
}

impl BenchmarkCounters {
    fn checked_delta(self, start: Self) -> Option<Self> {
        Some(Self {
            instructions: self.instructions.checked_sub(start.instructions)?,
            heap_bytes: self.heap_bytes.checked_sub(start.heap_bytes)?,
            memory_bytes: self.memory_bytes.checked_sub(start.memory_bytes)?,
            total_allocation: self.total_allocation.checked_sub(start.total_allocation)?,
        })
    }

    fn checked_add(self, other: Self) -> Result<Self, BenchmarkCounter> {
        Ok(Self {
            instructions: self
                .instructions
                .checked_add(other.instructions)
                .ok_or(BenchmarkCounter::Instructions)?,
            heap_bytes: self
                .heap_bytes
                .checked_add(other.heap_bytes)
                .ok_or(BenchmarkCounter::HeapBytes)?,
            memory_bytes: self
                .memory_bytes
                .checked_add(other.memory_bytes)
                .ok_or(BenchmarkCounter::MemoryBytes)?,
            total_allocation: self
                .total_allocation
                .checked_add(other.total_allocation)
                .ok_or(BenchmarkCounter::TotalAllocation)?,
        })
    }

    fn min_assign(&mut self, other: Self) {
        self.instructions = self.instructions.min(other.instructions);
        self.heap_bytes = self.heap_bytes.min(other.heap_bytes);
        self.memory_bytes = self.memory_bytes.min(other.memory_bytes);
        self.total_allocation = self.total_allocation.min(other.total_allocation);
    }

    fn max_assign(&mut self, other: Self) {
        self.instructions = self.instructions.max(other.instructions);
        self.heap_bytes = self.heap_bytes.max(other.heap_bytes);
        self.memory_bytes = self.memory_bytes.max(other.memory_bytes);
        self.total_allocation = self.total_allocation.max(other.total_allocation);
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RawBenchmarkEvent {
    pub prefix: String,
    pub label: String,
    pub suite: String,
    pub span_label: String,
    pub kind: BenchmarkEventKind,
    pub counters: BenchmarkCounters,
    pub source_line: usize,
    pub source: BenchmarkEventSource,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MalformedBenchmarkMarker {
    pub source_line: usize,
    pub source: BenchmarkEventSource,
    pub line: String,
    pub reason: String,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct BenchmarkParseReport {
    pub events: Vec<RawBenchmarkEvent>,
    pub malformed_markers: Vec<MalformedBenchmarkMarker>,
    pub ignored_line_count: usize,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkSpan {
    pub suite: String,
    pub span_label: String,
    pub start_line: usize,
    pub end_line: usize,
    pub start: BenchmarkCounters,
    pub end: BenchmarkCounters,
    pub delta: BenchmarkCounters,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum UnpairedBenchmarkMarkerKind {
    Start,
    End,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UnpairedBenchmarkMarker {
    pub event: RawBenchmarkEvent,
    pub kind: UnpairedBenchmarkMarkerKind,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct InvalidBenchmarkSpan {
    pub start: RawBenchmarkEvent,
    pub end: RawBenchmarkEvent,
    pub reason: String,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct BenchmarkSpanReport {
    pub spans: Vec<BenchmarkSpan>,
    pub unpaired_markers: Vec<UnpairedBenchmarkMarker>,
    pub invalid_spans: Vec<InvalidBenchmarkSpan>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkAggregateRow {
    pub span_label: String,
    pub runs: u64,
    pub total: BenchmarkCounters,
    pub min: BenchmarkCounters,
    pub max: BenchmarkCounters,
    pub peak_end: BenchmarkCounters,
    scope: AggregateScope,
}

/// Floor-rounded scaled means of exact aggregate totals and run counts.
///
/// Fields are `floor(total * scale / runs)`; choose a positive scale to
/// interpret them in the original counter unit. Scale zero projects zero.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct BenchmarkAverages {
    pub scale: u64,
    pub instructions: u128,
    pub heap_bytes: u128,
    pub memory_bytes: u128,
    pub total_allocation: u128,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct BenchmarkAggregateReport {
    pub rows: Vec<BenchmarkAggregateRow>,
}

/// Counter whose aggregate cannot be represented.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum BenchmarkCounter {
    Runs,
    Instructions,
    HeapBytes,
    MemoryBytes,
    TotalAllocation,
}

/// Overflow rejects the complete aggregation, without returning partial totals.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkAggregateError {
    pub span_label: String,
    pub counter: BenchmarkCounter,
    scope: AggregateScope,
}

impl BenchmarkAggregateError {
    /// Suite label, or `ALL` for the cross-suite aggregate.
    #[must_use]
    pub fn suite(&self) -> &str {
        self.scope.label()
    }

    /// Whether the overflow belongs to the cross-suite aggregate.
    #[must_use]
    pub const fn is_all_suites(&self) -> bool {
        matches!(self.scope, AggregateScope::All)
    }
}

impl std::fmt::Display for BenchmarkAggregateError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            formatter,
            "benchmark aggregate overflow: scope={:?}, span={}, counter={:?}",
            self.scope, self.span_label, self.counter
        )
    }
}

impl std::error::Error for BenchmarkAggregateError {}

/// Current and previous scaled means, preserving missing rows and measured zero.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkComparisonRow {
    pub span_label: String,
    pub current_runs: Option<u64>,
    pub previous_runs: Option<u64>,
    pub current_average: Option<BenchmarkAverages>,
    pub previous_average: Option<BenchmarkAverages>,
    scope: AggregateScope,
}

impl BenchmarkAggregateRow {
    /// Project exact totals at the caller's scale, rounding down.
    /// Zero runs and unrepresentable scaled results return IC Metrics' typed error.
    pub fn average(&self, scale: u64) -> Result<BenchmarkAverages, MeasurementRatioError> {
        Ok(BenchmarkAverages {
            scale,
            instructions: checked_scaled_ratio(self.total.instructions, self.runs, scale)?,
            heap_bytes: checked_scaled_ratio(self.total.heap_bytes, self.runs, scale)?,
            memory_bytes: checked_scaled_ratio(self.total.memory_bytes, self.runs, scale)?,
            total_allocation: checked_scaled_ratio(self.total.total_allocation, self.runs, scale)?,
        })
    }

    /// Suite label, or `ALL` for the cross-suite aggregate.
    #[must_use]
    pub fn suite(&self) -> &str {
        self.scope.label()
    }

    /// Report whether this row aggregates matching spans across every suite.
    #[must_use]
    pub const fn is_all_suites(&self) -> bool {
        matches!(self.scope, AggregateScope::All)
    }
}

impl BenchmarkComparisonRow {
    /// Suite label, or `ALL` for the cross-suite aggregate.
    #[must_use]
    pub fn suite(&self) -> &str {
        self.scope.label()
    }

    /// Report whether this row compares the aggregate across every suite.
    #[must_use]
    pub const fn is_all_suites(&self) -> bool {
        matches!(self.scope, AggregateScope::All)
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct BenchmarkComparisonReport {
    pub rows: Vec<BenchmarkComparisonRow>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct BenchmarkRunMetadata {
    pub timestamp: String,
    pub run_directory_name: String,
    pub run_index: u32,
    pub git_commit_hash: Option<String>,
    pub git_commit_short_hash: Option<String>,
    pub ic_testkit_version: String,
    pub pocket_ic_version: String,
    pub rustc_version: String,
    pub benchmark_command: Option<String>,
    pub selected_previous_run: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkRunReport {
    pub parse: BenchmarkParseReport,
    pub spans: BenchmarkSpanReport,
    pub aggregates: BenchmarkAggregateReport,
    pub comparison: Option<BenchmarkComparisonReport>,
    pub metadata: BenchmarkRunMetadata,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BenchmarkRunDirectory {
    pub path: PathBuf,
    pub directory_name: String,
    pub run_index: u32,
    pub git_commit_hash: Option<String>,
    pub git_commit_short_hash: Option<String>,
}

#[must_use]
pub fn format_marker(prefix: &str, label: &str, counters: BenchmarkCounters) -> String {
    format!(
        "{}|{}|{}|{}|{}|{}",
        prefix,
        label,
        counters.instructions,
        counters.heap_bytes,
        counters.memory_bytes,
        counters.total_allocation
    )
}

#[must_use]
pub fn benchmark_run_directory_name(
    timestamp: &str,
    git_commit_short_hash: Option<&str>,
    run_index: u32,
) -> String {
    let commit = git_commit_short_hash
        .filter(|hash| !hash.is_empty())
        .unwrap_or("unknown");
    format!("{timestamp}-{commit}-{run_index:04}")
}

/// Compute the next available benchmark run path for one timestamp and commit.
///
/// This function does not create or reserve the returned directory. Callers
/// allocating the same prefix concurrently must synchronize that shared
/// resource or provide unique timestamps.
/// A missing runs root starts at index one; other directory errors are returned.
pub fn next_benchmark_run_directory(
    runs_root: impl AsRef<Path>,
    timestamp: &str,
    git_commit_hash: Option<&str>,
) -> io::Result<BenchmarkRunDirectory> {
    let runs_root = runs_root.as_ref();
    let git_commit_hash = git_commit_hash.filter(|hash| !hash.is_empty());
    let git_commit_short_hash = git_commit_hash.map(short_commit_hash);
    let prefix = format!(
        "{}-{}-",
        timestamp,
        git_commit_short_hash.as_deref().unwrap_or("unknown")
    );
    let run_index = next_run_index_for_prefix(runs_root, &prefix)?;
    let directory_name =
        benchmark_run_directory_name(timestamp, git_commit_short_hash.as_deref(), run_index);

    Ok(BenchmarkRunDirectory {
        path: runs_root.join(&directory_name),
        directory_name,
        run_index,
        git_commit_hash: git_commit_hash.map(str::to_string),
        git_commit_short_hash,
    })
}

/// Find the latest earlier run with readable metadata and a matching command.
///
/// Names must follow [`benchmark_run_directory_name`]; indices are ordered
/// numerically even when they exceed four digits. Eligible runs are ranked by
/// metadata timestamp, directory prefix, then numeric index. An invalid current
/// name returns [`io::ErrorKind::InvalidInput`]. Unreadable or malformed
/// candidates are skipped.
/// A missing runs root has no previous run; other directory errors are returned.
pub fn find_latest_previous_run(
    runs_root: impl AsRef<Path>,
    current_run_directory_name: &str,
    benchmark_command: Option<&str>,
) -> io::Result<Option<PathBuf>> {
    let runs_root = runs_root.as_ref();
    let current_key = run_directory_sort_key(current_run_directory_name).ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "invalid benchmark run directory name",
        )
    })?;
    let mut latest: Option<((String, String, u32), PathBuf)> = None;

    let entries = match fs::read_dir(runs_root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error),
    };
    for entry in entries {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }

        let directory_name = entry.file_name().to_string_lossy().into_owned();
        let Some((prefix, index)) = run_directory_sort_key(&directory_name) else {
            continue;
        };
        if (prefix, index) >= current_key {
            continue;
        }

        let metadata_path = entry.path().join("metadata.json");
        let Ok(metadata) = read_benchmark_run_metadata(&metadata_path) else {
            continue;
        };

        if let Some(command) = benchmark_command
            && metadata.benchmark_command.as_deref() != Some(command)
        {
            continue;
        }

        let key = (metadata.timestamp, prefix.to_owned(), index);
        if latest
            .as_ref()
            .is_none_or(|(latest_key, _)| key >= *latest_key)
        {
            latest = Some((key, entry.path()));
        }
    }

    Ok(latest.map(|(_, path)| path))
}

pub fn read_benchmark_run_metadata(path: impl AsRef<Path>) -> io::Result<BenchmarkRunMetadata> {
    let input = fs::read_to_string(path)?;
    let value = serde_json::from_str::<Value>(&input).map_err(metadata_json_error)?;

    if !value.is_object() {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "benchmark metadata JSON must be an object",
        ));
    }
    serde_json::from_value(value).map_err(metadata_json_error)
}

#[must_use]
pub fn parse_benchmark_events(input: &str, config: &BenchmarkParserConfig) -> BenchmarkParseReport {
    parse_benchmark_events_from_source(input, config, BenchmarkEventSource::Unknown)
}

#[must_use]
pub fn parse_benchmark_events_from_source(
    input: &str,
    config: &BenchmarkParserConfig,
    source: BenchmarkEventSource,
) -> BenchmarkParseReport {
    let mut report = BenchmarkParseReport::default();
    append_benchmark_events(&mut report, input, config, source);
    report
}

fn append_benchmark_events(
    report: &mut BenchmarkParseReport,
    input: &str,
    config: &BenchmarkParserConfig,
    source: BenchmarkEventSource,
) {
    for (index, line) in input.lines().enumerate() {
        let source_line = index + 1;
        if !has_configured_prefix(line, &config.prefixes) {
            if config.strict && !line.trim().is_empty() {
                report.malformed_markers.push(malformed(
                    source_line,
                    source,
                    line,
                    "line does not use a configured marker prefix",
                ));
            } else {
                report.ignored_line_count += 1;
            }
            continue;
        }

        match parse_marker_line(line, source_line, source, config) {
            Ok(event) => report.events.push(event),
            Err(marker) => report.malformed_markers.push(marker),
        }
    }
}

/// Parse separately captured stdout and stderr.
///
/// Separate streams do not carry global ordering, so events are returned in
/// stdout-then-stderr order. If a benchmark span can start on one stream and end
/// on the other, capture combined process output and use [`parse_benchmark_events`].
#[must_use]
pub fn parse_benchmark_events_from_captured_output(
    stdout: &str,
    stderr: &str,
    config: &BenchmarkParserConfig,
) -> BenchmarkParseReport {
    let mut report = BenchmarkParseReport::default();
    append_benchmark_events(&mut report, stdout, config, BenchmarkEventSource::Stdout);
    append_benchmark_events(&mut report, stderr, config, BenchmarkEventSource::Stderr);
    report
}

#[must_use]
pub fn pair_benchmark_spans(events: &[RawBenchmarkEvent]) -> BenchmarkSpanReport {
    let mut report = BenchmarkSpanReport::default();
    let mut open_starts: BTreeMap<(&str, &str), Vec<&RawBenchmarkEvent>> = BTreeMap::new();

    for event in events {
        let key = (event.suite.as_str(), event.span_label.as_str());
        match event.kind {
            BenchmarkEventKind::Start => open_starts.entry(key).or_default().push(event),
            BenchmarkEventKind::End => match open_starts.entry(key) {
                Entry::Occupied(mut entry) => {
                    let start = entry
                        .get_mut()
                        .pop()
                        .expect("open marker stacks must remain nonempty");
                    if entry.get().is_empty() {
                        entry.remove();
                    }
                    push_paired_span(&mut report, start, event);
                }
                Entry::Vacant(_) => report.unpaired_markers.push(UnpairedBenchmarkMarker {
                    event: event.clone(),
                    kind: UnpairedBenchmarkMarkerKind::End,
                }),
            },
        }
    }

    for starts in open_starts.into_values() {
        for event in starts {
            report.unpaired_markers.push(UnpairedBenchmarkMarker {
                event: event.clone(),
                kind: UnpairedBenchmarkMarkerKind::Start,
            });
        }
    }

    report
}

/// Aggregate spans by named suite and across all suites.
///
/// Returns an error if any counter total or run count overflows. No partial
/// report is returned, and totals are never wrapped or saturated.
pub fn aggregate_benchmark_spans(
    spans: &[BenchmarkSpan],
) -> Result<BenchmarkAggregateReport, BenchmarkAggregateError> {
    let mut rows: BTreeMap<(AggregateScope, &str), AggregateBuilder> = BTreeMap::new();

    for span in spans {
        add_span_to_aggregate(&mut rows, AggregateScope::Suite(span.suite.clone()), span)?;
        add_span_to_aggregate(&mut rows, AggregateScope::All, span)?;
    }

    Ok(BenchmarkAggregateReport {
        rows: rows
            .into_iter()
            .map(|((scope, span_label), builder)| builder.finish(scope, span_label.to_owned()))
            .collect(),
    })
}

/// Pair current and previous means at the same caller-selected scale.
/// Returns a typed error if either side cannot supply the selected projection.
pub fn compare_benchmark_aggregates(
    current: &[BenchmarkAggregateRow],
    previous: &[BenchmarkAggregateRow],
    scale: u64,
) -> Result<BenchmarkComparisonReport, MeasurementRatioError> {
    let mut rows = BTreeMap::new();
    for row in current {
        rows.insert((&row.scope, row.span_label.as_str()), (Some(row), None));
    }
    for row in previous {
        rows.entry((&row.scope, row.span_label.as_str()))
            .or_insert((None, None))
            .1 = Some(row);
    }

    Ok(BenchmarkComparisonReport {
        rows: rows
            .into_iter()
            .map(|((scope, span_label), (current_row, previous_row))| {
                Ok(BenchmarkComparisonRow {
                    span_label: span_label.to_owned(),
                    current_runs: current_row.map(|row| row.runs),
                    previous_runs: previous_row.map(|row| row.runs),
                    current_average: current_row.map(|row| row.average(scale)).transpose()?,
                    previous_average: previous_row.map(|row| row.average(scale)).transpose()?,
                    scope: scope.clone(),
                })
            })
            .collect::<Result<_, MeasurementRatioError>>()?,
    })
}

/// Write one complete benchmark report into the requested directory.
///
/// The directory is caller-owned. Concurrent writers must use unique paths or
/// synchronize access to the same path outside this function.
pub fn write_benchmark_report_dir(
    report: &BenchmarkRunReport,
    path: impl AsRef<Path>,
) -> io::Result<()> {
    let path = path.as_ref();
    let suite_aggregates = aggregates_csv(
        report
            .aggregates
            .rows
            .iter()
            .filter(|row| !row.is_all_suites()),
    )?;
    let all_aggregates = aggregates_csv(
        report
            .aggregates
            .rows
            .iter()
            .filter(|row| row.is_all_suites()),
    )?;
    let summary = benchmark_summary_markdown(report)?;
    fs::create_dir_all(path)?;

    fs::write(
        path.join("raw-events.csv"),
        raw_events_csv(&report.parse.events),
    )?;
    fs::write(
        path.join("malformed-markers.csv"),
        malformed_markers_csv(&report.parse.malformed_markers),
    )?;
    fs::write(path.join("spans.csv"), spans_csv(&report.spans.spans))?;
    fs::write(
        path.join("unpaired-markers.csv"),
        unpaired_markers_csv(&report.spans.unpaired_markers),
    )?;
    fs::write(
        path.join("invalid-spans.csv"),
        invalid_spans_csv(&report.spans.invalid_spans),
    )?;
    fs::write(path.join("suite-aggregates.csv"), suite_aggregates)?;
    fs::write(path.join("all-aggregates.csv"), all_aggregates)?;
    fs::write(
        path.join("comparison.csv"),
        comparison_csv(report.comparison.as_ref()),
    )?;
    fs::write(path.join("bench-summary.md"), summary)?;
    fs::write(path.join("metadata.json"), metadata_json(&report.metadata))?;

    Ok(())
}

fn parse_marker_line(
    line: &str,
    source_line: usize,
    source: BenchmarkEventSource,
    config: &BenchmarkParserConfig,
) -> Result<RawBenchmarkEvent, MalformedBenchmarkMarker> {
    let mut parts = line.split('|');
    // Six fields and an absent seventh establish the complete marker shape
    // without collecting an arbitrary number of columns from malformed logs.
    let columns: [Option<&str>; 7] = std::array::from_fn(|_| parts.next());
    let [
        Some(prefix),
        Some(label),
        Some(instructions),
        Some(heap_bytes),
        Some(memory_bytes),
        Some(total_allocation),
        None,
    ] = columns
    else {
        return Err(malformed(
            source_line,
            source,
            line,
            "expected six pipe-separated columns",
        ));
    };

    if !config.prefixes.iter().any(|known| known == prefix) {
        return Err(malformed(
            source_line,
            source,
            line,
            "prefix is not configured",
        ));
    }

    if label.is_empty() {
        return Err(malformed(source_line, source, line, "label is empty"));
    }

    let (span_label, kind) = split_label_kind(label).ok_or_else(|| {
        malformed(
            source_line,
            source,
            line,
            "label must end in :start or :end",
        )
    })?;

    let counters = BenchmarkCounters {
        instructions: parse_counter(instructions, source_line, source, line, "instructions")?,
        heap_bytes: parse_counter(heap_bytes, source_line, source, line, "heap_bytes")?,
        memory_bytes: parse_counter(memory_bytes, source_line, source, line, "memory_bytes")?,
        total_allocation: parse_counter(
            total_allocation,
            source_line,
            source,
            line,
            "total_allocation",
        )?,
    };
    let suite = config.suite_derivation.derive_suite(span_label);

    Ok(RawBenchmarkEvent {
        prefix: prefix.to_string(),
        label: label.to_string(),
        suite,
        span_label: span_label.to_string(),
        kind,
        counters,
        source_line,
        source,
    })
}

fn parse_counter(
    value: &str,
    source_line: usize,
    source: BenchmarkEventSource,
    line: &str,
    name: &str,
) -> Result<u128, MalformedBenchmarkMarker> {
    if value.is_empty() {
        return Err(malformed(
            source_line,
            source,
            line,
            &format!("{name} counter is empty"),
        ));
    }

    value.parse::<u128>().map_err(|_| {
        malformed(
            source_line,
            source,
            line,
            &format!("{name} counter is not an unsigned integer"),
        )
    })
}

fn split_label_kind(label: &str) -> Option<(&str, BenchmarkEventKind)> {
    let start = label.strip_suffix(":start");
    let end = label.strip_suffix(":end");

    match (start, end) {
        (Some(span_label), None) if !span_label.is_empty() => {
            Some((span_label, BenchmarkEventKind::Start))
        }
        (None, Some(span_label)) if !span_label.is_empty() => {
            Some((span_label, BenchmarkEventKind::End))
        }
        _ => None,
    }
}

fn has_configured_prefix(line: &str, prefixes: &[String]) -> bool {
    prefixes.iter().any(|prefix| {
        line.strip_prefix(prefix)
            .is_some_and(|rest| rest.starts_with('|'))
    })
}

fn malformed(
    source_line: usize,
    source: BenchmarkEventSource,
    line: &str,
    reason: &str,
) -> MalformedBenchmarkMarker {
    MalformedBenchmarkMarker {
        source_line,
        source,
        line: line.to_string(),
        reason: reason.to_string(),
    }
}

fn push_paired_span(
    report: &mut BenchmarkSpanReport,
    start: &RawBenchmarkEvent,
    end: &RawBenchmarkEvent,
) {
    if let Some(delta) = end.counters.checked_delta(start.counters) {
        report.spans.push(BenchmarkSpan {
            suite: start.suite.clone(),
            span_label: start.span_label.clone(),
            start_line: start.source_line,
            end_line: end.source_line,
            start: start.counters,
            end: end.counters,
            delta,
        });
    } else {
        report.invalid_spans.push(InvalidBenchmarkSpan {
            start: start.clone(),
            end: end.clone(),
            reason: "end counter is lower than start counter".to_string(),
        });
    }
}

#[derive(Clone, Debug)]
struct AggregateBuilder {
    runs: u64,
    total: BenchmarkCounters,
    min: BenchmarkCounters,
    max: BenchmarkCounters,
    peak_end: BenchmarkCounters,
}

impl AggregateBuilder {
    const fn new(span: &BenchmarkSpan) -> Self {
        Self {
            runs: 1,
            total: span.delta,
            min: span.delta,
            max: span.delta,
            peak_end: span.end,
        }
    }

    fn push(&mut self, span: &BenchmarkSpan) -> Result<(), BenchmarkCounter> {
        let runs = self.runs.checked_add(1).ok_or(BenchmarkCounter::Runs)?;
        let total = self.total.checked_add(span.delta)?;
        self.runs = runs;
        self.total = total;
        self.min.min_assign(span.delta);
        self.max.max_assign(span.delta);
        self.peak_end.max_assign(span.end);
        Ok(())
    }

    const fn finish(self, scope: AggregateScope, span_label: String) -> BenchmarkAggregateRow {
        BenchmarkAggregateRow {
            span_label,
            runs: self.runs,
            total: self.total,
            min: self.min,
            max: self.max,
            peak_end: self.peak_end,
            scope,
        }
    }
}

fn add_span_to_aggregate<'a>(
    rows: &mut BTreeMap<(AggregateScope, &'a str), AggregateBuilder>,
    scope: AggregateScope,
    span: &'a BenchmarkSpan,
) -> Result<(), BenchmarkAggregateError> {
    match rows.entry((scope, span.span_label.as_str())) {
        Entry::Occupied(mut entry) => entry.get_mut().push(span).map_err(|counter| {
            let (scope, span_label) = entry.key();
            BenchmarkAggregateError {
                span_label: (*span_label).to_owned(),
                counter,
                scope: scope.clone(),
            }
        }),
        Entry::Vacant(entry) => {
            entry.insert(AggregateBuilder::new(span));
            Ok(())
        }
    }
}

fn raw_events_csv(events: &[RawBenchmarkEvent]) -> String {
    let mut out = String::from(
        "source_line,source,prefix,suite,label,span_label,kind,instructions,heap_bytes,memory_bytes,total_allocation\n",
    );
    for event in events {
        let _ = writeln!(
            out,
            "{},{},{},{},{},{},{},{},{},{},{}",
            event.source_line,
            event.source.as_str(),
            csv_cell(&event.prefix),
            csv_cell(&event.suite),
            csv_cell(&event.label),
            csv_cell(&event.span_label),
            kind_str(event.kind),
            event.counters.instructions,
            event.counters.heap_bytes,
            event.counters.memory_bytes,
            event.counters.total_allocation
        );
    }
    out
}

fn malformed_markers_csv(markers: &[MalformedBenchmarkMarker]) -> String {
    let mut out = String::from("source_line,source,reason,line\n");
    for marker in markers {
        let _ = writeln!(
            out,
            "{},{},{},{}",
            marker.source_line,
            marker.source.as_str(),
            csv_cell(&marker.reason),
            csv_cell(&marker.line)
        );
    }
    out
}

fn spans_csv(spans: &[BenchmarkSpan]) -> String {
    let mut out = String::from(
        "suite,span_label,start_line,end_line,instructions_delta,heap_bytes_delta,memory_bytes_delta,total_allocation_delta\n",
    );
    for span in spans {
        let _ = writeln!(
            out,
            "{},{},{},{},{},{},{},{}",
            csv_cell(&span.suite),
            csv_cell(&span.span_label),
            span.start_line,
            span.end_line,
            span.delta.instructions,
            span.delta.heap_bytes,
            span.delta.memory_bytes,
            span.delta.total_allocation
        );
    }
    out
}

fn unpaired_markers_csv(markers: &[UnpairedBenchmarkMarker]) -> String {
    let mut out = String::from("source_line,source,kind,suite,span_label,label\n");
    for marker in markers {
        let kind = match marker.kind {
            UnpairedBenchmarkMarkerKind::Start => "start",
            UnpairedBenchmarkMarkerKind::End => "end",
        };
        let _ = writeln!(
            out,
            "{},{},{},{},{},{}",
            marker.event.source_line,
            marker.event.source.as_str(),
            kind,
            csv_cell(&marker.event.suite),
            csv_cell(&marker.event.span_label),
            csv_cell(&marker.event.label)
        );
    }
    out
}

fn invalid_spans_csv(spans: &[InvalidBenchmarkSpan]) -> String {
    let mut out = String::from("start_line,end_line,suite,span_label,reason\n");
    for span in spans {
        let _ = writeln!(
            out,
            "{},{},{},{},{}",
            span.start.source_line,
            span.end.source_line,
            csv_cell(&span.start.suite),
            csv_cell(&span.start.span_label),
            csv_cell(&span.reason)
        );
    }
    out
}

fn aggregates_csv<'a>(rows: impl Iterator<Item = &'a BenchmarkAggregateRow>) -> io::Result<String> {
    let mut out = String::from(
        "suite,span_label,runs,instructions_total,instructions_mean,heap_bytes_total,heap_bytes_mean,memory_bytes_total,memory_bytes_mean,total_allocation_total,total_allocation_mean\n",
    );
    for row in rows {
        let average = row.average(1).map_err(io::Error::other)?;
        let _ = writeln!(
            out,
            "{},{},{},{},{},{},{},{},{},{},{}",
            csv_cell(row.suite()),
            csv_cell(&row.span_label),
            row.runs,
            row.total.instructions,
            average.instructions,
            row.total.heap_bytes,
            average.heap_bytes,
            row.total.memory_bytes,
            average.memory_bytes,
            row.total.total_allocation,
            average.total_allocation
        );
    }
    Ok(out)
}

fn comparison_csv(comparison: Option<&BenchmarkComparisonReport>) -> String {
    let mut out = String::from(
        "scope,suite,span_label,current_runs,previous_runs,current_mean_scale,previous_mean_scale,current_instructions_mean,previous_instructions_mean,current_heap_bytes_mean,previous_heap_bytes_mean,current_memory_bytes_mean,previous_memory_bytes_mean,current_total_allocation_mean,previous_total_allocation_mean\n",
    );
    if let Some(comparison) = comparison {
        for row in &comparison.rows {
            let current = row.current_average;
            let previous = row.previous_average;
            let value = |mean: Option<BenchmarkAverages>, field: fn(BenchmarkAverages) -> u128| {
                mean.map_or_else(String::new, |mean| field(mean).to_string())
            };
            let _ = writeln!(
                out,
                "{},{},{},{},{},{},{},{},{},{},{},{},{},{},{}",
                if row.is_all_suites() { "all" } else { "suite" },
                csv_cell(row.suite()),
                csv_cell(&row.span_label),
                optional_u64_cell(row.current_runs),
                optional_u64_cell(row.previous_runs),
                current.map_or_else(String::new, |mean| mean.scale.to_string()),
                previous.map_or_else(String::new, |mean| mean.scale.to_string()),
                value(current, |m| m.instructions),
                value(previous, |m| m.instructions),
                value(current, |m| m.heap_bytes),
                value(previous, |m| m.heap_bytes),
                value(current, |m| m.memory_bytes),
                value(previous, |m| m.memory_bytes),
                value(current, |m| m.total_allocation),
                value(previous, |m| m.total_allocation)
            );
        }
    }
    out
}

fn benchmark_summary_markdown(report: &BenchmarkRunReport) -> io::Result<String> {
    let mut out = String::from(
        "# Benchmark Summary\n\nMeans are floor-rounded in original counter units. Comparison CSV retains current and previous scaled means.\n\n| Benchmark | Runs | Instructions Mean | Heap Bytes Mean | Memory Bytes Mean | Allocation Mean |\n| --- | ---: | ---: | ---: | ---: | ---: |\n",
    );
    for row in report
        .aggregates
        .rows
        .iter()
        .filter(|row| !row.is_all_suites())
    {
        let average = row.average(1).map_err(io::Error::other)?;
        let _ = writeln!(
            out,
            "| {} | {} | {} | {} | {} | {} |",
            markdown_cell(&row.span_label),
            row.runs,
            average.instructions,
            average.heap_bytes,
            average.memory_bytes,
            average.total_allocation
        );
    }
    Ok(out)
}

fn metadata_json(metadata: &BenchmarkRunMetadata) -> String {
    let value = serde_json::to_value(metadata).expect("metadata JSON must serialize");

    let mut output = serde_json::to_string_pretty(&value).expect("metadata JSON must serialize");
    output.push('\n');
    output
}

fn next_run_index_for_prefix(runs_root: &Path, prefix: &str) -> io::Result<u32> {
    let entries = match fs::read_dir(runs_root) {
        Ok(entries) => entries,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(1),
        Err(error) => return Err(error),
    };
    let mut max_index = 0;
    for entry in entries {
        let entry = entry?;
        if !entry.file_type()?.is_dir() {
            continue;
        }

        if let Some(index) = entry
            .file_name()
            .to_str()
            .and_then(|name| name.strip_prefix(prefix))
            .and_then(parse_run_index)
        {
            max_index = max_index.max(index);
        }
    }

    max_index
        .checked_add(1)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "benchmark run index exhausted"))
}

fn parse_run_index(index: &str) -> Option<u32> {
    if index.len() >= 4 && index.chars().all(|char| char.is_ascii_digit()) {
        index.parse().ok()
    } else {
        None
    }
}

fn run_directory_sort_key(name: &str) -> Option<(&str, u32)> {
    let (prefix, index) = name.rsplit_once('-')?;
    Some((prefix, parse_run_index(index)?))
}

fn short_commit_hash(hash: &str) -> String {
    hash.chars().take(7).collect()
}

fn metadata_json_error(err: serde_json::Error) -> io::Error {
    io::Error::new(
        io::ErrorKind::InvalidData,
        format!("invalid benchmark metadata JSON: {err}"),
    )
}

const fn kind_str(kind: BenchmarkEventKind) -> &'static str {
    match kind {
        BenchmarkEventKind::Start => "start",
        BenchmarkEventKind::End => "end",
    }
}

fn csv_cell(value: &str) -> String {
    if value.contains([',', '"', '\n', '\r']) {
        format!("\"{}\"", value.replace('"', "\"\""))
    } else {
        value.to_string()
    }
}

fn optional_u64_cell(value: Option<u64>) -> String {
    value.map_or_else(String::new, |value| value.to_string())
}

fn markdown_cell(value: &str) -> String {
    value.replace('|', "\\|")
}

#[cfg(test)]
mod tests {
    use super::{AggregateBuilder, BenchmarkCounter, BenchmarkCounters, BenchmarkSpan};

    #[test]
    fn aggregate_run_count_overflow_is_rejected_without_changing_totals() {
        let span = BenchmarkSpan {
            suite: "probe".into(),
            span_label: "probe/run".into(),
            start_line: 1,
            end_line: 2,
            start: BenchmarkCounters::default(),
            end: BenchmarkCounters::default(),
            delta: BenchmarkCounters::default(),
        };
        let mut builder = AggregateBuilder::new(&span);
        builder.runs = u64::MAX;
        assert_eq!(builder.push(&span), Err(BenchmarkCounter::Runs));
        assert_eq!(builder.runs, u64::MAX);
        assert_eq!(builder.total, BenchmarkCounters::default());
    }
}