krabka_bench_driver/
qualification.rs

1//! Failover evidence checks and recovery analysis for benchmark runs.
2
3use {
4    crate::{
5        aggregate::{CellKey, stat},
6        ids::{MessageCount, TimeOffsetMs},
7        numeric::to_f64,
8        scenario::{Disturbance, RunOutput, Sample, Stack},
9    },
10    krabka_units::{fmt::Human as _, prelude::*},
11    std::collections::BTreeMap,
12};
13
14#[derive(Debug, Clone, Copy, PartialEq)]
15pub(crate) struct RateRecovery {
16    pub(crate) baseline: Frequency,
17    pub(crate) min_after_kill: Frequency,
18    pub(crate) recovery: Option<Time>,
19}
20
21/// Why one failover run carries no usable evidence, or `None` when it does.
22///
23/// A run that errored, that recorded no disturbance, that stamped a recovery
24/// before the kill, or that moved no traffic is not a measurement. Admitting
25/// one is what let the gate report PASS over a workload that died: a recovery
26/// stamp below the kill stamp saturates to 0 ms in
27/// [`crate::ids::TimeOffsetMs::since`], and a zero-rate baseline reads as a
28/// stack that never slowed down. Each of these is a gate violation and not a
29/// reason to drop the run quietly.
30fn failover_evidence_defect(run: &RunOutput) -> Option<String> {
31    if !run.errors.is_empty() {
32        return Some(format!(
33            "the run reported errors: {}",
34            run.errors.join("; ")
35        ));
36    }
37    let Some(disturbance) = run.disturbance.as_ref() else {
38        return Some("the run recorded no disturbance".to_owned());
39    };
40    if disturbance.kill_at_ms == TimeOffsetMs(0) {
41        return Some("the run recorded no broker kill".to_owned());
42    }
43    if disturbance.recovery_at_ms < disturbance.kill_at_ms {
44        return Some(format!(
45            "the recovery stamp {} is before the kill stamp {}",
46            disturbance.recovery_at_ms, disturbance.kill_at_ms
47        ));
48    }
49    if disturbance.recovery_at_ms == TimeOffsetMs(0) {
50        return Some("the run never acknowledged a record after the kill".to_owned());
51    }
52    if run.throughput.msgs_produced == MessageCount(0) {
53        return Some("the run produced no messages".to_owned());
54    }
55    if run.throughput.producer_rate <= Frequency::ZERO {
56        return Some("the run measured a zero producer rate".to_owned());
57    }
58    None
59}
60
61/// Evaluate failover evidence and comparisons without I/O. An empty result means the gate passes.
62#[must_use]
63pub fn failover_gate_violations(runs: &[RunOutput]) -> Vec<String> {
64    let mut violations = Vec::new();
65    let mut by_group: BTreeMap<CellKey, Vec<&RunOutput>> = BTreeMap::new();
66    for r in runs.iter().filter(|r| r.scenario.failover.is_some()) {
67        // Validate before grouping. A run with no usable evidence never
68        // reaches the comparison, and it fails the gate on its own account.
69        if let Some(defect) = failover_evidence_defect(r) {
70            violations.push(format!(
71                "{}: {} failover run is not a measurement: {defect}",
72                CellKey::of(r).label(),
73                r.stack.as_str(),
74            ));
75            continue;
76        }
77        by_group.entry(CellKey::of(r)).or_default().push(r);
78    }
79
80    if by_group.is_empty() {
81        violations.push("missing failover results".into());
82        return violations;
83    }
84
85    for (key, group) in by_group {
86        let krabka: Vec<&RunOutput> = group
87            .iter()
88            .copied()
89            .filter(|r| r.stack == Stack::Krabka)
90            .collect();
91        let kafka: Vec<&RunOutput> = group
92            .iter()
93            .copied()
94            .filter(|r| r.stack == Stack::Kafka)
95            .collect();
96        let label = key.label();
97
98        let Some(c_recovery) = mean_failover_recovery(&krabka) else {
99            violations.push(format!(
100                "{label}: missing Krabka failover disturbance result"
101            ));
102            continue;
103        };
104        let Some(k_recovery) = mean_failover_recovery(&kafka) else {
105            violations.push(format!(
106                "{label}: missing kafka failover disturbance result"
107            ));
108            continue;
109        };
110        if c_recovery > k_recovery {
111            violations.push(format!(
112                "{label}: Krabka recovery {} is slower than kafka {}",
113                c_recovery.human(),
114                k_recovery.human()
115            ));
116        }
117        compare_failover_smoothness(&mut violations, &label, &krabka, &kafka);
118
119        require_rate_samples(&mut violations, &label, "krabka", &krabka);
120        require_rate_samples(&mut violations, &label, "kafka", &kafka);
121        compare_rate_recovery(
122            &mut violations,
123            &label,
124            &krabka,
125            &kafka,
126            "producer",
127            producer_sample_rate,
128        );
129        compare_rate_recovery(
130            &mut violations,
131            &label,
132            &krabka,
133            &kafka,
134            "consumer",
135            consumer_sample_rate,
136        );
137    }
138
139    violations
140}
141
142fn compare_failover_smoothness(
143    violations: &mut Vec<String>,
144    label: &str,
145    krabka: &[&RunOutput],
146    kafka: &[&RunOutput],
147) {
148    if let (Some(c_drops), Some(k_drops)) =
149        (mean_failover_dropped(krabka), mean_failover_dropped(kafka))
150        && c_drops > k_drops
151    {
152        violations.push(format!(
153            "{label}: Krabka dropped {c_drops:.0} messages vs kafka {k_drops:.0}"
154        ));
155    }
156    if let (Some(c_spike), Some(k_spike)) = (
157        mean_failover_latency_spike(krabka),
158        mean_failover_latency_spike(kafka),
159    ) && c_spike > k_spike
160    {
161        violations.push(format!(
162            "{label}: Krabka latency spike {} is higher than kafka {}",
163            c_spike.human(),
164            k_spike.human()
165        ));
166    }
167}
168
169fn compare_rate_recovery(
170    violations: &mut Vec<String>,
171    label: &str,
172    krabka: &[&RunOutput],
173    kafka: &[&RunOutput],
174    metric: &str,
175    select: SampleRate,
176) {
177    let (Some(c_rate), Some(k_rate)) = (
178        mean_rate_recovery(krabka, select),
179        mean_rate_recovery(kafka, select),
180    ) else {
181        return;
182    };
183
184    match (c_rate.recovery, k_rate.recovery) {
185        (Some(c), Some(k)) if c > k => violations.push(format!(
186            "{label}: Krabka {metric} rate recovery {} is slower than kafka {}",
187            c.human(),
188            k.human()
189        )),
190        (None, Some(k)) => violations.push(format!(
191            "{label}: Krabka {metric} rate did not recover while kafka recovered in {}",
192            k.human()
193        )),
194        (None, None) => violations.push(format!("{label}: Krabka {metric} rate did not recover")),
195        _ => {}
196    }
197}
198
199fn require_rate_samples(
200    violations: &mut Vec<String>,
201    label: &str,
202    stack_label: &str,
203    runs: &[&RunOutput],
204) {
205    if runs.is_empty() {
206        return;
207    }
208    for (idx, run) in runs.iter().enumerate() {
209        if rate_recovery_for_run(run, producer_sample_rate).is_none()
210            || rate_recovery_for_run(run, consumer_sample_rate).is_none()
211        {
212            violations.push(format!(
213                "{label}: {stack_label} failover run is missing message-rate samples (run {})",
214                idx + 1
215            ));
216        }
217    }
218}
219
220fn mean_failover_value(runs: &[&RunOutput], value: impl Fn(&Disturbance) -> f64) -> Option<f64> {
221    let vals: Vec<f64> = runs
222        .iter()
223        .filter_map(|r| r.disturbance.as_ref())
224        .map(value)
225        .collect();
226    (!vals.is_empty()).then(|| stat(&vals).mean)
227}
228
229pub(crate) fn mean_failover_recovery(runs: &[&RunOutput]) -> Option<Time> {
230    mean_failover_value(runs, |d| d.recovery_at_ms.since(d.kill_at_ms).secs_f64())
231        .map(Time::from_secs_f64)
232}
233
234fn mean_failover_dropped(runs: &[&RunOutput]) -> Option<f64> {
235    mean_failover_value(runs, |d| to_f64(d.dropped.0))
236}
237
238fn mean_failover_latency_spike(runs: &[&RunOutput]) -> Option<Time> {
239    mean_failover_value(runs, |d| d.latency_spike_max.secs_f64()).map(Time::from_secs_f64)
240}
241
242pub(crate) fn mean_rate_recovery(runs: &[&RunOutput], select: SampleRate) -> Option<RateRecovery> {
243    let vals: Vec<RateRecovery> = runs
244        .iter()
245        .filter_map(|r| rate_recovery_for_run(r, select))
246        .collect();
247    if vals.is_empty() {
248        return None;
249    }
250    let baselines: Vec<f64> = vals.iter().map(|v| v.baseline.per_sec_f64()).collect();
251    let mins: Vec<f64> = vals
252        .iter()
253        .map(|v| v.min_after_kill.per_sec_f64())
254        .collect();
255    let recoveries: Vec<f64> = vals
256        .iter()
257        .filter_map(|v| v.recovery.map(TimeExt::secs_f64))
258        .collect();
259    Some(RateRecovery {
260        baseline: Frequency::from_per_sec(stat(&baselines).mean),
261        min_after_kill: Frequency::from_per_sec(stat(&mins).mean),
262        recovery: (!recoveries.is_empty()).then(|| Time::from_secs_f64(stat(&recoveries).mean)),
263    })
264}
265
266fn rate_recovery_for_run(r: &RunOutput, select: SampleRate) -> Option<RateRecovery> {
267    let failover = r.scenario.failover.as_ref()?;
268    if r.samples.is_empty() {
269        return None;
270    }
271    // Sample offsets are measured from the *measurement* window start, so the
272    // kill offset drops the warmup the scenario spent before it.
273    let kill_offset = failover.kill_after - r.scenario.warmup;
274    let rate_at = |s: &Sample| select(s).per_sec_f64();
275    let baseline: Vec<f64> = r
276        .samples
277        .iter()
278        .filter(|s| s.t_offset_ms.as_time() < kill_offset)
279        .map(rate_at)
280        .collect();
281    if baseline.is_empty() {
282        return None;
283    }
284    let baseline_rate = stat(&baseline).mean;
285    let after: Vec<&Sample> = r
286        .samples
287        .iter()
288        .filter(|s| s.t_offset_ms.as_time() >= kill_offset)
289        .collect();
290    if after.is_empty() {
291        return None;
292    }
293    let min_after_kill = after
294        .iter()
295        .map(|s| rate_at(s))
296        .fold(f64::INFINITY, f64::min);
297    let threshold = baseline_rate * 0.90;
298    let recovery = after
299        .iter()
300        .find(|s| rate_at(s) >= threshold)
301        .map(|s| s.t_offset_ms.as_time() - kill_offset);
302
303    Some(RateRecovery {
304        baseline: Frequency::from_per_sec(baseline_rate),
305        min_after_kill: Frequency::from_per_sec(min_after_kill),
306        recovery,
307    })
308}
309
310/// How one of the two client message rates is read out of a time-series sample.
311pub(crate) type SampleRate = fn(&Sample) -> Frequency;
312
313pub(crate) fn producer_sample_rate(s: &Sample) -> Frequency { s.producer_rate }
314
315pub(crate) fn consumer_sample_rate(s: &Sample) -> Frequency { s.consumer_rate }
krabka_bench_driver/qualification.rs

krabka_bench_driver/
qualification.rs
