krabka_verified/
producer_snapshot.rs

1//! Producer-snapshot selection, validation, replay, and deletion decisions.
2//!
3//! The rules are Kafka's, from three places:
4//!
5//! - `LogLoader.load` picks the log start the reload runs against, and first
6//!   calls `ProducerStateManager.removeStraySnapshots` with the base offset
7//!   of every local segment.
8//! - `ProducerStateManager.truncateAndReload(logStartOffset, logEndOffset, _)`
9//!   deletes every snapshot outside `(logStartOffset, logEndOffset]`, then
10//!   `loadFromSnapshot` loads the newest snapshot that is left.
11//! - `UnifiedLog.rebuildProducerState` replays each local segment from
12//!   `max(segment.baseOffset, mapEndOffset, logStartOffset)`.
13
14#[cfg(creusot)]
15use crate::producer::ProducerDecision;
16use creusot_std::prelude::*;
17
18// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
19#[cfg(creusot)]
20#[cfg_attr(test, mutants::skip)]
21#[logic(open)]
22/// A physically retained data row is valid and belongs to the requested producer.
23pub(crate) fn retained_producer_row(end: Int, row: ProducerSnapshotEntryFacts, pid: i64) -> bool {
24    pearlite! {
25        snapshot_entry_valid_model(end, row) && row.last_offset@ >= 0
26            && row.producer_id == pid
27    }
28}
29
30// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
31#[cfg(creusot)]
32#[cfg_attr(test, mutants::skip)]
33#[logic(open)]
34/// A retained snapshot row matches both endpoints of the requested sequence.
35pub fn snapshot_sequence_matches(row: ProducerSnapshotEntryFacts, base: Int, delta: Int) -> bool {
36    pearlite! {
37        base == crate::producer::sequence_modulo_2_31(row.last_sequence@ - row.offset_delta@)
38            && row.last_sequence@ == crate::producer::sequence_modulo_2_31(base + delta)
39    }
40}
41
42// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
43#[cfg(creusot)]
44#[cfg_attr(test, mutants::skip)]
45#[logic(open)]
46/// The original retained batch span and its exclusive acknowledgement frontier.
47pub(crate) fn recovered_batch_coordinates(
48    row: ProducerSnapshotEntryFacts,
49    base: Int,
50    frontier: Int,
51    end: Int,
52) -> bool {
53    pearlite! { base == row.last_offset@ - row.offset_delta@
54    && frontier == row.last_offset@ + 1
55    && 0 <= base && base < frontier && frontier <= end }
56}
57
58// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
59#[cfg(creusot)]
60#[cfg_attr(test, mutants::skip)]
61#[logic(open)]
62/// Classification after no retained batch matches the requested sequence.
63pub(crate) fn nonduplicate_snapshot_decision(
64    incoming_epoch: Int,
65    epoch: Int,
66    sequence: Int,
67    last_sequence: Int,
68) -> ProducerDecision {
69    pearlite! { if incoming_epoch < epoch { ProducerDecision::Fenced }
70    else if incoming_epoch > epoch {
71        if sequence == 0 { ProducerDecision::Append } else { ProducerDecision::OutOfOrder }
72    } else if sequence == crate::producer::sequence_modulo_2_31(last_sequence + 1) { ProducerDecision::Append }
73    else { ProducerDecision::OutOfOrder } }
74}
75
76/// The offsets one producer-state reload runs against.
77///
78/// These are three `i64` values with three different meanings, so they
79/// travel as one struct and a transposed call site does not compile.
80#[krabka_macros::model_type]
81pub struct ProducerReloadRange {
82    /// The `logStartOffset` Kafka hands `truncateAndReload`, as
83    /// [`producer_snapshot_reload_log_start`] picks it.
84    pub log_start: i64,
85    /// The first offset a local segment can serve: the greater of the
86    /// oldest local segment's base and the log start.
87    pub local_start: i64,
88    /// The log end offset the state is rebuilt up to.
89    pub log_end: i64,
90}
91
92/// One decoded producer-state snapshot entry, field for field as Kafka's
93/// `ProducerStateManager` snapshot schema (version 1) lays it out, minus the
94/// timestamp, which no validity rule reads.
95///
96/// Seven integers of four widths with seven meanings travel as one struct, so
97/// a transposed call site does not compile.
98#[krabka_macros::model_type(proof(std::clone::Clone, Copy))]
99pub struct ProducerSnapshotEntryFacts {
100    /// `ProducerId`.
101    pub producer_id: i64,
102    /// `ProducerEpoch`.
103    pub producer_epoch: i16,
104    /// `LastSequence`, or `-1` for a producer whose only record is a marker.
105    pub last_sequence: i32,
106    /// `LastOffset`, or `-1` with the same meaning.
107    pub last_offset: i64,
108    /// `OffsetDelta`: the last batch's offset count less one.
109    pub offset_delta: i32,
110    /// `CoordinatorEpoch`, or `-1` before any marker.
111    pub coordinator_epoch: i32,
112    /// `CurrentTxnFirstOffset`, or `-1` with no open transaction.
113    pub current_txn_first_offset: i64,
114}
115
116mod snapshot_transaction_valid;
117#[cfg(test)]
118pub(crate) use snapshot_transaction_valid::take_classifier_visits;
119#[cfg(creusot)]
120pub use snapshot_transaction_valid::{
121    is_segment_base, kafka_reload_keeps, kafka_replay_start, kafka_stray_removed,
122    snapshot_last_record_valid, snapshot_transaction_valid,
123};
124pub use snapshot_transaction_valid::{
125    producer_snapshot_latest_index, producer_snapshot_reload_keeps,
126    producer_snapshot_reload_log_start, producer_snapshot_stray, producer_snapshot_strays,
127};
128
129mod replay_start;
130#[cfg(creusot)]
131pub use replay_start::snapshot_entry_valid_model;
132pub use replay_start::{producer_snapshot_entry_valid, producer_snapshot_replay_start};
133
134// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
135#[cfg(creusot)]
136#[cfg_attr(test, mutants::skip)]
137#[logic(open)]
138/// Producer data rows preserve their strictly increasing physical completion offsets.
139pub(crate) fn producer_offsets_ordered(rows: Seq<ProducerSnapshotEntryFacts>) -> bool {
140    pearlite! { forall<i: Int, j: Int> 0 <= i && i < j && j < rows.len() ==> rows[i].last_offset@ < rows[j].last_offset@ }
141}
142
143#[cfg(test)]
144mod tests;
krabka_verified/producer_snapshot.rs

krabka_verified/
producer_snapshot.rs
