krabka_verified/transaction/
marker.rs

1#[cfg(creusot)]
2use super::{TransactionIdentity, TransactionSnapshot};
3use {
4    super::{
5        LogBatchKind, TransactionMarkerMaterializationDecision, TransactionMarkerPartitionState,
6        TransactionMarkerRequest,
7    },
8    creusot_std::prelude::*,
9};
10
11/// Transaction version 2 requires an epoch bump for a pending transaction's
12/// marker, except at the final marker epoch. Completed retries remain allowed.
13#[ensures(result == (transaction_version@ >= 2
14    && producer_epoch == current_producer_epoch
15    && has_pending_transaction && producer_epoch@ != i16::MAX@))]
16#[must_use]
17pub fn transaction_marker_equal_epoch_fenced(
18    transaction_version: i16,
19    producer_epoch: i16,
20    current_producer_epoch: i16,
21    has_pending_transaction: bool,
22) -> bool {
23    transaction_version >= 2
24        && producer_epoch == current_producer_epoch
25        && has_pending_transaction
26        && producer_epoch != i16::MAX
27}
28
29#[ensures((result == LogBatchKind::Data) == !is_control)]
30#[ensures((result == LogBatchKind::Abort) ==
31    (is_control && [email protected]() >= 4 && key@[2]@ == 0 && key@[3]@ == 0))]
32#[ensures((result == LogBatchKind::Commit) ==
33    (is_control && [email protected]() >= 4 && key@[2]@ == 0 && key@[3]@ == 1))]
34#[ensures((result == LogBatchKind::Barrier) ==
35    (is_control && [email protected]() >= 4 && key@[2]@ == 3 && key@[3]@ == 232))]
36#[ensures((result == LogBatchKind::OtherControl) == (is_control
37    && !([email protected]() >= 4 && ((key@[2]@ == 0 && (key@[3]@ == 0 || key@[3]@ == 1))
38        || (key@[2]@ == 3 && key@[3]@ == 232)))))]
39#[must_use]
40pub fn log_batch_kind(is_control: bool, key: &[u8]) -> LogBatchKind {
41    if !is_control {
42        return LogBatchKind::Data;
43    }
44    if key.len() < 4 {
45        return LogBatchKind::OtherControl;
46    }
47    match (key.get(2), key.get(3)) {
48        (Some(0), Some(0)) => LogBatchKind::Abort,
49        (Some(0), Some(1)) => LogBatchKind::Commit,
50        (Some(3), Some(232)) => LogBatchKind::Barrier,
51        _ => LogBatchKind::OtherControl,
52    }
53}
54
55// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
56#[cfg(creusot)]
57#[cfg_attr(test, mutants::skip)]
58#[logic(open)]
59/// The marker names a nonnegative producer identity, every epoch is at least
60/// the `-1` sentinel, and a pending transaction has a real producer epoch.
61pub fn marker_well_formed(
62    request: TransactionMarkerRequest,
63    current: TransactionMarkerPartitionState,
64) -> bool {
65    pearlite! {
66        request.producer_id@ >= 0
67            && request.producer_epoch@ >= 0
68            && request.coordinator_epoch@ >= -1
69            && current.producer_epoch@ >= -1
70            && current.coordinator_epoch@ >= -1
71            && (!current.has_pending_transaction || current.producer_epoch@ >= 0)
72    }
73}
74
75// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
76#[cfg(creusot)]
77#[cfg_attr(test, mutants::skip)]
78#[logic(open)]
79/// Neither the producer nor the coordinator generation of the marker is
80/// older than the partition's (`ProducerAppendInfo.checkProducerEpoch` and
81/// `appendEndTxnMarker`'s coordinator-epoch check).
82pub fn marker_generation_current(
83    request: TransactionMarkerRequest,
84    current: TransactionMarkerPartitionState,
85) -> bool {
86    pearlite! {
87        request.producer_epoch@ >= current.producer_epoch@
88            && request.coordinator_epoch@ >= current.coordinator_epoch@
89    }
90}
91
92// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
93#[cfg(creusot)]
94#[cfg_attr(test, mutants::skip)]
95#[logic(open)]
96/// The marker repeats the exact generation that already closed the
97/// producer's transaction on this partition.
98pub fn marker_completed_retry(
99    request: TransactionMarkerRequest,
100    current: TransactionMarkerPartitionState,
101) -> bool {
102    pearlite! {
103        !current.has_pending_transaction
104            && request.producer_epoch@ == current.producer_epoch@
105            && request.coordinator_epoch@ == current.coordinator_epoch@
106    }
107}
108
109// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
110#[cfg(creusot)]
111#[cfg_attr(test, mutants::skip)]
112#[logic(open)]
113/// Only a COMMIT that closes a pending transaction on `__consumer_offsets`
114/// publishes the transaction's offset commits.
115pub fn marker_publishes_offsets(
116    request: TransactionMarkerRequest,
117    current: TransactionMarkerPartitionState,
118) -> bool {
119    pearlite! {
120        current.has_pending_transaction && request.is_commit && request.is_offsets_partition
121    }
122}
123
124/// Fence a transaction marker against the partition's latest producer and
125/// coordinator generations, suppress an exact completed retry, and publish
126/// offsets only for a pending commit on `__consumer_offsets`.
127#[ensures((result == TransactionMarkerMaterializationDecision::RejectMalformed)
128    == !marker_well_formed(request, current))]
129#[ensures((result == TransactionMarkerMaterializationDecision::RejectProducerEpoch)
130    == (marker_well_formed(request, current)
131        && request.producer_epoch@ < current.producer_epoch@))]
132#[ensures((result == TransactionMarkerMaterializationDecision::RejectCoordinatorEpoch)
133    == (marker_well_formed(request, current)
134        && request.producer_epoch@ >= current.producer_epoch@
135        && request.coordinator_epoch@ < current.coordinator_epoch@))]
136#[ensures((result == TransactionMarkerMaterializationDecision::Retry)
137    == (marker_well_formed(request, current) && marker_completed_retry(request, current)))]
138#[ensures((result == TransactionMarkerMaterializationDecision::AppendAndPublishOffsets)
139    == (marker_well_formed(request, current)
140        && marker_generation_current(request, current)
141        && marker_publishes_offsets(request, current)))]
142#[ensures((result == TransactionMarkerMaterializationDecision::AppendWithoutOffsetPublication)
143    == (marker_well_formed(request, current)
144        && marker_generation_current(request, current)
145        && !marker_completed_retry(request, current)
146        && !marker_publishes_offsets(request, current)))]
147#[must_use]
148pub fn transaction_marker_materialization_decision(
149    request: TransactionMarkerRequest,
150    current: TransactionMarkerPartitionState,
151) -> TransactionMarkerMaterializationDecision {
152    if request.producer_id < 0
153        || request.producer_epoch < 0
154        || request.coordinator_epoch < -1
155        || current.producer_epoch < -1
156        || current.coordinator_epoch < -1
157        || (current.has_pending_transaction && current.producer_epoch == -1)
158    {
159        TransactionMarkerMaterializationDecision::RejectMalformed
160    } else if request.producer_epoch < current.producer_epoch {
161        TransactionMarkerMaterializationDecision::RejectProducerEpoch
162    } else if request.coordinator_epoch < current.coordinator_epoch {
163        TransactionMarkerMaterializationDecision::RejectCoordinatorEpoch
164    } else if !current.has_pending_transaction
165        && request.producer_epoch == current.producer_epoch
166        && request.coordinator_epoch == current.coordinator_epoch
167    {
168        TransactionMarkerMaterializationDecision::Retry
169    } else if current.has_pending_transaction && request.is_commit && request.is_offsets_partition {
170        TransactionMarkerMaterializationDecision::AppendAndPublishOffsets
171    } else {
172        TransactionMarkerMaterializationDecision::AppendWithoutOffsetPublication
173    }
174}
175
176// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
177#[cfg(creusot)]
178#[cfg_attr(test, mutants::skip)]
179#[logic(open)]
180/// `snapshot` names a producer identity: its PID and epoch are nonnegative.
181pub fn is_identity(snapshot: TransactionSnapshot) -> bool {
182    pearlite! { snapshot.pid@ >= 0 && snapshot.epoch@ >= 0 }
183}
184
185// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
186#[cfg(creusot)]
187#[cfg_attr(test, mutants::skip)]
188#[logic(open)]
189/// `snapshot` holds the producer identity `identity`.
190pub fn has_identity(snapshot: TransactionSnapshot, identity: TransactionIdentity) -> bool {
191    pearlite! { snapshot.pid == identity.pid && snapshot.epoch == identity.epoch }
192}
193
194// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
195#[cfg(creusot)]
196#[cfg_attr(test, mutants::skip)]
197#[logic(open)]
198/// Two snapshots agree on producer identity and state.
199pub fn snapshot_eq(left: TransactionSnapshot, right: TransactionSnapshot) -> bool {
200    pearlite! {
201        left.pid == right.pid && left.epoch == right.epoch && left.state == right.state
202    }
203}
krabka_verified/transaction/marker.rs

krabka_verified/transaction/
marker.rs
