veredictum 0.1.3

// SPDX-FileCopyrightText: Veredictum contributors
// SPDX-License-Identifier: Apache-2.0

//! The measured-window core: dispatch the built schedule open-loop, collect
//! per-operation HDR histograms, aggregate the re-checkable measurement
//! record.
//!
//! Shared by the class runs (conformance) and the stress ladder
//! (exploration).

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, mpsc};
use std::time::Instant;

use hdrhistogram::Histogram;

use crate::ixit::{Environment, Ixit};
use crate::perf::{
    ClassVerdict, Measurement, OperationMeasurement, PerfOp, PerformanceCase, Principal,
    class_verdict,
};
use crate::perf_run::client::PerfPrincipals;
use crate::perf_run::corpus::SeededCorpus;
use crate::perf_run::execute::{CaptureStore, perform};
use crate::perf_run::pack::JourneyPack;
use crate::perf_run::schedule::{JourneyWorkload, build_schedule};

/// Latency histograms record microseconds in `1 µs ..= 10 min` at 3
/// significant figures — far past the client timeout, so a timeout can
/// never saturate the range.
const HDR_MAX_US: u64 = 600_000_000;

/// How many failed arrivals report their observed wire status / reason
/// through the progress channel — the triage evidence a bare error count
/// cannot carry.
const FAILURE_SAMPLES: u32 = 16;

/// The per-operation aggregation a run collects.
struct OpRecorder {
    histogram: Histogram<u64>,
    errors: u64,
}

/// One completed arrival: operation, latency from the PLANNED instant, and
/// whether the wire outcome matched the binding's expected kind.
struct Completion {
    op: PerfOp,
    latency_us: u64,
    ok: bool,
    recorded: bool,
}

/// Why the generator could not fire a planned arrival.
///
/// A fault is the INSTRUMENT failing to drive the schedule, so it is never
/// a wire observation about the server: it fails the window instead of
/// landing in a histogram as an error arrival.
#[derive(Debug, Clone, Copy)]
enum GeneratorFault {
    /// The principal set driving the window declares no instance for the
    /// arrival's principal, so no request was ever sent.
    UndeclaredPrincipal(Principal),
}

impl std::fmt::Display for GeneratorFault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match *self {
            Self::UndeclaredPrincipal(principal) => write!(
                f,
                "the ixit declares no instance for the {principal:?} principal"
            ),
        }
    }
}

/// Everything one window's collector folded out of its arrival reports.
struct CollectedWindow {
    /// The per-operation recorders, in first-arrival order.
    recorders: Vec<(PerfOp, OpRecorder)>,
    /// The generator faults, counted per `<operation>: <reason>` line.
    generator_faults: BTreeMap<String, u64>,
}

/// What one planned arrival reports back to the collector.
enum ArrivalReport {
    /// The arrival fired and its response was timed.
    Completed(Completion),
    /// The generator could not fire the arrival at all.
    NotFired {
        /// The operation the arrival would have driven.
        op: PerfOp,
        /// Why the generator could not fire it.
        fault: GeneratorFault,
    },
}

/// One executed open-loop window's raw outcome — the shared core the class
/// runs (conformance) and the knee stress ladder (exploration) both drive.
#[derive(Debug)]
pub struct WindowOutcome {
    /// Measured arrivals over the actual measured span (arrivals/s).
    pub offered_load_sustained: f64,
    /// Per-operation records (encoded HDR histograms, re-checkable).
    pub operations: Vec<OperationMeasurement>,
    /// Whether the GENERATOR failed to hold the schedule (dispatch lagged
    /// more than 2% past the planned span) — the honest stop signal for a
    /// stress climb: beyond this point the instrument, not the SUT, is the
    /// bottleneck.
    pub generator_bound: bool,
}

/// Folds every arrival report the dispatcher sends into per-operation
/// recorders, returning them beside the generator faults the window hit.
///
/// Only a recorded completion reaches a histogram. An arrival the generator
/// could not fire carries no latency at all, so it is counted per
/// `<operation>: <reason>` line instead, warmup arrivals included: an
/// instrument that could not drive the schedule has no measurement to
/// publish about either span.
///
/// The empty prototype is built once and cloned per operation, so the only
/// bounds check the crate can refuse happens here rather than per arrival.
///
/// # Errors
/// A message when the latency-histogram bounds are refused.
fn collect_completions(rx: mpsc::Receiver<ArrivalReport>) -> Result<CollectedWindow, String> {
    let empty = Histogram::new_with_bounds(1, HDR_MAX_US, 3)
        .map_err(|e| format!("latency histogram bounds refused: {e}"))?;
    let mut recorders: Vec<(PerfOp, OpRecorder)> = Vec::new();
    let mut generator_faults: BTreeMap<String, u64> = BTreeMap::new();
    for report in rx {
        let done = match report {
            ArrivalReport::Completed(done) => done,
            ArrivalReport::NotFired { op, fault } => {
                let counted = generator_faults
                    .entry(format!("{}: {fault}", op.as_str()))
                    .or_default();
                *counted = counted.saturating_add(1);
                continue;
            }
        };
        if !done.recorded {
            continue;
        }
        let index = if let Some(index) = recorders.iter().position(|(op, _)| *op == done.op) {
            index
        } else {
            recorders.push((
                done.op,
                OpRecorder {
                    histogram: empty.clone(),
                    errors: 0,
                },
            ));
            recorders.len().saturating_sub(1)
        };
        if let Some((_, recorder)) = recorders.get_mut(index) {
            let value = done.latency_us.clamp(1, HDR_MAX_US);
            let _saturated = recorder.histogram.record(value);
            if !done.ok {
                recorder.errors = recorder.errors.saturating_add(1);
            }
        }
    }
    Ok(CollectedWindow {
        recorders,
        generator_faults,
    })
}

/// The window's refusal when the generator could not fire planned
/// arrivals: the total count, then one `<operation>: <reason> (N arrivals)`
/// clause per distinct fault. `None` when every planned arrival fired.
///
/// A fault is the instrument failing to drive the schedule, so the window
/// publishes no measurement at all instead of a record missing the
/// arrivals it lost.
fn generator_fault_refusal(faults: &BTreeMap<String, u64>) -> Option<String> {
    let faulted: u64 = faults.values().copied().sum();
    if faulted == 0 {
        return None;
    }
    let detail: Vec<String> = faults
        .iter()
        .map(|(reason, count)| format!("{reason} ({count} arrivals)"))
        .collect();
    Some(format!("{faulted} generator faults: {}", detail.join("; ")))
}

/// Execute one open-loop window: the journey workload at `rate` operation
/// arrivals/s for `warmup_s + duration_s`, recording only the post-warmup
/// span.
///
/// The workload's own [`PerfPrincipals`] is the one principal set: the same
/// declaration decides which journeys the schedule plans and which client
/// each arrival is fired through, so the schedule can never plan an arrival
/// the dispatcher has no client for.
///
/// # Errors
/// A message on schedule construction or aggregation failure, and one
/// naming the count when the generator could not fire an arrival. An
/// individual arrival's WIRE fault is an error observation instead, never a
/// run failure.
#[expect(
    clippy::too_many_lines,
    reason = "one measured-window procedure: schedule → collect → aggregate"
)]
pub fn run_window(
    corpus: &SeededCorpus,
    workload: &JourneyWorkload<'_>,
    rate: f64,
    warmup_s: u64,
    duration_s: u64,
    progress: &(dyn Fn(String) + Sync),
) -> Result<WindowOutcome, String> {
    let principals: &PerfPrincipals = workload.principals;
    let schedule = build_schedule(workload, rate, warmup_s, duration_s, corpus.ward.len())?;
    if !schedule.dropped_journeys.is_empty() {
        progress(format!(
            "journeys not scheduled (the ixit declares no principal for them; the remaining \
             shares were renormalized): {}",
            schedule.dropped_journeys.join(", ")
        ));
    }
    let total = schedule.arrivals.len();
    let captures = CaptureStore::new();

    let (tx, rx) = mpsc::channel::<ArrivalReport>();
    let collector = std::thread::spawn(move || collect_completions(rx));

    let start = Instant::now();
    let dispatched_measured = Arc::new(AtomicU64::new(0));
    // Failure sampling (the first FAILURE_SAMPLES failed arrivals), and the
    // same bound over the arrivals the generator could not fire.
    let failure_samples = Arc::new(AtomicU32::new(0));
    // The dispatcher alone reports an unfired arrival, so its own sample
    // counter needs no sharing.
    let mut fault_samples: u32 = 0;
    progress(format!(
        "open-loop schedule: {total} arrivals ({} measured) at {rate}/s aggregate \
         ({warmup_s}s warmup + {duration_s}s measured, {} journeys interleaved)",
        schedule.planned_measured,
        schedule
            .arrivals
            .last()
            .map_or(0, |a| a.journey.saturating_add(1))
    ));

    // The dispatch span (last arrival fired − start) is captured before the
    // scope waits for in-flight workers, so trailing responses never inflate
    // the sustained-load denominator.
    let dispatch_span = std::thread::scope(|scope| {
        for (i, planned_arrival) in schedule.arrivals.iter().enumerate() {
            // The principal the arrival is driven by. Without its client
            // nothing can be sent, so the arrival is a generator fault and
            // never a wire observation about the server.
            let principal = planned_arrival.op.principal();
            let Some(client) = principals.client(principal) else {
                let fault = GeneratorFault::UndeclaredPrincipal(principal);
                if fault_samples < FAILURE_SAMPLES {
                    fault_samples = fault_samples.saturating_add(1);
                    progress(format!(
                        "arrival not fired: {} at {:?}: {fault}",
                        planned_arrival.op.as_str(),
                        planned_arrival.at,
                    ));
                }
                let _closed = tx.send(ArrivalReport::NotFired {
                    op: planned_arrival.op,
                    fault,
                });
                continue;
            };
            let planned = start + planned_arrival.at;
            let now = Instant::now();
            if planned > now {
                std::thread::sleep(planned - now);
            }
            if planned_arrival.recorded {
                dispatched_measured.fetch_add(1, Ordering::Relaxed);
            }
            let tx = tx.clone();
            let client = client.clone();
            let captures = &captures;
            #[expect(
                clippy::as_conversions,
                reason = "the arrival index widens exactly: usize is at most 64 bits on every supported target"
            )]
            let arrival_index = i as u64;
            let failure_samples = Arc::clone(&failure_samples);
            scope.spawn(move || {
                let mut observed: Option<u16> = None;
                let outcome = perform(
                    &client,
                    arrival_index,
                    planned_arrival,
                    corpus,
                    workload.pack,
                    captures,
                    &mut observed,
                );
                let latency = planned.elapsed();
                let latency_us = u64::try_from(latency.as_micros().min(u128::from(HDR_MAX_US)))
                    .unwrap_or(HDR_MAX_US);
                // An unresolvable prerequisite (the SUT has not landed the
                // earlier stage) is an honest error observation.
                let ok = match &outcome {
                    Ok(ok) => *ok,
                    Err(_) => false,
                };
                if !ok && failure_samples.fetch_add(1, Ordering::Relaxed) < FAILURE_SAMPLES {
                    let detail = match (&outcome, observed) {
                        (Err(reason), _) => reason.clone(),
                        (Ok(_), Some(status)) => format!("unexpected wire status {status}"),
                        (Ok(_), None) => "no wire observation".to_owned(),
                    };
                    progress(format!(
                        "arrival failure sample: {} journey {} at {:?}: {detail}",
                        planned_arrival.op.as_str(),
                        planned_arrival.journey,
                        planned_arrival.at,
                    ));
                }
                let _closed = tx.send(ArrivalReport::Completed(Completion {
                    op: planned_arrival.op,
                    latency_us,
                    ok,
                    recorded: planned_arrival.recorded,
                }));
            });
            if i % 1000 == 999 {
                progress(format!("dispatched {}/{total} arrivals", i + 1));
            }
        }
        drop(tx);
        start.elapsed()
    });

    // A panic payload is `Box<dyn Any>`, not a `Display` error: recover the
    // message the panic carried so the run reports WHY the collector died.
    let collected = collector.join().map_err(|payload| {
        let detail = payload
            .downcast_ref::<&str>()
            .map(|message| (*message).to_owned())
            .or_else(|| payload.downcast_ref::<String>().cloned())
            .unwrap_or_else(|| "non-string panic payload".to_owned());
        format!("collector thread panicked: {detail}")
    })??;
    if let Some(refusal) = generator_fault_refusal(&collected.generator_faults) {
        return Err(refusal);
    }

    // Offered load actually sustained: measured arrivals over the actual
    // measured span (>= the planned window when the generator lagged, which
    // honestly deflates the sustained rate). Under the DIURNAL curve the
    // floor semantic is the busy hour (ITU-T E.500): the planned busy-hour
    // rate scaled by dispatch fidelity — off-peak troughs are the design,
    // never a shortfall; a lagging generator still deflates it.
    let planned_span_s = warmup_s.saturating_add(duration_s);
    #[expect(
        clippy::as_conversions,
        clippy::cast_precision_loss,
        reason = "spans/counts << 2^52"
    )]
    let (offered_load_sustained, generator_bound) = {
        let actual_span = dispatch_span.as_secs_f64().max(planned_span_s as f64);
        let measured_span = actual_span - warmup_s as f64;
        let dispatched = dispatched_measured.load(Ordering::Relaxed) as f64;
        let fidelity = (dispatched / schedule.planned_measured.max(1) as f64)
            .min(planned_span_s as f64 / actual_span);
        let offered = match workload.curve {
            crate::perf::ArrivalCurve::Uniform => {
                if measured_span > 0.0 {
                    dispatched / measured_span
                } else {
                    0.0
                }
            }
            crate::perf::ArrivalCurve::Diurnal => schedule.planned_busy_hour * fidelity,
        };
        let lagged = dispatch_span.as_secs_f64() > planned_span_s as f64 * 1.02;
        (offered, lagged)
    };

    let mut operations: Vec<OperationMeasurement> = Vec::new();
    for (op, recorder) in &collected.recorders {
        operations.push(OperationMeasurement::from_histogram(
            op.as_str(),
            &recorder.histogram,
            recorder.errors,
        )?);
    }
    operations.sort_by(|a, b| a.operation.cmp(&b.operation));

    Ok(WindowOutcome {
        offered_load_sustained,
        operations,
        generator_bound,
    })
}

/// Drive one performance case's open-loop hospital-simulation workload and
/// produce its re-checkable measurement record (verdict computed by
/// [`crate::perf::class_verdict`] from the encoded histograms).
///
/// `warmup_s` is the case's normative warmup; `duration_s` is the case's
/// sustained window or an officially EXTENDED one (the hours ladder — a
/// longer hold of the same offered load is a stricter demonstration).
/// The CLI never passes a window shorter than the case's; only the offline
/// test harness drives synthetic second-scale windows.
///
/// # Errors
/// As [`run_window`], plus the case's own invariant check and the class
/// verdict computation.
#[expect(clippy::too_many_arguments, reason = "the one case-drive seam")]
pub fn drive_case(
    case: &PerformanceCase,
    principals: &PerfPrincipals,
    corpus: &SeededCorpus,
    journey_pack: &JourneyPack,
    catalogue: &crate::perf::JourneyCatalogue,
    environment: &Environment,
    warmup_s: u64,
    duration_s: u64,
    progress: &(dyn Fn(String) + Sync),
) -> Result<Measurement, String> {
    case.check_invariants()?;
    let workload = JourneyWorkload {
        catalogue,
        shares: &case.workload.journeys,
        pack: journey_pack,
        curve: case.workload.arrival_curve,
        principals,
    };
    let window = run_window(
        corpus,
        &workload,
        case.workload.arrival_rate.0,
        warmup_s,
        duration_s,
        progress,
    )?;
    let (verdict, violations) =
        class_verdict(case, window.offered_load_sustained, &window.operations)?;
    Ok(Measurement {
        case: case.id.clone(),
        class: case.class,
        environment: environment.clone(),
        offered_load_sustained: window.offered_load_sustained,
        warmup_s,
        duration_s,
        operations: window.operations,
        verdict,
        violations,
        // The perf handler attaches the sampled telemetry after the window
        // (the sampler brackets this call); stress windows never carry one.
        resources: None,
    })
}

/// Convenience: the ixit precondition for a measured run — every principal
/// the party declares (the `sut` instance is mandatory) and a present
/// environment block.
///
/// # Errors
/// A message naming the missing piece (the environment block is mandatory
/// for performance runs).
pub fn measured_run_context(ixit: &Ixit) -> Result<(PerfPrincipals, &Environment), String> {
    let principals = PerfPrincipals::from_ixit(ixit)?;
    let environment = ixit.environment.as_ref().ok_or_else(|| {
        "ixit has no environment block (mandatory for performance runs)".to_owned()
    })?;
    Ok((principals, environment))
}

/// Whether a measurement's verdict re-derives to the same value from its
/// own embedded histograms (the tamper check the verdict pipeline runs).
///
/// # Errors
/// As [`class_verdict`].
pub fn rederive_verdict(
    case: &PerformanceCase,
    measurement: &Measurement,
) -> Result<(ClassVerdict, Vec<String>), String> {
    class_verdict(
        case,
        measurement.offered_load_sustained,
        &measurement.operations,
    )
}

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

    /// An arrival the generator could not fire carries no latency, so it
    /// never reaches a histogram: it is counted per
    /// `<operation>: <reason>` line, and the window refuses to publish a
    /// measurement naming both the count and the faulting operation.
    #[test]
    fn unfired_arrivals_are_counted_per_reason_and_refuse_the_window() {
        let (tx, rx) = mpsc::channel::<ArrivalReport>();
        let fault = GeneratorFault::UndeclaredPrincipal(Principal::Unauthenticated);
        for _ in 0..3 {
            tx.send(ArrivalReport::NotFired {
                op: PerfOp::UnauthenticatedProbe,
                fault,
            })
            .expect("the collector receiver is alive");
        }
        tx.send(ArrivalReport::Completed(Completion {
            op: PerfOp::EhrRead,
            latency_us: 1_500,
            ok: true,
            recorded: true,
        }))
        .expect("the collector receiver is alive");
        drop(tx);

        let collected = collect_completions(rx).expect("the histogram bounds are accepted");
        assert_eq!(collected.recorders.len(), 1);
        let refusal = generator_fault_refusal(&collected.generator_faults)
            .expect("three unfired arrivals refuse the window");
        assert!(
            refusal.starts_with("3 generator faults"),
            "the refusal does not name the fault count: {refusal}"
        );
        assert!(
            refusal.contains("unauthenticated_probe"),
            "the refusal does not name the faulting operation: {refusal}"
        );
        assert!(
            refusal.contains("(3 arrivals)"),
            "the refusal does not count the reason's arrivals: {refusal}"
        );
    }

    /// A window that fired every planned arrival has nothing to refuse.
    #[test]
    fn a_window_without_faults_produces_no_refusal() {
        assert_eq!(generator_fault_refusal(&BTreeMap::new()), None);
    }
}