krabka_verified/transaction/
producer_identity.rs

1#[cfg(creusot)]
2use super::has_identity;
3use {
4    super::{
5        EXHAUSTED_PRODUCER_EPOCH, InitProducerIdIdentityDecision, TransactionCompletionDecision,
6        TransactionIdentity, TransactionSnapshot,
7    },
8    creusot_std::prelude::*,
9};
10
11/// What an `InitProducerId` caller's producer identity may do to the entry it
12/// names (KIP-360).
13///
14/// A request producer id of `-1` supplies no identity, which every
15/// `InitProducerId` below v3 and every first initialisation does.
16///
17/// Kafka admits a supplied identity in `TransactionCoordinator`'s
18/// `isValidProducerId`: the identity names the entry's producer id, whatever
19/// its epoch, or it names the producer id from before the last rotation
20/// together with an exhausted epoch. The epoch then decides the outcome in
21/// `TransactionMetadata.prepareIncrementProducerEpoch`: the entry's own epoch
22/// bumps it, the epoch before the last bump is a retry of that bump, and
23/// every other epoch is fenced.
24///
25/// The retry rule also covers a failed epoch fence: that path records the
26/// epoch the producer still holds as the last epoch, so the producer that
27/// owns the transaction is the one the rule admits.
28///
29/// Kafka's `isValidProducerId` has a third admission clause, which this
30/// kernel does not model: `txnMetadata.producerEpoch ==
31/// RecordBatch.NO_PRODUCER_EPOCH` admits every supplied identity, and
32/// `prepareIncrementProducerEpoch` then bumps. Only the metadata that
33/// `handleInitProducerId` has just created for an unknown transactional ID
34/// carries that epoch. The host covers the clause: it answers a transactional
35/// ID with no entry by allocating a fresh identity without calling this
36/// kernel, and every entry it does pass here has a nonnegative epoch, because
37/// allocation hands out nonnegative epochs and
38/// [`crate::transaction::transaction_pid_install_decision`] rejects a replayed entry with a
39/// negative current or staged epoch.
40#[ensures((result == InitProducerIdIdentityDecision::BumpWithoutIdentity)
41    == (request_pid@ == -1))]
42#[ensures((result == InitProducerIdIdentityDecision::Bump)
43    == (request_pid@ != -1
44        && request_pid@ == entry_pid@
45        && request_epoch@ == entry_epoch@))]
46#[ensures((result == InitProducerIdIdentityDecision::Retry)
47    == (request_pid@ != -1
48        && (request_pid@ == entry_pid@
49            || (request_pid@ == prev_pid@
50                && request_epoch@ >= EXHAUSTED_PRODUCER_EPOCH@))
51        && !(request_pid@ == entry_pid@ && request_epoch@ == entry_epoch@)
52        && request_epoch@ == last_epoch@))]
53#[ensures((result == InitProducerIdIdentityDecision::Fenced)
54    == (request_pid@ != -1
55        && (!(request_pid@ == entry_pid@
56            || (request_pid@ == prev_pid@
57                && request_epoch@ >= EXHAUSTED_PRODUCER_EPOCH@))
58            || (!(request_pid@ == entry_pid@ && request_epoch@ == entry_epoch@)
59                && request_epoch@ != last_epoch@))))]
60#[must_use]
61pub fn init_producer_id_identity_decision(
62    entry_pid: i64,
63    entry_epoch: i16,
64    last_epoch: i16,
65    prev_pid: i64,
66    request_pid: i64,
67    request_epoch: i16,
68) -> InitProducerIdIdentityDecision {
69    if request_pid == -1 {
70        return InitProducerIdIdentityDecision::BumpWithoutIdentity;
71    }
72    let admitted = request_pid == entry_pid
73        || (request_pid == prev_pid && request_epoch >= EXHAUSTED_PRODUCER_EPOCH);
74    if !admitted {
75        return InitProducerIdIdentityDecision::Fenced;
76    }
77    if request_pid == entry_pid && request_epoch == entry_epoch {
78        return InitProducerIdIdentityDecision::Bump;
79    }
80    if request_epoch == last_epoch {
81        return InitProducerIdIdentityDecision::Retry;
82    }
83    InitProducerIdIdentityDecision::Fenced
84}
85
86/// Revalidate the transaction entry after the marker fan-out released its lock.
87///
88/// `expected` and `prepare_state` are what the caller prepared; `completion`
89/// and `complete_state` are the completion it intends to write. The live entry
90/// already holding that completion is an idempotent success; otherwise the
91/// entry must still hold the prepared identity (else it was fenced) and the
92/// prepared state (else another caller moved it).
93///
94/// The two state tags are one Kafka `Prepare* -> Complete*` pairing, which the
95/// host takes from a fixed mapping, so they differ.
96#[requires(prepare_state != complete_state)]
97#[ensures((result == TransactionCompletionDecision::AlreadyComplete)
98    == (has_identity(current, completion) && current.state == complete_state))]
99#[ensures((result == TransactionCompletionDecision::RejectStaleIdentity)
100    == (!(has_identity(current, completion) && current.state == complete_state)
101        && !has_identity(current, expected)))]
102#[ensures((result == TransactionCompletionDecision::RejectState)
103    == (!(has_identity(current, completion) && current.state == complete_state)
104        && has_identity(current, expected)
105        && current.state != prepare_state))]
106#[ensures((result == TransactionCompletionDecision::Proceed)
107    == (has_identity(current, expected) && current.state == prepare_state))]
108#[must_use]
109pub fn transaction_completion_decision(
110    current: TransactionSnapshot,
111    expected: TransactionIdentity,
112    completion: TransactionIdentity,
113    prepare_state: i8,
114    complete_state: i8,
115) -> TransactionCompletionDecision {
116    if current.pid == completion.pid
117        && current.epoch == completion.epoch
118        && current.state == complete_state
119    {
120        return TransactionCompletionDecision::AlreadyComplete;
121    }
122    if current.pid != expected.pid || current.epoch != expected.epoch {
123        return TransactionCompletionDecision::RejectStaleIdentity;
124    }
125    if current.state == prepare_state {
126        TransactionCompletionDecision::Proceed
127    } else {
128        TransactionCompletionDecision::RejectState
129    }
130}
krabka_verified/transaction/producer_identity.rs

krabka_verified/transaction/
producer_identity.rs
