krabka_bench_driver/soak/
deadline.rs

1//! The deadline checker: every fault ran, was cleaned up, and was recovered
2//! from in time; every path stayed inside its budgets.
3//!
4//! A fault the schedule named but the evidence does not record fails, as does
5//! a recovery the fault kind implies but nobody measured. The alternative --
6//! judging only what was recorded -- passes a run whose fault injector silently
7//! did nothing, which is the failure a soak is least able to notice by itself.
8
9use {
10    super::{
11        contract::{Contract, RecoveryCheck, Target, human},
12        evidence::{Evidence, FaultRecord, LatencySummary, RecoveryRecord},
13        schedule::ScheduledFault,
14    },
15    krabka_units::{prelude::*, serde_units},
16    serde::{Deserialize, Serialize},
17    std::fmt,
18};
19
20/// How one measured recovery compared with its deadline.
21#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum DeadlineOutcome {
24    Met,
25    Missed,
26    /// No measurement was recorded, or the harness gave up waiting.
27    NotObserved,
28}
29
30/// One recovery against its deadline.
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32pub struct DeadlineVerdict {
33    pub fault: String,
34    pub check: RecoveryCheck,
35    #[serde(with = "serde_units::human::time")]
36    pub deadline: Time,
37    #[serde(with = "serde_units::human::option_time")]
38    pub took: Option<Time>,
39    pub outcome: DeadlineOutcome,
40}
41
42impl fmt::Display for DeadlineVerdict {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        let took = self.took.map_or_else(|| "never".to_owned(), human);
45        write!(
46            f,
47            "{} {}: took {took} against {} ({:?})",
48            self.fault,
49            self.check,
50            human(self.deadline),
51            self.outcome
52        )
53    }
54}
55
56/// A scheduled fault the evidence shows was not executed cleanly.
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58pub struct FaultProblem {
59    pub fault: String,
60    pub problem: String,
61}
62
63/// One latency percentile or throughput floor against its budget.
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65pub struct BudgetVerdict {
66    pub path: String,
67    /// `produce p99`, `consume p50`, `throughput`, ...
68    pub measure: String,
69    pub budget: String,
70    pub measured: String,
71    pub within: bool,
72}
73
74/// How late a fault may start against its scheduled offset. The executor
75/// resolves the target and the cluster state first, so it never starts early,
76/// but it may start a few seconds late.
77pub const START_SLACK: Time = secs(30);
78
79/// The role a fault record must carry for a plan that targets by role.
80fn expected_role(target: Target) -> Option<&'static str> {
81    match target {
82        Target::Controller => Some("controller"),
83        Target::Leader => Some("leader"),
84        Target::Follower => Some("follower"),
85        Target::Any | Target::Broker(_) => None,
86    }
87}
88
89/// Every way `record` fails to be the execution of `fault`.
90fn execution_problems(
91    fault: &ScheduledFault,
92    record: &FaultRecord,
93    started_unix_ms: i64,
94) -> Vec<String> {
95    let mut problems = Vec::new();
96    if record.kind != fault.kind.name() {
97        problems.push(format!(
98            "recorded as a `{}` fault, but the schedule names `{}`",
99            record.kind,
100            fault.kind.name()
101        ));
102    }
103    // A role is recorded with the controller taking precedence, so a leader
104    // that also held the controller is recorded as the controller.
105    if let Some(role) = expected_role(fault.target)
106        && record.role != role
107        && !(role == "leader" && record.role == "controller")
108    {
109        problems.push(format!(
110            "landed on a {} ({}), but the plan targets the {role}",
111            record.role, record.target
112        ));
113    }
114    let offset = Time::from_millis(record.injected_unix_ms.saturating_sub(started_unix_ms));
115    // Records carry whole milliseconds and a jittered start does not, so the
116    // early bound allows one millisecond of rounding.
117    if offset + millis(1) < fault.start || offset > fault.start + START_SLACK {
118        problems.push(format!(
119            "injected at {} into the load, but scheduled at {} (at most {} late)",
120            human(offset),
121            human(fault.start),
122            human(START_SLACK)
123        ));
124    }
125    if let Some(error) = &record.cleanup_error {
126        problems.push(format!("cleanup failed: {error}"));
127    } else if record.cleared_unix_ms.is_none() {
128        problems.push("injected but never cleared".to_owned());
129    }
130    problems
131}
132
133/// The recoveries `fault` is judged on when it landed on a broker with `role`.
134///
135/// They are the recoveries that its plan names. A fault that moves leadership
136/// away from the active controller (`role` is `controller`) also needs a
137/// controller failover, whatever the plan's target selector said. The executor
138/// calls this with the role it recorded, so that it measures the same
139/// recoveries that the judge checks.
140#[must_use]
141pub fn checks_for(fault: &ScheduledFault, role: &str) -> Vec<RecoveryCheck> {
142    let mut checks = fault.checks.clone();
143    if role == "controller"
144        && fault.kind.moves_leadership()
145        && !checks.contains(&RecoveryCheck::ControllerFailover)
146    {
147        checks.push(RecoveryCheck::ControllerFailover);
148    }
149    checks
150}
151
152/// The recoveries `fault` is judged on, given its fault record. A fault with
153/// no record is judged on the recoveries its plan names.
154fn required_checks(fault: &ScheduledFault, record: Option<&FaultRecord>) -> Vec<RecoveryCheck> {
155    checks_for(fault, record.map_or("", |record| record.role.as_str()))
156}
157
158/// Check every scheduled fault's execution and recoveries.
159///
160/// A fault record must match its schedule: the same kind, the role the plan
161/// targets, and a start no earlier than its offset and at most
162/// [`START_SLACK`] after it. A self-check contract's `missed_deadline`
163/// replaces the deadline of its one named check on the first occurrence of
164/// its fault.
165#[must_use]
166pub fn check_recoveries(
167    contract: &Contract,
168    schedule: &[ScheduledFault],
169    evidence: &Evidence,
170) -> (Vec<DeadlineVerdict>, Vec<FaultProblem>) {
171    let mut verdicts = Vec::new();
172    let mut problems = Vec::new();
173    for fault in schedule {
174        let record = evidence.faults.iter().find(|record| record.id == fault.id);
175        match record {
176            None => problems.push(FaultProblem {
177                fault: fault.id.clone(),
178                problem: "scheduled but never injected".to_owned(),
179            }),
180            Some(record) => problems.extend(
181                execution_problems(fault, record, evidence.started_unix_ms)
182                    .into_iter()
183                    .map(|problem| FaultProblem {
184                        fault: fault.id.clone(),
185                        problem,
186                    }),
187            ),
188        }
189        for check in required_checks(fault, record) {
190            let deadline = deadline_for(contract, fault, check);
191            let took = evidence
192                .recoveries
193                .iter()
194                .find(|record| record.fault == fault.id && record.check == check)
195                .and_then(RecoveryRecord::took);
196            let outcome = match took {
197                None => DeadlineOutcome::NotObserved,
198                Some(took) if took > deadline => DeadlineOutcome::Missed,
199                Some(_) => DeadlineOutcome::Met,
200            };
201            verdicts.push(DeadlineVerdict {
202                fault: fault.id.clone(),
203                check,
204                deadline,
205                took,
206                outcome,
207            });
208        }
209    }
210    (verdicts, problems)
211}
212
213/// The deadline `check` is judged against after `fault`.
214#[must_use]
215pub fn deadline_for(contract: &Contract, fault: &ScheduledFault, check: RecoveryCheck) -> Time {
216    match &contract.self_check {
217        Some(self_check)
218            if self_check.missed_deadline.fault == fault.plan
219                && self_check.missed_deadline.check == check
220                && fault.id == format!("{}#0", fault.plan) =>
221        {
222            self_check.missed_deadline.deadline
223        }
224        _ => check.deadline(&contract.deadlines),
225    }
226}
227
228/// Check every path's steady-window latency and throughput against its budget.
229#[must_use]
230pub fn check_budgets(contract: &Contract, evidence: &Evidence) -> Vec<BudgetVerdict> {
231    let mut verdicts = Vec::new();
232    for path in &contract.paths {
233        let Some(measured) = evidence
234            .paths
235            .iter()
236            .find(|measured| measured.path == path.name)
237        else {
238            verdicts.push(BudgetVerdict {
239                path: path.name.clone(),
240                measure: "measurement".to_owned(),
241                budget: "present".to_owned(),
242                measured: "absent".to_owned(),
243                within: false,
244            });
245            continue;
246        };
247        for (direction, budget, summary) in [
248            ("produce", &path.produce_latency, &measured.produce_steady),
249            ("consume", &path.consume_latency, &measured.consume_steady),
250        ] {
251            push_latency(&mut verdicts, &path.name, direction, summary, [
252                ("p50", budget.p50, summary.p50),
253                ("p99", budget.p99, summary.p99),
254                ("p999", budget.p999, summary.p999),
255            ]);
256        }
257        verdicts.push(BudgetVerdict {
258            path: path.name.clone(),
259            measure: "throughput".to_owned(),
260            budget: format!(">= {}", rate(path.throughput_floor)),
261            measured: rate(measured.steady_rate),
262            within: measured.steady_rate >= path.throughput_floor,
263        });
264    }
265    verdicts
266}
267
268fn push_latency(
269    verdicts: &mut Vec<BudgetVerdict>,
270    path: &str,
271    direction: &str,
272    summary: &LatencySummary,
273    percentiles: [(&str, Time, Time); 3],
274) {
275    if summary.count == 0 {
276        verdicts.push(BudgetVerdict {
277            path: path.to_owned(),
278            measure: format!("{direction} latency"),
279            budget: "samples".to_owned(),
280            measured: "none".to_owned(),
281            within: false,
282        });
283        return;
284    }
285    for (name, budget, measured) in percentiles {
286        verdicts.push(BudgetVerdict {
287            path: path.to_owned(),
288            measure: format!("{direction} {name}"),
289            budget: format!("<= {}", human(budget)),
290            measured: human(measured),
291            within: measured <= budget,
292        });
293    }
294}
295
296fn rate(frequency: Frequency) -> String {
297    use krabka_units::fmt::Human as _;
298    frequency.human().to_string()
299}
300
301#[cfg(test)]
302mod tests {
303    use {
304        super::{DeadlineOutcome, check_budgets, check_recoveries, checks_for},
305        crate::soak::{
306            contract::{
307                FaultKind,
308                RecoveryCheck::{
309                    self, CatchUpAfterPause, ControllerFailover, IsrRejoin, LeaderFailover,
310                },
311                Target, fixture,
312            },
313            evidence::{Evidence, FaultRecord, LatencySummary, PathMeasurement, RecoveryRecord},
314            schedule::{ScheduledFault, expand},
315        },
316        assert2::check,
317        krabka_units::prelude::*,
318    };
319
320    /// The load starts here; the fixture schedules `short-pause#0` at 30s
321    /// and `leader-restart#0` at 90s from it.
322    const STARTED_MS: i64 = 1_000_000;
323
324    fn record(id: &str, kind: &str, role: &str, offset_ms: i64) -> FaultRecord {
325        FaultRecord {
326            id: id.to_owned(),
327            kind: kind.to_owned(),
328            target: "broker-1".to_owned(),
329            role: role.to_owned(),
330            node: None,
331            injected_unix_ms: STARTED_MS + offset_ms,
332            cleared_unix_ms: Some(STARTED_MS + offset_ms + 1_000),
333            cleanup_error: None,
334            detail: String::new(),
335        }
336    }
337
338    fn recovery(fault: &str, check: RecoveryCheck, took_ms: Option<i64>) -> RecoveryRecord {
339        RecoveryRecord {
340            fault: fault.to_owned(),
341            check,
342            from_unix_ms: 10_000,
343            recovered_unix_ms: took_ms.map(|took| 10_000 + took),
344        }
345    }
346
347    /// Every fault of the fixture executed and recovered in two seconds.
348    fn recovered() -> Evidence {
349        Evidence {
350            started_unix_ms: STARTED_MS,
351            faults: vec![
352                record("short-pause#0", "pause", "follower", 30_200),
353                record("leader-restart#0", "restart", "leader", 90_500),
354            ],
355            recoveries: vec![
356                recovery(
357                    "short-pause#0",
358                    RecoveryCheck::CatchUpAfterPause,
359                    Some(2_000),
360                ),
361                recovery(
362                    "leader-restart#0",
363                    RecoveryCheck::LeaderFailover,
364                    Some(2_000),
365                ),
366                recovery("leader-restart#0", RecoveryCheck::IsrRejoin, Some(2_000)),
367            ],
368            ..Evidence::default()
369        }
370    }
371
372    #[test]
373    fn the_self_check_override_makes_its_one_deadline_impossible() {
374        let contract = fixture::self_check();
375        let schedule = expand(&contract).unwrap();
376        let (verdicts, problems) = check_recoveries(&contract, &schedule, &recovered());
377        check!(problems.is_empty());
378        let missed: Vec<String> = verdicts
379            .iter()
380            .filter(|verdict| verdict.outcome != DeadlineOutcome::Met)
381            .map(|verdict| format!("{} {}", verdict.fault, verdict.check))
382            .collect();
383        check!(missed == ["leader-restart#0 isr_rejoin"]);
384    }
385
386    #[test]
387    fn a_fault_that_lands_on_the_controller_is_judged_on_controller_failover() {
388        let mut contract = fixture::self_check();
389        contract.self_check = None;
390        let schedule = expand(&contract).unwrap();
391        let mut evidence = recovered();
392        // `leader-restart` targets the leader, which turned out to be the
393        // active controller as well.
394        evidence.faults[1].role = "controller".into();
395        let (verdicts, _) = check_recoveries(&contract, &schedule, &evidence);
396        let unmeasured: Vec<String> = verdicts
397            .iter()
398            .filter(|verdict| verdict.outcome == DeadlineOutcome::NotObserved)
399            .map(|verdict| format!("{} {}", verdict.fault, verdict.check))
400            .collect();
401        check!(unmeasured == ["leader-restart#0 controller_failover"]);
402    }
403
404    type Edit = fn(&mut Evidence);
405
406    #[yare::parameterized(
407        late = { |e: &mut Evidence| e.recoveries[0].recovered_unix_ms = Some(10_000 + 31_000), "short-pause#0 catch_up_after_pause Missed" },
408        never = { |e: &mut Evidence| e.recoveries[1].recovered_unix_ms = None, "leader-restart#0 leader_failover NotObserved" },
409        unmeasured = { |e: &mut Evidence| { e.recoveries.remove(0); }, "short-pause#0 catch_up_after_pause NotObserved" },
410    )]
411    fn a_late_or_unmeasured_recovery_fails(edit: Edit, expected: &str) {
412        let mut contract = fixture::self_check();
413        contract.self_check = None;
414        let schedule = expand(&contract).unwrap();
415        let mut evidence = recovered();
416        edit(&mut evidence);
417        let (verdicts, _) = check_recoveries(&contract, &schedule, &evidence);
418        let failed: Vec<String> = verdicts
419            .iter()
420            .filter(|verdict| verdict.outcome != DeadlineOutcome::Met)
421            .map(|verdict| format!("{} {} {:?}", verdict.fault, verdict.check, verdict.outcome))
422            .collect();
423        check!(failed == [expected]);
424    }
425
426    #[yare::parameterized(
427        never_injected = { |e: &mut Evidence| { e.faults.remove(0); }, "scheduled but never injected" },
428        never_cleared = { |e: &mut Evidence| e.faults[0].cleared_unix_ms = None, "injected but never cleared" },
429        cleanup_failed = { |e: &mut Evidence| e.faults[0].cleanup_error = Some("SIGCONT refused".into()), "cleanup failed: SIGCONT refused" },
430        wrong_kind = { |e: &mut Evidence| e.faults[0].kind = "restart".into(), "recorded as a `restart` fault, but the schedule names `pause`" },
431        wrong_role = { |e: &mut Evidence| e.faults[0].role = "leader".into(), "landed on a leader (broker-1), but the plan targets the follower" },
432        early = { |e: &mut Evidence| e.faults[0].injected_unix_ms = STARTED_MS + 10_000, "injected at 10s into the load, but scheduled at 30s (at most 30s late)" },
433        late = { |e: &mut Evidence| e.faults[0].injected_unix_ms = STARTED_MS + 61_000, "injected at 61s into the load, but scheduled at 30s (at most 30s late)" },
434    )]
435    fn a_fault_that_did_not_run_cleanly_is_a_problem(edit: Edit, expected: &str) {
436        let contract = fixture::self_check();
437        let schedule = expand(&contract).unwrap();
438        let mut evidence = recovered();
439        edit(&mut evidence);
440        let (_, problems) = check_recoveries(&contract, &schedule, &evidence);
441        check!(problems.len() == 1);
442        check!(problems[0].fault == "short-pause#0");
443        check!(problems[0].problem == expected);
444    }
445
446    fn summary(p50: Time, p99: Time, p999: Time) -> LatencySummary {
447        LatencySummary {
448            p50,
449            p99,
450            p999,
451            count: 1_000,
452        }
453    }
454
455    #[yare::parameterized(
456        inside = { millis(400), per_sec(45), &[] },
457        slow_p99 = { millis(600), per_sec(45), &["classic-idempotent produce p99"] },
458        too_few = { millis(400), per_sec(30), &["classic-idempotent throughput"] },
459    )]
460    fn budgets_are_judged_on_the_steady_windows(p99: Time, steady: Frequency, outside: &[&str]) {
461        let contract = fixture::self_check();
462        let measured = |path: &str, p99, steady_rate| PathMeasurement {
463            path: path.to_owned(),
464            produce_steady: summary(millis(10), p99, secs(1)),
465            produce_disturbed: summary(secs(9), secs(9), secs(9)),
466            consume_steady: summary(millis(10), millis(100), secs(1)),
467            consume_disturbed: LatencySummary::default(),
468            steady_rate,
469        };
470        let evidence = Evidence {
471            paths: vec![
472                measured("classic-idempotent", p99, steady),
473                measured("classic-transactional", millis(400), per_sec(45)),
474            ],
475            ..Evidence::default()
476        };
477        let failed: Vec<String> = check_budgets(&contract, &evidence)
478            .iter()
479            .filter(|verdict| !verdict.within)
480            .map(|verdict| format!("{} {}", verdict.path, verdict.measure))
481            .collect();
482        check!(failed == outside);
483    }
484
485    #[test]
486    fn a_path_with_no_measurement_fails_its_budget() {
487        let contract = fixture::self_check();
488        let verdicts = check_budgets(&contract, &Evidence::default());
489        check!(verdicts.iter().all(|verdict| !verdict.within));
490        check!(verdicts.len() == contract.paths.len());
491    }
492
493    #[yare::parameterized(
494        restart_on_the_controller = { FaultKind::Restart { graceful: false }, Target::Leader, "controller", &[LeaderFailover, IsrRejoin, ControllerFailover] },
495        restart_on_a_leader = { FaultKind::Restart { graceful: false }, Target::Leader, "leader", &[LeaderFailover, IsrRejoin] },
496        failover_pause_on_the_controller = { FaultKind::Pause { intends_failover: true }, Target::Any, "controller", &[LeaderFailover, CatchUpAfterPause, ControllerFailover] },
497        short_pause_on_the_controller = { FaultKind::Pause { intends_failover: false }, Target::Any, "controller", &[CatchUpAfterPause] },
498        controller_plan_on_the_controller = { FaultKind::Restart { graceful: true }, Target::Controller, "controller", &[LeaderFailover, IsrRejoin, ControllerFailover] },
499        no_record = { FaultKind::Restart { graceful: true }, Target::Any, "", &[LeaderFailover, IsrRejoin] },
500    )]
501    fn checks_for_adds_controller_failover_only_when_leadership_left_the_controller(
502        kind: FaultKind,
503        target: Target,
504        role: &str,
505        expected: &[RecoveryCheck],
506    ) {
507        let fault = ScheduledFault {
508            id: "f#0".to_owned(),
509            plan: "f".to_owned(),
510            checks: kind.checks(target),
511            kind,
512            target,
513            start: secs(10),
514            duration: secs(5),
515        };
516        check!(checks_for(&fault, role) == expected);
517    }
518}
krabka_bench_driver/soak/deadline.rs

krabka_bench_driver/soak/
deadline.rs
