krabka_verified/
transaction.rs

1//! Transaction-completion fencing after the `EndTxn` marker fan-out.
2
3use creusot_std::prelude::*;
4
5// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
6#[cfg(creusot)]
7#[cfg_attr(test, mutants::skip)]
8#[logic(open)]
9/// The stable frontier is the minimum pending start, or the end of an idle log.
10pub(crate) fn first_unstable_frontier(starts: Seq<i64>, log_end: Int, lso: Int) -> bool {
11    pearlite! {
12        lso <= log_end
13        && (forall<i: Int> 0 <= i && i < starts.len() ==> starts[i]@ <= log_end)
14        && ((starts.len() == 0 && lso == log_end)
15            || (starts.len() > 0
16                && (exists<i: Int> 0 <= i && i < starts.len() && lso == starts[i]@)
17                && (forall<i: Int> 0 <= i && i < starts.len() ==> lso <= starts[i]@)))
18    }
19}
20
21/// Classification of the actual first control-record key, shared by live
22/// append and recovery. Versions are ignored as in Kafka's marker type parser.
23#[krabka_macros::model_type]
24pub enum LogBatchKind {
25    Data,
26    Abort,
27    Commit,
28    Barrier,
29    OtherControl,
30}
31
32/// Whether one transaction record may install its live producer identities.
33#[krabka_macros::model_type]
34pub enum TransactionPidInstallDecision {
35    RejectWrongPartition,
36    RejectCurrentIdentity,
37    RejectStagedIdentity,
38    RejectCollision,
39    Apply,
40}
41
42/// Whether one transaction marker may append and publish committed offsets.
43#[krabka_macros::model_type]
44pub enum TransactionMarkerMaterializationDecision {
45    RejectMalformed,
46    RejectProducerEpoch,
47    RejectCoordinatorEpoch,
48    Retry,
49    AppendWithoutOffsetPublication,
50    AppendAndPublishOffsets,
51}
52
53/// One transaction marker as it reaches one data partition.
54#[krabka_macros::model_type]
55pub struct TransactionMarkerRequest {
56    pub producer_id: i64,
57    pub producer_epoch: i16,
58    /// The sending coordinator's epoch; `-1` names no coordinator generation.
59    pub coordinator_epoch: i32,
60    /// The marker is a COMMIT rather than an ABORT.
61    pub is_commit: bool,
62    /// The partition is a `__consumer_offsets` partition.
63    pub is_offsets_partition: bool,
64}
65
66/// The partition's latest producer state for the marker's producer ID.
67#[krabka_macros::model_type]
68pub struct TransactionMarkerPartitionState {
69    /// The latest producer epoch, or `-1` when the partition has none.
70    pub producer_epoch: i16,
71    /// The latest coordinator epoch, or `-1` when the partition has none.
72    pub coordinator_epoch: i32,
73    /// The producer has an open transaction on this partition.
74    pub has_pending_transaction: bool,
75}
76
77/// Whether the idle reaper or the completion task may publish one prepared
78/// completion.
79#[krabka_macros::model_type]
80pub enum TransactionReaperCompletionDecision {
81    /// A snapshot names a negative producer ID or epoch.
82    RejectMalformed,
83    /// The live entry no longer holds the prepared producer identity.
84    RejectStaleIdentity,
85    /// The live entry holds the prepared identity but is no longer the exact
86    /// prepared snapshot.
87    RejectChangedPreparedState,
88    /// The live entry already holds the intended completion.
89    AlreadyComplete,
90    /// The live entry is the exact prepared snapshot.
91    Proceed,
92}
93
94/// Whether one transaction generation may persist a partition registration.
95///
96/// The ordering mirrors Kafka's `TransactionCoordinator.handleAddPartitionsToTransaction`:
97/// a pending transition (`pendingTransitionInProgress`) is checked first so it
98/// can complete before any identity check runs, then the producer id, then the
99/// producer epoch, then the `PrepareCommit`/`PrepareAbort` state check. Every
100/// one of those four rejections is retriable on Kafka's wire
101/// (`CONCURRENT_TRANSACTIONS`, `INVALID_PRODUCER_ID_MAPPING`, or
102/// `PRODUCER_FENCED`), which the host maps; this module only orders the facts.
103#[krabka_macros::model_type]
104pub enum TransactionRegistrationDecision {
105    RejectNotCoordinator,
106    RejectUnknownProducer,
107    RejectPendingTransition,
108    RejectProducerId,
109    RejectProducerEpoch,
110    RejectState,
111    PersistRetry,
112    PersistRegistration,
113}
114
115/// Facts used to fence one transaction partition registration.
116#[krabka_macros::model_type]
117pub struct TransactionRegistrationFacts {
118    pub ownership: TransactionRegistrationOwnershipFacts,
119    pub identity: TransactionRegistrationIdentityFacts,
120    pub state: TransactionRegistrationStateFacts,
121}
122
123#[krabka_macros::model_type]
124pub struct TransactionRegistrationOwnershipFacts {
125    pub is_coordinator: bool,
126    pub producer_id_valid: bool,
127    pub entry_exists: bool,
128}
129
130#[krabka_macros::model_type]
131pub struct TransactionRegistrationIdentityFacts {
132    /// Kafka's `txnMetadata.pendingTransitionInProgress`: a staged producer
133    /// identity is waiting on an older transaction to complete first.
134    pub pending_transition: bool,
135    pub matching: TransactionRegistrationIdentityMatchFacts,
136}
137
138#[krabka_macros::model_type]
139pub struct TransactionRegistrationIdentityMatchFacts {
140    pub transactional_id_matches: bool,
141    pub producer_id_matches: bool,
142    pub producer_epoch_matches: bool,
143}
144
145#[krabka_macros::model_type]
146pub struct TransactionRegistrationStateFacts {
147    pub state_allows_registration: bool,
148    /// Kafka's `txnMetadata.state == ONGOING`. The retry optimization below
149    /// applies only in that exact state, never against a stale partition set
150    /// left over from a completed or not-yet-started transaction.
151    pub state_is_ongoing: bool,
152    pub exact_partitions_registered: bool,
153}
154
155/// Whether an `EndTxn` caller may finalize the entry it prepared.
156#[krabka_macros::model_type]
157pub enum TransactionCompletionDecision {
158    Proceed,
159    AlreadyComplete,
160    RejectStaleIdentity,
161    RejectState,
162}
163
164/// Sentinel persisted for a transaction controlled by an external 2PC owner.
165pub const NO_TRANSACTION_TIMEOUT_MS: i32 = i32::MAX;
166
167/// State fact needed by the idle-transaction reaper.
168#[krabka_macros::model_type]
169pub enum IdleTransactionState {
170    Ongoing,
171    Other,
172}
173
174/// The persisted identity and state observed after marker fan-out.
175#[krabka_macros::model_type]
176pub struct TransactionSnapshot {
177    pub pid: i64,
178    pub epoch: i16,
179    pub state: i8,
180}
181
182/// A producer identity captured before marker fan-out.
183#[krabka_macros::model_type]
184pub struct TransactionIdentity {
185    pub pid: i64,
186    pub epoch: i16,
187}
188
189impl TransactionSnapshot {
190    /// The producer identity this snapshot holds.
191    #[ensures(result.pid == self.pid && result.epoch == self.epoch)]
192    #[must_use]
193    pub fn identity(self) -> TransactionIdentity {
194        TransactionIdentity {
195            pid: self.pid,
196            epoch: self.epoch,
197        }
198    }
199}
200
201/// What `InitProducerId` does with the producer identity a caller supplies.
202#[krabka_macros::model_type]
203pub enum InitProducerIdIdentityDecision {
204    /// The caller names no identity. Kafka bumps the epoch and records no
205    /// last epoch (`prepareIncrementProducerEpoch` with an empty expected
206    /// epoch).
207    BumpWithoutIdentity,
208    /// The caller names the entry's live epoch. Kafka bumps the epoch and
209    /// records the epoch it held as the last epoch.
210    Bump,
211    /// The caller names the epoch the entry held before its last bump. Kafka
212    /// treats this as a retry of the call that made that bump: it answers the
213    /// entry's identity and writes nothing.
214    Retry,
215    /// Kafka answers `PRODUCER_FENCED`.
216    Fenced,
217}
218
219/// The epoch at and above which Kafka treats a producer epoch as exhausted
220/// (`TransactionMetadata.isEpochExhausted`).
221///
222/// The coordinator keeps one epoch in hand to fence the producer with, so it
223/// never hands out `i16::MAX`.
224pub const EXHAUSTED_PRODUCER_EPOCH: i16 = i16::MAX - 1;
225
226mod marker;
227#[cfg(creusot)]
228pub use marker::{
229    has_identity, is_identity, marker_completed_retry, marker_generation_current,
230    marker_publishes_offsets, marker_well_formed, snapshot_eq,
231};
232pub use marker::{
233    log_batch_kind, transaction_marker_equal_epoch_fenced,
234    transaction_marker_materialization_decision,
235};
236
237mod reaper;
238pub use reaper::{
239    transaction_partition_registration, transaction_pid_install_decision,
240    transaction_reaper_completion_decision,
241};
242
243mod next_producer_identity;
244pub use next_producer_identity::{
245    aborted_transaction_interval, aborted_transaction_overlaps, first_unstable_offset,
246    next_producer_identity, should_abort_idle_transaction, transaction_marker_closes,
247};
248
249mod abort_rows;
250pub use abort_rows::unique_aborted_transaction_rows;
251
252mod producer_identity;
253pub use producer_identity::{init_producer_id_identity_decision, transaction_completion_decision};
254
255// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
256#[cfg(creusot)]
257#[cfg_attr(test, mutants::skip)]
258#[logic(open)]
259/// Every prefix bounded by the log and live transactions is below this limit.
260pub(crate) fn unstable_fetch_limit_maximal(
261    starts: Seq<i64>,
262    log_end: Int,
263    high_watermark: Int,
264    deliverable: Int,
265    limit: Int,
266) -> bool {
267    pearlite! { forall<v: Int> v <= log_end && v <= high_watermark && v <= deliverable
268    && (forall<i: Int> 0 <= i && i < starts.len() ==> v <= starts[i]@) ==> v <= limit }
269}
270
271// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
272#[cfg(creusot)]
273#[cfg_attr(test, mutants::skip)]
274#[logic(open)]
275/// Every coherent pending transaction bounds both stability and the consumer fetch prefix.
276pub(crate) fn pending_starts_bound_fetch(
277    starts: Seq<i64>,
278    end: Int,
279    stable: Int,
280    limit: Int,
281) -> bool {
282    pearlite! { forall<i: Int> 0 <= i && i < starts.len()
283    ==> starts[i]@ <= end && stable <= starts[i]@ && limit <= starts[i]@ }
284}
285
286// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
287#[cfg(creusot)]
288#[cfg_attr(test, mutants::skip)]
289#[logic(open)]
290/// Every remaining transaction starts inside the prefix preceding the marker.
291pub(crate) fn pending_starts_before_marker(starts: Seq<i64>, base: Int) -> bool {
292    pearlite! { forall<i: Int> 0 <= i && i < starts.len() ==> 0 <= starts[i]@ && starts[i]@ <= base }
293}
294
295#[cfg(test)]
296mod tests;
krabka_verified/transaction.rs

krabka_verified/
transaction.rs
