goose 0.19.0

//! Optional graph data collected during the load tests.
//!
//! If the HTML report is enabled the graph data will be collected and stored in
//! the [`GraphData`] structure during the load test. When the report is written
//! this data is converted into [`Graph`] structures and HTML markup is generated
//! based on them.

use crate::metrics::dashboard_snapshot::{
    Heatmap, HeatmapColumn, HeatmapCursor, RequestSeries, SeriesWindow, SERIES_WINDOW_SECS,
};
use crate::metrics::{buckets, GooseRequestMetricTimingData};
use crate::test_plan::{TestPlanHistory, TestPlanStepAction};
use chrono::prelude::*;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::convert::TryFrom;
use std::fmt::Write;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::OnceLock;

/// One key's counts by second, and for requests, which latency ring holds the
/// request's own response times. Laid out in order, so a request's pending
/// response times follow the fields a batch merge reads first.
#[derive(Clone)]
#[repr(C)]
struct KeySeries {
    counts: TimeSeries<u32, u32>,
    /// The request's response times from the newest batches, kept here,
    /// next to the count each batch already updates, until they move to its
    /// ring (see [`PendingLatency`]). Allocated the first time the heatmap
    /// records a response time for the request, so without the heatmap a
    /// name costs one pointer more than its counts, and errors never
    /// allocate one.
    pending: Option<Box<PendingLatency>>,
}

/// A request's batched response times for two consecutive seconds, counted
/// by fixed bucket index, waiting to move to its latency ring. It lives in
/// the request's [`KeySeries`], so merging a batch writes next to the count
/// the batch already updates instead of reaching into the ring's slot for
/// the second; the counts move to the ring once both seconds are past, or
/// when a heatmap is built. Laid out in order: what every merge writes
/// first, and both seconds' counts of a bucket side by side.
#[derive(Clone)]
#[repr(C)]
struct PendingLatency {
    /// The index of the request's ring in [`LatencyHistograms`], or `None`
    /// when it has no ring of its own, assigned when this is allocated.
    /// Keeping it here lets a batch merge find the ring with the same lookup
    /// that counts the request. Reset with the latency histograms, by
    /// [`GraphData::reset_rate_series`].
    ring: Option<usize>,
    /// The seconds held: `second` and `second + 1`, once `held`.
    second: usize,
    held: bool,
    /// Fastest, slowest and total in microseconds, and count, for each
    /// second held.
    totals: [[u64; 4]; 2],
    /// Rows past the fixed ones (from 61 seconds): which second held (0 or
    /// 1), the row and its count.
    overflow: Vec<(usize, usize, usize)>,
    /// Counts by fixed bucket index, for each second held.
    counts: [[u32; 2]; FIXED_BUCKETS],
}

impl PendingLatency {
    fn new(ring: Option<usize>) -> Self {
        PendingLatency {
            ring,
            second: 0,
            held: false,
            totals: [[0; 4]; 2],
            overflow: Vec::new(),
            counts: [[0; 2]; FIXED_BUCKETS],
        }
    }

    /// Add a batch's timing data for held second `at` (0 or 1).
    fn add(&mut self, at: usize, data: &GooseRequestMetricTimingData) {
        self.add_totals(at, data);
        for (&bucket, &count) in &data.times_us {
            self.add_bucket(at, bucket, count);
        }
    }

    /// Add a batch's fastest, slowest, total and count for held second `at`.
    fn add_totals(&mut self, at: usize, data: &GooseRequestMetricTimingData) {
        let totals = &mut self.totals[at];
        if data.counter > 0 && (totals[3] == 0 || data.minimum_time_us < totals[0]) {
            totals[0] = data.minimum_time_us;
        }
        totals[1] = totals[1].max(data.maximum_time_us);
        totals[2] += data.total_time_us;
        totals[3] += data.counter as u64;
    }

    /// Add `count` responses in the timing data bucket that starts at
    /// `bucket` microseconds to held second `at`.
    fn add_bucket(&mut self, at: usize, bucket: u64, count: usize) {
        let row = bucket_row(bucket);
        match self.counts.get_mut(row) {
            Some(counts) => {
                let held = &mut counts[at];
                *held = held.saturating_add(u32::try_from(count).unwrap_or(u32::MAX));
            }
            None => self.overflow.push((at, row, count)),
        }
    }

    /// Move everything held to `ring`, each second marked with `rev` of its
    /// own, and empty the counts.
    fn move_to(
        &mut self,
        ring: &mut LatencyRing,
        sink: &mut ColumnSink<'_>,
        mut next_rev: impl FnMut() -> u64,
    ) {
        for at in 0..2 {
            let [minimum, maximum, total, counter] = self.totals[at];
            if counter == 0 {
                continue;
            }
            let (counts, overflow) = (&self.counts, &self.overflow);
            ring.write(self.second + at, next_rev(), sink, |histogram| {
                for (index, counts) in counts.iter().enumerate() {
                    if counts[at] > 0 {
                        histogram.add_dense(index, counts[at] as usize);
                    }
                }
                for &(held, row, count) in overflow {
                    if held == at {
                        histogram.add_bucket(row, count);
                    }
                }
                histogram.add_totals(minimum, maximum, total, counter as usize);
            });
            for counts in self.counts.iter_mut() {
                counts[at] = 0;
            }
            self.totals[at] = [0; 4];
        }
        self.overflow.clear();
    }
}

impl std::ops::Deref for KeySeries {
    type Target = TimeSeries<u32, u32>;
    fn deref(&self) -> &Self::Target {
        &self.counts
    }
}

impl std::ops::DerefMut for KeySeries {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.counts
    }
}

#[derive(Clone)]
struct ItemsPerSecond {
    series: HashMap<String, KeySeries>,
    /// Absolute second used as `origin` for newly inserted keys.
    ///
    /// After dashboard pruning / rate-series reset this tracks the retained window
    /// so a late-seen request name does not allocate from absolute second 0.
    origin: usize,
}

impl ItemsPerSecond {
    fn new() -> ItemsPerSecond {
        ItemsPerSecond {
            series: Default::default(),
            origin: 0,
        }
    }

    fn with_origin(origin: usize) -> ItemsPerSecond {
        ItemsPerSecond {
            series: Default::default(),
            origin,
        }
    }

    #[inline(always)]
    fn initialize_or_increment(&mut self, key: &str, second: usize, value: u32) -> u32 {
        let data = self.increment(key, second, value);
        data.get(second)
    }

    /// Add `value` to `key`'s count for `second`, returning the key's entry.
    #[inline(always)]
    fn increment(&mut self, key: &str, second: usize, value: u32) -> &mut KeySeries {
        let origin = self.origin;
        let data = self
            .series
            .entry(key.to_string())
            .or_insert_with(|| KeySeries {
                counts: TimeSeries::with_origin(origin),
                pending: None,
            });
        data.increase_value(second, value);
        data
    }

    /// A report graph of every key's counts, in the map's iteration order.
    fn graph<'a>(
        &'a self,
        html_id: &'a str,
        y_axis_label: &'a str,
        granular_data: bool,
    ) -> Graph<'a, u32, u32> {
        Graph {
            html_id,
            y_axis_label,
            granular_data,
            left_out: None,
            data: self
                .series
                .iter()
                .map(|(label, series)| (label.as_str(), &series.counts))
                .collect(),
        }
    }

    #[cfg(test)]
    #[inline(always)]
    fn insert(&mut self, key: &str, time_series: TimeSeries<u32, u32>) {
        self.series.insert(
            key.to_string(),
            KeySeries {
                counts: time_series,
                pending: None,
            },
        );
    }

    #[cfg(test)]
    #[inline(always)]
    fn len(&self) -> usize {
        self.series.len()
    }

    #[cfg(test)]
    #[inline(always)]
    fn get(&self, key: &str) -> Option<TimeSeries<u32, u32>> {
        self.series.get(key).map(|data| data.counts.clone())
    }
}

/// Most request names that get their own latency series. Later names still
/// count toward the all requests series.
pub(crate) const MAX_REQUEST_SERIES: usize = 100;

/// The heatmap's fixed rows, from 0 µs up: where each band starts, the
/// width of its rows and how many it has, all in microseconds. 100 µs rows
/// under a millisecond, 1 ms rows to 64 ms, 2 ms to 100 ms, 10 ms to
/// 500 ms, 50 ms to 1 s, 100 ms to 5 s and 1 s to 61 s. No row is narrower
/// than the timing data buckets it takes, so every row can fill.
const ROW_BANDS: [(u64, u64, usize); 7] = [
    (0, 100, 10),
    (1_000, 1_000, 63),
    (64_000, 2_000, 18),
    (100_000, 10_000, 40),
    (500_000, 50_000, 10),
    (1_000_000, 100_000, 40),
    (5_000_000, 1_000_000, 56),
];

/// The fixed rows, every time up to 61 seconds, within the 256 a
/// [`SecondHistogram`] entry's 8 bit index holds.
const FIXED_BUCKETS: usize = 237;

/// Where the fixed rows end. Past it, possible with a longer `--timeout`,
/// each timing data bucket is a row of its own.
const FIXED_END_US: u64 = 61_000_000;

/// The heatmap row of a time in microseconds: its fixed row, or past 61
/// seconds the row of its timing data bucket.
fn display_index(us: u64) -> usize {
    let mut first = 0;
    for &(start, width, rows) in &ROW_BANDS {
        if us < start + width * rows as u64 {
            return first + ((us - start) / width) as usize;
        }
        first += rows;
    }
    FIXED_BUCKETS + buckets::bucket_index(us) - buckets::bucket_index(FIXED_END_US)
}

/// The first and the last microsecond of heatmap row `index`.
fn display_range(index: usize) -> (u64, u64) {
    let mut first = 0;
    for &(start, width, rows) in &ROW_BANDS {
        if index < first + rows {
            let lowest = start + width * (index - first) as u64;
            return (lowest, lowest + width - 1);
        }
        first += rows;
    }
    // The first starts at 61 seconds, though its bucket may start before.
    let bucket = buckets::bucket_at(buckets::bucket_index(FIXED_END_US) + index - FIXED_BUCKETS);
    (bucket.max(FIXED_END_US), buckets::bucket_end(bucket) - 1)
}

/// The time, in microseconds, heatmap row `index` stands for: its middle.
fn display_value(index: usize) -> u64 {
    let (lowest, highest) = display_range(index);
    lowest + (highest + 1 - lowest) / 2
}

/// The heatmap row of a timing data bucket that starts at `bucket`
/// microseconds: the row of its last microsecond. A bucket is never wider
/// than a row, so it reaches at most into the next one, and a whole
/// millisecond time, which starts a row, stays in that row.
fn bucket_row(bucket: u64) -> usize {
    // Looked up for every bucket a batch merges, so the rows of the buckets
    // up to 61 seconds are worked out once.
    static ROWS: OnceLock<Vec<u8>> = OnceLock::new();
    let rows = ROWS.get_or_init(|| {
        (0..=buckets::bucket_index(FIXED_END_US))
            .map(|index| {
                let row = display_index(buckets::bucket_end(buckets::bucket_at(index)) - 1);
                u8::try_from(row).expect("a row the packed index holds")
            })
            .collect()
    });
    match rows.get(buckets::bucket_index(bucket)) {
        Some(&row) => row as usize,
        None => display_index(buckets::bucket_end(bucket) - 1),
    }
}

/// Bits of a [`SecondHistogram`] entry that hold the count; the bucket index
/// is in the top 8 bits.
const COUNT_BITS: u32 = 24;
/// Most one entry counts. A bucket that fills an entry continues in the next.
const COUNT_MAX: u32 = (1 << COUNT_BITS) - 1;

// Every fixed row's index fits in an entry's top bits.
const _: () = assert!(FIXED_BUCKETS <= 1 << (32 - COUNT_BITS));

/// One second's response times, counted by heatmap row (see
/// [`ROW_BANDS`]), held compactly: a sorted list with 4 bytes for each row
/// the second has seen (its index and its count), so a quiet name costs
/// little and a busy one at most about 1 KB a second. Rows past 61 seconds,
/// possible with a longer `--timeout`, go to a short sorted overflow list.
/// The metrics processor merges each batch's response times for a request
/// name (the batch's
/// [`GooseRequestMetricTimingData`](crate::metrics::GooseRequestMetricTimingData),
/// all from one second) into the rings, each of its buckets into the row
/// it ends in. A ring counts its newest seconds in a flat array
/// instead (see [`LatencyRing`]), so merging a batch is an addition per
/// bucket rather than a search, and packs each second once it stops
/// filling.
#[derive(Clone, Debug, Default)]
pub(crate) struct SecondHistogram {
    entries: Vec<u32>,
    overflow: Vec<(usize, u32)>,
    /// While a ring's second is still filling: its count for each fixed
    /// row, with `entries` empty.
    open: Option<Box<[usize; FIXED_BUCKETS]>>,
    /// The fastest response in microseconds, not rounded.
    pub(crate) minimum_time_us: u64,
    /// The slowest response in microseconds, not rounded.
    pub(crate) maximum_time_us: u64,
    /// Sum of all response times, in microseconds.
    pub(crate) total_time_us: u64,
    /// Number of responses.
    pub(crate) counter: usize,
}

/// Two histograms are equal when they hold the same response times, however
/// each holds them.
impl PartialEq for SecondHistogram {
    fn eq(&self, other: &Self) -> bool {
        self.minimum_time_us == other.minimum_time_us
            && self.maximum_time_us == other.maximum_time_us
            && self.total_time_us == other.total_time_us
            && self.counter == other.counter
            && self
                .buckets()
                .coalesce(merge_bucket)
                .eq(other.buckets().coalesce(merge_bucket))
    }
}

/// A row as `(index, count)`.
type Bucket = (usize, usize);

/// Join consecutive pieces of one row (a row past [`COUNT_MAX`]).
fn merge_bucket(a: Bucket, b: Bucket) -> Result<Bucket, (Bucket, Bucket)> {
    if a.0 == b.0 {
        Ok((a.0, a.1 + b.1))
    } else {
        Err((a, b))
    }
}

impl SecondHistogram {
    /// An empty histogram for a ring's newest second, counted by fixed index
    /// until [`pack`](Self::pack).
    fn open() -> Self {
        Self::open_with(Box::new([0; FIXED_BUCKETS]))
    }

    /// [`Self::open`], counting in `counts`, which must be all zero.
    fn open_with(counts: Box<[usize; FIXED_BUCKETS]>) -> Self {
        debug_assert!(counts.iter().all(|&count| count == 0));
        SecondHistogram {
            open: Some(counts),
            ..Default::default()
        }
    }

    /// Add `count` to the row with fixed index `index`, searching from
    /// entry `from` on. Returns where the search for a higher index can
    /// start.
    fn add_indexed(&mut self, from: usize, index: usize, mut count: usize) -> usize {
        debug_assert!(index < FIXED_BUCKETS && FIXED_BUCKETS <= 1 << (32 - COUNT_BITS));
        let bucket = index as u32;
        let key = bucket << COUNT_BITS;
        let mut at = from + self.entries[from..].partition_point(|&entry| entry < key);
        while count > 0 {
            if at < self.entries.len() && self.entries[at] >> COUNT_BITS == bucket {
                let room = (COUNT_MAX - (self.entries[at] & COUNT_MAX)) as usize;
                let add = room.min(count);
                self.entries[at] += add as u32;
                count -= add;
            } else {
                let add = count.min(COUNT_MAX as usize);
                self.entries.insert(at, key | add as u32);
                count -= add;
            }
            at += 1;
        }
        at
    }

    /// Add `count` responses to heatmap row `row`.
    fn add_bucket(&mut self, row: usize, count: usize) {
        if row < FIXED_BUCKETS {
            self.add_dense(row, count);
        } else {
            self.add_overflow(row, count);
        }
    }

    /// Add `count` to the row with fixed index `index`.
    fn add_dense(&mut self, index: usize, count: usize) {
        match self.open.as_mut() {
            Some(open) => open[index] += count,
            None => {
                self.add_indexed(0, index, count);
            }
        }
    }

    /// Add `count` to a row past the fixed ones.
    fn add_overflow(&mut self, row: usize, count: usize) {
        let count = u32::try_from(count).unwrap_or(u32::MAX);
        match self.overflow.binary_search_by_key(&row, |&(row, _)| row) {
            Ok(at) => self.overflow[at].1 = self.overflow[at].1.saturating_add(count),
            Err(at) => self.overflow.insert(at, (row, count)),
        }
    }

    fn add_totals(&mut self, minimum: u64, maximum: u64, total: u64, counter: usize) {
        // The counters, not a minimum of 0, say whether there are times yet:
        // a time can be 0 µs.
        if counter > 0 && (self.counter == 0 || minimum < self.minimum_time_us) {
            self.minimum_time_us = minimum;
        }
        if maximum > self.maximum_time_us {
            self.maximum_time_us = maximum;
        }
        self.total_time_us += total;
        self.counter += counter;
    }

    /// Record one response time in whole milliseconds (unit tests).
    #[cfg(test)]
    pub(crate) fn record(&mut self, time: u64) {
        self.record_us(time.saturating_mul(1000));
    }

    /// Record one response time in microseconds, in the row of its timing
    /// data bucket, as a batch merge counts it.
    pub(crate) fn record_us(&mut self, us: u64) {
        self.add_totals(us, us, us, 1);
        self.add_bucket(bucket_row(buckets::bucket_floor(us)), 1);
    }

    /// Add another second's response times.
    pub(crate) fn merge(&mut self, other: &SecondHistogram) {
        if let Some(open) = self.open.as_mut() {
            for (index, count) in other.indexed() {
                open[index] += count;
            }
        } else {
            // Both are sorted, so each search starts where the last one ended.
            let mut from = 0;
            for (index, count) in other.indexed() {
                from = self.add_indexed(from, index, count);
            }
        }
        for &(row, count) in &other.overflow {
            self.add_overflow(row, count as usize);
        }
        self.add_totals(
            other.minimum_time_us,
            other.maximum_time_us,
            other.total_time_us,
            other.counter,
        );
    }

    /// Add a batch's response times for one request name, all from this
    /// second: each bucket into its row, and the unrounded minimum,
    /// maximum, total and count, as
    /// [`GooseRequestMetricTimingData::merge`] adds them.
    pub(crate) fn merge_timing(&mut self, other: &GooseRequestMetricTimingData) {
        for (&bucket, &count) in &other.times_us {
            self.add_bucket(bucket_row(bucket), count);
        }
        self.add_totals(
            other.minimum_time_us,
            other.maximum_time_us,
            other.total_time_us,
            other.counter,
        );
    }

    /// Once the second is unlikely to grow: move the counts of an open
    /// second into the compact list, allocated once at its size, and release
    /// spare capacity. Returns the emptied counts of the open second, for
    /// the next second to reuse.
    fn pack(&mut self) -> Option<Box<[usize; FIXED_BUCKETS]>> {
        let mut open = self.open.take()?;
        if self.entries.is_empty() {
            let buckets = open.iter().filter(|&&count| count > 0).count();
            self.entries.reserve_exact(buckets);
        }
        let mut from = 0;
        for (index, count) in open.iter_mut().enumerate() {
            if *count > 0 {
                from = self.add_indexed(from, index, *count);
                *count = 0;
            }
        }
        self.entries.shrink_to_fit();
        self.overflow.shrink_to_fit();
        Some(open)
    }

    /// Count a packed histogram by fixed index again, in `counts`, which
    /// must be all zero, until the next [`pack`](Self::pack).
    fn reopen(&mut self, mut counts: Box<[usize; FIXED_BUCKETS]>) {
        debug_assert!(self.open.is_none() && counts.iter().all(|&count| count == 0));
        for (index, count) in self.indexed() {
            counts[index] += count;
        }
        self.entries = Vec::new();
        self.open = Some(counts);
    }

    /// The heatmap's rows of this second: each non empty row as
    /// `index, count`, flattened, by ascending index.
    fn heat_cells(&self) -> Vec<u64> {
        let mut cells: Vec<u64> = Vec::new();
        for (index, count) in self.buckets() {
            let (index, count) = (index as u64, count as u64);
            // A row past COUNT_MAX continues in the next entry.
            match cells.len().checked_sub(2).map(|at| cells[at]) {
                Some(last) if last == index => *cells.last_mut().expect("a count") += count,
                _ => cells.extend([index, count]),
            }
        }
        cells
    }

    /// Non empty fixed rows as `(index, count)`, fastest first. A row past
    /// [`COUNT_MAX`] appears more than once, consecutively, unless the
    /// second is open.
    fn indexed(&self) -> impl Iterator<Item = (usize, usize)> + '_ {
        let open = self.open.iter().flat_map(|open| {
            open.iter()
                .enumerate()
                .filter(|&(_, &count)| count > 0)
                .map(|(index, &count)| (index, count))
        });
        let packed = self
            .entries
            .iter()
            .map(|&entry| ((entry >> COUNT_BITS) as usize, (entry & COUNT_MAX) as usize));
        open.chain(packed)
    }

    /// Non empty rows, fixed and past them, as `(index, count)`, fastest
    /// first. A row past [`COUNT_MAX`] may appear more than once,
    /// consecutively.
    fn buckets(&self) -> impl Iterator<Item = (usize, usize)> + '_ {
        self.indexed().chain(
            self.overflow
                .iter()
                .map(|&(row, count)| (row, count as usize)),
        )
    }

    /// [`Self::percentile_us`] in whole milliseconds, truncated (unit
    /// tests).
    #[cfg(test)]
    pub(crate) fn percentile(&self, percent: f32) -> usize {
        (self.percentile_us(percent) / 1000) as usize
    }

    /// The response time `percent` of these requests finished within, in
    /// microseconds: the middle of its row, kept between the fastest and
    /// slowest, by the rule the whole run's percentiles follow.
    pub(crate) fn percentile_us(&self, percent: f32) -> u64 {
        crate::metrics::response_time_percentile(
            self.buckets()
                .map(|(index, count)| (display_value(index), count)),
            self.counter,
            self.minimum_time_us,
            self.maximum_time_us,
            percent,
        )
    }
}

/// Heatmap row `index` as `[time, lowest, highest]` in microseconds.
fn row_entry_us(index: usize) -> [u64; 3] {
    let (lowest, highest) = display_range(index);
    [display_value(index), lowest, highest]
}

/// The heatmap's row table, `[time, lowest, highest]` in microseconds
/// for each index from 0 to `last`, and always every fixed index.
fn bucket_table_us(last: usize) -> Vec<[u64; 3]> {
    (0..=last.max(FIXED_BUCKETS - 1))
        .map(row_entry_us)
        .collect()
}

/// A new generation for a set of latency histograms: unique in the process
/// and, starting from the clock, unlikely to repeat one a client kept from an
/// earlier Goose.
fn next_heatmap_generation() -> u64 {
    static NEXT: OnceLock<AtomicU64> = OnceLock::new();
    NEXT.get_or_init(|| {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(1, |since| since.as_millis() as u64);
        AtomicU64::new(now_ms * 1000)
    })
    .fetch_add(1, AtomicOrdering::Relaxed)
}

/// One second of a [`LatencyRing`].
#[derive(Clone)]
struct RingSlot {
    second: usize,
    /// The histograms' revision when this second last changed.
    rev: u64,
    data: SecondHistogram,
}

/// Most columns the report's heatmap keeps for all requests.
const ALL_REQUESTS_COLUMNS: usize = 600;
/// Most columns the report's heatmap keeps for each request name with a ring.
const REQUEST_COLUMNS: usize = 60;

/// A series' response times over the whole run, for the report's heatmap:
/// one histogram for each column of a width in seconds that
/// [`LatencyHistograms`] keeps, from its origin. When a series fills its
/// budget, adjacent pairs merge and the width doubles, so memory stops
/// growing however long the run. The newest two columns stay open, counted
/// by fixed index as a ring's newest seconds are, so folding a second into
/// them is additions.
#[derive(Clone, Default)]
pub(crate) struct Columns {
    columns: Vec<SecondHistogram>,
    /// Columns whose histograms are still open.
    open: Vec<usize>,
    /// Emptied counts of packed columns, for new columns to count in.
    #[allow(clippy::vec_box)]
    spare: Vec<Box<[usize; FIXED_BUCKETS]>>,
}

impl Columns {
    /// The histogram of `column`. A column after every other opens, and
    /// those two or more before it pack.
    fn at(&mut self, column: usize) -> &mut SecondHistogram {
        if column >= self.columns.len() {
            self.columns
                .resize_with(column + 1, SecondHistogram::default);
        }
        if column + 1 == self.columns.len() && !self.open.contains(&column) {
            let (columns, spare) = (&mut self.columns, &mut self.spare);
            self.open.retain(|&older| {
                if older + 2 > column {
                    return true;
                }
                spare.extend(columns[older].pack());
                false
            });
            let counts = self
                .spare
                .pop()
                .unwrap_or_else(|| Box::new([0; FIXED_BUCKETS]));
            self.columns[column].reopen(counts);
            self.open.push(column);
        }
        &mut self.columns[column]
    }

    /// The highest heatmap bucket index any column uses.
    fn last_bucket(&self) -> Option<usize> {
        self.columns
            .iter()
            .filter_map(|column| {
                let cells = column.heat_cells();
                cells.len().checked_sub(2).map(|at| cells[at] as usize)
            })
            .max()
    }

    /// Merge adjacent pairs of columns, for a width twice as wide.
    fn double(&mut self) {
        for column in self.open.drain(..) {
            self.spare.extend(self.columns[column].pack());
        }
        let mut pairs = std::mem::take(&mut self.columns).into_iter();
        while let Some(mut first) = pairs.next() {
            if let Some(second) = pairs.next() {
                first.merge(&second);
                first.entries.shrink_to_fit();
            }
            self.columns.push(first);
        }
    }
}

/// Where a ring's seconds go once folded: its own [`Columns`], if it has
/// them, and for a measured ring all requests' columns, each with its width
/// in seconds, all counted from `origin`. Passed in by the ring's caller in
/// [`LatencyHistograms`], which keeps the origin and the widths.
pub(crate) struct ColumnSink<'a> {
    origin: usize,
    width: usize,
    all: Option<(&'a mut Columns, usize)>,
}

impl ColumnSink<'_> {
    /// For a ring that feeds only its own columns, `width` seconds wide.
    fn own(origin: usize, width: usize) -> Self {
        ColumnSink {
            origin,
            width,
            all: None,
        }
    }

    /// For a ring with no columns to feed (unit tests).
    #[cfg(test)]
    fn none() -> Self {
        Self::own(0, 1)
    }

    /// Apply `add` to the column holding `second` in `own`, the ring's own
    /// columns, and in all requests' columns. A second before the origin
    /// goes to the first column.
    fn feed(
        &mut self,
        own: Option<&mut Columns>,
        second: usize,
        mut add: impl FnMut(&mut SecondHistogram),
    ) {
        let since = second.saturating_sub(self.origin);
        if let Some(own) = own {
            add(own.at(since / self.width));
        }
        if let Some((all, width)) = self.all.as_mut() {
            add(all.at(since / *width));
        }
    }
}

/// One response time histogram for each of the last [`SERIES_WINDOW_SECS`]
/// seconds. Second `s` lives in slot `s % len`, tagged with `s` so a slot that
/// still holds an older second is recognized and reused, and with the
/// revision of its last change so a heatmap can send only what changed. The
/// newest seconds stay open, counted by fixed index, until a second two later
/// begins. A ring with whole run columns folds each second into them once,
/// when it packs, and sends anything written to a second after that, or too
/// old for the ring, to them directly.
#[derive(Clone)]
pub(crate) struct LatencyRing {
    slots: Vec<Option<RingSlot>>,
    /// Seconds whose histograms are still open.
    open: Vec<usize>,
    /// Emptied counts of packed seconds, for new seconds to count in, so a
    /// ring does not allocate and free them every second. Boxed, as a
    /// histogram holds them, so they move between the two without copying.
    #[allow(clippy::vec_box)]
    spare: Vec<Box<[usize; FIXED_BUCKETS]>>,
    /// This ring's whole run columns, for the report's heatmap.
    columns: Option<Columns>,
}

impl LatencyRing {
    pub(crate) fn new() -> Self {
        Self::with_columns(false)
    }

    /// A ring that keeps whole run columns when `columns` is set.
    fn with_columns(columns: bool) -> Self {
        LatencyRing {
            slots: (0..SERIES_WINDOW_SECS).map(|_| None).collect(),
            open: Vec::new(),
            spare: Vec::new(),
            columns: columns.then(Columns::default),
        }
    }

    /// The histogram for `second`, replacing an older second's, marked as
    /// changed at `rev`. `None` when the slot already holds a newer second:
    /// the sample is too old to show. Seconds packed here fold into `sink`.
    fn slot_mut(
        &mut self,
        second: usize,
        rev: u64,
        sink: &mut ColumnSink<'_>,
    ) -> Option<&mut SecondHistogram> {
        // The window is a constant, so the remainder needs no division.
        let len = SERIES_WINDOW_SECS as usize;
        debug_assert_eq!(self.slots.len(), len);
        let at = second % len;
        match &self.slots[at] {
            Some(slot) if slot.second > second => return None,
            Some(slot) if slot.second == second => {}
            _ => {
                // A new second has begun: those two or more seconds back
                // have all but their slowest requests, so pack them.
                let (slots, spare, columns) = (&mut self.slots, &mut self.spare, &mut self.columns);
                self.open.retain(|&older| {
                    if older + 2 > second {
                        return true;
                    }
                    if let Some(slot) = &mut slots[older % len] {
                        if slot.second == older {
                            Self::pack_slot(slot, spare, columns.as_mut(), sink);
                        }
                    }
                    false
                });
                let data = match self.spare.pop() {
                    Some(counts) => SecondHistogram::open_with(counts),
                    None => SecondHistogram::open(),
                };
                self.slots[at] = Some(RingSlot { second, rev, data });
                self.open.push(second);
            }
        }
        self.slots[at].as_mut().map(|slot| {
            slot.rev = rev;
            &mut slot.data
        })
    }

    /// Pack an open second and fold it into the columns.
    #[allow(clippy::vec_box)]
    fn pack_slot(
        slot: &mut RingSlot,
        spare: &mut Vec<Box<[usize; FIXED_BUCKETS]>>,
        columns: Option<&mut Columns>,
        sink: &mut ColumnSink<'_>,
    ) {
        spare.extend(slot.data.pack());
        let data = &slot.data;
        sink.feed(columns, slot.second, |column| column.merge(data));
    }

    /// Pack every open second, folding each into the columns, so they hold
    /// every time written so far.
    fn pack_open(&mut self, sink: &mut ColumnSink<'_>) {
        let len = SERIES_WINDOW_SECS as usize;
        for older in self.open.drain(..) {
            if let Some(slot) = &mut self.slots[older % len] {
                if slot.second == older {
                    Self::pack_slot(slot, &mut self.spare, self.columns.as_mut(), sink);
                }
            }
        }
    }

    /// Apply `add` to `second`'s histogram, at `rev`; and to the columns
    /// when the second is already packed, and so folded, or too old for the
    /// ring.
    fn write(
        &mut self,
        second: usize,
        rev: u64,
        sink: &mut ColumnSink<'_>,
        mut add: impl FnMut(&mut SecondHistogram),
    ) {
        let feeds = self.columns.is_some() || sink.all.is_some();
        let folded = match self.slot_mut(second, rev, sink) {
            Some(histogram) => {
                add(histogram);
                histogram.open.is_none()
            }
            None => true,
        };
        if feeds && folded {
            sink.feed(self.columns.as_mut(), second, add);
        }
    }

    /// Record one response time in whole milliseconds in `second`, at
    /// `rev`, for a ring with no columns to feed (unit tests).
    #[cfg(test)]
    pub(crate) fn record(&mut self, second: usize, time: u64, rev: u64) {
        self.record_to(
            second,
            time.saturating_mul(1000),
            rev,
            &mut ColumnSink::none(),
        );
    }

    /// Record one response time in microseconds in `second`, at `rev`.
    fn record_to(&mut self, second: usize, us: u64, rev: u64, sink: &mut ColumnSink<'_>) {
        self.write(second, rev, sink, |histogram| histogram.record_us(us));
    }

    /// Add a batch's response times for one request name to `second`, at
    /// `rev`.
    fn merge_timing_to(
        &mut self,
        second: usize,
        other: &GooseRequestMetricTimingData,
        rev: u64,
        sink: &mut ColumnSink<'_>,
    ) {
        self.write(second, rev, sink, |histogram| histogram.merge_timing(other));
    }

    /// Add a batch's response times for one request name to `second`, at
    /// `rev` (unit tests).
    #[cfg(test)]
    pub(crate) fn merge_timing(
        &mut self,
        second: usize,
        other: &GooseRequestMetricTimingData,
        rev: u64,
    ) {
        self.merge_timing_to(second, other, rev, &mut ColumnSink::none());
    }

    /// Forget every second, and empty the columns.
    pub(crate) fn clear(&mut self) {
        for slot in self.slots.iter_mut() {
            *slot = None;
        }
        self.open.clear();
        if let Some(columns) = self.columns.as_mut() {
            *columns = Columns::default();
        }
    }

    /// Fold every packed second from `from` on into the columns; open
    /// seconds fold in as they pack.
    fn fold_packed(&mut self, from: usize, sink: &mut ColumnSink<'_>) {
        let (slots, columns) = (&self.slots, &mut self.columns);
        for slot in slots.iter().flatten() {
            if slot.second >= from && slot.data.open.is_none() {
                sink.feed(columns.as_mut(), slot.second, |column| {
                    column.merge(&slot.data)
                });
            }
        }
    }

    /// [`Self::slot`], to build and keep its heatmap cells.
    fn slot_at_mut(&mut self, second: usize) -> Option<&mut RingSlot> {
        self.slots[second % SERIES_WINDOW_SECS as usize]
            .as_mut()
            .filter(|slot| slot.second == second)
    }

    fn slot(&self, second: usize) -> Option<&RingSlot> {
        self.slots[second % SERIES_WINDOW_SECS as usize]
            .as_ref()
            .filter(|slot| slot.second == second)
    }

    /// The histogram for `second`, if the ring still holds it (unit tests).
    #[cfg(test)]
    pub(crate) fn get(&self, second: usize) -> Option<&SecondHistogram> {
        self.slot(second).map(|slot| &slot.data)
    }
}

/// Where a heatmap's measured columns come from.
trait MeasuredColumns {
    /// The revision `second` last changed at, if it has response times.
    fn rev(&self, second: usize) -> Option<u64>;
    /// `second`'s heatmap cells.
    fn cells(&mut self, second: usize) -> Vec<u64>;
}

impl MeasuredColumns for LatencyRing {
    fn rev(&self, second: usize) -> Option<u64> {
        self.slot(second).map(|slot| slot.rev)
    }

    fn cells(&mut self, second: usize) -> Vec<u64> {
        self.slot_at_mut(second)
            .map_or_else(Vec::new, |slot| slot.data.heat_cells())
    }
}

/// All requests: every name's ring and the ring of names without one,
/// added up for a column only when a heatmap asks for it.
struct AllRequests<'a> {
    rings: &'a [LatencyRing],
    others: &'a LatencyRing,
    /// Each second's cells as of the revision they were built at.
    cells: &'a mut [Option<(usize, u64, Vec<u64>)>],
}

impl AllRequests<'_> {
    fn slots(&self, second: usize) -> impl Iterator<Item = &RingSlot> + '_ {
        self.rings
            .iter()
            .chain([self.others])
            .filter_map(move |ring| ring.slot(second))
    }
}

impl MeasuredColumns for AllRequests<'_> {
    fn rev(&self, second: usize) -> Option<u64> {
        self.slots(second).map(|slot| slot.rev).max()
    }

    fn cells(&mut self, second: usize) -> Vec<u64> {
        let Some(rev) = self.rev(second) else {
            return Vec::new();
        };
        let at = second % self.cells.len();
        match &self.cells[at] {
            Some((kept, built, cells)) if *kept == second && *built == rev => cells.clone(),
            _ => {
                let mut all = SecondHistogram::open();
                for slot in self.slots(second) {
                    all.merge(&slot.data);
                }
                let cells = all.heat_cells();
                self.cells[at] = Some((second, rev, cells.clone()));
                cells
            }
        }
    }
}

/// Per second response time histograms for the dashboard's heatmap: each of
/// the first [`MAX_REQUEST_SERIES`] request names, the names after them
/// together, all requests as their sum, and the same for synthetic times
/// while coordinated omission mitigation is on. A batch's response times for
/// a name go to one ring; the all requests column of a second is only added
/// up when a heatmap needs it.
#[derive(Clone)]
pub(crate) struct LatencyHistograms {
    /// Requests whose names have no ring of their own.
    others: LatencyRing,
    /// The all requests heatmap's cells, kept by second (see [`AllRequests`]).
    all_cells: Vec<Option<(usize, u64, Vec<u64>)>>,
    /// Each admitted request name's ring, in the order names were admitted.
    rings: Vec<LatencyRing>,
    /// Admitted request names and the index of their ring in `rings`.
    by_request: HashMap<String, usize>,
    synthetic: Option<(LatencyRing, HashMap<String, LatencyRing>)>,
    /// Changes when the rings are cleared, so a client starts over.
    generation: u64,
    /// Counts changes: each record or merge marks the seconds it changes
    /// with the next value.
    rev: u64,
    /// All requests' whole run columns for the report's heatmap, when the
    /// rings keep columns: each name's are in its ring, and synthetic times'
    /// in theirs.
    all_columns: Option<Columns>,
    /// The second every series' columns count from: 0, or the second of the
    /// last metrics reset on the series clock.
    origin: usize,
    /// Seconds in a column of all requests, measured and synthetic.
    all_width: usize,
    /// Seconds in a column of each request name, measured and synthetic.
    request_width: usize,
    /// Most columns of all requests and of each request name.
    budgets: (usize, usize),
    /// The first second that fits in no column at the current widths; past
    /// every second when no columns are kept.
    limit: usize,
}

impl LatencyHistograms {
    /// `synthetic` keeps rings for coordinated omission mitigation's
    /// synthetic times too; pass it when mitigation is not `disabled`.
    /// `columns` keeps whole run columns for the report's heatmap.
    pub(crate) fn new(synthetic: bool, columns: bool) -> Self {
        let mut histograms = LatencyHistograms {
            others: LatencyRing::new(),
            all_cells: (0..SERIES_WINDOW_SECS).map(|_| None).collect(),
            rings: Vec::new(),
            by_request: HashMap::new(),
            synthetic: synthetic.then(|| (LatencyRing::with_columns(columns), HashMap::new())),
            generation: next_heatmap_generation(),
            rev: 0,
            all_columns: columns.then(Columns::default),
            origin: 0,
            all_width: 1,
            request_width: 1,
            budgets: (ALL_REQUESTS_COLUMNS, REQUEST_COLUMNS),
            limit: usize::MAX,
        };
        histograms.set_limit();
        histograms
    }

    /// [`Self::new`] keeping columns, with budgets of `all` columns for all
    /// requests and `request` for each name, so a test sees several merges.
    #[cfg(test)]
    pub(crate) fn with_budgets(synthetic: bool, all: usize, request: usize) -> Self {
        let mut histograms = Self::new(synthetic, true);
        histograms.budgets = (all, request);
        histograms.set_limit();
        histograms
    }

    /// Recompute the first second that fits in no column.
    fn set_limit(&mut self) {
        let (all, request) = self.budgets;
        self.limit = match self.all_columns {
            Some(_) => {
                (self.origin + all * self.all_width).min(self.origin + request * self.request_width)
            }
            None => usize::MAX,
        };
    }

    /// Before writing `second`: double the widths until it fits in a
    /// column. Every second a ring later folds was written, and so checked,
    /// here, and widths never narrow until the next reset.
    #[inline(always)]
    fn fit_columns(&mut self, second: usize) {
        if second >= self.limit {
            self.widen(second);
        }
    }

    /// Double the width of every series of a budget `second` does not fit,
    /// as often as it takes.
    #[cold]
    fn widen(&mut self, second: usize) {
        let (all, request) = self.budgets;
        while second >= self.origin + all * self.all_width {
            self.all_width *= 2;
            let synthetic = self.synthetic.as_mut().map(|(all, _)| all);
            for columns in self
                .all_columns
                .iter_mut()
                .chain(synthetic.and_then(|ring| ring.columns.as_mut()))
            {
                columns.double();
            }
        }
        while second >= self.origin + request * self.request_width {
            self.request_width *= 2;
            let synthetic = self
                .synthetic
                .iter_mut()
                .flat_map(|(_, by_request)| by_request.values_mut());
            for ring in self.rings.iter_mut().chain(synthetic) {
                if let Some(columns) = ring.columns.as_mut() {
                    columns.double();
                }
            }
        }
        self.set_limit();
    }

    /// The revision of a change about to be made.
    fn next_rev(&mut self) -> u64 {
        self.rev += 1;
        self.rev
    }

    /// The index of `key`'s ring, admitting the name while there is room.
    fn ring_index(&mut self, key: &str) -> Option<usize> {
        if let Some(&index) = self.by_request.get(key) {
            return Some(index);
        }
        if self.rings.len() >= MAX_REQUEST_SERIES {
            return None;
        }
        self.rings
            .push(LatencyRing::with_columns(self.all_columns.is_some()));
        self.by_request
            .insert(key.to_string(), self.rings.len() - 1);
        Some(self.rings.len() - 1)
    }

    /// `key`'s pending response times in `entry`, allocated the first time
    /// the request has a response time to record, which is when its ring is
    /// assigned: so the first [`MAX_REQUEST_SERIES`] names seen get rings,
    /// in the order they are seen.
    fn pending<'a>(&mut self, key: &str, entry: &'a mut KeySeries) -> &'a mut PendingLatency {
        let ring = match entry.pending {
            Some(_) => None,
            None => self.ring_index(key),
        };
        entry
            .pending
            .get_or_insert_with(|| Box::new(PendingLatency::new(ring)))
    }

    /// The ring of the name with ring `index`, or of the names without one,
    /// and where its seconds fold: its own columns, at the request width,
    /// and all requests' columns.
    fn ring(&mut self, index: Option<usize>) -> (&mut LatencyRing, ColumnSink<'_>) {
        let ring = match index {
            Some(index) => &mut self.rings[index],
            None => &mut self.others,
        };
        let all_width = self.all_width;
        let sink = ColumnSink {
            origin: self.origin,
            width: self.request_width,
            all: self.all_columns.as_mut().map(|all| (all, all_width)),
        };
        (ring, sink)
    }

    #[cfg(test)]
    fn record(&mut self, key: &str, second: usize, us: u64) {
        self.fit_columns(second);
        let rev = self.next_rev();
        let index = self.ring_index(key);
        let (ring, mut sink) = self.ring(index);
        ring.record_to(second, us, rev, &mut sink);
    }

    /// Record one response time in microseconds for `key`, whose entry is
    /// `entry`.
    fn record_in(&mut self, key: &str, entry: &mut KeySeries, second: usize, us: u64) {
        self.fit_columns(second);
        let index = self.pending(key, entry).ring;
        let rev = self.next_rev();
        let (ring, mut sink) = self.ring(index);
        ring.record_to(second, us, rev, &mut sink);
    }

    /// Merge a batch's timing data for `key` in `second` into the key's
    /// pending counts, moving what they held to its ring first when `second`
    /// is past both seconds they hold. Times for a second before them go to
    /// the ring directly.
    fn merge_pending(
        &mut self,
        key: &str,
        entry: &mut KeySeries,
        second: usize,
        data: &GooseRequestMetricTimingData,
    ) {
        self.fit_columns(second);
        let pending = self.pending(key, entry);
        match self.pending_at(pending, second) {
            Some(at) => pending.add(at, data),
            None => self.merge_older(pending, second, data),
        }
    }

    /// Which of the two seconds `key`'s pending counts hold is `second`,
    /// after moving what they held to its ring when `second` is past both;
    /// `None` when `second` is before them.
    fn pending_at(&mut self, pending: &mut PendingLatency, second: usize) -> Option<usize> {
        if !pending.held {
            pending.held = true;
            pending.second = second;
        }
        if second < pending.second {
            return None;
        }
        if second > pending.second + 1 {
            self.move_pending(pending);
            pending.second = second;
        }
        Some(second - pending.second)
    }

    /// Merge timing data for a second before those `key`'s pending counts
    /// hold straight into its ring.
    fn merge_older(
        &mut self,
        pending: &PendingLatency,
        second: usize,
        data: &GooseRequestMetricTimingData,
    ) {
        let rev = self.next_rev();
        let (ring, mut sink) = self.ring(pending.ring);
        ring.merge_timing_to(second, data, rev, &mut sink);
    }

    /// Move `pending` to its request's ring.
    fn move_pending(&mut self, pending: &mut PendingLatency) {
        let mut rev = self.rev;
        let (ring, mut sink) = self.ring(pending.ring);
        pending.move_to(ring, &mut sink, || {
            rev += 1;
            rev
        });
        self.rev = rev;
    }

    /// Pack every ring's open seconds, folding them into the columns, when
    /// columns are kept.
    fn pack_open(&mut self) {
        if self.all_columns.is_none() {
            return;
        }
        for index in (0..self.rings.len()).map(Some).chain([None]) {
            let (ring, mut sink) = self.ring(index);
            ring.pack_open(&mut sink);
        }
        let (origin, all_width, request_width) = (self.origin, self.all_width, self.request_width);
        if let Some((all, by_request)) = self.synthetic.as_mut() {
            all.pack_open(&mut ColumnSink::own(origin, all_width));
            for ring in by_request.values_mut() {
                ring.pack_open(&mut ColumnSink::own(origin, request_width));
            }
        }
    }

    /// All requests in `second`: every ring's histogram for it added up, if
    /// any ring holds it.
    #[cfg(test)]
    pub(crate) fn all_second(&self, second: usize) -> Option<SecondHistogram> {
        let mut slots = self
            .rings
            .iter()
            .chain([&self.others])
            .filter_map(|ring| ring.slot(second))
            .peekable();
        slots.peek()?;
        let mut all = SecondHistogram::default();
        for slot in slots {
            all.merge(&slot.data);
        }
        Some(all)
    }

    /// Record a synthetic time, in microseconds. A name gets a synthetic
    /// ring only if it has a measured one.
    fn record_synthetic(&mut self, key: &str, second: usize, us: u64) {
        self.fit_columns(second);
        let rev = self.next_rev();
        let (origin, columns) = (self.origin, self.all_columns.is_some());
        let Some((all, by_request)) = self.synthetic.as_mut() else {
            return;
        };
        all.record_to(
            second,
            us,
            rev,
            &mut ColumnSink::own(origin, self.all_width),
        );
        if !self.by_request.contains_key(key) {
            return;
        }
        let mut sink = ColumnSink::own(origin, self.request_width);
        if let Some(ring) = by_request.get_mut(key) {
            ring.record_to(second, us, rev, &mut sink);
        } else {
            let mut ring = LatencyRing::with_columns(columns);
            ring.record_to(second, us, rev, &mut sink);
            by_request.insert(key.to_string(), ring);
        }
    }

    /// Start or stop keeping whole run columns for the report's heatmap
    /// while the run goes on, `now` being the current second. Started, the
    /// columns count from the oldest second the rings can hold and take every
    /// second they hold; stopped, the columns go.
    fn keep_columns(&mut self, keep: bool, now: usize) {
        if keep == self.all_columns.is_some() {
            return;
        }
        let columns = || keep.then(Columns::default);
        self.all_columns = columns();
        for ring in self.rings.iter_mut() {
            ring.columns = columns();
        }
        if let Some((all, by_request)) = self.synthetic.as_mut() {
            all.columns = columns();
            for ring in by_request.values_mut() {
                ring.columns = columns();
            }
        }
        self.origin = now.saturating_sub(SERIES_WINDOW_SECS as usize - 1);
        self.all_width = 1;
        self.request_width = 1;
        self.set_limit();
        if !keep {
            return;
        }
        self.fit_columns(now);
        let origin = self.origin;
        for index in (0..self.rings.len()).map(Some).chain([None]) {
            let (ring, mut sink) = self.ring(index);
            ring.fold_packed(origin, &mut sink);
        }
        let (all_width, request_width) = (self.all_width, self.request_width);
        if let Some((all, by_request)) = self.synthetic.as_mut() {
            all.fold_packed(origin, &mut ColumnSink::own(origin, all_width));
            for ring in by_request.values_mut() {
                ring.fold_packed(origin, &mut ColumnSink::own(origin, request_width));
            }
        }
    }

    /// Clear every ring and its columns and forget every name, so names are
    /// admitted again, and start a new generation. Columns count from
    /// `origin`, the series second of the reset, one second wide.
    fn clear(&mut self, origin: usize) {
        self.others.clear();
        self.all_cells.iter_mut().for_each(|cells| *cells = None);
        self.rings.clear();
        self.by_request.clear();
        if let Some((all, by_request)) = self.synthetic.as_mut() {
            all.clear();
            by_request.clear();
        }
        if let Some(columns) = self.all_columns.as_mut() {
            *columns = Columns::default();
        }
        self.origin = origin;
        self.all_width = 1;
        self.request_width = 1;
        self.set_limit();
        self.generation = next_heatmap_generation();
    }

    /// The heatmap of one measured ring and, when `co_active` and synthetic
    /// rings are kept, its synthetic counterpart, over `start..=end`: only
    /// the seconds changed after `since` when it is a cursor of these
    /// histograms, otherwise every second with data.
    #[allow(clippy::too_many_arguments)]
    ///
    /// `generation` and `rev` are the histograms' own, passed in so the
    /// rings can be borrowed to build and keep each column's cells.
    fn heatmap(
        (generation, rev): (u64, u64),
        measured: &mut impl MeasuredColumns,
        mut synthetic: Option<&mut LatencyRing>,
        start: usize,
        end: usize,
        hold_from: usize,
        since: Option<HeatmapCursor>,
    ) -> Heatmap {
        let since = match since {
            Some(cursor) if cursor.generation == generation && cursor.rev <= rev => cursor.rev,
            _ => 0,
        };
        let mut columns = Vec::new();
        let mut last_index = FIXED_BUCKETS - 1;
        for second in start..=end {
            let real = measured.rev(second);
            let added = synthetic
                .as_deref_mut()
                .and_then(|ring| ring.slot_at_mut(second));
            if !real.is_some_and(|real| real > since) && !added.iter().any(|slot| slot.rev > since)
            {
                continue;
            }
            let real = match real {
                Some(_) => measured.cells(second),
                None => Vec::new(),
            };
            let added = added.map_or_else(Vec::new, |slot| slot.data.heat_cells());
            for cells in [&real, &added] {
                if let Some(&index) = cells.iter().rev().nth(1) {
                    last_index = last_index.max(index as usize);
                }
            }
            columns.push(HeatmapColumn {
                second: second as u64,
                measured: real,
                synthetic: added,
            });
        }
        Heatmap {
            generation,
            rev,
            since,
            start_second: start as u64,
            end_second: end as u64,
            hold_from: hold_from as u64,
            buckets_us: bucket_table_us(last_index),
            columns,
        }
    }

    fn heatmap_all(
        &mut self,
        start: usize,
        end: usize,
        hold_from: usize,
        since: Option<HeatmapCursor>,
        co_active: bool,
    ) -> Heatmap {
        let synthetic = self
            .synthetic
            .as_mut()
            .filter(|_| co_active)
            .map(|(all, _)| all);
        let mut all = AllRequests {
            rings: &self.rings,
            others: &self.others,
            cells: &mut self.all_cells,
        };
        Self::heatmap(
            (self.generation, self.rev),
            &mut all,
            synthetic,
            start,
            end,
            hold_from,
            since,
        )
    }

    fn heatmap_request(
        &mut self,
        key: &str,
        start: usize,
        end: usize,
        hold_from: usize,
        since: Option<HeatmapCursor>,
        co_active: bool,
    ) -> Option<Heatmap> {
        let measured = &mut self.rings[*self.by_request.get(key)?];
        let synthetic = self
            .synthetic
            .as_mut()
            .filter(|_| co_active)
            .and_then(|(_, by_request)| by_request.get_mut(key));
        Some(Self::heatmap(
            (self.generation, self.rev),
            measured,
            synthetic,
            start,
            end,
            hold_from,
            since,
        ))
    }
}

/// The report's response time heatmap, embedded in the report as JSON for
/// its chart script: every series with whole run columns, each column's
/// cells and percentiles worked out here. `S` is the series: a
/// [`SeriesStream`] for the report, which builds each as it is written.
#[derive(Serialize)]
struct ReportHeatmap<S> {
    /// When the test started, in milliseconds since the epoch, as the
    /// report's other graphs count it.
    start: i64,
    /// The Goose host's offset from UTC at the start, in milliseconds, so
    /// the chart shows times in its zone, as the other graphs do.
    offset: i64,
    /// The second, from the start, that column 0 begins at.
    origin: usize,
    /// `[time, lowest, highest]` of each bucket index used, in
    /// microseconds.
    buckets_us: Vec<[u64; 3]>,
    series: S,
}

/// The heatmap's series, built one at a time as they are serialized, so
/// only one series' columns are held at once.
struct SeriesStream<'a> {
    latency: &'a LatencyHistograms,
    co_active: bool,
}

impl Serialize for SeriesStream<'_> {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.collect_seq(
            self.latency
                .report_sources(self.co_active)
                .map(ReportSeries::new),
        )
    }
}

/// One series of the report's heatmap.
#[derive(Serialize)]
struct ReportSeries<'a> {
    /// `All requests`, or a request's `METHOD name`.
    name: &'a str,
    /// Seconds in each column.
    width: usize,
    measured: ReportView,
    /// Measured and synthetic times together, when coordinated omission
    /// mitigation recorded synthetic times and the series has columns of
    /// them.
    adjusted: Option<ReportView>,
}

/// Where a series of the report's heatmap comes from: its name, its column
/// width, its measured columns, and its synthetic columns when it offers
/// an Adjusted view.
type ReportSource<'a> = (&'a str, usize, &'a Columns, Option<&'a Columns>);

impl<'a> ReportSeries<'a> {
    fn new((name, width, measured, added): ReportSource<'a>) -> Self {
        ReportSeries {
            name,
            width,
            measured: ReportView::new(measured, None),
            adjusted: added.map(|added| ReportView::new(measured, Some(added))),
        }
    }
}

/// The columns of one view of a series, and the sparsest and busiest cell
/// counts its colours are scaled over.
#[derive(Serialize)]
struct ReportView {
    min: usize,
    max: usize,
    /// Each column with requests, flattened: its index, its p50, p95 and
    /// p99 in microseconds, then each non empty bucket as its index, its
    /// count and the count's place from 0 (`min`) to 1000 (`max`) on a
    /// logarithmic scale.
    columns: Vec<Vec<usize>>,
}

impl ReportView {
    /// The view of `measured` and, when given, `synthetic` columns added to
    /// them.
    fn new(measured: &Columns, synthetic: Option<&Columns>) -> Self {
        let mut columns = Vec::new();
        let mut adjusted;
        let synthetic_column =
            |index: usize| synthetic.and_then(|columns| columns.columns.get(index));
        let len = measured
            .columns
            .len()
            .max(synthetic.map_or(0, |columns| columns.columns.len()));
        for index in 0..len {
            let measured = measured.columns.get(index);
            let column = match synthetic_column(index) {
                Some(added) => {
                    adjusted = measured.cloned().unwrap_or_default();
                    adjusted.merge(added);
                    &adjusted
                }
                None => match measured {
                    Some(measured) => measured,
                    None => continue,
                },
            };
            if column.counter == 0 {
                continue;
            }
            let mut flat = vec![
                index,
                column.percentile_us(0.5) as usize,
                column.percentile_us(0.95) as usize,
                column.percentile_us(0.99) as usize,
            ];
            for pair in column.heat_cells().chunks(2) {
                flat.extend([pair[0] as usize, pair[1] as usize, 0]);
            }
            columns.push(flat);
        }
        let counts = || {
            columns
                .iter()
                .flat_map(|column| column[4..].chunks(3).map(|cell| cell[1]))
        };
        let (min, max) = (counts().min().unwrap_or(0), counts().max().unwrap_or(0));
        for column in columns.iter_mut() {
            for cell in column[4..].chunks_mut(3) {
                cell[2] = Self::level(cell[1], min, max);
            }
        }
        ReportView { min, max, columns }
    }

    /// Where `count` sits from 0, for `min`, to 1000, for `max`, on a
    /// logarithmic scale, as the dashboard places its colours. When every
    /// cell holds the same count, all are the highest.
    fn level(count: usize, min: usize, max: usize) -> usize {
        if max <= min {
            return 1000;
        }
        let place = (count as f64 / min as f64).ln() / (max as f64 / min as f64).ln();
        (place.clamp(0.0, 1.0) * 1000.0).round() as usize
    }
}

impl LatencyHistograms {
    /// Where each series of the report's heatmap comes from: all requests
    /// first, then each request name with columns, by name. `co_active`
    /// says coordinated omission mitigation recorded synthetic times.
    fn report_sources(&self, co_active: bool) -> impl Iterator<Item = ReportSource<'_>> {
        let synthetic = self.synthetic.as_ref().filter(|_| co_active);
        // All requests' synthetic columns outlive a reset empty, and then
        // offer no view; a name's synthetic ring exists only once it has a
        // time.
        let all = self.all_columns.as_ref().map(|all| {
            let added = synthetic
                .and_then(|(all, _)| all.columns.as_ref())
                .filter(|columns| columns.columns.iter().any(|column| column.counter > 0));
            ("All requests", self.all_width, all, added)
        });
        let mut names: Vec<(&String, &usize)> = self.by_request.iter().collect();
        names.sort();
        let names = names.into_iter().filter_map(move |(name, &index)| {
            let measured = self.rings[index].columns.as_ref()?;
            let added = synthetic
                .and_then(|(_, by_request)| by_request.get(name))
                .and_then(|ring| ring.columns.as_ref());
            Some((name.as_str(), self.request_width, measured, added))
        });
        all.into_iter().chain(names)
    }

    /// The report's heatmap, when whole run columns are kept, with `series`
    /// as its series.
    fn report_heatmap_with<S>(
        &self,
        co_active: bool,
        started: DateTime<Utc>,
        series: S,
    ) -> Option<ReportHeatmap<S>> {
        self.all_columns.as_ref()?;
        let last = self
            .report_sources(co_active)
            .flat_map(|(_, _, measured, added)| [Some(measured), added])
            .flatten()
            .filter_map(Columns::last_bucket)
            .max()
            .unwrap_or(0);
        Some(ReportHeatmap {
            start: started.timestamp() * 1000,
            offset: i64::from(
                Local
                    .offset_from_utc_datetime(&started.naive_utc())
                    .local_minus_utc(),
            ) * 1000,
            origin: self.origin,
            buckets_us: bucket_table_us(last),
            series,
        })
    }

    /// The report's heatmap with every series built (unit tests).
    #[cfg(test)]
    fn report_heatmap(
        &self,
        co_active: bool,
        started: DateTime<Utc>,
    ) -> Option<ReportHeatmap<Vec<ReportSeries<'_>>>> {
        let series = self
            .report_sources(co_active)
            .map(ReportSeries::new)
            .collect();
        self.report_heatmap_with(co_active, started, series)
    }
}

/// Writes JSON that is safe inside a `<script>` element: `<`, `>` and `&`
/// escaped, so no string in it can close the element or start markup, and
/// U+2028 and U+2029 escaped, which older script parsers end a line at.
/// Only strings can hold them, so only string fragments are checked.
struct ScriptFormatter;

impl serde_json::ser::Formatter for ScriptFormatter {
    fn write_string_fragment<W: ?Sized + std::io::Write>(
        &mut self,
        writer: &mut W,
        fragment: &str,
    ) -> std::io::Result<()> {
        let mut from = 0;
        for (at, c) in fragment.char_indices() {
            let escaped = match c {
                '<' => "\\u003c",
                '>' => "\\u003e",
                '&' => "\\u0026",
                '\u{2028}' => "\\u2028",
                '\u{2029}' => "\\u2029",
                _ => continue,
            };
            writer.write_all(&fragment.as_bytes()[from..at])?;
            writer.write_all(escaped.as_bytes())?;
            from = at + c.len_utf8();
        }
        writer.write_all(&fragment.as_bytes()[from..])
    }
}

/// The report heatmap's chart script. It reads the data from the JSON
/// element before it and draws the selected series with an ECharts
/// `custom` series, one rectangle per cell. Request names reach the page
/// only as the text of the select's options.
const REPORT_HEATMAP_SCRIPT: &str = include_str!("report/heatmap.js");

/// The report heatmap card's markup before its JSON.
const REPORT_HEATMAP_HEAD: &str = r#"<div class="graph heatmap">
                <style>
                    .graph.heatmap {
                        break-inside: avoid;
                    }
                    .heatmap-controls button {
                        border: 1px solid #2c664f;
                        background: white;
                        padding: 0.2em 0.8em;
                    }
                    .heatmap-controls button.active {
                        background: #2c664f;
                        color: white;
                    }
                    .heatmap-legend {
                        display: flex;
                        align-items: center;
                        gap: 0.5em;
                        width: 1000px;
                        font-size: 0.9em;
                    }
                    .heatmap-ramp {
                        display: inline-block;
                        width: 200px;
                        height: 0.8em;
                        print-color-adjust: exact;
                        -webkit-print-color-adjust: exact;
                    }
                </style>
                <h3>Response time heatmap</h3>
                <div class="heatmap-controls">
                    <label>Series <select id="graph-latency-heatmap-series"></select></label>
                    <span id="graph-latency-heatmap-views" hidden>
                        <button type="button" data-view="adjusted">Adjusted</button>
                        <button type="button" data-view="measured">Measured</button>
                    </span>
                </div>
                <p id="graph-latency-heatmap-width" class="heatmap-note"></p>
                <div id="graph-latency-heatmap" style="width: 1000px; height:500px; background: white;"></div>
                <div class="heatmap-legend">
                    <span id="graph-latency-heatmap-min"></span>
                    <span id="graph-latency-heatmap-ramp" class="heatmap-ramp"></span>
                    <span id="graph-latency-heatmap-max"></span>
                </div>
                <script type="application/json" id="graph-latency-heatmap-data">"#;
/// The markup between the card's JSON and its chart script.
const REPORT_HEATMAP_MIDDLE: &str = r#"</script>
                <script type="text/javascript">"#;
/// The markup after the card's chart script.
const REPORT_HEATMAP_TAIL: &str = r#"</script>
            </div>"#;

/// The report heatmap's card, with `heatmap` as its JSON, or nothing when it
/// can't be written.
pub(crate) fn heatmap_card<H: Serialize + ?Sized>(heatmap: &H) -> String {
    let mut card = REPORT_HEATMAP_HEAD.as_bytes().to_vec();
    let mut serializer = serde_json::Serializer::with_formatter(&mut card, ScriptFormatter);
    if heatmap.serialize(&mut serializer).is_err() {
        return String::new();
    }
    card.extend_from_slice(REPORT_HEATMAP_MIDDLE.as_bytes());
    card.extend_from_slice(REPORT_HEATMAP_SCRIPT.as_bytes());
    card.extend_from_slice(REPORT_HEATMAP_TAIL.as_bytes());
    card.shrink_to_fit();
    String::from_utf8(card).unwrap_or_default()
}

/// The per second data the HTML report graphs, as the JSON report holds it
/// with `--report-graph-data`: each series as its first second and its
/// values, request names in order, and the heatmap as the HTML report embeds
/// it.
#[derive(Serialize)]
pub(crate) struct ReportGraphs<'a> {
    requests_per_second: BTreeMap<&'a str, &'a TimeSeries<u32, u32>>,
    errors_per_second: BTreeMap<&'a str, &'a TimeSeries<u32, u32>>,
    average_response_time: BTreeMap<&'a str, &'a TimeSeries<MovingAverage, f32>>,
    transactions_per_second: &'a TimeSeries<usize, usize>,
    scenarios_per_second: &'a TimeSeries<usize, usize>,
    users_per_second: &'a TimeSeries<usize, usize>,
    #[serde(skip_serializing_if = "Option::is_none")]
    heatmap: Option<ReportGraphsHeatmap<'a>>,
}

/// The heatmap in the JSON report: built from the histograms, or read back
/// from a JSON report.
enum ReportGraphsHeatmap<'a> {
    Live(ReportHeatmap<SeriesStream<'a>>),
    Recorded(&'a serde_json::Value),
}

impl Serialize for ReportGraphsHeatmap<'_> {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        match self {
            ReportGraphsHeatmap::Live(heatmap) => heatmap.serialize(serializer),
            ReportGraphsHeatmap::Recorded(heatmap) => heatmap.serialize(serializer),
        }
    }
}

/// The graph data of a JSON report read back, to render its HTML report.
#[derive(Deserialize)]
pub(crate) struct RecordedGraphs {
    #[serde(default)]
    requests_per_second: HashMap<String, TimeSeries<u32, u32>>,
    #[serde(default)]
    errors_per_second: HashMap<String, TimeSeries<u32, u32>>,
    #[serde(default)]
    average_response_time: HashMap<String, TimeSeries<MovingAverage, f32>>,
    #[serde(default = "TimeSeries::new")]
    transactions_per_second: TimeSeries<usize, usize>,
    #[serde(default = "TimeSeries::new")]
    scenarios_per_second: TimeSeries<usize, usize>,
    #[serde(default = "TimeSeries::new")]
    users_per_second: TimeSeries<usize, usize>,
    #[serde(default)]
    heatmap: Option<serde_json::Value>,
}

impl RecordedGraphs {
    /// The graph data, and the heatmap as the JSON report holds it.
    pub(crate) fn into_parts(self) -> (GraphData, Option<serde_json::Value>) {
        let items = |series: HashMap<String, TimeSeries<u32, u32>>| ItemsPerSecond {
            series: series
                .into_iter()
                .map(|(name, counts)| {
                    (
                        name,
                        KeySeries {
                            counts,
                            pending: None,
                        },
                    )
                })
                .collect(),
            origin: 0,
        };
        let graph_data = GraphData {
            requests_per_second: items(self.requests_per_second),
            errors_per_second: items(self.errors_per_second),
            average_response_time_per_second: self.average_response_time,
            transactions_per_second: self.transactions_per_second,
            scenarios_per_second: self.scenarios_per_second,
            users_per_second: self.users_per_second,
            latency: None,
        };
        (graph_data, self.heatmap)
    }
}

/// Used to collect graph data during a load test. Cloned for Save report now,
/// which writes a report while the run goes on.
#[derive(Clone)]
pub(crate) struct GraphData {
    /// Counts requests per second for each request type.
    requests_per_second: ItemsPerSecond,
    /// Counts errors per second.
    errors_per_second: ItemsPerSecond,
    /// Maintains average response time per second.
    average_response_time_per_second: HashMap<String, TimeSeries<MovingAverage, f32>>,
    /// Number of transactions at the end of each second of the test.
    transactions_per_second: TimeSeries<usize, usize>,
    /// Number of scenarios at the end of each second of the test.
    scenarios_per_second: TimeSeries<usize, usize>,
    /// Number of users at the end of each second of the test.
    users_per_second: TimeSeries<usize, usize>,
    /// Response time histograms by second for the dashboard's heatmap.
    /// Only the metrics processor's copy has them, and only while the
    /// dashboard tracks percentiles over time.
    latency: Option<LatencyHistograms>,
}

impl GraphData {
    /// Create a new GraphData object.
    pub(crate) fn new() -> Self {
        trace!("new graph");
        GraphData {
            requests_per_second: ItemsPerSecond::new(),
            errors_per_second: ItemsPerSecond::new(),
            average_response_time_per_second: HashMap::new(),
            transactions_per_second: TimeSeries::new(),
            scenarios_per_second: TimeSeries::new(),
            users_per_second: TimeSeries::new(),
            latency: None,
        }
    }

    /// Create GraphData for the metrics processor: with latency histograms
    /// when `latency_series` is set, synthetic ones too when
    /// `synthetic_latency` (coordinated omission mitigation is on), and
    /// whole run columns for the report's heatmap when `latency_columns`.
    pub(crate) fn for_metrics_processor(
        latency_series: bool,
        synthetic_latency: bool,
        latency_columns: bool,
    ) -> Self {
        let mut graph_data = GraphData::new();
        if latency_series {
            graph_data.latency = Some(LatencyHistograms::new(synthetic_latency, latency_columns));
        }
        graph_data
    }

    /// Reset per-second *rate* series after the increase→maintain metrics reset.
    ///
    /// Request/error/latency/transaction/scenario buckets restart so running
    /// metrics and HTML report rates reflect steady-state load only.
    ///
    /// **Active users are a level over the whole run**, not a rate: the users
    /// series is preserved so the live dashboard (and HTML active-users graph)
    /// still show the initial ramp and any later increases/decreases. Wiping
    /// users and reseeding second 0 with the final count made the chart look
    /// like the test had always been at full load.
    ///
    /// When dashboard pruning has already advanced the users series `origin`,
    /// rate series are recreated with that same origin so the next sample at
    /// absolute second *S* does not expand vectors from 0..S.
    ///
    /// The heatmap's whole run columns restart from `series_elapsed_secs`,
    /// the reset's second on the series clock.
    pub(crate) fn reset_rate_series(&mut self, series_elapsed_secs: usize) {
        let origin = self.users_per_second.origin;
        self.requests_per_second = ItemsPerSecond::with_origin(origin);
        self.errors_per_second = ItemsPerSecond::with_origin(origin);
        self.average_response_time_per_second = HashMap::new();
        self.transactions_per_second = TimeSeries::with_origin(origin);
        self.scenarios_per_second = TimeSeries::with_origin(origin);
        if let Some(latency) = self.latency.as_mut() {
            latency.clear(series_elapsed_secs);
        }
    }

    /// Record one measured response time in whole milliseconds for the
    /// heatmap, without counting a request (unit tests).
    #[cfg(test)]
    pub(crate) fn record_latency(&mut self, key: &str, second: usize, time: u64) {
        self.record_latency_us(key, second, time.saturating_mul(1000));
    }

    /// [`Self::record_latency`] in microseconds (unit tests).
    #[cfg(test)]
    pub(crate) fn record_latency_us(&mut self, key: &str, second: usize, us: u64) {
        if let Some(latency) = self.latency.as_mut() {
            latency.record(key, second, us);
        }
    }

    /// [`Self::record_request_per_second_and_latency_us`] with the response
    /// time in whole milliseconds (unit tests).
    #[cfg(test)]
    pub(crate) fn record_request_per_second_and_latency(
        &mut self,
        key: &str,
        second: usize,
        latency: Option<u64>,
    ) {
        self.record_request_per_second_and_latency_us(
            key,
            second,
            latency.map(|time| time.saturating_mul(1000)),
        );
    }

    /// Count one request in `second` and, when `latency` (in microseconds)
    /// is given and the histograms are kept, record its response time for
    /// the heatmap.
    pub(crate) fn record_request_per_second_and_latency_us(
        &mut self,
        key: &str,
        second: usize,
        latency: Option<u64>,
    ) {
        let data = self.requests_per_second.increment(key, second, 1);
        debug!(
            "incremented second {second} for requests per second counter: {}",
            data.get(second)
        );
        if let (Some(time), Some(histograms)) = (latency, self.latency.as_mut()) {
            histograms.record_in(key, data, second, time);
        }
    }

    /// Merge a batch's count of requests in `second` and, when the
    /// histograms are kept and `times` is given, their response times (the
    /// batch's timing data for this request, every time from `second`),
    /// keeping them beside the count the same lookup finds.
    pub(crate) fn record_requests_per_second_and_latency_batch(
        &mut self,
        key: &str,
        second: usize,
        count: u32,
        times: Option<&GooseRequestMetricTimingData>,
    ) {
        let data = self.requests_per_second.increment(key, second, count);
        if let (Some(times), Some(histograms)) = (times, self.latency.as_mut()) {
            if times.counter > 0 {
                histograms.merge_pending(key, data, second, times);
            }
        }
    }

    /// Count a batch's requests for `key` in `second`, and merge their
    /// response times (the batch's timing data for the request, all from
    /// `second`) into `whole_run`, the request's timing data for the whole
    /// run, and, when the histograms are kept, into its pending counts,
    /// walking the batch's buckets once for both.
    pub(crate) fn record_requests_per_second_merging_latency(
        &mut self,
        key: &str,
        second: usize,
        data: &GooseRequestMetricTimingData,
        whole_run: &mut GooseRequestMetricTimingData,
    ) {
        let entry = self
            .requests_per_second
            .increment(key, second, data.counter as u32);
        let Some(histograms) = self.latency.as_mut().filter(|_| data.counter > 0) else {
            return whole_run.merge(data);
        };
        let pending = histograms.pending(key, entry);
        match histograms.pending_at(pending, second) {
            Some(at) => {
                pending.add_totals(at, data);
                whole_run.merge_also(data, |bucket, count| pending.add_bucket(at, bucket, count));
            }
            None => {
                histograms.merge_older(pending, second, data);
                whole_run.merge(data);
            }
        }
    }

    /// Move every request's pending response times to its ring, so the
    /// rings hold every time merged so far.
    fn move_pending_latency(&mut self) {
        let Some(histograms) = self.latency.as_mut() else {
            return;
        };
        for entry in self.requests_per_second.series.values_mut() {
            if let Some(pending) = entry.pending.as_deref_mut().filter(|pending| pending.held) {
                histograms.move_pending(pending);
            }
        }
    }

    /// Before a batch writes response times from seconds up to `second`:
    /// widen the heatmap's whole run columns if they do not reach it.
    pub(crate) fn fit_latency_columns(&mut self, second: usize) {
        if let Some(histograms) = self.latency.as_mut() {
            histograms.fit_columns(second);
        }
    }

    /// The first second of the run the graph data holds: 0, unless the
    /// dashboard's window dropped older seconds.
    pub(crate) fn first_second(&self) -> usize {
        self.users_per_second.origin
    }

    /// Whether the heatmap keeps whole run columns (unit tests).
    #[cfg(test)]
    pub(crate) fn has_latency_columns_for_test(&self) -> bool {
        self.latency
            .as_ref()
            .is_some_and(|latency| latency.all_columns.is_some())
    }

    /// Start or stop keeping the report heatmap's whole run columns while the
    /// run goes on, as Save report is turned on or off; `now` is the current
    /// second on the series clock.
    pub(crate) fn keep_latency_columns(&mut self, keep: bool, now: usize) {
        if keep {
            // Every response time held moves to its ring first, so the
            // columns take it.
            self.move_pending_latency();
        }
        if let Some(histograms) = self.latency.as_mut() {
            histograms.keep_columns(keep, now);
        }
    }

    /// At the end of the run: move every pending response time to its ring
    /// and pack every open second, so the heatmap's whole run columns hold
    /// every time recorded.
    pub(crate) fn finish_latency_columns(&mut self) {
        self.move_pending_latency();
        if let Some(histograms) = self.latency.as_mut() {
            histograms.pack_open();
        }
    }

    /// The report's response time heatmap card, or nothing when whole run
    /// columns are not kept. `co_active` says coordinated omission
    /// mitigation recorded synthetic times, so the card offers Adjusted and
    /// Measured views; `started` is when the test started.
    pub(crate) fn get_latency_heatmap_markup(
        &self,
        co_active: bool,
        started: DateTime<Utc>,
    ) -> String {
        // A run with no requests since the start or the last reset has no
        // heatmap to show.
        let Some(latency) = self.latency.as_ref().filter(|latency| {
            latency
                .all_columns
                .as_ref()
                .is_some_and(|all| all.columns.iter().any(|column| column.counter > 0))
        }) else {
            return String::new();
        };
        let Some(heatmap) =
            latency.report_heatmap_with(co_active, started, SeriesStream { latency, co_active })
        else {
            return String::new();
        };
        // The JSON goes straight into the card, one series at a time, so
        // no other copy of it, or of every series' columns, is ever held.
        heatmap_card(&heatmap)
    }

    /// The report's heatmap, for the JSON report, when whole run columns are
    /// kept and hold a response time.
    fn report_heatmap(
        &self,
        co_active: bool,
        started: DateTime<Utc>,
    ) -> Option<ReportHeatmap<SeriesStream<'_>>> {
        let latency = self.latency.as_ref().filter(|latency| {
            latency
                .all_columns
                .as_ref()
                .is_some_and(|all| all.columns.iter().any(|column| column.counter > 0))
        })?;
        latency.report_heatmap_with(co_active, started, SeriesStream { latency, co_active })
    }

    /// The per second data the HTML report graphs, for the JSON report. A
    /// heatmap read back from a JSON report, `recorded`, goes in as it is.
    pub(crate) fn report_graphs<'a>(
        &'a self,
        co_active: bool,
        started: DateTime<Utc>,
        recorded: Option<&'a serde_json::Value>,
    ) -> ReportGraphs<'a> {
        let counts = |items: &'a ItemsPerSecond| {
            items
                .series
                .iter()
                .map(|(name, series)| (name.as_str(), &series.counts))
                .collect()
        };
        ReportGraphs {
            requests_per_second: counts(&self.requests_per_second),
            errors_per_second: counts(&self.errors_per_second),
            average_response_time: self
                .average_response_time_per_second
                .iter()
                .map(|(name, series)| (name.as_str(), series))
                .collect(),
            transactions_per_second: &self.transactions_per_second,
            scenarios_per_second: &self.scenarios_per_second,
            users_per_second: &self.users_per_second,
            heatmap: match recorded {
                Some(recorded) => Some(ReportGraphsHeatmap::Recorded(recorded)),
                None => self
                    .report_heatmap(co_active, started)
                    .map(ReportGraphsHeatmap::Live),
            },
        }
    }

    /// Whether response times are kept by second for the heatmaps.
    pub(crate) fn records_latency(&self) -> bool {
        self.latency.is_some()
    }

    /// Record one synthetic response time from coordinated omission
    /// mitigation, in whole milliseconds as mitigation makes them, when
    /// synthetic histograms are kept.
    pub(crate) fn record_synthetic_latency(&mut self, key: &str, second: usize, time: u64) {
        if let Some(latency) = self.latency.as_mut() {
            latency.record_synthetic(key, second, time.saturating_mul(1000));
        }
    }

    /// Whether synthetic response times are kept for the heatmap.
    pub(crate) fn records_synthetic_latency(&self) -> bool {
        self.latency
            .as_ref()
            .is_some_and(|latency| latency.synthetic.is_some())
    }

    /// Keys (`"METHOD name"`) of the requests with their own latency series.
    pub(crate) fn latency_series_keys(&self) -> HashSet<&str> {
        self.latency
            .as_ref()
            .map(|latency| latency.by_request.keys().map(String::as_str).collect())
            .unwrap_or_default()
    }

    /// The latency histograms, if kept (unit tests).
    #[cfg(test)]
    pub(crate) fn latency_histograms(&mut self) -> Option<&LatencyHistograms> {
        self.latency.as_ref()
    }

    /// The measured histogram for `second`: all requests when `key` is
    /// `None`, otherwise `key`'s own ring (unit tests).
    #[cfg(test)]
    pub(crate) fn latency_second_for_test(
        &mut self,
        key: Option<&str>,
        second: usize,
    ) -> Option<SecondHistogram> {
        self.move_pending_latency();
        let latency = self.latency.as_ref()?;
        match key {
            None => latency.all_second(second),
            Some(key) => latency.rings[*latency.by_request.get(key)?]
                .get(second)
                .cloned(),
        }
    }

    /// GraphData keeping whole run columns with budgets of `all` columns for
    /// all requests and `request` for each name, and synthetic columns too
    /// when `synthetic` (unit tests).
    #[cfg(test)]
    pub(crate) fn with_latency_budgets(synthetic: bool, all: usize, request: usize) -> Self {
        let mut graph = GraphData::new();
        graph.latency = Some(LatencyHistograms::with_budgets(synthetic, all, request));
        graph
    }

    /// The whole run columns of all requests (`key` `None`) or of `key`, as
    /// `(column, requests)` for each non empty column, and their width
    /// (unit tests).
    #[cfg(test)]
    pub(crate) fn latency_columns_for_test(
        &self,
        key: Option<&str>,
    ) -> Option<(usize, Vec<(usize, usize)>)> {
        let latency = self.latency.as_ref()?;
        let (columns, width) = match key {
            None => (latency.all_columns.as_ref()?, latency.all_width),
            Some(key) => (
                latency.rings[*latency.by_request.get(key)?]
                    .columns
                    .as_ref()?,
                latency.request_width,
            ),
        };
        let counts = columns
            .columns
            .iter()
            .enumerate()
            .filter(|(_, column)| column.counter > 0)
            .map(|(index, column)| (index, column.counter))
            .collect();
        Some((width, counts))
    }

    /// The average response time for `key` in `second`, if recorded (unit
    /// tests).
    #[cfg(test)]
    pub(crate) fn average_in_second_for_test(&self, key: &str, second: usize) -> Option<f32> {
        self.average_response_time_per_second
            .get(key)
            .map(|series| series.get(second))
            .filter(|average| average.count > 0)
            .map(|average| average.average)
    }

    /// How many response times the average for `key` in `second` holds
    /// (unit tests).
    #[cfg(test)]
    pub(crate) fn average_count_in_second_for_test(&self, key: &str, second: usize) -> u32 {
        self.average_response_time_per_second
            .get(key)
            .map_or(0, |series| series.get(second).count)
    }

    /// The requests per second count for `key` in `second` (unit tests).
    #[cfg(test)]
    pub(crate) fn requests_in_second_for_test(&self, key: &str, second: usize) -> u32 {
        self.requests_per_second
            .series
            .get(key)
            .map_or(0, |series| series.get(second))
    }

    /// The errors per second count for `key` in `second` (unit tests).
    #[cfg(test)]
    pub(crate) fn errors_in_second_for_test(&self, key: &str, second: usize) -> u32 {
        self.errors_per_second
            .series
            .get(key)
            .map_or(0, |series| series.get(second))
    }

    /// Last absolute second with a users sample, if any.
    pub(crate) fn users_last_absolute_second(&self) -> Option<usize> {
        self.users_per_second.last_absolute_second()
    }

    /// Record requests per second metric, without latency (unit tests).
    #[cfg(test)]
    pub(crate) fn record_requests_per_second(&mut self, key: &str, second: usize) {
        let value = self
            .requests_per_second
            .initialize_or_increment(key, second, 1);
        debug!("incremented second {second} for requests per second counter: {value}");
    }

    /// Record errors per second metric.
    pub(crate) fn record_errors_per_second(&mut self, key: &str, second: usize) {
        let value = self
            .errors_per_second
            .initialize_or_increment(key, second, 1);

        debug!("incremented second {second} for errors per second counter: {value}");
    }

    /// Take back an error recorded in `second`, for a failed request that
    /// `set_success()` marked a success. Does nothing when the second is no
    /// longer held.
    pub(crate) fn remove_error_per_second(&mut self, key: &str, second: usize) {
        if let Some(series) = self.errors_per_second.series.get_mut(key) {
            series.decrement(second);
        }
    }

    /// Record average response time per second metric.
    #[cfg(test)]
    pub(crate) fn record_average_response_time_per_second(
        &mut self,
        key: String,
        second: usize,
        response_time: u64,
    ) {
        self.record_average_response_time_per_second_us(
            key,
            second,
            response_time.saturating_mul(1000),
        );
    }

    /// Record one response time, in microseconds, in the average response
    /// time per second, which is in milliseconds.
    pub(crate) fn record_average_response_time_per_second_us(
        &mut self,
        key: String,
        second: usize,
        us: u64,
    ) {
        let origin = self.users_per_second.origin;
        let data = self
            .average_response_time_per_second
            .entry(key)
            .or_insert_with(|| TimeSeries::with_origin(origin));
        data.increase_value(second, (us as f64 / 1000.0) as f32);

        debug!(
            "updated second {} for average response time per second: {}",
            second,
            data.get(second).average
        );
    }

    /// Record transactions per second metric.
    pub(crate) fn record_transactions_per_second(&mut self, second: usize) {
        self.transactions_per_second.increase_value(second, 1);

        debug!(
            "incremented second {} for transactions per second counter: {}",
            second,
            self.transactions_per_second.get(second)
        );
    }

    /// Record scenarios per second metric.
    pub(crate) fn record_scenarios_per_second(&mut self, second: usize) {
        self.scenarios_per_second.increase_value(second, 1);

        debug!(
            "incremented second {} for scenarios per second counter: {}",
            second,
            self.scenarios_per_second.get(second)
        );
    }

    /// Record pre-aggregated average response time for batch merging.
    ///
    /// Performs a proper weighted merge of the batch's total response time and
    /// count into the per-second moving average, avoiding the precision loss
    /// that would result from truncating the average to an integer. `next`
    /// is the batch's total response time and count for `second + 1`, merged
    /// through the same lookup.
    pub(crate) fn record_average_response_time_per_second_batch(
        &mut self,
        key: &str,
        second: usize,
        total_time: f64,
        count: u32,
    ) {
        if count == 0 {
            return;
        }
        let origin = self.users_per_second.origin;
        let data = self
            .average_response_time_per_second
            .entry(key.to_string())
            .or_insert_with(|| TimeSeries::with_origin(origin));
        let batch_avg = MovingAverage {
            count,
            average: total_time as f32 / count as f32,
        };
        data.merge_value(second, &batch_avg);
    }

    /// Record multiple transactions per second for batch merging.
    pub(crate) fn record_transactions_per_second_batch(&mut self, second: usize, count: usize) {
        self.transactions_per_second.increase_value(second, count);
    }

    /// Record multiple scenarios per second for batch merging.
    pub(crate) fn record_scenarios_per_second_batch(&mut self, second: usize, count: usize) {
        self.scenarios_per_second.increase_value(second, count);
    }

    /// Records number of users for a current second.
    pub(crate) fn record_users_per_second(&mut self, users: usize, second: usize) {
        self.users_per_second.set_and_maintain_last(second, users);
    }

    /// Record active users at `second` only if time is not going backwards.
    ///
    /// Used by the dashboard snapshot path to pin the current level at "now"
    /// without rewinding past samples (which would recreate mid-ramp V-glitches
    /// when combined with gap-fill).
    pub(crate) fn record_users_monotonic(&mut self, users: usize, second: usize) {
        if let Some(last) = self.users_per_second.last_absolute_second() {
            if second < last {
                return;
            }
        }
        self.users_per_second.set_and_maintain_last(second, users);
    }

    /// Convenience wrapper for unit tests (no end-second hint).
    ///
    /// Production paths always pass a series-clock hint via
    /// [`Self::export_series_window_until`].
    #[cfg(test)]
    pub(crate) fn export_series_window(&mut self, window_secs: u32) -> SeriesWindow {
        self.export_series_window_until(window_secs, None)
    }

    /// [`Self::export_series_window_with_heatmap`] for a client with no
    /// heatmap yet and coordinated omission mitigation without events (unit
    /// tests).
    #[cfg(test)]
    pub(crate) fn export_series_window_until(
        &mut self,
        window_secs: u32,
        end_second_hint: Option<usize>,
    ) -> SeriesWindow {
        self.export_series_window_with_heatmap(window_secs, end_second_hint, false, None)
    }

    /// The trailing window's first and last second, as
    /// [`Self::export_series_window_with_heatmap`] defines them, or `None`
    /// before anything is recorded.
    fn series_window_bounds(
        &self,
        window_secs: u32,
        end_second_hint: Option<usize>,
    ) -> Option<(usize, usize)> {
        if window_secs == 0 {
            return None;
        }
        let mut end_second: Option<usize> = end_second_hint;
        for series in self.requests_per_second.series.values() {
            if let Some(last) = series.last_absolute_second() {
                end_second = Some(end_second.map_or(last, |e| e.max(last)));
            }
        }
        if let Some(last) = self.users_per_second.last_absolute_second() {
            end_second = Some(end_second.map_or(last, |e| e.max(last)));
        }
        let end_second = end_second?;
        Some((
            end_second.saturating_sub(window_secs as usize - 1),
            end_second,
        ))
    }

    /// First second the heatmap does not draw yet: the newest seconds may
    /// still be missing their slowest requests.
    fn latency_hold_from(end_second: usize, end_second_hint: Option<usize>) -> usize {
        end_second_hint.unwrap_or(end_second).saturating_sub(1)
    }

    /// Export a trailing aggregate series window for the live dashboard.
    ///
    /// Normative algorithm:
    /// 1. `end_second` = max second index across RPS keys, users series, and
    ///    optional `end_second_hint` (series clock "now" from the main loop)
    /// 2. `start_second` = `end_second.saturating_sub(window_secs - 1)`
    /// 3. Allocate fixed-length RPS/FPS/latency buffers; for each per-key series
    ///    add its overlapping range into those buffers (and accumulate
    ///    request-count-weighted latency the same way)
    /// 4. Missing RPS/error/latency seconds contribute 0
    /// 5. **Users are a level, not a rate**: missing samples (including trailing
    ///    seconds where request metrics advanced past the last `RecordUsers`)
    ///    hold the last known active-user count — never drop to 0 solely because
    ///    the users series lags RPS by a few seconds.
    /// 6. With latency histograms kept, `heatmap` covers the same seconds:
    ///    the seconds that changed after `heatmap_since` when it is a cursor
    ///    of these histograms, otherwise every second, with synthetic times
    ///    only when `co_active` (coordinated omission mitigation has recorded
    ///    an event) and synthetic histograms are kept. Without latency
    ///    histograms it is `None`.
    ///
    /// `end_second_hint` extends the window through series-clock "now" when the
    /// last stored sample is a few seconds older (main-loop cadence / no recent
    /// requests).
    pub(crate) fn export_series_window_with_heatmap(
        &mut self,
        window_secs: u32,
        end_second_hint: Option<usize>,
        co_active: bool,
        heatmap_since: Option<HeatmapCursor>,
    ) -> SeriesWindow {
        self.move_pending_latency();
        let Some((start_second, end_second)) =
            self.series_window_bounds(window_secs, end_second_hint)
        else {
            let mut empty = SeriesWindow::empty();
            empty.heatmap = self
                .latency
                .as_mut()
                .map(|latency| latency.heatmap_all(0, 0, 0, heatmap_since, co_active));
            return empty;
        };
        let len = end_second - start_second + 1;

        // Allocate fixed-length outputs once, then accumulate per key into the
        // overlapping range. This is O(stored_samples_in_window) rather than
        // O(window_secs × unique_keys), which matters at ~1 Hz with many request
        // names.
        let mut rps = vec![0.0f64; len];
        let mut fps = vec![0.0f64; len];
        let mut avg_latency_ms = vec![0.0f64; len];
        let mut latency_weight = vec![0u64; len];

        for series in self.requests_per_second.series.values() {
            accumulate_rate_series_into_window(series, start_second, end_second, &mut rps);
        }
        for series in self.errors_per_second.series.values() {
            accumulate_rate_series_into_window(series, start_second, end_second, &mut fps);
        }

        // Weighted latency: for each key, walk the overlap of its RPS and latency
        // series with the export window and add into fixed output vectors.
        for (key, latency_series) in &self.average_response_time_per_second {
            let Some(rps_series) = self.requests_per_second.series.get(key) else {
                continue;
            };
            if latency_series.data.is_empty() || rps_series.data.is_empty() {
                continue;
            }
            let series_start = latency_series
                .origin
                .max(rps_series.origin)
                .max(start_second);
            let series_end = (latency_series.origin + latency_series.data.len() - 1)
                .min(rps_series.origin + rps_series.data.len() - 1)
                .min(end_second);
            if series_start > series_end {
                continue;
            }
            for s in series_start..=series_end {
                let count = rps_series.data[s - rps_series.origin] as u64;
                if count == 0 {
                    continue;
                }
                let avg = latency_series.data[s - latency_series.origin].average as f64;
                let idx = s - start_second;
                avg_latency_ms[idx] += avg * count as f64;
                latency_weight[idx] += count;
            }
        }
        for i in 0..len {
            if latency_weight[i] > 0 {
                avg_latency_ms[i] /= latency_weight[i] as f64;
            }
        }

        // Level hold: do not paint "0 users" for trailing seconds where only
        // RPS has advanced (main-loop RecordUsers lags drain_pending). Users are
        // a single series so O(window) is fine.
        let mut users = Vec::with_capacity(len);
        for s in start_second..=end_second {
            users.push(self.users_per_second.get_held(s) as u64);
        }

        let hold_from = Self::latency_hold_from(end_second, end_second_hint);
        let heatmap = self.latency.as_mut().map(|latency| {
            latency.heatmap_all(
                start_second,
                end_second,
                hold_from,
                heatmap_since,
                co_active,
            )
        });

        SeriesWindow {
            start_second: start_second as u64,
            rps,
            fps,
            users,
            avg_latency_ms,
            heatmap,
        }
    }

    /// One request's heatmap for `GET /api/v1/series`, over the same seconds
    /// as [`Self::export_series_window_with_heatmap`] and sent the same way.
    /// `None` when the request has no histograms of its own.
    pub(crate) fn export_request_series(
        &mut self,
        method: String,
        name: String,
        window_secs: u32,
        end_second_hint: Option<usize>,
        co_active: bool,
        since: Option<HeatmapCursor>,
    ) -> Option<RequestSeries> {
        self.latency.as_ref()?;
        self.move_pending_latency();
        let key = format!("{method} {name}");
        let (start_second, end_second) = self
            .series_window_bounds(window_secs, end_second_hint)
            .unwrap_or((0, 0));
        let latency = self.latency.as_mut()?;
        let heatmap = latency.heatmap_request(
            &key,
            start_second,
            end_second,
            Self::latency_hold_from(end_second, end_second_hint),
            since,
            co_active,
        )?;
        Some(RequestSeries {
            method,
            name,
            heatmap,
        })
    }

    /// Drop series samples older than the trailing `keep_secs` window.
    ///
    /// Used when the live dashboard is enabled without `--report-file`, so
    /// multi-hour soaks do not retain per-second history that the dashboard
    /// never exports (export window is [`SERIES_WINDOW_SECS`](crate::metrics::dashboard_snapshot::SERIES_WINDOW_SECS)).
    /// No-op when `keep_secs == 0`.
    ///
    /// After length-based pruning, all per-key series are **aligned** to
    /// `users_per_second.origin`. Length-only prune leaves idle keys (short
    /// `data`, origin still 0) alone; if such a key becomes active again at a
    /// large absolute second, `increase_value` would expand from the stale
    /// origin and defeat the memory bound. Alignment drops pre-window samples
    /// and advances each series origin to the retained window.
    pub(crate) fn prune_to_window(&mut self, keep_secs: u32) {
        let keep = keep_secs as usize;
        if keep == 0 {
            return;
        }
        for series in self.requests_per_second.series.values_mut() {
            series.prune_to_window(keep);
        }
        for series in self.errors_per_second.series.values_mut() {
            series.prune_to_window(keep);
        }
        for series in self.average_response_time_per_second.values_mut() {
            series.prune_to_window(keep);
        }
        self.transactions_per_second.prune_to_window(keep);
        self.scenarios_per_second.prune_to_window(keep);
        self.users_per_second.prune_to_window(keep);

        // Align every series (and the ItemsPerSecond new-key origin) to the
        // retained users window so idle keys cannot re-expand from second 0.
        let origin = self.users_per_second.origin;
        for series in self.requests_per_second.series.values_mut() {
            series.align_origin_to(origin);
        }
        for series in self.errors_per_second.series.values_mut() {
            series.align_origin_to(origin);
        }
        for series in self.average_response_time_per_second.values_mut() {
            series.align_origin_to(origin);
        }
        self.transactions_per_second.align_origin_to(origin);
        self.scenarios_per_second.align_origin_to(origin);
        self.requests_per_second.origin = origin;
        self.errors_per_second.origin = origin;
    }

    /// Max stored samples across RPS/users series (unit tests / diagnostics).
    #[cfg(test)]
    pub(crate) fn max_series_len_for_test(&self) -> usize {
        let mut max_len = self.users_per_second.data.len();
        for series in self.requests_per_second.series.values() {
            max_len = max_len.max(series.data.len());
        }
        for series in self.errors_per_second.series.values() {
            max_len = max_len.max(series.data.len());
        }
        max_len
            .max(self.transactions_per_second.data.len())
            .max(self.scenarios_per_second.data.len())
    }

    /// Generate active users graph.
    pub(crate) fn get_active_users_graph(&self, granular_data: bool) -> Graph<'_, usize, usize> {
        self.create_graph_from_single_data(
            "graph-active-users",
            "Active users #",
            granular_data,
            &self.users_per_second,
        )
    }

    /// Generate requests per second graph.
    pub(crate) fn get_requests_per_second_graph(&self, granular_data: bool) -> Graph<'_, u32, u32> {
        self.requests_per_second
            .graph("graph-rps", "Requests #", granular_data)
    }

    /// Generate average response time graph.
    pub(crate) fn get_average_response_time_graph(
        &self,
        granular_data: bool,
    ) -> Graph<'_, MovingAverage, f32> {
        self.create_graph_from_data(
            "graph-avg-response-time",
            "Response time [ms]",
            granular_data,
            &self.average_response_time_per_second,
        )
    }

    /// Generate active transactions graph.
    pub(crate) fn get_transactions_per_second_graph(
        &self,
        granular_data: bool,
    ) -> Graph<'_, usize, usize> {
        self.create_graph_from_single_data(
            "graph-tps",
            "Transactions #",
            granular_data,
            &self.transactions_per_second,
        )
    }

    /// Generate active scenarios graph.
    pub(crate) fn get_scenarios_per_second_graph(
        &self,
        granular_data: bool,
    ) -> Graph<'_, usize, usize> {
        self.create_graph_from_single_data(
            "graph-sps",
            "Scenarios #",
            granular_data,
            &self.scenarios_per_second,
        )
    }

    /// Generate errors per second graph.
    pub(crate) fn get_errors_per_second_graph(&self, granular_data: bool) -> Graph<'_, u32, u32> {
        self.errors_per_second
            .graph("graph-eps", "Errors #", granular_data)
    }

    /// Creates a Graph from granular data.
    fn create_graph_from_data<
        'a,
        T: Clone + TimeSeriesValue<T, U>,
        U: Serialize + Copy + PartialEq + PartialOrd,
    >(
        &self,
        html_id: &'a str,
        y_axis_label: &'a str,
        granular_data: bool,
        data: &'a HashMap<String, TimeSeries<T, U>>,
    ) -> Graph<'a, T, U> {
        Graph::new(html_id, y_axis_label, granular_data, data)
    }

    /// Creates a Graph from single (just total numbers, not granular) data.
    fn create_graph_from_single_data<
        'a,
        T: Clone + TimeSeriesValue<T, U>,
        U: Serialize + Copy + PartialEq + PartialOrd,
    >(
        &self,
        html_id: &'a str,
        y_axis_label: &'a str,
        granular_data: bool,
        data: &'a TimeSeries<T, U>,
    ) -> Graph<'a, T, U> {
        Graph {
            html_id,
            y_axis_label,
            granular_data,
            left_out: None,
            data: vec![("Total", data)],
        }
    }
}

/// Defines the HTML graph data.
#[derive(Debug)]
pub(crate) struct Graph<'a, T: Clone + TimeSeriesValue<T, U>, U: Serialize + Copy> {
    /// HTML ID of the graph's main wrapper.
    html_id: &'a str,
    /// Label of the y axis.
    y_axis_label: &'a str,
    /// Indicates whether the granular data should be displayed on graphs.
    granular_data: bool,
    /// Graph data, borrowed from [`GraphData`] in the iteration order of its
    /// map, so rendering a report does not copy every series.
    data: Vec<(&'a str, &'a TimeSeries<T, U>)>,
    /// A second, from the start of the run, that every line leaves out.
    left_out: Option<usize>,
}

impl<'a, T: Clone + TimeSeriesValue<T, U>, U: Serialize + Copy + PartialEq + PartialOrd>
    Graph<'a, T, U>
{
    /// Creates a new Graph object.
    #[allow(clippy::too_many_arguments)]
    fn new(
        html_id: &'a str,
        y_axis_label: &'a str,
        granular_data: bool,
        data: &'a HashMap<String, TimeSeries<T, U>>,
    ) -> Graph<'a, T, U> {
        Graph {
            html_id,
            y_axis_label,
            granular_data,
            left_out: None,
            data: data
                .iter()
                .map(|(label, series)| (label.as_str(), series))
                .collect(),
        }
    }

    /// Leaves out the run's final second, `partial`, when the run stopped
    /// partway through it: a count there holds only part of a second, so the
    /// line would end in a drop. Every line leaves out the same second, so a
    /// request whose last one came earlier keeps its last full second.
    pub(crate) fn without_partial_second(mut self, partial: Option<usize>) -> Self {
        self.left_out = partial;
        self
    }

    /// The series with this label, for tests.
    #[cfg(test)]
    fn series(&self, label: &str) -> Option<&TimeSeries<T, U>> {
        self.data
            .iter()
            .find(|(l, _)| *l == label)
            .map(|(_, series)| *series)
    }

    /// Helper function to build HTML charts powered by the
    /// [ECharts](https://echarts.apache.org) library.
    pub(crate) fn get_markup(
        self,
        history: &[TestPlanHistory],
        test_started_time: DateTime<Utc>,
    ) -> String {
        let mut steps = String::new();
        for step in history.windows(2) {
            let started = Local
                .timestamp_opt(step[0].timestamp.timestamp(), 0)
                // @TODO: Error handling
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S")
                .to_string();
            let stopped = Local
                .timestamp_opt(step[1].timestamp.timestamp(), 0)
                // @TODO: Error handling
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S")
                .to_string();
            match &step[0].action {
                // For increasing show the current number of users to the new number of users.
                TestPlanStepAction::Increasing => {
                    let _ = write!(
                        steps,
                        r#"[
                            {{
                                xAxis: '{started}',
                                itemStyle: {{ borderColor: 'rgba(44, 102, 79, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{started}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{started}',
                                itemStyle: {{ color: 'rgba(44, 102, 79, 0.05)' }},
                            }},
                            {{
                                xAxis: '{stopped}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{stopped}',
                                itemStyle: {{ borderColor: 'rgba(44, 102, 79, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{stopped}'
                            }}
                        ],"#,
                    );
                }
                // For decreasing show the new number of users from the current number of users.
                TestPlanStepAction::Decreasing | TestPlanStepAction::Canceling => {
                    let _ = write!(
                        steps,
                        r#"[
                            {{
                                xAxis: '{started}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{started}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{started}',
                                itemStyle: {{ color: 'rgba(179, 65, 65, 0.05)' }},
                            }},
                            {{
                                xAxis: '{stopped}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{stopped}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{stopped}'
                            }}
                        ],"#,
                    );
                }
                _ => {}
            }
        }

        let mut total_values: TimeSeries<T, U> = TimeSeries::new();
        if self.data.is_empty() {
            "<!-- no data, no legend -->".to_string()
        } else {
            let (legend, main_label, main_values, other_values) = if self.data.len() > 1 {
                // If we are dealing with a metric with granular data we need to calculate totals.
                for (_, single_data) in &self.data {
                    total_values.add_time_series(single_data);
                }

                // We will have multiple lines. We need to prepare the legend section on the graph
                // and create data series for all of them.
                let mut legend = vec!["Total"];

                let mut other_values = String::new();
                if self.granular_data {
                    // In order for this to sort correctly we need to flip label and time series since tuples
                    // are sorted lexicographically so we want time series to be the first element of the tuple.
                    for (sub_data, label) in self
                        .data
                        .iter()
                        .map(|&(label, sub_data)| (sub_data, label))
                        .sorted()
                        .rev()
                    {
                        legend.push(label);

                        let formatted_line = format!(
                            r#"{{
                                name: '{label}',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: {values},
                            }},
                        "#,
                            label = label,
                            values = json!(self.add_timestamp_to_html_graph_data(
                                &sub_data.get_graph_data(),
                                sub_data.origin,
                                test_started_time
                            ))
                        );
                        other_values += formatted_line.as_str();
                    }
                    (
                        format!(
                            r#"legend: {{
                            type: '{legend_type}',
                            width: '75%',
                            data: {data},
                        }},"#,
                            legend_type = if self.data.len() > 4 {
                                "scroll"
                            } else {
                                "plain"
                            },
                            data = json!(legend)
                        ),
                        "Total",
                        &total_values,
                        other_values,
                    )
                } else {
                    ("".to_string(), "Total", &total_values, "".to_string())
                }
            } else {
                // If there is only one data series in the metric we simply display it.
                (
                    "".to_string(),
                    self.data[0].0,
                    self.data[0].1,
                    "".to_string(),
                )
            };

            format!(
                r#"<div class="graph">
                <div id="{html_id}" style="width: 1000px; height:500px; background: white;"></div>

                <script type="text/javascript">
                    var chartDom = document.getElementById('{html_id}');
                    var myChart = echarts.init(chartDom);

                    myChart.setOption({{
                        color: ['#2c664f', '#5470c6', '#91cc75', '#fac858', '#ee6666', '#73c0de', '#3ba272', '#fc8452', '#9a60b4', '#ea7ccc'],
                        tooltip: {{ trigger: 'axis' }},
                        toolbox: {{
                            feature: {{
                                dataZoom: {{ yAxisIndex: 'none' }},
                                restore: {{}},
                                saveAsImage: {{}}
                            }}
                        }},
                        dataZoom: [
                            {{
                                type: 'inside',
                                start: 0,
                                end: 100,
                                fillerColor: 'rgba(34, 80, 61, 0.25)',
                                selectedDataBackground: {{
                                    lineStyle: {{ color: '#2c664f' }},
                                    areaStyle: {{ color: '#378063' }}
                                }}
                            }},
                            {{
                                start: 0,
                                end: 100,
                                fillerColor: 'rgba(34, 80, 61, 0.25)',
                                selectedDataBackground: {{
                                    lineStyle: {{ color: '#2c664f' }},
                                    areaStyle: {{ color: '#378063' }}
                                }}
                            }},
                        ],
                        xAxis: {{ type: 'time' }},
                        yAxis: {{
                            name: '{y_axis_label}',
                            nameLocation: 'center',
                            nameRotate: 90,
                            nameGap: 45,
                            type: 'value'
                        }},
                        {legend}
                        series: [
                            {{
                                name: '{main_label}',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                lineStyle: {{ color: '#2c664f' }},
                                areaStyle: {{ color: '#378063' }},
                                markArea: {{
                                    data: [
                                        {steps}
                                    ]
                                }},
                                data: {main_values},
                            }},
                            {other_values}
                        ]
                    }});
                </script>
            </div>"#,
                html_id = self.html_id,
                main_values = json!(self.add_timestamp_to_html_graph_data(
                    &main_values.get_graph_data(),
                    main_values.origin,
                    test_started_time
                )),
                y_axis_label = self.y_axis_label,
                main_label = main_label,
                legend = legend,
                other_values = other_values,
                steps = steps,
            )
        }
    }

    /// Adds timestamps to the graph data series to ensure correct time display on x axis.
    ///
    /// Will take a vector of (generally numerical) values and convert them into tuples where
    /// the second element will be the data point and the first element will be formatted time
    /// it belongs to.
    fn add_timestamp_to_html_graph_data(
        &self,
        data: &[Option<U>],
        origin: usize,
        started: DateTime<Utc>,
    ) -> Vec<(String, U)> {
        // A second no clock can show, from an origin read back from a JSON
        // report, is left out rather than drawn.
        data.iter()
            .enumerate()
            .filter_map(|(index, value)| {
                let second = origin
                    .checked_add(index)
                    .filter(|&second| Some(second) != self.left_out)
                    .and_then(|second| i64::try_from(second).ok())
                    .and_then(|second| second.checked_add(started.timestamp()))?;
                let time = Local.timestamp_opt(second, 0).single()?;
                value.map(|data| (time.format("%Y-%m-%d %H:%M:%S").to_string(), data))
            })
            .collect::<Vec<_>>()
    }
}

/// Add one rate series' samples into a fixed-length export window buffer.
///
/// Only visits the absolute-second overlap between the series storage and
/// `[start_second, end_second]`. Missing buckets stay at their pre-zeroed value.
fn accumulate_rate_series_into_window(
    series: &TimeSeries<u32, u32>,
    start_second: usize,
    end_second: usize,
    out: &mut [f64],
) {
    if series.data.is_empty() {
        return;
    }
    let series_end = series.origin + series.data.len() - 1;
    let overlap_start = start_second.max(series.origin);
    let overlap_end = end_second.min(series_end);
    if overlap_start > overlap_end {
        return;
    }
    for s in overlap_start..=overlap_end {
        out[s - start_second] += series.data[s - series.origin] as f64;
    }
}

/// Data structure to represent time series data.
#[derive(Debug, Clone)]
struct TimeSeries<T: TimeSeriesValue<T, U>, U> {
    /// Time series data.
    ///
    /// Each element represents the value for one second in the time series.
    /// `data[i]` holds the sample for absolute second `origin + i`.
    ///
    /// Stored as a [`VecDeque`] so dashboard pruning can drop a prefix with
    /// amortized O(dropped) cost (`pop_front`) instead of O(retained) shifts
    /// from `Vec::drain(0..n)` on every 1 Hz prune across many request keys.
    data: VecDeque<T>,
    /// Absolute second corresponding to `data[0]`. Non-zero only after pruning
    /// (dashboard-only long soaks). Report-file collection never prunes, so this
    /// stays 0 and indices remain absolute seconds from test start.
    origin: usize,
    /// Total value of the time series (sum of all elements).
    total: T,
    /// Phantom data indicates to the compiler that the "U" generic data type has zero size.
    phantom: PhantomData<U>,
}

impl<T: Clone + TimeSeriesValue<T, U>, U: PartialEq + PartialOrd> Ord for TimeSeries<T, U> {
    fn cmp(&self, other: &Self) -> Ordering {
        let self_total = self.total();
        let other_total = other.total();

        if self_total > other_total {
            Ordering::Greater
        } else if self_total < other_total {
            Ordering::Less
        } else {
            Ordering::Equal
        }
    }
}

impl<T: Clone + TimeSeriesValue<T, U>, U: PartialEq + PartialOrd> PartialOrd for TimeSeries<T, U> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<T: Clone + TimeSeriesValue<T, U>, U: PartialEq> Eq for TimeSeries<T, U> {}

impl<T: Clone + TimeSeriesValue<T, U>, U: PartialEq> PartialEq for TimeSeries<T, U> {
    fn eq(&self, other: &Self) -> bool {
        self.total() == other.total()
    }
}

impl<T: Clone + TimeSeriesValue<T, U>, U> TimeSeries<T, U> {
    /// Creates a new TimeSeries object.
    fn new() -> TimeSeries<T, U> {
        Self::with_origin(0)
    }

    /// Creates an empty series whose first bucket is absolute second `origin`.
    ///
    /// Used after dashboard pruning / rate reset so inserts at absolute second *S*
    /// store at index `S - origin` instead of expanding from 0.
    fn with_origin(origin: usize) -> TimeSeries<T, U> {
        TimeSeries {
            data: VecDeque::new(),
            origin,
            phantom: PhantomData,
            total: T::initial_value(),
        }
    }

    /// Last absolute second that has a stored sample, if any.
    fn last_absolute_second(&self) -> Option<usize> {
        if self.data.is_empty() {
            None
        } else {
            Some(self.origin + self.data.len() - 1)
        }
    }

    /// Drop samples older than the trailing `keep_secs` window (by absolute second).
    ///
    /// After pruning, `origin` advances and `data` holds only the retained window so
    /// multi-hour dashboard-only soaks do not grow unbounded. No-op when the series
    /// is already within the window.
    ///
    /// Uses repeated [`VecDeque::pop_front`] so cost is O(dropped) rather than
    /// O(retained) from a front `Vec::drain` — important when many per-key series
    /// are pruned at ~1 Hz after already sitting at the keep window.
    ///
    /// **Limitation:** when `data.len() <= keep_secs` this is a no-op even if the
    /// series still has a stale low `origin` (idle keys). Callers that need a
    /// shared window must follow up with [`Self::align_origin_to`].
    fn prune_to_window(&mut self, keep_secs: usize) {
        if keep_secs == 0 || self.data.len() <= keep_secs {
            return;
        }
        let drop = self.data.len() - keep_secs;
        for _ in 0..drop {
            self.data.pop_front();
        }
        self.origin = self.origin.saturating_add(drop);
        // `total` still reflects historical increments (including dropped samples).
        // Dashboard export uses per-second buckets only; HTML reports never prune.
    }

    /// Advance `origin` to `target_origin`, dropping any samples strictly before it.
    ///
    /// Complements length-based [`Self::prune_to_window`]: idle per-key series keep
    /// a short `data` buffer and never length-prune, so without this step a late
    /// reappearance at absolute second *S* would expand from the stale origin.
    /// No-op when `target_origin <= self.origin`.
    fn align_origin_to(&mut self, target_origin: usize) {
        if target_origin <= self.origin {
            return;
        }
        if self.data.is_empty() {
            self.origin = target_origin;
            return;
        }
        let last = self.origin + self.data.len() - 1;
        if last < target_origin {
            // Entire series is before the retained window.
            self.data.clear();
            self.origin = target_origin;
            return;
        }
        let drop = target_origin - self.origin;
        for _ in 0..drop {
            self.data.pop_front();
        }
        self.origin = target_origin;
    }

    /// Increases the the value for a given second.
    fn increase_value(&mut self, second: usize, value: U) {
        // Samples that fall before the pruned window are dropped (dashboard-only).
        if second < self.origin {
            return;
        }
        self.expand(second, T::initial_value());
        let idx = second - self.origin;
        self.data[idx].increase_value(&value);
        self.total.increase_value(&value);
    }

    /// Merges a pre-aggregated value into a given second.
    fn merge_value(&mut self, second: usize, value: &T) {
        // Samples that fall before the pruned window are dropped (dashboard-only).
        if second < self.origin {
            return;
        }
        self.expand(second, T::initial_value());
        let idx = second - self.origin;
        self.data[idx].merge(value);
        self.total.merge(value);
    }

    /// Adds another time series.
    fn add_time_series(&mut self, other: &TimeSeries<T, U>) {
        for (offset, other_item) in other.data.iter().enumerate() {
            let second = other.origin + offset;
            if second < self.origin {
                continue;
            }
            self.expand(second, T::initial_value());
            let idx = second - self.origin;
            self.data.get_mut(idx).unwrap().merge(other_item);
            self.total.merge(other_item);
        }
    }

    /// Sets a value for a given second and maintains last recorded value if
    /// there is a gap in the time series.
    fn set_and_maintain_last(&mut self, second: usize, value: U) {
        if second < self.origin {
            return;
        }
        self.expand(second, self.last());
        self.total.increase_value(&value);
        let idx = second - self.origin;
        self.data[idx].set_value(value);
    }

    /// Returns a value for a given second.
    fn get(&self, second: usize) -> T {
        if second < self.origin {
            return T::initial_value();
        }
        match self.data.get(second - self.origin) {
            Some(value) => value.clone(),
            None => T::initial_value(),
        }
    }

    /// Level-style read: hold the nearest known sample when `second` is outside
    /// the stored range (before origin or past the last sample).
    ///
    /// Used for active-user counts, which are continuous levels. Rate series
    /// (RPS/errors) correctly use [`Self::get`] so missing buckets stay 0.
    fn get_held(&self, second: usize) -> T {
        if self.data.is_empty() {
            return T::initial_value();
        }
        if second < self.origin {
            return self.data[0].clone();
        }
        let last_abs = self.origin + self.data.len() - 1;
        if second > last_abs {
            return self.last();
        }
        self.data[second - self.origin].clone()
    }

    /// Returns the last value in the time series or initial value if the time
    /// series is empty.
    fn last(&self) -> T {
        match self.data.back() {
            Some(last) => last.clone(),
            None => T::initial_value(),
        }
    }

    /// Gets time series suitable for usage in HTML graphs (generally each value
    /// becomes some kind of a scalar).
    ///
    /// Report generation never prunes (`origin == 0`), so the returned vector is
    /// indexed by absolute second from test start. If pruning were used with
    /// reports, callers would need to pad `origin` leading empties; that path is
    /// intentionally unsupported.
    fn get_graph_data(&self) -> Vec<Option<U>> {
        self.data
            .iter()
            .map(|value| value.get_graph_value())
            .collect::<Vec<_>>()
    }

    /// Expands vectors that collect per-second data for HTML report graphs with a
    /// default value.
    ///
    /// We need to do that since we don't know for how long the load test will run
    /// and we can't initialize these vectors at the beginning. It is also
    /// better to do it as we go to save memory.
    ///
    /// `second` is absolute (from test start). Storage index is `second - origin`.
    fn expand(&mut self, second: usize, initial: T) {
        debug_assert!(second >= self.origin);
        let idx = second - self.origin;
        if self.data.len() <= idx {
            for _ in 0..(idx - self.data.len() + 1) {
                self.data.push_back(initial.clone());
                self.total.merge(&initial);
            }
        };
    }

    /// Gets time series total value (sum of all values).
    fn total(&self) -> U {
        self.total.get_total_value()
    }
}

/// A series in the JSON report: its first second and its values.
impl<T: Serialize + TimeSeriesValue<T, U>, U> Serialize for TimeSeries<T, U> {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        use serde::ser::SerializeStruct;
        let mut series = serializer.serialize_struct("TimeSeries", 2)?;
        series.serialize_field("origin", &self.origin)?;
        series.serialize_field("data", &self.data)?;
        series.end()
    }
}

/// A series read back from the JSON report, its total worked out again.
impl<'de, T, U> Deserialize<'de> for TimeSeries<T, U>
where
    T: Deserialize<'de> + Clone + TimeSeriesValue<T, U>,
{
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        struct Stored<T> {
            #[serde(default)]
            origin: usize,
            data: VecDeque<T>,
        }
        let stored = Stored::<T>::deserialize(deserializer)?;
        let mut total = T::initial_value();
        for value in &stored.data {
            total.merge(value);
        }
        Ok(TimeSeries {
            data: stored.data,
            origin: stored.origin,
            total,
            phantom: PhantomData,
        })
    }
}

impl TimeSeries<u32, u32> {
    /// Takes one back from a given second, stopping at zero. Does nothing for
    /// a second the series does not hold.
    fn decrement(&mut self, second: usize) {
        if second < self.origin {
            return;
        }
        if let Some(held) = self.data.get_mut(second - self.origin) {
            if *held > 0 {
                *held -= 1;
                self.total -= 1;
            }
        }
    }
}

/// Defines a single value in a TimeSeries.
pub trait TimeSeriesValue<T, U> {
    /// Initial ("zero") value.
    fn initial_value() -> T;
    /// Adds the given value to the current value.
    fn increase_value(&mut self, value: &U);
    /// Sets the value (and drops existing one if present).
    fn set_value(&mut self, value: U);
    /// Merges (adds) another TimeSeriesValue.
    fn merge(&mut self, other: &T);
    /// Gets representation of the value suitable for HTML graphs (generally a scalar).
    fn get_graph_value(&self) -> Option<U>;
    fn get_total_value(&self) -> U;
}

impl TimeSeriesValue<usize, usize> for usize {
    fn initial_value() -> usize {
        0
    }
    fn increase_value(&mut self, value: &usize) {
        *self += *value;
    }
    fn set_value(&mut self, value: usize) {
        *self = value;
    }
    fn merge(&mut self, other: &usize) {
        *self += *other;
    }
    fn get_graph_value(&self) -> Option<usize> {
        match *self == 0 {
            true => None,
            false => Some(*self),
        }
    }
    fn get_total_value(&self) -> usize {
        *self
    }
}

impl TimeSeriesValue<u32, u32> for u32 {
    fn initial_value() -> u32 {
        0
    }
    fn increase_value(&mut self, value: &u32) {
        *self += *value;
    }
    fn set_value(&mut self, value: u32) {
        *self = value;
    }
    fn merge(&mut self, other: &u32) {
        *self += *other;
    }
    fn get_graph_value(&self) -> Option<u32> {
        match *self == 0 {
            true => None,
            false => Some(*self),
        }
    }
    fn get_total_value(&self) -> u32 {
        *self
    }
}

impl TimeSeriesValue<MovingAverage, f32> for MovingAverage {
    fn initial_value() -> MovingAverage {
        MovingAverage::new()
    }
    fn increase_value(&mut self, value: &f32) {
        self.add_item(value);
    }
    fn set_value(&mut self, _: f32) {
        panic!("TimeSeriesValue::set() is not supported for MovingAverage");
    }
    fn merge(&mut self, other: &MovingAverage) {
        let total_count = self.count + other.count;
        if total_count == 0 {
            self.average = 0.;
        } else {
            self.average = self.average * (self.count as f32 / total_count as f32)
                + other.average * (other.count as f32 / total_count as f32);
        };
        self.count = total_count;
    }
    fn get_graph_value(&self) -> Option<f32> {
        match self.average == 0f32 {
            true => None,
            false => Some(self.average),
        }
    }
    fn get_total_value(&self) -> f32 {
        self.average
    }
}

/// Data structure to maintain moving averages.
///
/// It will maintain the current average and the number of data items that
/// were used to compute it. The JSON report holds it as `[count, average]`.
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(from = "(u32, f32)", into = "(u32, f32)")]
pub(crate) struct MovingAverage {
    /// Number of data items that were used to compute the current average.
    count: u32,
    /// Current average.
    average: f32,
}

impl From<(u32, f32)> for MovingAverage {
    fn from((count, average): (u32, f32)) -> Self {
        MovingAverage { count, average }
    }
}

impl From<MovingAverage> for (u32, f32) {
    fn from(average: MovingAverage) -> Self {
        (average.count, average.average)
    }
}

impl MovingAverage {
    /// Create a new MovingAverage object.
    fn new() -> Self {
        MovingAverage {
            count: 0,
            average: 0.,
        }
    }

    /// Adds a new data item and calculates the new average.
    fn add_item(&mut self, item: &f32) {
        self.count += 1;
        self.average += (*item - self.average) / self.count as f32;
    }
}

impl Default for MovingAverage {
    /// Creates an empty moving average.
    fn default() -> MovingAverage {
        MovingAverage::new()
    }
}

impl Ord for MovingAverage {
    fn cmp(&self, other: &Self) -> Ordering {
        if self.average > other.average {
            Ordering::Greater
        } else if self.average < other.average {
            Ordering::Less
        } else {
            Ordering::Equal
        }
    }
}

impl PartialOrd for MovingAverage {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl PartialEq for MovingAverage {
    fn eq(&self, other: &Self) -> bool {
        self.average == other.average
    }
}

impl Eq for MovingAverage {}

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

    #[test]
    fn test_graph_setters() {
        let mut graph = GraphData::new();
        graph.requests_per_second.insert(
            "GET /",
            TimeSeries {
                origin: 0,
                data: VecDeque::from(vec![123, 234, 345, 456, 567]),
                phantom: PhantomData,
                total: 0,
            },
        );
        graph.users_per_second = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 456, 567, 123, 234]),
            phantom: PhantomData,
            total: 0,
        };
        graph.average_response_time_per_second.insert(
            "GET /".to_string(),
            TimeSeries {
                origin: 0,
                data: VecDeque::from(vec![
                    MovingAverage {
                        count: 123,
                        average: 1.23,
                    },
                    MovingAverage {
                        count: 234,
                        average: 2.34,
                    },
                    MovingAverage {
                        count: 345,
                        average: 3.45,
                    },
                    MovingAverage {
                        count: 456,
                        average: 4.56,
                    },
                    MovingAverage {
                        count: 567,
                        average: 5.67,
                    },
                ]),
                phantom: PhantomData,
                total: MovingAverage {
                    count: 0,
                    average: 0.,
                },
            },
        );
        graph.transactions_per_second = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 123, 234, 456, 567]),
            phantom: PhantomData,
            total: 0,
        };

        graph.scenarios_per_second = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 123, 234, 456, 567]),
            phantom: PhantomData,
            total: 0,
        };

        graph.errors_per_second.insert(
            "GET /",
            TimeSeries {
                origin: 0,
                data: VecDeque::from(vec![567, 123, 234, 345, 456]),
                phantom: PhantomData,
                total: 0,
            },
        );

        let rps_graph = graph.get_requests_per_second_graph(true);
        let expected_time_series: TimeSeries<u32, u32> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![123, 234, 345, 456, 567]),
            phantom: PhantomData,
            total: 0,
        };
        assert_eq!(
            rps_graph.series("GET /").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(rps_graph.html_id, "graph-rps");
        assert_eq!(rps_graph.y_axis_label, "Requests #");

        let users_graph = graph.get_active_users_graph(true);
        let expected_time_series: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 456, 567, 123, 234]),
            phantom: PhantomData,
            total: 0,
        };
        assert_eq!(
            users_graph.series("Total").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(users_graph.html_id, "graph-active-users");
        assert_eq!(users_graph.y_axis_label, "Active users #");

        let avg_rt_graph = graph.get_average_response_time_graph(true);
        let expected_time_series: TimeSeries<MovingAverage, f32> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![
                MovingAverage {
                    count: 123,
                    average: 1.23,
                },
                MovingAverage {
                    count: 234,
                    average: 2.34,
                },
                MovingAverage {
                    count: 345,
                    average: 3.45,
                },
                MovingAverage {
                    count: 456,
                    average: 4.56,
                },
                MovingAverage {
                    count: 567,
                    average: 5.67,
                },
            ]),
            phantom: PhantomData,
            total: MovingAverage {
                count: 0,
                average: 0.,
            },
        };
        assert_eq!(
            avg_rt_graph.series("GET /").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(avg_rt_graph.html_id, "graph-avg-response-time");
        assert_eq!(avg_rt_graph.y_axis_label, "Response time [ms]");

        let transactions_graph = graph.get_transactions_per_second_graph(true);
        let expected_time_series: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 123, 234, 456, 567]),
            phantom: PhantomData,
            total: 0,
        };
        assert_eq!(
            transactions_graph.series("Total").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(transactions_graph.html_id, "graph-tps");
        assert_eq!(transactions_graph.y_axis_label, "Transactions #");

        let scenarios_graph = graph.get_scenarios_per_second_graph(true);
        let expected_time_series: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![345, 123, 234, 456, 567]),
            phantom: PhantomData,
            total: 0,
        };
        assert_eq!(
            scenarios_graph.series("Total").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(scenarios_graph.html_id, "graph-sps");
        assert_eq!(scenarios_graph.y_axis_label, "Scenarios #");

        let errors_graph = graph.get_errors_per_second_graph(true);
        let expected_time_series: TimeSeries<u32, u32> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![567, 123, 234, 345, 456]),
            phantom: PhantomData,
            total: 0,
        };

        assert_eq!(
            errors_graph.series("GET /").unwrap().clone(),
            expected_time_series
        );
        assert_eq!(errors_graph.html_id, "graph-eps");
        assert_eq!(errors_graph.y_axis_label, "Errors #");
    }

    #[test]
    fn test_record_requests_per_second() {
        // Should be initialized with empty requests per second vector.
        let mut graph = GraphData::new();
        assert_eq!(graph.requests_per_second.len(), 0);

        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 1);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 2);
        assert_eq!(
            graph.requests_per_second.get("GET /").unwrap().data.len(),
            3
        );
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[0], 3);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[1], 1);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[2], 5);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().total(), 9);

        graph.record_requests_per_second("GET /", 100);
        graph.record_requests_per_second("GET /", 100);
        graph.record_requests_per_second("GET /", 100);
        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 1);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 5);
        assert_eq!(
            graph.requests_per_second.get("GET /").unwrap().data.len(),
            101
        );
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[0], 4);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[1], 2);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[2], 6);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[3], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[4], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[5], 1);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[100], 3);
        for second in 6..100 {
            assert_eq!(
                graph.requests_per_second.get("GET /").unwrap().data[second],
                0
            );
        }
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().total(), 16);

        graph.record_requests_per_second("GET /user", 0);
        graph.record_requests_per_second("GET /user", 1);
        graph.record_requests_per_second("GET /user", 1);
        graph.record_requests_per_second("GET /", 2);
        graph.record_requests_per_second("GET /", 5);
        assert_eq!(
            graph
                .requests_per_second
                .get("GET /user")
                .unwrap()
                .data
                .len(),
            2
        );
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().data[0],
            1
        );
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().data[1],
            2
        );

        assert_eq!(
            graph.requests_per_second.get("GET /").unwrap().data.len(),
            101
        );
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[0], 4);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[1], 2);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[2], 7);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[3], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[4], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[5], 2);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[100], 3);
        for second in 6..100 {
            assert_eq!(
                graph.requests_per_second.get("GET /").unwrap().data[second],
                0
            );
        }
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().total(), 18);
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().total(),
            3
        );

        graph.record_requests_per_second("GET /user", 100);
        graph.record_requests_per_second("GET /user", 0);
        graph.record_requests_per_second("GET /user", 1);
        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 1);
        assert_eq!(
            graph
                .requests_per_second
                .get("GET /user")
                .unwrap()
                .data
                .len(),
            101
        );
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().data[0],
            2
        );
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().data[1],
            3
        );
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().data[100],
            1
        );
        for second in 6..100 {
            assert_eq!(
                graph.requests_per_second.get("GET /user").unwrap().data[second],
                0
            );
        }

        assert_eq!(
            graph.requests_per_second.get("GET /").unwrap().data.len(),
            101
        );
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[0], 5);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[1], 3);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[2], 7);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[3], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[4], 0);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[5], 2);
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().data[100], 3);
        for second in 6..100 {
            assert_eq!(
                graph.requests_per_second.get("GET /").unwrap().data[second],
                0
            );
        }
        assert_eq!(graph.requests_per_second.get("GET /").unwrap().total(), 20);
        assert_eq!(
            graph.requests_per_second.get("GET /user").unwrap().total(),
            6
        );
    }

    #[test]
    fn test_record_errors_per_second() {
        // Should be initialized with empty errors per second vector.
        let mut graph = GraphData::new();
        assert_eq!(graph.errors_per_second.len(), 0);

        graph.record_errors_per_second("GET /", 0);
        graph.record_errors_per_second("GET /", 0);
        graph.record_errors_per_second("GET /", 0);
        graph.record_errors_per_second("GET /", 1);
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 2);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data.len(), 3);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[0], 3);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[1], 1);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[2], 5);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().total(), 9);

        graph.record_errors_per_second("GET /", 100);
        graph.record_errors_per_second("GET /", 100);
        graph.record_errors_per_second("GET /", 100);
        graph.record_errors_per_second("GET /", 0);
        graph.record_errors_per_second("GET /", 1);
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 5);
        assert_eq!(
            graph.errors_per_second.get("GET /").unwrap().data.len(),
            101
        );
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[0], 4);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[1], 2);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[2], 6);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[3], 0);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[4], 0);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[5], 1);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data[100], 3);
        for second in 6..100 {
            assert_eq!(
                graph.errors_per_second.get("GET /").unwrap().data[second],
                0
            );
        }
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().total(), 16);
    }

    #[test]
    fn test_remove_error_per_second() {
        let mut graph = GraphData::new();
        graph.record_errors_per_second("GET /", 2);
        graph.record_errors_per_second("GET /", 2);

        graph.remove_error_per_second("GET /", 2);
        assert_eq!(graph.errors_in_second_for_test("GET /", 2), 1);
        graph.remove_error_per_second("GET /", 2);
        graph.remove_error_per_second("GET /", 2);
        assert_eq!(graph.errors_in_second_for_test("GET /", 2), 0);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().total(), 0);

        // A second past the series and a key it does not have change nothing.
        graph.remove_error_per_second("GET /", 50);
        assert_eq!(graph.errors_per_second.get("GET /").unwrap().data.len(), 3);
        graph.remove_error_per_second("GET /other", 2);
        assert_eq!(graph.errors_per_second.len(), 1);

        // Nor does a second before a pruned series' origin.
        graph.record_errors_per_second("GET /", 9);
        graph
            .errors_per_second
            .series
            .get_mut("GET /")
            .unwrap()
            .prune_to_window(2);
        graph.remove_error_per_second("GET /", 2);
        assert_eq!(graph.errors_in_second_for_test("GET /", 9), 1);
    }

    #[test]
    #[allow(clippy::float_cmp)]
    fn test_record_average_response_time_per_second() {
        // Should be initialized with empty average response time per second vector.
        let mut graph = GraphData::new();
        assert_eq!(graph.average_response_time_per_second.len(), 0);

        graph.record_average_response_time_per_second("GET /".to_string(), 0, 5);
        graph.record_average_response_time_per_second("GET /".to_string(), 0, 4);
        graph.record_average_response_time_per_second("GET /".to_string(), 0, 3);
        graph.record_average_response_time_per_second("GET /".to_string(), 1, 1);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 4);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 8);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 12);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 4);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 4);
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data
                .len(),
            3
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[0]
                .average,
            4.
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[1]
                .average,
            1.
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[2]
                .average,
            6.4
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .total(),
            5.0000005
        );

        graph.record_average_response_time_per_second("GET /".to_string(), 100, 5);
        graph.record_average_response_time_per_second("GET /".to_string(), 100, 9);
        graph.record_average_response_time_per_second("GET /".to_string(), 100, 7);
        graph.record_average_response_time_per_second("GET /".to_string(), 0, 2);
        graph.record_average_response_time_per_second("GET /".to_string(), 1, 2);
        graph.record_average_response_time_per_second("GET /".to_string(), 2, 5);
        graph.record_average_response_time_per_second("GET /".to_string(), 5, 2);
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data
                .len(),
            101
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[0]
                .average,
            3.5
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[1]
                .average,
            1.5
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[2]
                .average,
            6.166667
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[3]
                .average,
            0.
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[4]
                .average,
            0.
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[5]
                .average,
            2.
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .data[100]
                .average,
            7.
        );
        for second in 6..100 {
            assert_eq!(
                graph
                    .average_response_time_per_second
                    .get("GET /")
                    .unwrap()
                    .data[second]
                    .average,
                0.
            );
        }
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /")
                .unwrap()
                .total(),
            4.8125005
        );
    }

    #[test]
    #[allow(clippy::float_cmp)]
    fn test_record_average_response_time_per_second_batch() {
        let mut graph = GraphData::new();

        // Batch merge: 3 requests with total_time 60.0 → avg 20.0
        graph.record_average_response_time_per_second_batch("GET /", 0, 60.0, 3);
        let avg = graph
            .average_response_time_per_second
            .get("GET /")
            .unwrap()
            .get(0);
        assert_eq!(avg.average, 20.0);
        assert_eq!(avg.count, 3);

        // Add a single individual request with response time 40.
        graph.record_average_response_time_per_second("GET /".to_string(), 0, 40);
        let avg = graph
            .average_response_time_per_second
            .get("GET /")
            .unwrap()
            .get(0);
        // Weighted merge: (3 * 20 + 1 * 40) / 4 = 25.0
        assert_eq!(avg.average, 25.0);
        assert_eq!(avg.count, 4);

        // Another batch at a different second.
        graph.record_average_response_time_per_second_batch("GET /", 5, 100.0, 2);
        let avg = graph
            .average_response_time_per_second
            .get("GET /")
            .unwrap()
            .get(5);
        assert_eq!(avg.average, 50.0);
        assert_eq!(avg.count, 2);

        // Merge a second batch into the same second.
        graph.record_average_response_time_per_second_batch("GET /", 5, 200.0, 2);
        let avg = graph
            .average_response_time_per_second
            .get("GET /")
            .unwrap()
            .get(5);
        // Weighted: (2 * 50 + 2 * 100) / 4 = 75.0
        assert_eq!(avg.average, 75.0);
        assert_eq!(avg.count, 4);
    }

    #[test]
    fn test_record_transactions_per_second() {
        // Should be initialized with empty transactions per second vector.
        let mut graph = GraphData::new();
        assert_eq!(graph.transactions_per_second.data.len(), 0);

        graph.record_transactions_per_second(0);
        graph.record_transactions_per_second(0);
        graph.record_transactions_per_second(0);
        graph.record_transactions_per_second(1);
        graph.record_transactions_per_second(2);
        graph.record_transactions_per_second(2);
        graph.record_transactions_per_second(2);
        graph.record_transactions_per_second(2);
        graph.record_transactions_per_second(2);
        assert_eq!(graph.transactions_per_second.data.len(), 3);
        assert_eq!(graph.transactions_per_second.data[0], 3);
        assert_eq!(graph.transactions_per_second.data[1], 1);
        assert_eq!(graph.transactions_per_second.data[2], 5);
        assert_eq!(graph.transactions_per_second.total(), 9);

        graph.record_transactions_per_second(100);
        graph.record_transactions_per_second(100);
        graph.record_transactions_per_second(100);
        graph.record_transactions_per_second(0);
        graph.record_transactions_per_second(1);
        graph.record_transactions_per_second(2);
        graph.record_transactions_per_second(5);
        assert_eq!(graph.transactions_per_second.data.len(), 101);
        assert_eq!(graph.transactions_per_second.data[0], 4);
        assert_eq!(graph.transactions_per_second.data[1], 2);
        assert_eq!(graph.transactions_per_second.data[2], 6);
        assert_eq!(graph.transactions_per_second.data[3], 0);
        assert_eq!(graph.transactions_per_second.data[4], 0);
        assert_eq!(graph.transactions_per_second.data[5], 1);
        assert_eq!(graph.transactions_per_second.data[100], 3);
        for second in 6..100 {
            assert_eq!(graph.transactions_per_second.data[second], 0);
        }
        assert_eq!(graph.transactions_per_second.total(), 16);
    }

    #[test]
    fn test_record_users_per_second() {
        // Should be initialized with empty transactions per second vector.
        let mut graph = GraphData::new();
        assert_eq!(graph.users_per_second.data.len(), 0);

        graph.record_users_per_second(1, 0);
        graph.record_users_per_second(1, 1);
        graph.record_users_per_second(2, 2);
        assert_eq!(graph.users_per_second.data.len(), 3);
        assert_eq!(graph.users_per_second.data[0], 1);
        assert_eq!(graph.users_per_second.data[1], 1);
        assert_eq!(graph.users_per_second.data[2], 2);
        assert_eq!(graph.users_per_second.total(), 6);

        graph.record_users_per_second(5, 5);
        graph.record_users_per_second(10, 37);
        assert_eq!(graph.users_per_second.data.len(), 38);
        assert_eq!(graph.users_per_second.data[0], 1);
        assert_eq!(graph.users_per_second.data[1], 1);
        assert_eq!(graph.users_per_second.data[2], 2);
        assert_eq!(graph.users_per_second.data[3], 2);
        assert_eq!(graph.users_per_second.data[4], 2);
        assert_eq!(graph.users_per_second.data[5], 5);
        assert_eq!(graph.users_per_second.data[37], 10);
        for second in 6..36 {
            assert_eq!(graph.users_per_second.data[second], 5);
        }
        assert_eq!(graph.users_per_second.total(), 187);
    }

    #[test]
    fn test_export_series_window_empty() {
        let mut graph = GraphData::new();
        let window = graph.export_series_window(300);
        assert_eq!(window, SeriesWindow::empty());
        assert_eq!(window.start_second, 0);
        assert!(window.rps.is_empty());
        assert!(window.fps.is_empty());
        assert!(window.users.is_empty());
        assert!(window.avg_latency_ms.is_empty());
    }

    #[test]
    fn test_export_series_window_multi_key_reduction() {
        let mut graph = GraphData::new();

        // Key A: 2 requests at second 0 (latency 10ms each), 1 request at second 2 (latency 30ms).
        graph.record_requests_per_second("GET /a", 0);
        graph.record_requests_per_second("GET /a", 0);
        graph.record_average_response_time_per_second("GET /a".to_string(), 0, 10);
        graph.record_average_response_time_per_second("GET /a".to_string(), 0, 10);
        graph.record_requests_per_second("GET /a", 2);
        graph.record_average_response_time_per_second("GET /a".to_string(), 2, 30);

        // Key B: 3 requests at second 2 only (latency 50ms), disjoint from A's second-0 traffic.
        graph.record_requests_per_second("GET /b", 2);
        graph.record_requests_per_second("GET /b", 2);
        graph.record_requests_per_second("GET /b", 2);
        graph.record_average_response_time_per_second("GET /b".to_string(), 2, 50);
        graph.record_average_response_time_per_second("GET /b".to_string(), 2, 50);
        graph.record_average_response_time_per_second("GET /b".to_string(), 2, 50);

        // One error on key B at second 2.
        graph.record_errors_per_second("GET /b", 2);

        graph.record_users_per_second(1, 0);
        graph.record_users_per_second(4, 2);

        let window = graph.export_series_window(300);
        assert_eq!(window.start_second, 0);
        assert_eq!(window.rps.len(), 3);
        assert_eq!(window.fps.len(), 3);
        assert_eq!(window.users.len(), 3);
        assert_eq!(window.avg_latency_ms.len(), 3);

        // Second 0: 2 RPS from A only; latency 10ms.
        assert!((window.rps[0] - 2.0).abs() < f64::EPSILON);
        assert!((window.fps[0] - 0.0).abs() < f64::EPSILON);
        assert_eq!(window.users[0], 1);
        assert!((window.avg_latency_ms[0] - 10.0).abs() < 0.001);

        // Second 1: empty → 0 (gap filled by users maintain-last, but RPS/latency 0).
        assert!((window.rps[1] - 0.0).abs() < f64::EPSILON);
        assert!((window.fps[1] - 0.0).abs() < f64::EPSILON);
        // users_per_second.set_and_maintain_last fills gap with last value (1).
        assert_eq!(window.users[1], 1);
        assert!((window.avg_latency_ms[1] - 0.0).abs() < f64::EPSILON);

        // Second 2: 1 (A) + 3 (B) = 4 RPS; 1 error; weighted latency = (30*1 + 50*3)/4 = 45.
        assert!((window.rps[2] - 4.0).abs() < f64::EPSILON);
        assert!((window.fps[2] - 1.0).abs() < f64::EPSILON);
        assert_eq!(window.users[2], 4);
        assert!((window.avg_latency_ms[2] - 45.0).abs() < 0.001);
    }

    #[test]
    fn test_export_series_window_trailing_window_and_empty_seconds() {
        let mut graph = GraphData::new();
        // Activity only at seconds 0 and 5; window of 3 ending at 5 → seconds 3,4,5.
        graph.record_requests_per_second("GET /", 0);
        graph.record_requests_per_second("GET /", 5);
        graph.record_requests_per_second("GET /", 5);
        graph.record_average_response_time_per_second("GET /".to_string(), 0, 100);
        graph.record_average_response_time_per_second("GET /".to_string(), 5, 20);
        graph.record_average_response_time_per_second("GET /".to_string(), 5, 40);
        graph.record_users_per_second(2, 0);
        graph.record_users_per_second(8, 5);

        let window = graph.export_series_window(3);
        assert_eq!(window.start_second, 3);
        assert_eq!(window.rps.len(), 3);
        // Seconds 3 and 4 empty for RPS → 0.
        assert!((window.rps[0] - 0.0).abs() < f64::EPSILON);
        assert!((window.rps[1] - 0.0).abs() < f64::EPSILON);
        assert!((window.rps[2] - 2.0).abs() < f64::EPSILON);
        assert!((window.avg_latency_ms[0] - 0.0).abs() < f64::EPSILON);
        assert!((window.avg_latency_ms[1] - 0.0).abs() < f64::EPSILON);
        // Average of 20 and 40 = 30.
        assert!((window.avg_latency_ms[2] - 30.0).abs() < 0.001);
        // Users: 2 at second 0, then set_and_maintain fills 1..4 with 2 when
        // recording 8 at second 5. Window seconds 3,4,5 → [2, 2, 8].
        assert_eq!(window.users, vec![2, 2, 8]);
    }

    #[test]
    fn test_export_series_users_hold_when_rps_leads() {
        // Regression: Active users chart dropped to 0 near the right edge because
        // end_second came from RPS while users series lagged (RecordUsers is
        // main-loop only; drain_pending can advance RPS further).
        let mut graph = GraphData::new();
        for s in 0..=50 {
            graph.record_users_per_second(10, s);
        }
        // RPS continues through second 56 with no further RecordUsers.
        for s in 0..=56 {
            graph.record_requests_per_second("GET /", s);
        }

        let window = graph.export_series_window(300);
        assert_eq!(window.start_second, 0);
        assert_eq!(window.rps.len(), 57);
        assert_eq!(window.users.len(), 57);
        // Trailing seconds past last RecordUsers must hold 10, not drop to 0.
        for (i, &u) in window.users.iter().enumerate() {
            assert_eq!(u, 10, "users[{i}] should hold last known count 10, got {u}");
        }
        // RPS still present at the end (not all zeros).
        assert!(window.rps[56] > 0.0);
    }

    #[test]
    fn test_export_series_users_monotonic_ramp_without_forward_stamp() {
        // Regression: stamping active_users forward to the RPS/duration second
        // (set_and_maintain_last gap-fill with a stale count, then partial
        // overwrites from real RecordUsers) produced V-shaped glitches mid-ramp.
        // Only sequential RecordUsers samples should define the series; export
        // holds the last known level past the end.
        let mut graph = GraphData::new();
        // Steady 10 users.
        for s in 0..=60 {
            graph.record_users_per_second(10, s);
        }
        // Clean increase 10 → 20 over seconds 61..70.
        for (s, users) in (61..=70).zip(11..=20) {
            graph.record_users_per_second(users, s);
        }
        // RPS leads past the last users sample (as drain_pending does).
        for s in 0..=78 {
            graph.record_requests_per_second("GET /", s);
        }

        let window = graph.export_series_window(300);
        assert_eq!(window.users.len(), 79);
        // Steady region.
        for s in 0..=60 {
            assert_eq!(window.users[s], 10, "steady users[{s}]");
        }
        // Monotonic ramp — no dips.
        for s in 61..=70 {
            let expected = (s - 60 + 10) as u64; // 11..20
            assert_eq!(window.users[s], expected, "ramp users[{s}]");
            if s > 61 {
                assert!(
                    window.users[s] >= window.users[s - 1],
                    "ramp must be non-decreasing at {s}: {} < {}",
                    window.users[s],
                    window.users[s - 1]
                );
            }
        }
        // Trailing hold at 20 (not 0, not a dip).
        for s in 71..=78 {
            assert_eq!(window.users[s], 20, "hold users[{s}]");
        }
    }

    #[test]
    fn test_out_of_order_users_write_creates_v_glitch() {
        // Documents why user stamps must be monotonic in time: gap-fill with a
        // jump forward, then overwriting intermediate seconds, leaves stale dips.
        let mut graph = GraphData::new();
        graph.record_users_per_second(10, 60);
        // Bad stamp: jump to second 70 with live count 20 (fills 61..69 with 10).
        graph.record_users_per_second(20, 70);
        // Real hatch samples only overwrite some intermediate seconds.
        graph.record_users_per_second(15, 65);
        graph.record_users_per_second(18, 68);

        let window = graph.export_series_window(300);
        // 60=10, 61-64=10 (stale fill), 65=15, 66-67=10 (stale!), 68=18, 69=10, 70=20
        assert_eq!(window.users[60], 10);
        assert_eq!(window.users[65], 15);
        assert_eq!(
            window.users[66], 10,
            "stale gap-fill creates a dip after 15"
        );
        assert_eq!(window.users[68], 18);
        assert_eq!(
            window.users[69], 10,
            "stale gap-fill creates a dip before 20"
        );
        assert_eq!(window.users[70], 20);
        // Confirm the V: 15 → 10 → 18 is non-monotonic mid-ramp.
        assert!(window.users[66] < window.users[65] && window.users[66] < window.users[68]);
    }

    #[test]
    fn test_record_users_monotonic_rejects_backwards_writes() {
        let mut graph = GraphData::new();
        graph.record_users_per_second(10, 5);
        graph.record_users_per_second(20, 10);
        // Would create a V-glitch if applied — must be ignored.
        graph.record_users_monotonic(15, 7);
        let window = graph.export_series_window(300);
        assert_eq!(window.users[7], 10, "backwards write must not rewrite past");
        assert_eq!(window.users[10], 20);
        // Forward monotonic stamp is allowed and pins the edge.
        graph.record_users_monotonic(25, 12);
        let window = graph.export_series_window(300);
        assert_eq!(window.users[10], 20);
        assert_eq!(window.users[11], 20);
        assert_eq!(window.users[12], 25);
    }

    #[test]
    fn test_reset_rate_series_preserves_users_history() {
        let mut graph = GraphData::new();
        for (s, users) in (0..=5).zip(1..=6) {
            graph.record_users_per_second(users, s);
        }
        graph.record_requests_per_second("GET /", 3);
        graph.record_errors_per_second("GET /", 3);
        graph.record_average_response_time_per_second("GET /".into(), 3, 42);
        graph.record_transactions_per_second(3);
        graph.record_scenarios_per_second(3);

        graph.reset_rate_series(0);

        let window = graph.export_series_window(300);
        // Users ramp intact.
        assert_eq!(window.users, vec![1, 2, 3, 4, 5, 6]);
        // Rate series wiped.
        assert!(window.rps.iter().all(|&v| v == 0.0));
        assert!(window.fps.iter().all(|&v| v == 0.0));
        assert!(window.avg_latency_ms.iter().all(|&v| v == 0.0));
    }

    #[test]
    fn test_reset_rate_series_preserves_pruned_origin() {
        // After a long dashboard-only soak, prune advances origin. A subsequent
        // increase→maintain reset must not recreate rate series at origin=0 or the
        // next absolute-second sample allocates a multi-hour-sized vector.
        let mut graph = GraphData::new();
        for s in 0..=900 {
            graph.record_users_per_second(10, s);
            graph.record_requests_per_second("GET /", s);
        }
        graph.prune_to_window(600);
        assert!(graph.users_per_second.origin > 0);
        let origin_before = graph.users_per_second.origin;

        graph.reset_rate_series(0);
        assert_eq!(graph.requests_per_second.origin, origin_before);
        assert_eq!(graph.transactions_per_second.origin, origin_before);

        // Sample at a late absolute second: storage length stays near the window,
        // not ~absolute second from 0.
        let late = origin_before + 50;
        graph.record_requests_per_second("GET /new", late);
        let stored = graph
            .requests_per_second
            .series
            .get("GET /new")
            .expect("series")
            .data
            .len();
        assert!(
            stored <= 51,
            "expected compact storage after origin-preserving reset, got len={stored}",
            stored = stored
        );
    }

    #[test]
    fn test_prune_aligns_idle_key_origin_on_reappear() {
        // Idle request names only receive early samples, so length-based prune
        // leaves their origin at 0. After a long soak, re-activating that key
        // must not expand from absolute second 0.
        let mut graph = GraphData::new();
        graph.record_requests_per_second("GET /idle", 0);
        graph.record_requests_per_second("GET /idle", 1);
        graph.record_errors_per_second("GET /idle", 0);
        graph.record_average_response_time_per_second("GET /idle".into(), 0, 10);

        for s in 0..=900 {
            graph.record_users_per_second(10, s);
            graph.record_requests_per_second("GET /active", s);
            graph.record_transactions_per_second(s);
            graph.record_scenarios_per_second(s);
        }
        graph.prune_to_window(600);

        let origin = graph.users_per_second.origin;
        assert!(origin > 0, "users origin should advance after prune");
        assert_eq!(
            graph
                .requests_per_second
                .series
                .get("GET /idle")
                .expect("idle rps")
                .origin,
            origin
        );
        assert_eq!(
            graph
                .errors_per_second
                .series
                .get("GET /idle")
                .expect("idle errors")
                .origin,
            origin
        );
        assert_eq!(
            graph
                .average_response_time_per_second
                .get("GET /idle")
                .expect("idle latency")
                .origin,
            origin
        );

        let late = origin + 50;
        graph.record_requests_per_second("GET /idle", late);
        graph.record_errors_per_second("GET /idle", late);
        graph.record_average_response_time_per_second("GET /idle".into(), late, 12);

        for (label, len) in [
            (
                "rps",
                graph
                    .requests_per_second
                    .series
                    .get("GET /idle")
                    .unwrap()
                    .data
                    .len(),
            ),
            (
                "errors",
                graph
                    .errors_per_second
                    .series
                    .get("GET /idle")
                    .unwrap()
                    .data
                    .len(),
            ),
            (
                "latency",
                graph
                    .average_response_time_per_second
                    .get("GET /idle")
                    .unwrap()
                    .data
                    .len(),
            ),
        ] {
            assert!(
                len <= 51,
                "idle {label} series must stay compact after reappear, got len={len}",
                label = label,
                len = len
            );
        }
    }

    #[test]
    fn test_export_series_window_until_extends_to_hint() {
        let mut graph = GraphData::new();
        graph.record_users_per_second(7, 2);
        // No RPS; without hint end would be second 2.
        let window = graph.export_series_window_until(300, Some(8));
        assert_eq!(window.users.len(), 9); // seconds 0..8
        assert_eq!(window.users[2], 7);
        // Hold last known level through the hint ("now").
        for s in 3..=8 {
            assert_eq!(window.users[s], 7, "hold users[{s}]");
        }
    }

    /// GraphData with latency histograms, synthetic ones too when `synthetic`.
    fn latency_graph(synthetic: bool) -> GraphData {
        GraphData::for_metrics_processor(true, synthetic, false)
    }

    fn histograms(graph: &mut GraphData) -> &LatencyHistograms {
        graph.move_pending_latency();
        graph.latency_histograms().expect("latency histograms")
    }

    #[test]
    fn latency_batch_merge_equals_direct_recording() {
        let times: [(usize, u64); 6] = [(3, 12), (3, 140), (3, 1400), (4, 7), (4, 90), (4, 2600)];
        // Two batches, so the second finds the request's ring through the
        // index cached on its requests per second entry.
        let mut batches: [HashMap<usize, GooseRequestMetricTimingData>; 2] = Default::default();
        for (i, &(second, time)) in times.iter().enumerate() {
            batches[i % 2].entry(second).or_default().record_time(time);
        }
        let mut merged = latency_graph(false);
        for batch in &batches {
            for (second, data) in batch {
                merged.record_requests_per_second_and_latency_batch(
                    "GET /a",
                    *second,
                    data.counter as u32,
                    Some(data),
                );
            }
        }
        let mut direct = latency_graph(false);
        for &(second, time) in &times {
            direct.record_request_per_second_and_latency("GET /a", second, Some(time));
        }
        assert_eq!(
            merged.requests_per_second.get("GET /a").unwrap().data,
            direct.requests_per_second.get("GET /a").unwrap().data
        );
        for second in [3, 4] {
            let merged = histograms(&mut merged);
            let direct = histograms(&mut direct);
            assert_eq!(merged.all_second(second), direct.all_second(second));
            assert_eq!(
                merged.rings[merged.by_request["GET /a"]].get(second),
                direct.rings[direct.by_request["GET /a"]].get(second)
            );
            let mut expected = SecondHistogram::default();
            for &(_, time) in times.iter().filter(|(s, _)| *s == second) {
                expected.record(time);
            }
            assert_eq!(merged.all_second(second), Some(expected.clone()));
        }
    }

    /// Batches merged through a request's pending counts, in and out of
    /// order, with a jump of several seconds, an earlier second and a time
    /// past 60 seconds, give the rings and the heatmap the same response
    /// times as recording each request directly, whenever they are read.
    #[test]
    fn pending_latency_equals_direct_recording() {
        let batches: [(usize, &[u64]); 8] = [
            (5, &[3, 40]),
            (6, &[7]),
            (5, &[61_500, 2]),
            (6, &[0, 900]),
            (9, &[12]),
            (4, &[33]),
            (10, &[5, 5]),
            (9, &[250]),
        ];
        let mut merged = latency_graph(false);
        let mut direct = latency_graph(false);
        for (i, &(second, times)) in batches.iter().enumerate() {
            let mut data = GooseRequestMetricTimingData::new(None);
            for &time in times {
                data.record_time(time);
                direct.record_request_per_second_and_latency("GET /a", second, Some(time));
            }
            merged.record_requests_per_second_and_latency_batch(
                "GET /a",
                second,
                data.counter as u32,
                Some(&data),
            );
            // Read part way through as well as at the end.
            if i == 3 {
                let merged = histograms(&mut merged);
                let direct = histograms(&mut direct);
                for second in 4..=6 {
                    assert_eq!(merged.all_second(second), direct.all_second(second));
                }
            }
        }
        let window = |graph: &mut GraphData| {
            graph
                .export_series_window_until(300, Some(12))
                .heatmap
                .unwrap()
                .columns
        };
        assert_eq!(window(&mut merged), window(&mut direct));
        for second in 3..=11 {
            let merged = histograms(&mut merged);
            let direct = histograms(&mut direct);
            assert_eq!(
                merged.all_second(second),
                direct.all_second(second),
                "{second}"
            );
            assert_eq!(
                merged.rings[merged.by_request["GET /a"]].get(second),
                direct.rings[direct.by_request["GET /a"]].get(second),
                "{second}"
            );
        }
    }

    #[test]
    fn second_histogram_matches_timing_data() {
        // Every row maps to its own index and back, and the rows meet end to
        // end, past the fixed ones too.
        for index in 0..FIXED_BUCKETS + 50 {
            let (lowest, highest) = display_range(index);
            assert_eq!(display_index(lowest), index, "{index}");
            assert_eq!(display_index(highest), index, "{index}");
            assert_eq!(display_index(highest + 1), index + 1, "{index}");
            assert_eq!(display_index(display_value(index)), index, "{index}");
        }
        assert_eq!(display_index(60_999_999), FIXED_BUCKETS - 1);

        // The same totals as the whole run timing data, each bucket in the
        // row of its last microsecond, and each percentile in the row the
        // whole run's percentile bucket is drawn in, including times past 61
        // seconds (a longer --timeout).
        let mut times: Vec<u64> = Vec::new();
        let mut seed: u64 = 7;
        for _ in 0..5000 {
            seed = seed
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            times.push((seed >> 33) % 90_000 / (1 + (seed >> 60)));
        }
        let mut reference = crate::metrics::GooseRequestMetricTimingData::default();
        let mut histogram = SecondHistogram::default();
        for &time in &times {
            reference.record_time(time);
            histogram.record(time);
        }
        assert!(!histogram.overflow.is_empty());
        let buckets: Vec<(usize, usize)> = histogram.buckets().collect();
        let mut expected: Vec<(usize, usize)> = Vec::new();
        for (&bucket, &count) in &reference.times_us {
            let row = display_index(crate::metrics::buckets::bucket_end(bucket) - 1);
            match expected.last_mut() {
                Some(last) if last.0 == row => last.1 += count,
                _ => expected.push((row, count)),
            }
        }
        assert_eq!(buckets, expected);
        assert_eq!(histogram.counter, reference.counter);
        assert_eq!(histogram.total_time_us, reference.total_time_us);
        assert_eq!(histogram.minimum_time_us, reference.minimum_time_us);
        assert_eq!(histogram.maximum_time_us, reference.maximum_time_us);
        for percent in [0.0, 0.5, 0.95, 0.99, 1.0] {
            assert_eq!(
                display_index(histogram.percentile_us(percent)),
                bucket_row(crate::metrics::buckets::bucket_floor(
                    reference.percentile_us(percent)
                )),
                "p{percent}"
            );
        }

        // Batches merged into an empty histogram give the same histogram.
        let (first, second) = times.split_at(times.len() / 3);
        let mut via_batch = SecondHistogram::default();
        for part in [first, second] {
            let mut batch = SecondHistogram::default();
            for &time in part {
                batch.record(time);
            }
            via_batch.merge(&batch);
        }
        assert_eq!(via_batch, histogram);

        // A bucket past the most one entry holds continues in the next.
        let mut full = SecondHistogram::default();
        full.add_indexed(0, 5, COUNT_MAX as usize + 10);
        full.add_indexed(0, 5, 3);
        full.add_indexed(0, 4, 1);
        assert_eq!(
            full.buckets().collect::<Vec<_>>(),
            vec![(4, 1), (5, COUNT_MAX as usize), (5, 13)]
        );

        // Merging two halves equals recording everything.
        let (first, second) = times.split_at(times.len() / 2);
        let mut a = SecondHistogram::default();
        first.iter().for_each(|&time| a.record(time));
        let mut b = SecondHistogram::default();
        second.iter().for_each(|&time| b.record(time));
        a.merge(&b);
        assert_eq!(a, histogram);
    }

    #[test]
    fn open_second_equals_packed_second() {
        let mut times: Vec<u64> = Vec::new();
        let mut seed: u64 = 11;
        for _ in 0..3000 {
            seed = seed
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            times.push((seed >> 33) % 90_000 / (1 + (seed >> 60)));
        }
        let mut packed = SecondHistogram::default();
        times.iter().for_each(|&time| packed.record(time));

        // Recorded and merged into an open second, read open and packed.
        let mut open = SecondHistogram::open();
        let (first, second) = times.split_at(times.len() / 2);
        first.iter().for_each(|&time| open.record(time));
        let mut batch = SecondHistogram::default();
        second.iter().for_each(|&time| batch.record(time));
        open.merge(&batch);
        assert!(open.entries.is_empty());
        assert_eq!(open, packed);
        assert_eq!(
            open.buckets().collect::<Vec<_>>(),
            packed.buckets().collect::<Vec<_>>()
        );
        for percent in [0.0, 0.5, 0.95, 0.99, 1.0] {
            assert_eq!(open.percentile(percent), packed.percentile(percent));
        }
        open.pack();
        assert!(open.open.is_none());
        assert_eq!(open.entries, packed.entries);
        assert_eq!(open.overflow, packed.overflow);

        // A packed second keeps taking late responses.
        open.record(7);
        packed.record(7);
        assert_eq!(open.entries, packed.entries);

        // An open bucket past the most one entry holds packs into two.
        let mut full = SecondHistogram::open();
        full.add_bucket(5, COUNT_MAX as usize + 10);
        full.add_bucket(4, 1);
        full.pack();
        assert_eq!(
            full.buckets().collect::<Vec<_>>(),
            vec![(4, 1), (5, COUNT_MAX as usize), (5, 10)]
        );
    }

    /// The first [`MAX_REQUEST_SERIES`] request names the metrics thread
    /// sees with a response time get rings, in the order it sees them, even
    /// when their pending counts first move to a ring in another order
    /// (#713).
    #[test]
    fn rings_go_to_the_first_names_seen() {
        let mut graph = GraphData::for_metrics_processor(true, false, false);
        let mut times = GooseRequestMetricTimingData::new(None);
        times.record_time(25);
        let mut whole_run = GooseRequestMetricTimingData::new(None);
        let names: Vec<String> = (0..150).map(|i| format!("GET /n{i}")).collect();
        // First sightings in order, all in second 0.
        for name in &names {
            graph.record_requests_per_second_merging_latency(name, 0, &times, &mut whole_run);
        }
        // Then second 5 in reverse order, which moves each name's pending
        // counts to its ring, last name first.
        for name in names.iter().rev() {
            graph.record_requests_per_second_merging_latency(name, 5, &times, &mut whole_run);
        }
        let histograms = graph.latency.as_ref().unwrap();
        let mut ringed: Vec<&String> = histograms.by_request.keys().collect();
        ringed.sort_by_key(|name| names.iter().position(|n| n == *name));
        let first: Vec<&String> = names.iter().take(MAX_REQUEST_SERIES).collect();
        assert_eq!(ringed, first);
        // In the order seen.
        for (i, name) in names.iter().take(MAX_REQUEST_SERIES).enumerate() {
            assert_eq!(histograms.by_request[name], i, "{}", name);
        }
    }

    /// A request name's entry stays small, and its pending response times
    /// are allocated only when the heatmap records one: never without the
    /// heatmap, and never for an error (#713).
    #[test]
    fn pending_latency_is_allocated_only_for_the_heatmap() {
        assert!(
            std::mem::size_of::<KeySeries>() <= 64,
            "KeySeries is {} bytes",
            std::mem::size_of::<KeySeries>()
        );
        let mut times = GooseRequestMetricTimingData::new(None);
        times.record_time(12);
        times.record_time(70_000);
        for heatmap in [false, true] {
            let mut graph = GraphData::for_metrics_processor(heatmap, heatmap, false);
            let mut whole_run = GooseRequestMetricTimingData::new(None);
            for second in 0..5 {
                graph.record_requests_per_second_merging_latency(
                    "GET /a",
                    second,
                    &times,
                    &mut whole_run,
                );
                graph.record_requests_per_second_and_latency_batch(
                    "GET /b",
                    second,
                    2,
                    Some(&times),
                );
                graph.record_request_per_second_and_latency("GET /c", second, Some(9));
                graph.record_errors_per_second("GET /c 500 Internal Server Error", second);
            }
            graph.move_pending_latency();
            for (key, entry) in &graph.requests_per_second.series {
                assert_eq!(entry.pending.is_some(), heatmap, "{}", key);
            }
            assert_eq!(graph.errors_per_second.series.len(), 1);
            for (key, entry) in &graph.errors_per_second.series {
                assert!(entry.pending.is_none(), "{}", key);
            }
        }
    }

    #[test]
    fn latency_ring_packs_seconds_two_back() {
        let mut ring = LatencyRing::new();
        ring.record(10, 20, 1);
        ring.record(11, 20, 2);
        let is_open = |ring: &LatencyRing, second| ring.get(second).unwrap().open.is_some();
        assert!(is_open(&ring, 10) && is_open(&ring, 11));
        ring.record(12, 20, 3);
        assert!(!is_open(&ring, 10));
        assert!(is_open(&ring, 11) && is_open(&ring, 12));
        // A late response to a packed second still counts, and marks it
        // changed.
        ring.record(10, 30, 4);
        assert_eq!(ring.get(10).unwrap().counter, 2);
        assert_eq!(ring.slot(10).unwrap().rev, 4);
        assert_eq!(ring.slot(11).unwrap().rev, 2);
        // After a gap every older open second is packed.
        ring.record(40, 20, 5);
        assert!(!is_open(&ring, 11) && !is_open(&ring, 12));
        assert_eq!(ring.open, vec![40]);
        ring.clear();
        assert!(ring.open.is_empty());
    }

    #[test]
    fn latency_ring_drops_samples_older_than_its_slot() {
        let mut ring = LatencyRing::new();
        ring.record(400, 25, 1);
        // Second 100 maps to the same slot as 400 (300 slots).
        ring.record(100, 10, 2);
        ring.merge_timing(
            100,
            &{
                let mut data = GooseRequestMetricTimingData::default();
                data.record_time(10);
                data
            },
            3,
        );
        assert!(ring.get(100).is_none());
        let held = ring.get(400).expect("slot keeps second 400");
        assert_eq!(held.counter, 1);
        assert_eq!(held.maximum_time_us, 25_000);
        // A sample too old to keep does not mark the slot changed.
        assert_eq!(ring.slot(400).unwrap().rev, 1);
        // A newer second replaces the older one.
        ring.record(700, 5, 4);
        assert!(ring.get(400).is_none());
        assert_eq!(ring.get(700).map(|d| d.counter), Some(1));
    }

    /// The `(index, count)` pairs of `second`'s measured buckets.
    fn measured_pairs(heatmap: &Heatmap, second: u64) -> Vec<(u64, u64)> {
        heatmap
            .column(second)
            .map(|column| HeatmapColumn::pairs(&column.measured))
            .unwrap_or_default()
    }

    /// Rows from 5 to 61 seconds are 1 second wide, so a 5.1 s time is in
    /// the 5 to 6 s row, not a 5 to 7 s one.
    #[test]
    fn heatmap_rows_from_five_seconds_are_one_second_wide() {
        assert_eq!(
            display_range(display_index(5_100_000)),
            (5_000_000, 5_999_999)
        );
        assert_eq!(
            display_range(display_index(60_999_999)),
            (60_000_000, 60_999_999)
        );
        assert_eq!(display_index(60_999_999), FIXED_BUCKETS - 1);
        assert_eq!(FIXED_BUCKETS, 237);
    }

    #[test]
    fn heatmap_bucket_tables_follow_the_rows() {
        let table = bucket_table_us(FIXED_BUCKETS);
        assert_eq!(table.len(), FIXED_BUCKETS + 1);
        // Every measured time from 0 to past 61 seconds falls in exactly the
        // row whose range holds it, and the fixed rows meet end to end.
        for (index, entry) in table.iter().enumerate() {
            let [value, lowest, highest] = *entry;
            assert!(lowest <= value && value <= highest, "{:?}", entry);
            for time in [lowest, value, highest] {
                assert_eq!(display_index(time), index, "{entry:?}");
            }
            if index > 0 && index < FIXED_BUCKETS {
                assert_eq!(table[index - 1][2] + 1, lowest, "{entry:?}");
            }
        }
        for (index, entry) in [
            (0, [50, 0, 99]),
            (9, [950, 900, 999]),
            (10, [1_500, 1_000, 1_999]),
            (72, [63_500, 63_000, 63_999]),
            (73, [65_000, 64_000, 65_999]),
            (91, [105_000, 100_000, 109_999]),
            (131, [525_000, 500_000, 549_999]),
            (141, [1_050_000, 1_000_000, 1_099_999]),
            (145, [1_450_000, 1_400_000, 1_499_999]),
            (146, [1_550_000, 1_500_000, 1_599_999]),
            (181, [5_500_000, 5_000_000, 5_999_999]),
            (182, [6_500_000, 6_000_000, 6_999_999]),
            (236, [60_500_000, 60_000_000, 60_999_999]),
            (237, [61_170_848, 61_000_000, 61_341_695]),
        ] {
            assert_eq!(table[index], entry, "index {index}");
        }
        assert_eq!(bucket_table_us(0).len(), FIXED_BUCKETS);
    }

    /// Every fixed row takes at least one timing data bucket, so none stays
    /// empty under a spread of times.
    #[test]
    fn every_fixed_row_takes_a_bucket() {
        let mut taken = [false; FIXED_BUCKETS];
        let mut bucket = 0;
        while bucket < FIXED_END_US {
            if let Some(row) = taken.get_mut(bucket_row(bucket)) {
                *row = true;
            }
            bucket = crate::metrics::buckets::bucket_end(bucket);
        }
        let empty: Vec<usize> = (0..FIXED_BUCKETS).filter(|&row| !taken[row]).collect();
        assert!(empty.is_empty(), "rows no bucket lands in: {:?}", empty);
    }

    /// A column's cells are kept from one heatmap to the next, and rebuilt
    /// when its second changes.
    #[test]
    fn heatmap_columns_rebuild_only_when_changed() {
        let mut graph = latency_graph(false);
        graph.record_requests_per_second("GET /", 1);
        graph.record_latency("GET /", 1, 5);
        let first = graph
            .export_series_window_until(300, Some(4))
            .heatmap
            .unwrap();
        // 5 ms is row 14, past the ten rows under a millisecond.
        assert_eq!(measured_pairs(&first, 1), [(14, 1)]);
        let again = graph
            .export_series_window_until(300, Some(4))
            .heatmap
            .unwrap();
        assert_eq!(again.columns, first.columns);
        graph.record_latency("GET /", 1, 6);
        let changed = graph
            .export_series_window_until(300, Some(4))
            .heatmap
            .unwrap();
        assert_eq!(measured_pairs(&changed, 1), [(14, 1), (15, 1)]);
    }

    #[test]
    fn heatmap_sends_only_changed_seconds() {
        let mut graph = latency_graph(false);
        for (second, time) in [(1, 5), (2, 7), (2, 7), (2, 120)] {
            graph.record_requests_per_second("GET /", second);
            graph.record_latency("GET /", second, time);
        }
        let first = graph
            .export_series_window_until(300, Some(4))
            .heatmap
            .unwrap();
        assert_eq!(first.since, 0);
        assert_eq!(
            (first.start_second, first.end_second, first.hold_from),
            (0, 4, 3)
        );
        assert_eq!(first.columns.len(), 2);
        assert_eq!(measured_pairs(&first, 1), [(14, 1)]);
        // 7 ms is row 16; 120 ms is in the 120 to 130 ms row, index 93.
        assert_eq!(measured_pairs(&first, 2), [(16, 2), (93, 1)]);
        assert!(first
            .columns
            .iter()
            .all(|column| column.synthetic.is_empty()));
        assert_eq!(first.buckets_us.len(), FIXED_BUCKETS);

        // Nothing changed: no columns.
        let cursor = HeatmapCursor {
            generation: first.generation,
            rev: first.rev,
        };
        let same = graph
            .export_series_window_with_heatmap(300, Some(4), false, Some(cursor))
            .heatmap
            .unwrap();
        assert_eq!((same.since, same.rev), (first.rev, first.rev));
        assert!(same.columns.is_empty());

        // A late response to second 1 and a new second 3: only those, whole.
        graph.record_latency("GET /", 1, 6);
        graph.record_latency("GET /", 3, 75_000);
        let delta = graph
            .export_series_window_with_heatmap(300, Some(4), false, Some(cursor))
            .heatmap
            .unwrap();
        assert_eq!(delta.since, first.rev);
        assert_eq!(
            delta
                .columns
                .iter()
                .map(|column| column.second)
                .collect::<Vec<_>>(),
            [1, 3]
        );
        assert_eq!(measured_pairs(&delta, 1), [(14, 1), (15, 1)]);
        // Past 61 seconds a row per timing data bucket: 75 s is index 256,
        // and the table grows to cover it.
        assert_eq!(measured_pairs(&delta, 3), [(256, 1)]);
        assert_eq!(delta.buckets_us.len(), 257);
        assert_eq!(delta.buckets_us[256], [74_973_184, 74_448_896, 75_497_471]);

        // A cursor of another generation, or from the future, gets the whole
        // window.
        for stale in [
            HeatmapCursor {
                generation: cursor.generation + 1,
                rev: cursor.rev,
            },
            HeatmapCursor {
                generation: cursor.generation,
                rev: delta.rev + 1,
            },
        ] {
            let whole = graph
                .export_series_window_with_heatmap(300, Some(4), false, Some(stale))
                .heatmap
                .unwrap();
            assert_eq!(whole.since, 0);
            assert_eq!(whole.columns.len(), 3);
        }

        // A metrics reset starts a new generation with no columns.
        graph.reset_rate_series(0);
        let reset = graph
            .export_series_window_with_heatmap(300, Some(4), false, Some(cursor))
            .heatmap
            .unwrap();
        assert_ne!(reset.generation, cursor.generation);
        assert_eq!(reset.since, 0);
        assert!(reset.columns.is_empty());
    }

    #[test]
    fn heatmap_slow_second_and_split_buckets() {
        let mut graph = latency_graph(false);
        for _ in 0..10 {
            graph.record_requests_per_second("GET /slow", 5);
            graph.record_latency("GET /slow", 5, 1400);
        }
        let heatmap = graph
            .export_series_window_until(300, Some(7))
            .heatmap
            .unwrap();
        assert_eq!(measured_pairs(&heatmap, 5), [(145, 10)]);

        // A bucket past the most one packed entry holds is still one cell.
        let mut full = SecondHistogram::default();
        full.add_indexed(0, 5, COUNT_MAX as usize + 10);
        full.add_indexed(0, 4, 1);
        assert_eq!(full.heat_cells(), [4, 1, 5, COUNT_MAX as u64 + 10]);
    }

    #[test]
    fn heatmap_window_gaps_and_hold() {
        let mut graph = latency_graph(false);
        for second in [2, 4, 9, 10] {
            graph.record_requests_per_second("GET /", second);
            graph.record_latency("GET /", second, 20);
        }
        let heatmap = graph
            .export_series_window_until(300, Some(10))
            .heatmap
            .unwrap();
        // Seconds without requests have no column.
        assert_eq!(
            heatmap.columns.iter().map(|c| c.second).collect::<Vec<_>>(),
            [2, 4, 9, 10]
        );
        // The two newest seconds, from hint - 1, are held back.
        assert_eq!(heatmap.hold_from, 9);
        // A request filed one second past the hint extends the window; the
        // hold still starts at hint - 1.
        graph.record_requests_per_second("GET /", 11);
        graph.record_latency("GET /", 11, 20);
        let heatmap = graph
            .export_series_window_until(300, Some(10))
            .heatmap
            .unwrap();
        assert_eq!((heatmap.end_second, heatmap.hold_from), (11, 9));
        // Nothing recorded yet: an empty heatmap, not none.
        let empty = latency_graph(false)
            .export_series_window_until(300, None)
            .heatmap
            .unwrap();
        assert!(empty.columns.is_empty());
    }

    #[test]
    fn latency_keeps_first_hundred_names() {
        let mut graph = latency_graph(true);
        for i in 0..=MAX_REQUEST_SERIES {
            graph.record_latency(&format!("GET /{i}"), 1, 10);
        }
        let latency = histograms(&mut graph);
        assert_eq!(latency.by_request.len(), MAX_REQUEST_SERIES);
        assert!(!latency.by_request.contains_key("GET /100"));
        // The 101st name still counts toward all requests.
        assert_eq!(
            latency.all_second(1).map(|d| d.counter),
            Some(MAX_REQUEST_SERIES + 1)
        );
        // No synthetic ring for a name without a measured one.
        graph.record_synthetic_latency("GET /100", 1, 500);
        graph.record_synthetic_latency("GET /0", 1, 500);
        let (all, by_request) = histograms(&mut graph).synthetic.as_ref().unwrap();
        assert_eq!(all.get(1).map(|d| d.counter), Some(2));
        assert!(by_request.contains_key("GET /0"));
        assert!(!by_request.contains_key("GET /100"));

        let keys = graph.latency_series_keys();
        assert_eq!(keys.len(), MAX_REQUEST_SERIES);
        assert!(keys.contains("GET /99"));
        assert!(!keys.contains("GET /100"));

        // A metrics reset empties both maps and every ring, so names are
        // admitted again.
        graph.reset_rate_series(0);
        let latency = histograms(&mut graph);
        assert!(latency.by_request.is_empty());
        assert!(latency.all_second(1).is_none());
        let (all, by_request) = latency.synthetic.as_ref().unwrap();
        assert!(by_request.is_empty());
        assert!(all.get(1).is_none());
        graph.record_latency("GET /100", 2, 10);
        assert!(graph.latency_series_keys().contains("GET /100"));
    }

    /// The heatmap's JSON in the worst case, every row in every second of
    /// the window: what changes in a second stays within 200 KB and a 60
    /// second synthetic backfill within 220 KB. A whole window, sent once to
    /// a client with no heatmap, can exceed both, which is why only changes
    /// are sent.
    #[test]
    fn heatmap_worst_case_sizes() {
        let mut graph = latency_graph(true);
        let every_bucket = |count: usize| {
            let mut data = GooseRequestMetricTimingData::default();
            for index in 0..FIXED_BUCKETS {
                data.times_us.insert(
                    crate::metrics::buckets::bucket_floor(display_value(index)),
                    count,
                );
            }
            data.counter = FIXED_BUCKETS * count;
            data.maximum_time_us = 60_000_000;
            data
        };
        let data = every_bucket(12_345);
        for second in 0..300 {
            graph.record_requests_per_second_and_latency_batch(
                "GET /",
                second,
                data.counter as u32,
                Some(&data),
            );
        }
        let size = |heatmap: &Heatmap| serde_json::to_vec(heatmap).unwrap().len();
        let whole = graph
            .export_series_window_with_heatmap(300, Some(299), true, None)
            .heatmap
            .unwrap();
        assert_eq!(whole.columns.len(), 300);
        let cursor = HeatmapCursor {
            generation: whole.generation,
            rev: whole.rev,
        };
        // The next second: a new column, and late responses to the two
        // before it.
        for second in [298, 299, 300] {
            graph.record_requests_per_second_and_latency_batch(
                "GET /",
                second,
                data.counter as u32,
                Some(&data),
            );
        }
        let next = graph
            .export_series_window_with_heatmap(300, Some(300), true, Some(cursor))
            .heatmap
            .unwrap();
        assert_eq!(next.columns.len(), 3);
        // A 60 second stall backfilled with synthetic times in every bucket.
        let cursor = HeatmapCursor {
            generation: next.generation,
            rev: next.rev,
        };
        for second in 241..=300 {
            for index in 0..FIXED_BUCKETS {
                graph.record_synthetic_latency("GET /", second, display_value(index) / 1000);
            }
        }
        let backfill = graph
            .export_series_window_with_heatmap(300, Some(300), true, Some(cursor))
            .heatmap
            .unwrap();
        assert_eq!(backfill.columns.len(), 60);
        eprintln!(
            "worst case heatmap JSON: whole window {} bytes, a second's changes {} bytes, a 60 s backfill {} bytes",
            size(&whole),
            size(&next),
            size(&backfill)
        );
        assert!(size(&next) <= 200 * 1024);
        assert!(size(&backfill) <= 220 * 1024);
        assert!(size(&whole) > 200 * 1024);
    }

    #[test]
    fn heatmap_off_exports_none() {
        let mut graph = GraphData::new();
        graph.record_requests_per_second("GET /", 1);
        graph.record_latency("GET /", 1, 10);
        let window = graph.export_series_window_with_heatmap(300, Some(3), true, None);
        assert_eq!(window.rps.len(), 4);
        assert_eq!(window.heatmap, None);
        assert!(graph.latency_series_keys().is_empty());
        assert!(graph
            .export_request_series("GET".into(), "/".into(), 300, Some(3), true, None)
            .is_none());
    }

    #[test]
    fn heatmap_request_series_export() {
        let mut graph = latency_graph(true);
        for second in [1, 2] {
            for _ in 0..3 {
                graph.record_requests_per_second("GET /a", second);
                graph.record_latency("GET /a", second, 30);
            }
            graph.record_requests_per_second("GET /b", second);
            graph.record_latency("GET /b", second, 900);
        }
        graph.record_synthetic_latency("GET /a", 1, 4000);
        let series = graph
            .export_request_series("GET".into(), "/a".into(), 300, Some(5), true, None)
            .expect("series for /a");
        assert_eq!(series.method, "GET");
        assert_eq!(series.name, "/a");
        let heatmap = &series.heatmap;
        assert_eq!((heatmap.start_second, heatmap.end_second), (0, 5));
        assert_eq!(measured_pairs(heatmap, 1), [(39, 3)]);
        assert_eq!(measured_pairs(heatmap, 2), [(39, 3)]);
        // 4 s is index 171.
        assert_eq!(
            HeatmapColumn::pairs(&heatmap.column(1).unwrap().synthetic),
            [(171, 1)]
        );
        // Not co_active: no synthetic times.
        let series = graph
            .export_request_series("GET".into(), "/a".into(), 300, Some(5), false, None)
            .unwrap();
        assert!(series
            .heatmap
            .columns
            .iter()
            .all(|c| c.synthetic.is_empty()));
        // All requests: both names.
        let all = graph
            .export_series_window_until(300, Some(5))
            .heatmap
            .unwrap();
        assert_eq!(all.counts(1), (4, 0));
        assert!(graph
            .export_request_series("GET".into(), "/missing".into(), 300, Some(5), true, None)
            .is_none());
    }

    #[test]
    fn test_moving_average() {
        let mut moving_average = MovingAverage::new();
        assert_eq!(
            moving_average,
            MovingAverage {
                count: 0,
                average: 0.
            }
        );

        moving_average.add_item(&1.23);
        assert_eq!(
            moving_average,
            MovingAverage {
                count: 1,
                average: 1.23
            }
        );

        moving_average.add_item(&2.34);
        assert_eq!(
            moving_average,
            MovingAverage {
                count: 2,
                average: 1.785
            }
        );

        moving_average.add_item(&89.23);
        assert_eq!(
            moving_average,
            MovingAverage {
                count: 3,
                average: 30.933332
            }
        );

        moving_average.add_item(&12.34);
        assert_eq!(
            moving_average,
            MovingAverage {
                count: 4,
                average: 26.285
            }
        );
    }

    #[test]
    fn test_moving_average_cmp() {
        assert!(
            MovingAverage {
                count: 0,
                average: 0.
            } < MovingAverage {
                count: 0,
                average: 0.1,
            }
        );

        assert!(
            MovingAverage {
                count: 0,
                average: 1.1,
            } > MovingAverage {
                count: 0,
                average: 0.1,
            }
        );

        assert_eq!(
            MovingAverage {
                count: 1,
                average: 1.1,
            },
            MovingAverage {
                count: 2,
                average: 1.1,
            }
        );

        assert!(
            MovingAverage {
                count: 0,
                average: 1.1,
            } != MovingAverage {
                count: 0,
                average: 1.0,
            }
        );
    }

    #[test]
    fn test_add_timestamp_to_html_graph_data() {
        let data = vec![Some(123), Some(234), Some(345), Some(456), Some(567)];
        let empty = HashMap::new();
        let graph: Graph<usize, usize> = Graph::new("html_id", "Label", true, &empty);

        // A series that starts later, as after Save report is turned on during
        // a run, is drawn from its own first second.
        let later = graph.add_timestamp_to_html_graph_data(
            &data[2..],
            2,
            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 23).unwrap(),
        );
        let all = graph.add_timestamp_to_html_graph_data(
            &data,
            0,
            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 23).unwrap(),
        );
        assert_eq!(later, all[2..]);

        assert_eq!(
            graph.add_timestamp_to_html_graph_data(
                &data,
                0,
                Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 23).unwrap()
            ),
            vec![
                (
                    Local
                        .timestamp_opt(
                            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 23)
                                .unwrap()
                                .timestamp(),
                            0
                        )
                        .unwrap()
                        .format("%Y-%m-%d %H:%M:%S")
                        .to_string(),
                    123
                ),
                (
                    Local
                        .timestamp_opt(
                            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 24)
                                .unwrap()
                                .timestamp(),
                            0
                        )
                        .unwrap()
                        .format("%Y-%m-%d %H:%M:%S")
                        .to_string(),
                    234
                ),
                (
                    Local
                        .timestamp_opt(
                            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 25)
                                .unwrap()
                                .timestamp(),
                            0
                        )
                        .unwrap()
                        .format("%Y-%m-%d %H:%M:%S")
                        .to_string(),
                    345
                ),
                (
                    Local
                        .timestamp_opt(
                            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 26)
                                .unwrap()
                                .timestamp(),
                            0
                        )
                        .unwrap()
                        .format("%Y-%m-%d %H:%M:%S")
                        .to_string(),
                    456
                ),
                (
                    Local
                        .timestamp_opt(
                            Utc.with_ymd_and_hms(2021, 12, 14, 15, 12, 27)
                                .unwrap()
                                .timestamp(),
                            0
                        )
                        .unwrap()
                        .format("%Y-%m-%d %H:%M:%S")
                        .to_string(),
                    567
                )
            ]
        );
    }

    fn expected_graph_html_prefix(
        html_id: &str,
        y_axis_label: &str,
        main_series_label: &str,
    ) -> String {
        format!(
            r#"<div class="graph">
                <div id="{html_id}" style="width: 1000px; height:500px; background: white;"></div>

                <script type="text/javascript">
                    var chartDom = document.getElementById('{html_id}');
                    var myChart = echarts.init(chartDom);

                    myChart.setOption({{
                        color: ['#2c664f', '#5470c6', '#91cc75', '#fac858', '#ee6666', '#73c0de', '#3ba272', '#fc8452', '#9a60b4', '#ea7ccc'],
                        tooltip: {{ trigger: 'axis' }},
                        toolbox: {{
                            feature: {{
                                dataZoom: {{ yAxisIndex: 'none' }},
                                restore: {{}},
                                saveAsImage: {{}}
                            }}
                        }},
                        dataZoom: [
                            {{
                                type: 'inside',
                                start: 0,
                                end: 100,
                                fillerColor: 'rgba(34, 80, 61, 0.25)',
                                selectedDataBackground: {{
                                    lineStyle: {{ color: '#2c664f' }},
                                    areaStyle: {{ color: '#378063' }}
                                }}
                            }},
                            {{
                                start: 0,
                                end: 100,
                                fillerColor: 'rgba(34, 80, 61, 0.25)',
                                selectedDataBackground: {{
                                    lineStyle: {{ color: '#2c664f' }},
                                    areaStyle: {{ color: '#378063' }}
                                }}
                            }},
                        ],
                        xAxis: {{ type: 'time' }},
                        yAxis: {{
                            name: '{y_axis_label}',
                            nameLocation: 'center',
                            nameRotate: 90,
                            nameGap: 45,
                            type: 'value'
                        }},
                        
                        series: [
                            {{
                                name: '{main_series_label}',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                lineStyle: {{ color: '#2c664f' }},
                                areaStyle: {{ color: '#378063' }},
                                markArea: {{
"#
        )
    }

    #[test]
    fn test_graph_markup() {
        let expected_prefix = expected_graph_html_prefix("graph-rps", "Requests #", "GET /");

        let data: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![123, 111, 99, 134]),
            phantom: PhantomData,
            total: 0,
        };

        let mut graph = HashMap::new();
        graph.insert("GET /".to_string(), data);

        let mut expected = expected_prefix.to_owned();
        expected += format!(
            r#"                                    data: [
                                        
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",123],["{data_series_prefix}:33",111],["{data_series_prefix}:34",99],["{data_series_prefix}:35",134]],
                            }},
                            
                        ]
                    }});
                </script>
            </div>"#,
            data_series_prefix = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M")
        ).as_str();

        assert_eq!(
            Graph::new("graph-rps", "Requests #", true, &graph,).get_markup(
                &Vec::new(),
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );

        // It should make no difference if we disable granular graphs, since we only have one
        // request.
        assert_eq!(
            Graph::new("graph-rps", "Requests #", false, &graph,).get_markup(
                &Vec::new(),
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );

        let mut expected = expected_prefix;
        expected += format!(
            r#"                                    data: [
                                        [
                            {{
                                xAxis: '{increasing}',
                                itemStyle: {{ borderColor: 'rgba(44, 102, 79, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{increasing}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{increasing}',
                                itemStyle: {{ color: 'rgba(44, 102, 79, 0.05)' }},
                            }},
                            {{
                                xAxis: '{decreasing}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{decreasing}',
                                itemStyle: {{ borderColor: 'rgba(44, 102, 79, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{decreasing}'
                            }}
                        ],[
                            {{
                                xAxis: '{decreasing}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{decreasing}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{decreasing}',
                                itemStyle: {{ color: 'rgba(179, 65, 65, 0.05)' }},
                            }},
                            {{
                                xAxis: '{cancelling}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{cancelling}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{cancelling}'
                            }}
                        ],[
                            {{
                                xAxis: '{cancelling}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{cancelling}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{cancelling}',
                                itemStyle: {{ color: 'rgba(179, 65, 65, 0.05)' }},
                            }},
                            {{
                                xAxis: '{finishing}'
                            }}
                        ],
                        [
                            {{
                                xAxis: '{finishing}',
                                itemStyle: {{ borderColor: 'rgba(179, 65, 65, 0.25)', borderWidth: 1 }},
                            }},
                            {{
                                xAxis: '{finishing}'
                            }}
                        ],
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",123],["{data_series_prefix}:33",111],["{data_series_prefix}:34",99],["{data_series_prefix}:35",134]],
                            }},
                            
                        ]
                    }});
                </script>
            </div>"#,
            data_series_prefix = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
            increasing = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S"),
            decreasing = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 33).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S"),
            cancelling = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 34).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S"),
            finishing = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 35).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S"),
        ).as_str();

        let steps = vec![
            TestPlanHistory {
                action: TestPlanStepAction::Increasing,
                timestamp: Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap(),
                users: 123,
            },
            TestPlanHistory {
                action: TestPlanStepAction::Decreasing,
                timestamp: Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 33).unwrap(),
                users: 123,
            },
            TestPlanHistory {
                action: TestPlanStepAction::Canceling,
                timestamp: Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 34).unwrap(),
                users: 123,
            },
            TestPlanHistory {
                action: TestPlanStepAction::Finished,
                timestamp: Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 35).unwrap(),
                users: 123,
            },
        ];

        assert_eq!(
            Graph::new("graph-rps", "Requests #", true, &graph,).get_markup(
                &steps,
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );

        // It should make no difference if we disable granular graphs, since we only have one
        // request.
        assert_eq!(
            Graph::new("graph-rps", "Requests #", false, &graph,).get_markup(
                &steps,
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );

        let user_data: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![23, 12, 44, 22]),
            phantom: PhantomData,
            total: 0,
        };
        graph.insert("GET /user".to_string(), user_data);

        let markup = Graph::new("graph-rps", "Requests #", true, &graph).get_markup(
            &Vec::new(),
            Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap(),
        );
        let expected_legend = r#"
                        legend: {
                            type: 'plain',
                            width: '75%',
                            data: ["Total","GET /"#;
        assert!(
            markup.contains(expected_legend),
            "legend {} not found in {}",
            expected_legend,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'Total',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                lineStyle: {{ color: '#2c664f' }},
                                areaStyle: {{ color: '#378063' }},
                                markArea: {{
                                    data: [
                                        
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",146],["{data_series_prefix}:33",123],["{data_series_prefix}:34",143],["{data_series_prefix}:35",156]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",123],["{data_series_prefix}:33",111],["{data_series_prefix}:34",99],["{data_series_prefix}:35",134]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /user',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",23],["{data_series_prefix}:33",12],["{data_series_prefix}:34",44],["{data_series_prefix}:35",22]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        // Now same graph with granular data disabled.
        let expected_nongranular_prefix =
            expected_graph_html_prefix("graph-rps", "Requests #", "Total");
        let mut expected = expected_nongranular_prefix.to_owned();
        expected += format!(
            r#"                                    data: [
                                        
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",146],["{data_series_prefix}:33",123],["{data_series_prefix}:34",143],["{data_series_prefix}:35",156]],
                            }},
                            
                        ]
                    }});
                </script>
            </div>"#,
            data_series_prefix = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M")
        ).as_str();

        assert_eq!(
            Graph::new("graph-rps", "Requests #", false, &graph,).get_markup(
                &Vec::new(),
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );

        let more_data: TimeSeries<usize, usize> = TimeSeries {
            origin: 0,
            data: VecDeque::from(vec![1, 1, 1, 1]),
            phantom: PhantomData,
            total: 0,
        };
        graph.insert("GET /one".to_string(), more_data.clone());
        graph.insert("GET /two".to_string(), more_data.clone());
        graph.insert("GET /three".to_string(), more_data);

        let markup = Graph::new("graph-rps", "Requests #", true, &graph).get_markup(
            &Vec::new(),
            Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap(),
        );
        let expected_legend = r#"
                        legend: {
                            type: 'scroll',
                            width: '75%',
                            data: ["Total","GET /"#;
        assert!(
            markup.contains(expected_legend),
            "legend {} not found in {}",
            expected_legend,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'Total',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                lineStyle: {{ color: '#2c664f' }},
                                areaStyle: {{ color: '#378063' }},
                                markArea: {{
                                    data: [
                                        
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",149],["{data_series_prefix}:33",126],["{data_series_prefix}:34",146],["{data_series_prefix}:35",159]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",123],["{data_series_prefix}:33",111],["{data_series_prefix}:34",99],["{data_series_prefix}:35",134]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /user',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",23],["{data_series_prefix}:33",12],["{data_series_prefix}:34",44],["{data_series_prefix}:35",22]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /one',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",1],["{data_series_prefix}:33",1],["{data_series_prefix}:34",1],["{data_series_prefix}:35",1]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /two',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",1],["{data_series_prefix}:33",1],["{data_series_prefix}:34",1],["{data_series_prefix}:35",1]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        let expected_line = format!(
            r#"{{
                                name: 'GET /three',
                                type: 'line',
                                symbol: 'none',
                                sampling: 'lttb',
                                data: [["{data_series_prefix}:32",1],["{data_series_prefix}:33",1],["{data_series_prefix}:34",1],["{data_series_prefix}:35",1]],
                            }},"#,
            data_series_prefix = Local
                .timestamp_opt(
                    Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32)
                        .unwrap()
                        .timestamp(),
                    0
                )
                .unwrap()
                .format("%Y-%m-%d %H:%M"),
        );
        assert!(
            markup.contains(expected_line.as_str()),
            "line {} not found in {}",
            expected_line,
            markup
        );

        // Now same graph with granular data disabled.
        let mut expected = expected_nongranular_prefix;
        expected += format!(
            r#"                                    data: [
                                        
                                    ]
                                }},
                                data: [["{data_series_prefix}:32",149],["{data_series_prefix}:33",126],["{data_series_prefix}:34",146],["{data_series_prefix}:35",159]],
                            }},
                            
                        ]
                    }});
                </script>
            </div>"#,
            data_series_prefix = Local
                .timestamp_opt(Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap().timestamp(), 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M")
        ).as_str();

        assert_eq!(
            Graph::new("graph-rps", "Requests #", false, &graph,).get_markup(
                &Vec::new(),
                Utc.with_ymd_and_hms(2021, 11, 21, 21, 20, 32).unwrap()
            ),
            expected
        );
    }

    // --- The report heatmap's whole run columns (#746) ---

    fn latency(graph: &GraphData) -> &LatencyHistograms {
        graph.latency.as_ref().expect("latency histograms")
    }

    /// The non empty columns of `columns`, by index.
    fn filled(columns: &Columns) -> Vec<(usize, SecondHistogram)> {
        columns
            .columns
            .iter()
            .enumerate()
            .filter(|(_, column)| column.counter > 0)
            .map(|(index, column)| (index, column.clone()))
            .collect()
    }

    /// Columns of `width` seconds from second 0, built directly from
    /// `(second, time)` pairs.
    fn reference_columns(times: &[(usize, u64)], width: usize) -> Vec<(usize, SecondHistogram)> {
        let mut columns: std::collections::BTreeMap<usize, SecondHistogram> = Default::default();
        for &(second, time) in times {
            columns.entry(second / width).or_default().record(time);
        }
        columns.into_iter().collect()
    }

    /// A seeded response time: mostly fast, some slow, a few past 60 s.
    fn seeded_time(state: &mut u64) -> u64 {
        *state = state
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1442695040888963407);
        match (*state >> 33) % 100 {
            0 => 61_000 + (*state >> 40) % 5_000,
            1..=9 => 200 + (*state >> 40) % 3_000,
            _ => (*state >> 40) % 120,
        }
    }

    /// Merging columns as a budget fills gives the same counts and the
    /// same p50, p95 and p99 as building the wider columns directly from
    /// the raw times, after several merges of both budgets.
    #[test]
    fn merged_columns_match_columns_built_directly() {
        let mut graph = GraphData::with_latency_budgets(false, 4, 2);
        let mut state = 746;
        let mut times = Vec::new();
        let mut by_name: HashMap<&str, Vec<(usize, u64)>> = HashMap::new();
        for second in 0..40 {
            for (i, &name) in ["GET /a", "GET /b"].iter().enumerate() {
                let mut batch = GooseRequestMetricTimingData::new(None);
                for _ in 0..(3 + i * 2) {
                    let time = seeded_time(&mut state);
                    batch.record_time(time);
                    times.push((second, time));
                    by_name.entry(name).or_default().push((second, time));
                }
                graph.record_requests_per_second_and_latency_batch(
                    name,
                    second,
                    batch.counter as u32,
                    Some(&batch),
                );
            }
            // One request at a time too.
            let time = seeded_time(&mut state);
            graph.record_request_per_second_and_latency("GET /a", second, Some(time));
            times.push((second, time));
            by_name.entry("GET /a").or_default().push((second, time));
        }
        graph.finish_latency_columns();
        let latency = latency(&graph);
        // 40 seconds in 4 columns: 1, 2, 4, 8 then 16 seconds wide; in 2
        // columns, 32 seconds wide.
        assert_eq!((latency.all_width, latency.request_width), (16, 32));
        let all = latency.all_columns.as_ref().unwrap();
        assert!(all.columns.len() <= 4);
        let expected = reference_columns(&times, 16);
        assert_eq!(filled(all), expected);
        for (name, times) in &by_name {
            let columns = latency.rings[latency.by_request[*name]]
                .columns
                .as_ref()
                .unwrap();
            assert!(columns.columns.len() <= 2);
            assert_eq!(filled(columns), reference_columns(times, 32), "{name}");
        }
        for ((_, merged), (_, direct)) in filled(all).iter().zip(&expected) {
            for percent in [0.5, 0.95, 0.99] {
                assert_eq!(merged.percentile(percent), direct.percentile(percent));
            }
        }
    }

    /// A write to a second the ring already packed, and so folded, or that
    /// the ring refuses as too old, reaches the columns once: no time is
    /// lost or counted twice.
    #[test]
    fn late_writes_reach_the_columns_once() {
        let mut graph = GraphData::with_latency_budgets(false, 1_000, 1_000);
        let one = |time: u64| {
            let mut data = GooseRequestMetricTimingData::new(None);
            data.record_time(time);
            data
        };
        let batch = |graph: &mut GraphData, second: usize, time: u64| {
            graph.record_requests_per_second_and_latency_batch(
                "GET /a",
                second,
                1,
                Some(&one(time)),
            );
        };
        batch(&mut graph, 10, 5);
        // Seconds 11 to 13 move 10 to the ring, and 13 packs it.
        for second in 11..=13 {
            batch(&mut graph, second, 5);
            graph.move_pending_latency();
        }
        // A late batch for packed second 10, then one directly to it.
        batch(&mut graph, 10, 7);
        graph.record_request_per_second_and_latency("GET /a", 10, Some(9));
        // Second 20 leaves the ring once second 320 takes its slot.
        batch(&mut graph, 20, 5);
        batch(&mut graph, 320, 5);
        graph.move_pending_latency();
        batch(&mut graph, 20, 11);
        graph.finish_latency_columns();
        let latency = latency(&graph);
        let all = latency.all_columns.as_ref().unwrap();
        let own = latency.rings[0].columns.as_ref().unwrap();
        for columns in [all, own] {
            let mut ten = SecondHistogram::default();
            for time in [5, 7, 9] {
                ten.record(time);
            }
            assert_eq!(columns.columns[10], ten);
            let mut twenty = SecondHistogram::default();
            twenty.record(5);
            twenty.record(11);
            assert_eq!(columns.columns[20], twenty);
            let total: usize = columns.columns.iter().map(|column| column.counter).sum();
            assert_eq!(total, 9);
        }
        // The ring still shows second 10 as it was, for the dashboard.
        assert_eq!(latency.rings[0].get(10).map(|data| data.counter), Some(3));
    }

    /// Synthetic times land in the columns of the second they belong to:
    /// for a second the ring no longer holds, and in columns that have
    /// merged, both for all requests and for the name; the adjusted view
    /// adds them to the measured columns.
    #[test]
    fn synthetic_times_land_in_their_columns() {
        let mut graph = GraphData::with_latency_budgets(true, 100, 50);
        for second in 0..700 {
            graph.record_request_per_second_and_latency("GET /a", second, Some(10));
        }
        // Second 305 takes slot 5 in the synthetic rings, so they refuse
        // second 5, which goes to the columns directly.
        graph.record_synthetic_latency("GET /a", 305, 1_000);
        graph.record_synthetic_latency("GET /a", 5, 2_000);
        // Second 653 packs 650, so a later time for 650 is a late write.
        graph.record_synthetic_latency("GET /a", 650, 3_000);
        graph.record_synthetic_latency("GET /a", 653, 1_000);
        graph.record_synthetic_latency("GET /a", 650, 3_000);
        // A name without a measured ring adds only to all requests.
        graph.record_synthetic_latency("GET /none", 653, 1_000);
        graph.finish_latency_columns();
        let latency = latency(&graph);
        assert_eq!((latency.all_width, latency.request_width), (8, 16));
        let (all, by_request) = latency.synthetic.as_ref().unwrap();
        let synthetic_all = all.columns.as_ref().unwrap();
        let synthetic_a = by_request["GET /a"].columns.as_ref().unwrap();
        let counts = |columns: &Columns| -> Vec<(usize, usize)> {
            filled(columns)
                .into_iter()
                .map(|(index, column)| (index, column.counter))
                .collect()
        };
        assert_eq!(counts(synthetic_all), [(0, 1), (305 / 8, 1), (650 / 8, 4)]);
        assert_eq!(counts(synthetic_a), [(0, 1), (305 / 16, 1), (650 / 16, 3)]);
        assert!(!by_request.contains_key("GET /none"));
        assert_eq!(synthetic_all.columns[0].maximum_time_us, 2_000_000);
        assert_eq!(synthetic_a.columns[650 / 16].maximum_time_us, 3_000_000);

        let heatmap = latency.report_heatmap(true, Utc::now()).unwrap();
        let all = &heatmap.series[0];
        let adjusted = all.adjusted.as_ref().unwrap();
        // Column 0 holds 8 measured seconds and one synthetic time.
        let column = &adjusted.columns[0];
        assert_eq!(column[0], 0);
        let total: usize = column[4..].chunks(3).map(|cell| cell[1]).sum();
        assert_eq!(total, 9);
        let measured: usize = all.measured.columns[0][4..]
            .chunks(3)
            .map(|cell| cell[1])
            .sum();
        assert_eq!(measured, 8);
        // Without mitigation events there is no adjusted view.
        let heatmap = latency.report_heatmap(false, Utc::now()).unwrap();
        assert!(heatmap
            .series
            .iter()
            .all(|series| series.adjusted.is_none()));
    }

    /// After a reset that cleared every synthetic time, no series offers
    /// an Adjusted view, even with mitigation events recorded before it.
    #[test]
    fn no_adjusted_view_without_synthetic_times() {
        let mut graph = GraphData::with_latency_budgets(true, 600, 120);
        graph.record_request_per_second_and_latency("GET /a", 1, Some(10));
        graph.record_synthetic_latency("GET /a", 1, 900);
        graph.reset_rate_series(5);
        graph.record_request_per_second_and_latency("GET /a", 6, Some(10));
        graph.finish_latency_columns();
        let heatmap = latency(&graph).report_heatmap(true, Utc::now()).unwrap();
        assert_eq!(heatmap.series.len(), 2);
        assert!(heatmap
            .series
            .iter()
            .all(|series| series.adjusted.is_none()));
    }

    /// With budgets that fill at different seconds, each series widens when
    /// its own budget fills: no series ever holds more columns than its
    /// budget, after any write.
    #[test]
    fn each_budget_widens_its_own_series() {
        let mut graph = GraphData::with_latency_budgets(true, 6, 2);
        for second in 0..40 {
            graph.record_request_per_second_and_latency("GET /a", second, Some(10));
            graph.record_synthetic_latency("GET /a", second, 500);
            let latency = latency(&graph);
            let own = latency.rings[0].columns.as_ref().unwrap();
            let synthetic = latency.synthetic.as_ref().unwrap();
            let synthetic_own = synthetic.1["GET /a"].columns.as_ref().unwrap();
            let synthetic_all = synthetic.0.columns.as_ref().unwrap();
            assert!(
                own.columns.len() <= 2 && synthetic_own.columns.len() <= 2,
                "{}",
                second
            );
            assert!(
                latency.all_columns.as_ref().unwrap().columns.len() <= 6,
                "{}",
                second
            );
            assert!(synthetic_all.columns.len() <= 6, "{}", second);
        }
        let latency = latency(&graph);
        assert_eq!((latency.all_width, latency.request_width), (8, 32));
    }

    /// A synthetic time from a second later than any measured one widens
    /// the columns before it is written.
    #[test]
    fn a_synthetic_time_fits_its_second_in_the_columns() {
        let mut graph = GraphData::with_latency_budgets(true, 4, 2);
        graph.record_request_per_second_and_latency("GET /a", 0, Some(10));
        graph.record_synthetic_latency("GET /a", 9, 500);
        let latency = latency(&graph);
        assert_eq!((latency.all_width, latency.request_width), (4, 8));
        let synthetic = latency.synthetic.as_ref().unwrap();
        assert!(synthetic.0.columns.as_ref().unwrap().columns.len() <= 4);
        assert!(
            synthetic.1["GET /a"]
                .columns
                .as_ref()
                .unwrap()
                .columns
                .len()
                <= 2
        );
    }

    /// A reset also empties the columns synthetic times were folded into:
    /// none survive into the new run's Adjusted view.
    #[test]
    fn a_reset_clears_folded_synthetic_times() {
        let mut graph = GraphData::with_latency_budgets(true, 600, 120);
        for second in 0..5 {
            graph.record_request_per_second_and_latency("GET /a", second, Some(10));
            graph.record_synthetic_latency("GET /a", second, 900);
        }
        graph.finish_latency_columns();
        graph.reset_rate_series(10);
        graph.record_request_per_second_and_latency("GET /a", 11, Some(10));
        graph.finish_latency_columns();
        let heatmap = latency(&graph).report_heatmap(true, Utc::now()).unwrap();
        assert!(heatmap
            .series
            .iter()
            .all(|series| series.adjusted.is_none()));
    }

    /// A count's place on the colour scale is logarithmic, as the
    /// dashboard's: 10 between 1 and 100 sits in the middle.
    #[test]
    fn report_heatmap_colours_on_a_log_scale() {
        assert_eq!(ReportView::level(1, 1, 100), 0);
        assert_eq!(ReportView::level(10, 1, 100), 500);
        assert_eq!(ReportView::level(100, 1, 100), 1000);
        // Every cell holding the same count is the highest.
        assert_eq!(ReportView::level(7, 7, 7), 1000);
    }

    /// A run with no requests since the last reset shows no heatmap card.
    #[test]
    fn no_card_without_requests() {
        let mut graph = GraphData::with_latency_budgets(false, 600, 120);
        graph.finish_latency_columns();
        assert_eq!(graph.get_latency_heatmap_markup(false, Utc::now()), "");
        graph.record_request_per_second_and_latency("GET /a", 1, Some(10));
        graph.finish_latency_columns();
        assert!(graph
            .get_latency_heatmap_markup(false, Utc::now())
            .contains("Response time heatmap"));
    }

    /// A name first seen after the per name columns widened gets columns of
    /// the same width, so its measured and synthetic columns line up with
    /// every other name's.
    #[test]
    fn a_late_name_gets_the_current_width() {
        let mut graph = GraphData::with_latency_budgets(true, 10, 5);
        for second in 0..40 {
            graph.record_request_per_second_and_latency("GET /early", second, Some(10));
        }
        graph.record_request_per_second_and_latency("GET /late", 39, Some(10));
        graph.record_synthetic_latency("GET /late", 38, 500);
        graph.finish_latency_columns();
        let latency = latency(&graph);
        assert_eq!(latency.request_width, 8);
        let late = latency.rings[latency.by_request["GET /late"]]
            .columns
            .as_ref()
            .unwrap();
        assert_eq!(
            filled(late).iter().map(|(i, _)| *i).collect::<Vec<_>>(),
            [4]
        );
        let synthetic = latency.synthetic.as_ref().unwrap().1["GET /late"]
            .columns
            .as_ref()
            .unwrap();
        assert_eq!(
            filled(synthetic)
                .iter()
                .map(|(i, _)| *i)
                .collect::<Vec<_>>(),
            [4]
        );
    }

    /// A reset empties the columns, restores one second columns and starts
    /// them at its second: a time from before it, a request in flight at
    /// the reset, goes to column 0, once.
    #[test]
    fn reset_clears_the_columns() {
        let mut graph = GraphData::with_latency_budgets(false, 4, 4);
        for second in 0..20 {
            graph.record_request_per_second_and_latency("GET /a", second, Some(10));
        }
        assert!(latency(&graph).all_width > 1);
        graph.reset_rate_series(20);
        let latency_after = latency(&graph);
        assert_eq!(
            (
                latency_after.origin,
                latency_after.all_width,
                latency_after.request_width
            ),
            (20, 1, 1)
        );
        assert!(filled(latency_after.all_columns.as_ref().unwrap()).is_empty());
        graph.record_request_per_second_and_latency("GET /a", 19, Some(30));
        graph.record_request_per_second_and_latency("GET /a", 21, Some(10));
        graph.finish_latency_columns();
        let latency = latency(&graph);
        let all = latency.all_columns.as_ref().unwrap();
        let counts: Vec<(usize, usize)> = filled(all)
            .into_iter()
            .map(|(index, column)| (index, column.counter))
            .collect();
        assert_eq!(counts, [(0, 1), (1, 1)]);
        assert_eq!(all.columns[0].maximum_time_us, 30_000);
        // The report places column 0 at the reset's second.
        assert_eq!(
            latency.report_heatmap(false, Utc::now()).unwrap().origin,
            20
        );
    }

    /// The first 100 names get columns of their own; later names count
    /// only toward all requests, and the report lists only names with
    /// columns.
    #[test]
    fn only_the_first_names_get_columns() {
        let mut graph = GraphData::with_latency_budgets(false, 600, 120);
        for name in 0..105 {
            graph.record_request_per_second_and_latency(&format!("GET /{name}"), 1, Some(10));
        }
        graph.finish_latency_columns();
        let latency = latency(&graph);
        assert_eq!(latency.rings.len(), MAX_REQUEST_SERIES);
        assert!(latency.rings.iter().all(|ring| ring
            .columns
            .as_ref()
            .is_some_and(|columns| columns.columns[1].counter == 1)));
        assert_eq!(
            latency.all_columns.as_ref().unwrap().columns[1].counter,
            105
        );
        let heatmap = latency.report_heatmap(false, Utc::now()).unwrap();
        assert_eq!(heatmap.series.len(), MAX_REQUEST_SERIES + 1);
        assert!(heatmap
            .series
            .iter()
            .all(|series| series.name != "GET /104"));
    }

    /// At the end of the run the columns hold every time: the last two
    /// seconds, still open in the rings, and a name last requested 10
    /// seconds before the end, whose times waited in its pending counts.
    #[test]
    fn the_end_of_the_run_is_in_the_columns() {
        let mut graph = GraphData::with_latency_budgets(false, 600, 120);
        let one = |time: u64| {
            let mut data = GooseRequestMetricTimingData::new(None);
            data.record_time(time);
            data
        };
        for second in 0..=30 {
            graph.record_requests_per_second_and_latency_batch("GET /a", second, 1, Some(&one(10)));
            if second <= 20 {
                graph.record_requests_per_second_and_latency_batch(
                    "GET /b",
                    second,
                    1,
                    Some(&one(20)),
                );
            }
        }
        graph.finish_latency_columns();
        let latency = latency(&graph);
        let columns = |key: &str| {
            latency.rings[latency.by_request[key]]
                .columns
                .as_ref()
                .unwrap()
        };
        for second in 0..=30 {
            assert_eq!(columns("GET /a").columns[second].counter, 1, "{second}");
            let all = &latency.all_columns.as_ref().unwrap().columns[second];
            assert_eq!(all.counter, if second <= 20 { 2 } else { 1 }, "{second}");
        }
        assert_eq!(columns("GET /b").columns.len(), 21);
        assert_eq!(columns("GET /b").columns[20].counter, 1);
    }

    /// A report whose graphs start after the run did, as after Save report
    /// is turned on during a run, draws them from their first second, also
    /// when it is rendered again from its JSON report.
    #[test]
    fn graphs_that_start_late_keep_their_first_second() {
        let mut graph = GraphData::new();
        for second in 0..=900 {
            graph.record_users_per_second(10, second);
            graph.record_requests_per_second("GET /a", second);
            graph.record_requests_per_second("GET /b", second);
        }
        graph.prune_to_window(600);
        let origin = graph.first_second();
        assert!(origin > 0);
        let started = Utc.with_ymd_and_hms(2026, 10, 22, 10, 0, 0).unwrap();
        let at = |second: usize| {
            Local
                .timestamp_opt(started.timestamp() + second as i64, 0)
                .unwrap()
                .format("%Y-%m-%d %H:%M:%S")
                .to_string()
        };
        let check = |graph: &GraphData| {
            for markup in [
                graph.get_active_users_graph(false).get_markup(&[], started),
                graph
                    .get_requests_per_second_graph(true)
                    .get_markup(&[], started),
            ] {
                assert!(markup.contains(&at(origin)), "{}", markup);
                assert!(!markup.contains(&at(origin - 1)), "{}", markup);
            }
        };
        check(&graph);
        let json = serde_json::to_string(&graph.report_graphs(false, started, None)).unwrap();
        let (again, _) = serde_json::from_str::<RecordedGraphs>(&json)
            .unwrap()
            .into_parts();
        assert_eq!(again.first_second(), origin);
        check(&again);
    }

    /// Without whole run columns, as for the dashboard with Save report off,
    /// the rings work as before and the report has no heatmap.
    #[test]
    fn no_columns_without_a_report() {
        let configuration =
            crate::GooseConfiguration::parse_args_default(&["--dashboard"]).unwrap();
        assert!(configuration.records_latency_series());
        let mut graph = GraphData::for_metrics_processor(
            configuration.records_latency_series(),
            configuration.records_synthetic_latency(),
            false,
        );
        for second in 0..10 {
            graph.record_request_per_second_and_latency("GET /a", second, Some(10));
        }
        graph.finish_latency_columns();
        let latency = latency(&graph);
        assert!(latency.all_columns.is_none());
        assert!(latency.rings.iter().all(|ring| ring.columns.is_none()));
        assert_eq!(latency.limit, usize::MAX);
        assert_eq!(graph.get_latency_heatmap_markup(false, Utc::now()), "");

        // A report under the off switch keeps no histograms at all.
        for args in [
            vec!["--no-percentiles-over-time"],
            vec!["--no-percentiles-over-time", "--report-file", "r.html"],
        ] {
            let configuration = crate::GooseConfiguration::parse_args_default(&args).unwrap();
            let graph = GraphData::for_metrics_processor(
                configuration.records_latency_series(),
                configuration.records_synthetic_latency(),
                configuration.records_latency_columns(),
            );
            assert!(!graph.records_latency(), "{:?}", args);
        }
    }

    /// Each column's p50, p95 and p99 in the report's data are the ones
    /// Rust computes from the same counts, for the measured and the
    /// adjusted view, and the count's place on the colour scale runs from
    /// 0 for the sparsest cell to 1000 for the busiest.
    #[test]
    fn report_heatmap_percentiles_come_from_the_columns() {
        let mut graph = GraphData::with_latency_budgets(true, 8, 4);
        let mut state = 2026;
        let mut measured = Vec::new();
        let mut synthetic = Vec::new();
        for second in 0..30 {
            for _ in 0..20 {
                let time = seeded_time(&mut state);
                graph.record_request_per_second_and_latency("GET /a", second, Some(time));
                measured.push((second, time));
            }
            if second % 7 == 0 {
                graph.record_synthetic_latency("GET /a", second, 4_000);
                synthetic.push((second, 4_000));
            }
        }
        graph.finish_latency_columns();
        let heatmap = latency(&graph).report_heatmap(true, Utc::now()).unwrap();
        let all = &heatmap.series[0];
        assert_eq!(all.width, 4);
        let check = |view: &ReportView, times: &[(usize, u64)], width: usize| {
            let expected = reference_columns(times, width);
            assert_eq!(view.columns.len(), expected.len());
            for (column, (index, direct)) in view.columns.iter().zip(&expected) {
                assert_eq!(column[0], *index);
                assert_eq!(
                    column[1..4],
                    [
                        direct.percentile_us(0.5) as usize,
                        direct.percentile_us(0.95) as usize,
                        direct.percentile_us(0.99) as usize
                    ]
                );
                let cells: Vec<u64> = column[4..]
                    .chunks(3)
                    .flat_map(|cell| [cell[0] as u64, cell[1] as u64])
                    .collect();
                assert_eq!(cells, direct.heat_cells());
            }
            let levels: Vec<usize> = view
                .columns
                .iter()
                .flat_map(|column| column[4..].chunks(3).map(|cell| (cell[1], cell[2])))
                .filter(|&(count, _)| count == view.min || count == view.max)
                .map(|(count, level)| {
                    if count == view.min {
                        level
                    } else {
                        1000 - level
                    }
                })
                .collect();
            assert!(levels.iter().all(|&level| level == 0));
        };
        check(&all.measured, &measured, 4);
        let both: Vec<(usize, u64)> = measured.iter().chain(&synthetic).copied().collect();
        check(all.adjusted.as_ref().unwrap(), &both, 4);
        let a = &heatmap.series[1];
        assert_eq!((a.name, a.width), ("GET /a", 8));
        check(&a.measured, &measured, 8);
        check(a.adjusted.as_ref().unwrap(), &both, 8);
    }

    /// The card's JSON, written one series at a time, is the heatmap with
    /// every series built, which the other tests check.
    #[test]
    fn card_json_is_the_built_heatmap() {
        let mut graph = GraphData::with_latency_budgets(true, 4, 2);
        let mut state = 763;
        for second in 0..20 {
            for name in ["GET /a", "GET /b"] {
                graph.record_request_per_second_and_latency(
                    name,
                    second,
                    Some(seeded_time(&mut state)),
                );
            }
            if second % 3 == 0 {
                graph.record_synthetic_latency("GET /a", second, 2_500);
            }
        }
        graph.finish_latency_columns();
        let started = Utc::now();
        let markup = graph.get_latency_heatmap_markup(true, started);
        let start = markup.find("graph-latency-heatmap-data\">").unwrap() + 28;
        let end = start + markup[start..].find("</script>").unwrap();
        let card: serde_json::Value = serde_json::from_str(&markup[start..end]).unwrap();
        let built = latency(&graph).report_heatmap(true, started).unwrap();
        assert_eq!(card, serde_json::to_value(&built).unwrap());
        assert_eq!(card["series"].as_array().unwrap().len(), 3);
        assert!(!card["series"][1]["adjusted"].is_null());
    }

    /// A hostile request name stays text: the report's JSON has no `<`,
    /// `>`, `&`, U+2028 or U+2029, parses back to the same name, and the
    /// script builds the select from it with `textContent` only.
    #[test]
    fn report_heatmap_escapes_request_names() {
        let name = "GET /</script><img src=x onerror=alert(1)>\"'&\u{2028}\u{2029}";
        let mut graph = GraphData::with_latency_budgets(false, 600, 120);
        graph.record_request_per_second_and_latency(name, 1, Some(10));
        graph.finish_latency_columns();
        let markup = graph.get_latency_heatmap_markup(false, Utc::now());
        let start = markup.find("graph-latency-heatmap-data\">").unwrap() + 28;
        let end = start + markup[start..].find("</script>").unwrap();
        let json = &markup[start..end];
        for unsafe_char in ['<', '>', '&', '\u{2028}', '\u{2029}'] {
            assert!(!json.contains(unsafe_char), "{:?} in {}", unsafe_char, json);
        }
        let parsed: serde_json::Value = serde_json::from_str(json).unwrap();
        assert_eq!(parsed["series"][1]["name"], name);
        assert!(!REPORT_HEATMAP_SCRIPT.contains("innerHTML"));
        assert!(REPORT_HEATMAP_SCRIPT.contains("option.textContent = series.name"));
    }
}

/// The heatmap's rows under a millisecond and its percentile lines in
/// microseconds (#585).
#[cfg(test)]
mod microsecond_tests {
    use super::*;
    use crate::metrics::dashboard_snapshot::HeatmapColumn;

    /// The default `--timeout`, 60 seconds, in microseconds.
    const DEFAULT_TIMEOUT_US: u64 = 60_000_000;

    #[test]
    fn display_rows_are_100_microseconds_under_a_millisecond() {
        for us in 0..1000 {
            assert_eq!(display_index(us), (us / 100) as usize, "{us} µs");
        }
        assert_eq!(display_index(1000), 10);
        // Every time to the default timeout has a row the packed index holds.
        let mut previous = 0;
        for us in (0..=DEFAULT_TIMEOUT_US).step_by(997) {
            let index = display_index(us);
            assert!(index >= previous, "row falls at {} µs", us);
            assert!(index < 256, "{} µs has row {}", us, index);
            previous = index;
        }
        assert!(display_index(DEFAULT_TIMEOUT_US) < 256);
    }

    #[test]
    fn heatmap_rows_under_a_millisecond_are_filled() {
        let mut graph = GraphData::for_metrics_processor(true, false, false);
        for us in [50, 400, 400, 950] {
            graph.record_requests_per_second("GET /", 1);
            graph.record_latency_us("GET /", 1, us);
        }
        let heatmap = graph
            .export_series_window_until(300, Some(4))
            .heatmap
            .unwrap();
        let column = heatmap.column(1).expect("second 1");
        let pairs = HeatmapColumn::pairs(&column.measured);
        assert_eq!(pairs.len(), 3, "{pairs:?}");
        for ((index, count), (us, expected)) in pairs.iter().zip([(50, 1), (400, 2), (950, 1)]) {
            let [_, lowest, highest] = heatmap.buckets_us[*index as usize];
            assert!(
                lowest <= us && us <= highest,
                "{} µs in {} to {}",
                us,
                lowest,
                highest
            );
            assert_eq!(highest - lowest, 99);
            assert_eq!(*count, expected);
        }
    }

    #[test]
    fn heatmap_percentiles_follow_the_whole_run_to_a_row() {
        let mut histogram = SecondHistogram::default();
        let mut whole = GooseRequestMetricTimingData::default();
        let mut seed: u64 = 585;
        for i in 0..10_000u64 {
            seed = seed
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            let us = if i % 10 < 7 {
                30 + (seed >> 33) % 66
            } else {
                400 + (seed >> 33) % 501
            };
            histogram.record_us(us);
            whole.record_time_us(us);
        }
        for percent in [0.5, 0.95, 0.99] {
            let row = histogram.percentile_us(percent);
            assert_eq!(
                display_index(row),
                bucket_row(crate::metrics::buckets::bucket_floor(
                    whole.percentile_us(percent)
                )),
                "p{percent}"
            );
            assert!(row < 1000);
        }
    }
}