krabka_verified/producer/
decision.rs

1#[cfg(creusot)]
2use super::{
3    first_retained_duplicate, retained_duplicate_exists, retained_matches, sequence_decision,
4};
5use {
6    super::{ProducerDecision, ProducerEntryFacts, RetainedSequenceRange, increment_sequence},
7    creusot_std::prelude::*,
8};
9
10/// Classify an incoming batch against a producer's tracked entry and the
11/// sequence ranges of the batches it retains.
12///
13/// This is Kafka's `UnifiedLog.analyzeAndValidateProducerState` for a client
14/// append. It first looks the batch up among the retained batches
15/// (`ProducerStateEntry.findDuplicateBatch`): at the entry's epoch, a batch
16/// whose base sequence and last sequence (`base_sequence + last_offset_delta`
17/// modulo `2^31`) equal a retained batch's is a duplicate of the first such
18/// batch. `None` slots never match. Only a batch that repeats no retained
19/// batch goes on to `ProducerAppendInfo.checkProducerEpoch` and
20/// `checkSequence`:
21///
22/// - a producer with no entry can start at any sequence, except that under
23///   `trunk_rules` it must start at sequence 0 on a partition whose log is
24///   empty (`log_empty`) and is otherwise out of order;
25/// - a lower epoch is fenced;
26/// - a higher epoch must start at sequence 0 and is otherwise out of order;
27/// - the same epoch must continue at the last sequence plus one modulo `2^31`
28///   and is otherwise out of order.
29///
30/// The empty-log rule is Kafka trunk's (KAFKA-15591, in
31/// `ProducerAppendInfo.checkSequence`); Kafka 4.3.1 does not have it.
32/// `log_empty` is Kafka's `ProducerStateManager.mapEndOffset() == 0`: no
33/// record has ever been appended to the partition, so no producer state can
34/// have been lost, and `trunk_rules` says whether the host serves Kafka
35/// trunk's behaviour.
36///
37/// The host supplies the retained batches at the entry's epoch; a batch at
38/// any other epoch never matches them.
39#[ensures(match result {
40    ProducerDecision::Duplicate { retained: index } => first_retained_duplicate(
41        entry, retained@, producer_epoch, base_sequence, last_offset_delta, index@),
42    _ => !retained_duplicate_exists(
43            entry, retained@, producer_epoch, base_sequence, last_offset_delta)
44        && result == sequence_decision(
45            entry, producer_epoch, base_sequence, log_empty, trunk_rules),
46})]
47#[must_use]
48pub fn producer_decision(
49    entry: Option<ProducerEntryFacts>,
50    retained: &[Option<RetainedSequenceRange>],
51    producer_epoch: i16,
52    base_sequence: i32,
53    last_offset_delta: i32,
54    log_empty: bool,
55    trunk_rules: bool,
56) -> ProducerDecision {
57    let Some(tracked) = entry else {
58        if trunk_rules && log_empty && base_sequence != 0 {
59            return ProducerDecision::OutOfOrder;
60        }
61        return ProducerDecision::Append;
62    };
63    if producer_epoch == tracked.epoch {
64        let last_sequence = increment_sequence(base_sequence, last_offset_delta);
65        let mut i = 0usize;
66        #[invariant(i@ <= [email protected]())]
67        #[invariant(forall<j: Int> 0 <= j && j < i@
68            ==> !retained_matches(retained@[j], base_sequence, last_offset_delta))]
69        #[variant([email protected]() - i@)]
70        while i < retained.len() {
71            if let Some(&Some(range)) = retained.get(i)
72                && range.base_sequence == base_sequence
73                && range.last_sequence == last_sequence
74            {
75                return ProducerDecision::Duplicate { retained: i };
76            }
77            i += 1;
78        }
79    }
80    if producer_epoch < tracked.epoch {
81        return ProducerDecision::Fenced;
82    }
83    if producer_epoch > tracked.epoch {
84        if base_sequence == 0 {
85            return ProducerDecision::Append;
86        }
87        return ProducerDecision::OutOfOrder;
88    }
89    if base_sequence == increment_sequence(tracked.last_sequence, 1) {
90        return ProducerDecision::Append;
91    }
92    ProducerDecision::OutOfOrder
93}
krabka_verified/producer/decision.rs

krabka_verified/producer/
decision.rs
