krabka_verified/transaction/
reaper.rs

1#[cfg(creusot)]
2use super::{is_identity, snapshot_eq};
3use {
4    super::{
5        TransactionCompletionDecision, TransactionPidInstallDecision,
6        TransactionReaperCompletionDecision, TransactionRegistrationDecision,
7        TransactionRegistrationFacts, TransactionSnapshot, transaction_completion_decision,
8    },
9    creusot_std::prelude::*,
10};
11
12/// Recheck a prepared snapshot after the abort- or commit-marker fan-out.
13///
14/// This is [`transaction_completion_decision`] with two additions: every
15/// snapshot must name a nonnegative producer identity, because the live entry
16/// and the prepared one come from the replayed transaction log; and
17/// `exact_prepared_snapshot` must hold for `Proceed`. The host computes it
18/// from equality over the complete persisted transaction entry, including its
19/// staged identity, partition set, timeout, and timestamps, so a registration
20/// or recovery that kept the identity and state still blocks the completion.
21///
22/// `prepared.state` and `completion.state` are the prepare and complete tags
23/// of one Kafka completion, which the host takes from a fixed
24/// `Prepare* -> Complete*` pairing, so they differ.
25#[requires(prepared.state != completion.state)]
26#[ensures((result == TransactionReaperCompletionDecision::RejectMalformed)
27    == !(is_identity(current) && is_identity(prepared) && is_identity(completion)))]
28#[ensures((result == TransactionReaperCompletionDecision::AlreadyComplete)
29    == (is_identity(current)
30        && is_identity(prepared)
31        && is_identity(completion)
32        && snapshot_eq(current, completion)))]
33#[ensures((result == TransactionReaperCompletionDecision::RejectStaleIdentity)
34    == (is_identity(current)
35        && is_identity(prepared)
36        && is_identity(completion)
37        && !snapshot_eq(current, completion)
38        && !(current.pid == prepared.pid && current.epoch == prepared.epoch)))]
39#[ensures((result == TransactionReaperCompletionDecision::RejectChangedPreparedState)
40    == (is_identity(current)
41        && is_identity(prepared)
42        && is_identity(completion)
43        && !snapshot_eq(current, completion)
44        && current.pid == prepared.pid
45        && current.epoch == prepared.epoch
46        && !(current.state == prepared.state && exact_prepared_snapshot)))]
47#[ensures((result == TransactionReaperCompletionDecision::Proceed)
48    == (is_identity(current)
49        && is_identity(prepared)
50        && is_identity(completion)
51        && snapshot_eq(current, prepared)
52        && exact_prepared_snapshot))]
53#[must_use]
54pub fn transaction_reaper_completion_decision(
55    current: TransactionSnapshot,
56    prepared: TransactionSnapshot,
57    completion: TransactionSnapshot,
58    exact_prepared_snapshot: bool,
59) -> TransactionReaperCompletionDecision {
60    if current.pid < 0
61        || current.epoch < 0
62        || prepared.pid < 0
63        || prepared.epoch < 0
64        || completion.pid < 0
65        || completion.epoch < 0
66    {
67        return TransactionReaperCompletionDecision::RejectMalformed;
68    }
69    match transaction_completion_decision(
70        current,
71        prepared.identity(),
72        completion.identity(),
73        prepared.state,
74        completion.state,
75    ) {
76        TransactionCompletionDecision::AlreadyComplete => {
77            TransactionReaperCompletionDecision::AlreadyComplete
78        }
79        TransactionCompletionDecision::RejectStaleIdentity => {
80            TransactionReaperCompletionDecision::RejectStaleIdentity
81        }
82        TransactionCompletionDecision::RejectState => {
83            TransactionReaperCompletionDecision::RejectChangedPreparedState
84        }
85        TransactionCompletionDecision::Proceed => {
86            if exact_prepared_snapshot {
87                TransactionReaperCompletionDecision::Proceed
88            } else {
89                TransactionReaperCompletionDecision::RejectChangedPreparedState
90            }
91        }
92    }
93}
94
95/// Admit only a well-formed, uniquely owned producer-ID pair from the
96/// transaction log partition selected by its transactional ID.
97#[ensures((result == TransactionPidInstallDecision::RejectWrongPartition)
98    == !partition_matches)]
99#[ensures((result == TransactionPidInstallDecision::RejectCurrentIdentity)
100    == (partition_matches && (producer_id@ < 0 || producer_epoch@ < 0)))]
101#[ensures((result == TransactionPidInstallDecision::RejectStagedIdentity)
102    == (partition_matches
103        && producer_id@ >= 0
104        && producer_epoch@ >= 0
105        && !((next_producer_id@ == -1 && next_producer_epoch@ == -1)
106            || (next_producer_id@ >= 0 && next_producer_epoch@ >= 0))))]
107#[ensures((result == TransactionPidInstallDecision::RejectCollision)
108    == (pid_install_input_valid(partition_matches, producer_id@, producer_epoch@, next_producer_id@, next_producer_epoch@)
109        && (!current_owner_matches
110            || (next_producer_id@ >= 0 && !next_owner_matches))))]
111#[ensures((result == TransactionPidInstallDecision::Apply)
112    == (pid_install_input_valid(partition_matches, producer_id@, producer_epoch@, next_producer_id@, next_producer_epoch@)
113        && current_owner_matches
114        && (next_producer_id@ < 0 || next_owner_matches)))]
115#[must_use]
116pub fn transaction_pid_install_decision(
117    partition_matches: bool,
118    producer_id: i64,
119    producer_epoch: i16,
120    next_producer_id: i64,
121    next_producer_epoch: i16,
122    current_owner_matches: bool,
123    next_owner_matches: bool,
124) -> TransactionPidInstallDecision {
125    if !partition_matches {
126        TransactionPidInstallDecision::RejectWrongPartition
127    } else if producer_id < 0 || producer_epoch < 0 {
128        TransactionPidInstallDecision::RejectCurrentIdentity
129    } else if !((next_producer_id == -1 && next_producer_epoch == -1)
130        || (next_producer_id >= 0 && next_producer_epoch >= 0))
131    {
132        TransactionPidInstallDecision::RejectStagedIdentity
133    } else if !current_owner_matches || (next_producer_id >= 0 && !next_owner_matches) {
134        TransactionPidInstallDecision::RejectCollision
135    } else {
136        TransactionPidInstallDecision::Apply
137    }
138}
139
140// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
141#[cfg(creusot)]
142#[cfg_attr(test, mutants::skip)]
143#[logic(open)]
144fn registration_generation_ready(facts: TransactionRegistrationFacts) -> bool {
145    pearlite! { facts.ownership.is_coordinator
146    && facts.ownership.producer_id_valid
147    && facts.ownership.entry_exists
148    && facts.identity.matching.transactional_id_matches
149    && !facts.identity.pending_transition }
150}
151
152/// Fence a partition registration against coordinator ownership and one exact
153/// transactional-id, producer-id, and producer-epoch generation.
154#[ensures((result == TransactionRegistrationDecision::RejectNotCoordinator)
155    == !facts.ownership.is_coordinator)]
156#[ensures((result == TransactionRegistrationDecision::RejectUnknownProducer)
157    == (facts.ownership.is_coordinator
158        && (!facts.ownership.producer_id_valid
159            || !facts.ownership.entry_exists
160            || !facts.identity.matching.transactional_id_matches)))]
161#[ensures((result == TransactionRegistrationDecision::RejectPendingTransition)
162    == (facts.ownership.is_coordinator
163        && facts.ownership.producer_id_valid
164        && facts.ownership.entry_exists
165        && facts.identity.matching.transactional_id_matches
166        && facts.identity.pending_transition))]
167#[ensures((result == TransactionRegistrationDecision::RejectProducerId)
168    == (registration_generation_ready(facts)
169        && !facts.identity.matching.producer_id_matches))]
170#[ensures((result == TransactionRegistrationDecision::RejectProducerEpoch)
171    == (registration_generation_ready(facts)
172        && facts.identity.matching.producer_id_matches
173        && !facts.identity.matching.producer_epoch_matches))]
174#[ensures((result == TransactionRegistrationDecision::RejectState)
175    == (registration_generation_ready(facts)
176        && facts.identity.matching.producer_id_matches
177        && facts.identity.matching.producer_epoch_matches
178        && !facts.state.state_allows_registration))]
179#[ensures((result == TransactionRegistrationDecision::PersistRetry)
180    == (registration_generation_ready(facts)
181        && facts.identity.matching.producer_id_matches
182        && facts.identity.matching.producer_epoch_matches
183        && facts.state.state_allows_registration
184        && facts.state.state_is_ongoing
185        && facts.state.exact_partitions_registered))]
186#[ensures((result == TransactionRegistrationDecision::PersistRegistration)
187    == (registration_generation_ready(facts)
188        && facts.identity.matching.producer_id_matches
189        && facts.identity.matching.producer_epoch_matches
190        && facts.state.state_allows_registration
191        && !(facts.state.state_is_ongoing && facts.state.exact_partitions_registered)))]
192#[must_use]
193pub fn transaction_partition_registration(
194    facts: TransactionRegistrationFacts,
195) -> TransactionRegistrationDecision {
196    if !facts.ownership.is_coordinator {
197        TransactionRegistrationDecision::RejectNotCoordinator
198    } else if !facts.ownership.producer_id_valid
199        || !facts.ownership.entry_exists
200        || !facts.identity.matching.transactional_id_matches
201    {
202        TransactionRegistrationDecision::RejectUnknownProducer
203    } else if facts.identity.pending_transition {
204        TransactionRegistrationDecision::RejectPendingTransition
205    } else if !facts.identity.matching.producer_id_matches {
206        TransactionRegistrationDecision::RejectProducerId
207    } else if !facts.identity.matching.producer_epoch_matches {
208        TransactionRegistrationDecision::RejectProducerEpoch
209    } else if !facts.state.state_allows_registration {
210        TransactionRegistrationDecision::RejectState
211    } else if facts.state.state_is_ongoing && facts.state.exact_partitions_registered {
212        TransactionRegistrationDecision::PersistRetry
213    } else {
214        TransactionRegistrationDecision::PersistRegistration
215    }
216}
217
218// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
219#[cfg(creusot)]
220#[cfg_attr(test, mutants::skip)]
221#[logic(open)]
222/// The partition matches and both producer-ID pairs are well formed before ownership checks.
223fn pid_install_input_valid(
224    partition_matches: bool,
225    producer: Int,
226    epoch: Int,
227    next_producer: Int,
228    next_epoch: Int,
229) -> bool {
230    pearlite! { partition_matches && producer >= 0 && epoch >= 0
231    && ((next_producer == -1 && next_epoch == -1) || (next_producer >= 0 && next_epoch >= 0)) }
232}
krabka_verified/transaction/reaper.rs

krabka_verified/transaction/
reaper.rs
