krabka_verified/producer_snapshot/
replay_start.rs

1#[cfg(creusot)]
2use super::{
3    kafka_reload_keeps, kafka_replay_start, snapshot_last_record_valid, snapshot_transaction_valid,
4};
5use {
6    super::{ProducerReloadRange, ProducerSnapshotEntryFacts, producer_snapshot_reload_keeps},
7    creusot_std::prelude::*,
8};
9
10// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
11#[cfg(creusot)]
12#[cfg_attr(test, mutants::skip)]
13#[logic(open)]
14/// What a snapshot at `snapshot_offset` can truthfully say about one
15/// producer: a real producer identity, a coordinator epoch that is real or
16/// Kafka's `-1`, and last-record and transaction fields that describe only
17/// records before the snapshot's offset.
18pub fn snapshot_entry_valid_model(snapshot_offset: Int, entry: ProducerSnapshotEntryFacts) -> bool {
19    pearlite! {
20        snapshot_offset >= 0
21            && entry.producer_id@ >= 0
22            && entry.producer_epoch@ >= 0
23            && entry.coordinator_epoch@ >= -1
24            && snapshot_last_record_valid(snapshot_offset, entry)
25            && snapshot_transaction_valid(snapshot_offset, entry)
26    }
27}
28
29/// Validate one decoded producer entry against its snapshot's exclusive
30/// offset boundary: the result is `snapshot_entry_valid_model`.
31#[ensures(result == snapshot_entry_valid_model(snapshot_offset@, entry))]
32#[must_use]
33pub fn producer_snapshot_entry_valid(
34    snapshot_offset: i64,
35    entry: ProducerSnapshotEntryFacts,
36) -> bool {
37    let last_record_valid =
38        (entry.last_offset == -1 && entry.last_sequence == -1 && entry.offset_delta == 0)
39            || (entry.last_offset >= 0
40                && entry.last_sequence >= 0
41                && entry.offset_delta >= 0
42                && entry.last_offset >= i64::from(entry.offset_delta)
43                && entry.last_offset < snapshot_offset);
44    let transaction_valid = entry.current_txn_first_offset == -1
45        || (entry.current_txn_first_offset >= 0
46            && entry.current_txn_first_offset < snapshot_offset
47            && entry.current_txn_first_offset <= entry.last_offset);
48    snapshot_offset >= 0
49        && entry.producer_id >= 0
50        && entry.producer_epoch >= 0
51        && entry.coordinator_epoch >= -1
52        && last_record_valid
53        && transaction_valid
54}
55
56/// Select the replay cursor after a reload loaded `snapshot`, or none.
57///
58/// Kafka's `UnifiedLog.rebuildProducerState` replays each local segment from
59/// `max(segment.baseOffset, mapEndOffset, logStartOffset)` (see
60/// `kafka_replay_start`). The range must be nonnegative and ordered, with
61/// the local start at or below the log end, and a loaded snapshot must be one
62/// the reload keeps; anything else is rejected as corrupt.
63///
64/// The cursor can land inside a batch when a trim did. Kafka's
65/// `LogSegment.read(startOffset, ..)` then starts from the batch that holds
66/// the cursor, and replaying that whole batch is the host's job.
67#[ensures((result != None) == (range.log_start@ >= 0
68    && range.log_start@ <= range.log_end@
69    && range.local_start@ >= 0
70    && range.local_start@ <= range.log_end@
71    && match snapshot {
72        Some(offset) => kafka_reload_keeps(offset@, range.log_start@, range.log_end@),
73        None => true,
74    }))]
75#[ensures(forall<start: i64> result == Some(start) ==> start@ == kafka_replay_start(range, snapshot))]
76#[must_use]
77pub fn producer_snapshot_replay_start(
78    range: ProducerReloadRange,
79    snapshot: Option<i64>,
80) -> Option<i64> {
81    if range.log_start < 0
82        || range.log_end < range.log_start
83        || range.local_start < 0
84        || range.log_end < range.local_start
85    {
86        return None;
87    }
88    let map_end = match snapshot {
89        Some(offset) if producer_snapshot_reload_keeps(offset, range) => offset,
90        Some(_) => return None,
91        None => range.log_start,
92    };
93    Some(map_end.max(range.local_start))
94}
krabka_verified/producer_snapshot/replay_start.rs

krabka_verified/producer_snapshot/
replay_start.rs
