krabka_bench_driver/soak/
synthetic.rs

1//! Synthetic evidence: the run a contract describes, as if every record
2//! arrived and every recovery took half its deadline.
3//!
4//! This is the hermetic half of the self-check. It needs no cluster, so it
5//! runs on every pull request, and it proves two things the live self-check
6//! cannot prove alone: that every checked-in contract is internally consistent
7//! (a perfect run of it passes), and that the judge's verdict depends on the
8//! evidence rather than on the contract alone (the same run with one record
9//! removed fails, naming that record).
10
11use {
12    super::{
13        contract::{Contract, FaultKind, Isolation, LatencyBudget, Semantics, Target},
14        evidence::{
15            CommitCheck, Evidence, FINAL_COMMIT_CHECK, FaultRecord, InjectedLoss, LatencySummary,
16            PathMeasurement, ProducerLedger, ReaderLedger, RecoveryRecord, TxnLedger, TxnOutcome,
17        },
18        ledger::SeqSet,
19        lifecycle::{Checkpoint, CheckpointRead, Side, is_transition, rollback},
20        recovery::{PathRecovery, RecoveryEvidence},
21        schedule::{ScheduledFault, expand},
22    },
23    krabka_units::prelude::*,
24    num_traits::ToPrimitive as _,
25    std::collections::BTreeMap,
26};
27
28/// Records each synthetic producer acknowledges.
29const RECORDS: u64 = 200;
30
31/// The wall-clock instant a synthetic run starts at: 2026-10-15T00:00:00Z.
32const EPOCH_MS: i64 = 1_792_022_400_000;
33
34/// A run of `contract` in which nothing went wrong.
35#[must_use]
36pub fn clean_run(contract: &Contract) -> Evidence {
37    let mut evidence = Evidence {
38        contract: contract.name.clone(),
39        contract_digest: contract.digest(),
40        seed: contract.seed,
41        broker_image: "synthetic".to_owned(),
42        started_unix_ms: EPOCH_MS,
43        ended_unix_ms: EPOCH_MS + contract.duration.millis_i64(),
44        ..Evidence::default()
45    };
46    let schedule = expand(contract).unwrap_or_default();
47    for path in &contract.paths {
48        let producer = format!("{}-0", path.name);
49        let mut ledger = ProducerLedger {
50            path: path.name.clone(),
51            producer: producer.clone(),
52            semantics: path.semantics,
53            attempted: RECORDS,
54            acked: SeqSet::range(0, RECORDS),
55            failed: SeqSet::new(),
56            retried: SeqSet::new(),
57            transactions: Vec::new(),
58        };
59        let mut committed = SeqSet::range(0, RECORDS);
60        if path.semantics == Semantics::Transactional {
61            committed = SeqSet::new();
62            let size = u64::from(contract.clients.transactions.records_per_transaction);
63            let abort_every = u64::from(contract.clients.transactions.abort_every);
64            for (id, start) in (0..RECORDS)
65                .step_by(usize::try_from(size).unwrap_or(1))
66                .enumerate()
67            {
68                let id = u64::try_from(id).unwrap_or(u64::MAX);
69                let end = (start + size).min(RECORDS);
70                let seqs = SeqSet::range(start, end);
71                let outcome = if (id + 1) % abort_every == 0 {
72                    TxnOutcome::Aborted
73                } else {
74                    committed.insert_range(start, end);
75                    TxnOutcome::Committed
76                };
77                ledger.transactions.push(TxnLedger { id, seqs, outcome });
78            }
79        }
80        for &isolation in &path.consumers {
81            let seen = match (path.semantics, isolation) {
82                (Semantics::Transactional, Isolation::ReadCommitted) => committed.clone(),
83                _ => SeqSet::range(0, RECORDS),
84            };
85            let mut latest_by_key = BTreeMap::new();
86            if let Some(space) = path.key_space {
87                for seq in 0..RECORDS {
88                    latest_by_key.insert(seq % space, seq);
89                }
90            }
91            evidence.readers.push(ReaderLedger {
92                path: path.name.clone(),
93                isolation,
94                producer: producer.clone(),
95                seen,
96                duplicates: Vec::new(),
97                latest_by_key,
98            });
99        }
100        if path.commit_offsets {
101            let checkpoints = schedule
102                .iter()
103                .map(|fault| fault.id.clone())
104                .chain([FINAL_COMMIT_CHECK.to_owned()]);
105            for fault in checkpoints {
106                evidence.commits.push(CommitCheck {
107                    group: format!("{}-committer", path.name),
108                    topic: path.name.clone(),
109                    partition: 0,
110                    fault,
111                    acknowledged: 10,
112                    fetched: Some(10),
113                    log_end: 20,
114                });
115            }
116        }
117        evidence.producers.push(ledger);
118        let half = |budget: &LatencyBudget| LatencySummary {
119            p50: budget.p50 * 0.5,
120            p99: budget.p99 * 0.5,
121            p999: budget.p999 * 0.5,
122            count: RECORDS,
123        };
124        evidence.paths.push(PathMeasurement {
125            path: path.name.clone(),
126            produce_steady: half(&path.produce_latency),
127            produce_disturbed: half(&path.produce_latency),
128            consume_steady: half(&path.consume_latency),
129            consume_disturbed: half(&path.consume_latency),
130            steady_rate: path.rate,
131        });
132    }
133    for fault in &schedule {
134        let injected = EPOCH_MS + fault.start.millis_i64();
135        let cleared = EPOCH_MS + fault.end().millis_i64();
136        evidence.faults.push(FaultRecord {
137            id: fault.id.clone(),
138            kind: fault.kind.name().to_owned(),
139            target: "synthetic-broker".to_owned(),
140            role: match fault.target {
141                Target::Controller => "controller",
142                Target::Leader => "leader",
143                Target::Follower => "follower",
144                Target::Any | Target::Broker(_) => "broker",
145            }
146            .to_owned(),
147            node: synthetic_node(fault.target),
148            injected_unix_ms: injected,
149            cleared_unix_ms: Some(cleared),
150            cleanup_error: None,
151            detail: String::new(),
152        });
153        for &check in &fault.checks {
154            let deadline = check.deadline(&contract.deadlines);
155            evidence.recoveries.push(RecoveryRecord {
156                fault: fault.id.clone(),
157                check,
158                from_unix_ms: cleared,
159                recovered_unix_ms: Some(cleared + (deadline * 0.5).millis_i64()),
160            });
161        }
162    }
163    if let Some(objectives) = &contract.disaster_recovery {
164        // The capture starts `post_capture_load` before the end of the load,
165        // the source is lost at its end, and the restored cluster serves half
166        // the recovery time objective later. Every record comes back.
167        let disaster = evidence.ended_unix_ms;
168        let held = objectives.post_capture_load.secs_f64();
169        evidence.recovery = Some(RecoveryEvidence {
170            capture_started_unix_ms: disaster - objectives.post_capture_load.millis_i64(),
171            disaster_unix_ms: disaster,
172            serving_unix_ms: Some(disaster + (objectives.rto * 0.5).millis_i64()),
173            paths: evidence
174                .producers
175                .iter()
176                .map(|producer| PathRecovery {
177                    path: producer.path.clone(),
178                    producer: producer.producer.clone(),
179                    // The load kept its paced rate after the capture, and every
180                    // record of it came back.
181                    acknowledged: producer.acked.len()
182                        + after_boundary(contract, &producer.path, held),
183                    before_boundary: producer.acked.len(),
184                    lost_beyond_boundary: SeqSet::new(),
185                    oldest_loss_unix_ms: None,
186                })
187                .collect(),
188        });
189    }
190    lifecycle_run(contract, &schedule, &mut evidence);
191    evidence
192}
193
194/// The node id a synthetic run gives broker index `i`: `i + 1`, as the
195/// executor's fixture numbers its nodes.
196fn synthetic_node(target: Target) -> Option<i32> {
197    match target {
198        Target::Broker(index) => i32::try_from(index).ok().map(|index| index + 1),
199        Target::Controller | Target::Leader | Target::Follower | Target::Any => None,
200    }
201}
202
203/// The releases, the source's advertised ranges and one clean checkpoint per
204/// transition of a lifecycle contract: every broker starts on the source, a
205/// roll moves its broker to that side's image, nothing is finalized, and every
206/// acknowledged record is read back.
207fn lifecycle_run(contract: &Contract, schedule: &[ScheduledFault], evidence: &mut Evidence) {
208    let Some(lifecycle) = &contract.lifecycle else {
209        return;
210    };
211    let image = |side: Side| format!("synthetic-{side}");
212    for side in [Side::Source, Side::Target] {
213        evidence.releases.insert(side.to_string(), image(side));
214    }
215    evidence
216        .source_supports
217        .clone_from(&lifecycle.source.supports);
218    let brokers = i32::try_from(contract.cluster.brokers).unwrap_or(i32::MAX);
219    let mut images: BTreeMap<i32, String> = (1..=brokers)
220        .map(|node| (node, image(Side::Source)))
221        .collect();
222    for fault in schedule.iter().filter(|fault| is_transition(&fault.kind)) {
223        if let (FaultKind::Roll { to }, Some(node)) = (&fault.kind, synthetic_node(fault.target)) {
224            images.insert(node, image(*to));
225        }
226        let finalized = BTreeMap::new();
227        evidence.checkpoints.push(Checkpoint {
228            fault: fault.id.clone(),
229            unix_ms: EPOCH_MS + fault.end().millis_i64(),
230            images: images.clone(),
231            rollback: rollback(&lifecycle.source, &finalized),
232            finalized,
233            records: contract
234                .paths
235                .iter()
236                .map(|path| CheckpointRead {
237                    path: path.name.clone(),
238                    semantics: path.semantics,
239                    key_space: path.key_space,
240                    acknowledged: SeqSet::range(0, RECORDS),
241                    reader: ReaderLedger {
242                        path: path.name.clone(),
243                        isolation: Isolation::ReadUncommitted,
244                        producer: format!("{}-0", path.name),
245                        seen: SeqSet::range(0, RECORDS),
246                        duplicates: Vec::new(),
247                        latest_by_key: path.key_space.map_or_else(BTreeMap::new, |space| {
248                            (0..RECORDS).map(|seq| (seq % space, seq)).collect()
249                        }),
250                    },
251                })
252                .collect(),
253            config_mismatches: Vec::new(),
254        });
255    }
256}
257
258/// Records `path` acknowledges at its paced rate over `seconds`.
259fn after_boundary(contract: &Contract, path: &str, seconds: f64) -> u64 {
260    let rate = contract
261        .path(path)
262        .map_or(0.0, |path| path.rate.per_sec_f64());
263    (rate * seconds).ceil().to_u64().unwrap_or(0)
264}
265
266/// Remove the first acknowledged record of the self-check's path from every
267/// reader, and record that it was removed: the synthetic stand-in for the
268/// `DeleteRecords` the live self-check issues.
269pub fn inject_missing_record(contract: &Contract, evidence: &mut Evidence) {
270    let Some(self_check) = &contract.self_check else {
271        return;
272    };
273    let Some(producer) = evidence
274        .producers
275        .iter()
276        .find(|producer| producer.path == self_check.missing_record_path)
277    else {
278        return;
279    };
280    let Some(seq) = producer.acked.iter().next() else {
281        return;
282    };
283    let producer = producer.producer.clone();
284    for reader in &mut evidence.readers {
285        if reader.producer == producer {
286            reader.seen.remove(seq);
287        }
288    }
289    evidence.injected.push(InjectedLoss {
290        path: self_check.missing_record_path.clone(),
291        producer,
292        seq,
293        partition: 0,
294        offset: 0,
295    });
296}
krabka_bench_driver/soak/synthetic.rs

krabka_bench_driver/soak/
synthetic.rs
