krabka_verified/transaction/
next_producer_identity.rs

1#[cfg(creusot)]
2use super::first_unstable_frontier;
3use {
4    super::{IdleTransactionState, NO_TRANSACTION_TIMEOUT_MS},
5    creusot_std::prelude::*,
6};
7
8/// Select the first unstable transaction offset, or the log end when no
9/// transaction is open. A pending start beyond the log end is rejected.
10///
11/// `Some` and `None` partition the inputs: the result is `Some` exactly when
12/// every start is at or below the log end.
13#[ensures(match result {
14    Some(lso) => first_unstable_frontier(starts@, log_end@, lso@),
15    None => exists<i: Int> 0 <= i && i < [email protected]() && starts@[i]@ > log_end@,
16})]
17#[must_use]
18pub fn first_unstable_offset(starts: &[i64], log_end: i64) -> Option<i64> {
19    let mut lso = log_end;
20    let mut index = 0usize;
21    #[invariant(index@ <= [email protected]())]
22    #[invariant(lso@ <= log_end@)]
23    #[invariant(forall<i: Int> 0 <= i && i < index@ ==> starts@[i]@ <= log_end@)]
24    #[invariant(index@ == 0 ==> lso@ == log_end@)]
25    #[invariant(index@ > 0 ==> exists<i: Int> 0 <= i && i < index@ && lso@ == starts@[i]@)]
26    #[invariant(forall<i: Int> 0 <= i && i < index@ ==> lso@ <= starts@[i]@)]
27    #[variant([email protected]() - index@)]
28    while index < starts.len() {
29        let &start = starts.get(index)?;
30        if start > log_end {
31            return None;
32        }
33        if start < lso {
34            lso = start;
35        }
36        index += 1;
37    }
38    Some(lso)
39}
40
41/// A valid COMMIT or ABORT marker closes state only for its matching pending
42/// producer.
43#[ensures(result == ((is_abort || is_commit) && !(is_abort && is_commit) && has_pending))]
44#[must_use]
45pub fn transaction_marker_closes(is_abort: bool, is_commit: bool, has_pending: bool) -> bool {
46    (is_abort || is_commit) && !(is_abort && is_commit) && has_pending
47}
48
49/// Construct one aborted transaction's inclusive interval only from a live,
50/// nonnegative producer and ordered marker bounds.
51#[ensures(match result {
52    Some((start, last)) => producer_id@ >= 0
53        && pending_start == Some(start)
54        && last@ == marker_last@
55        && start@ <= last@,
56    None => producer_id@ < 0
57        || pending_start == None
58        || match pending_start { Some(start) => start@ > marker_last@, None => false },
59})]
60#[must_use]
61pub fn aborted_transaction_interval(
62    pending_start: Option<i64>,
63    marker_last: i64,
64    producer_id: i64,
65) -> Option<(i64, i64)> {
66    if producer_id < 0 {
67        return None;
68    }
69    let start = pending_start?;
70    if start > marker_last {
71        return None;
72    }
73    Some((start, marker_last))
74}
75
76/// Whether a valid inclusive aborted interval intersects a nonempty half-open
77/// Fetch range.
78#[ensures(result == (entry_start@ <= entry_last@
79    && query_start@ < query_end@
80    && entry_start@ < query_end@
81    && entry_last@ >= query_start@))]
82#[must_use]
83pub fn aborted_transaction_overlaps(
84    entry_start: i64,
85    entry_last: i64,
86    query_start: i64,
87    query_end: i64,
88) -> bool {
89    entry_start <= entry_last
90        && query_start < query_end
91        && entry_start < query_end
92        && entry_last >= query_start
93}
94
95/// Whether the idle reaper may abort one persisted transaction.
96///
97/// This is Kafka's `TransactionStateManager.timedOutTransactions`: the
98/// transaction is `ONGOING`, it is not a KIP-939 two-phase-commit transaction
99/// (`TransactionMetadata.isDistributedTwoPhaseCommitTxn`, a timeout of
100/// `Integer.MAX_VALUE`), and `txnStartTimestamp + txnTimeoutMs < now`. The
101/// comparison is strict, so a transaction is reapable only once its timeout
102/// has passed, not when it is reached. Kafka evaluates the sum in a Java
103/// `long`; the kernel compares exactly, without overflow.
104///
105/// The decision is total over every persisted timeout. Kafka's
106/// `TransactionLog.read` accepts any `TransactionTimeoutMs`, and
107/// `InitProducerId` refuses a timeout that is not positive
108/// (`validateTransactionTimeoutMs`), so a zero or negative timeout reaches the
109/// reaper only from a transaction-log record another writer produced. Kafka
110/// treats that timeout arithmetically, and so does this kernel.
111///
112/// A backwards clock never aborts a transaction with a nonnegative timeout,
113/// which is every timeout `InitProducerId` persists.
114#[ensures(result == (state == IdleTransactionState::Ongoing
115    && txn_timeout_ms@ != NO_TRANSACTION_TIMEOUT_MS@
116    && start_ms@ + txn_timeout_ms@ < now_ms@))]
117#[ensures(now_ms@ <= start_ms@ && txn_timeout_ms@ >= 0 ==> !result)]
118#[must_use]
119pub fn should_abort_idle_transaction(
120    state: IdleTransactionState,
121    txn_timeout_ms: i32,
122    start_ms: i64,
123    now_ms: i64,
124) -> bool {
125    let ongoing = match state {
126        IdleTransactionState::Ongoing => true,
127        IdleTransactionState::Other => false,
128    };
129    // Saturation keeps the order against any `i32` timeout: an elapsed time
130    // above `i64::MAX` exceeds it, and one below `i64::MIN` does not.
131    ongoing
132        && txn_timeout_ms != NO_TRANSACTION_TIMEOUT_MS
133        && now_ms.saturating_sub(start_ms) > i64::from(txn_timeout_ms)
134}
135
136#[logic(open)]
137fn identity_after_increment(result: Option<(i64, i16)>, pid: i64, epoch: i16) -> bool {
138    pearlite! { match result {
139        Some((result_pid, result_epoch)) => result_pid == pid && result_epoch@ == epoch@ + 1,
140        None => false,
141    }}
142}
143
144/// Choose the producer identity exposed after transaction completion.
145///
146/// Verified normal completion reserves `i16::MAX` for the transaction marker,
147/// while a staged recovery identity may use that epoch once before rotating.
148#[ensures(!verified ==> result == Some((pid, epoch)))]
149#[ensures(verified && !recovery && epoch@ < i16::MAX@ - 1 ==>
150    identity_after_increment(result, pid, epoch))]
151#[ensures(verified && recovery && epoch@ < i16::MAX@ ==>
152    identity_after_increment(result, pid, epoch))]
153#[ensures(verified
154    && ((!recovery && epoch@ >= i16::MAX@ - 1) || (recovery && epoch@ >= i16::MAX@)) ==>
155    match (result, fresh) {
156        (Some((result_pid, result_epoch)), Some(fresh_pid)) =>
157            result_pid == fresh_pid && result_epoch@ == 0,
158        (None, None) => true,
159        _ => false,
160    })]
161#[must_use]
162pub fn next_producer_identity(
163    verified: bool,
164    recovery: bool,
165    pid: i64,
166    epoch: i16,
167    fresh: Option<i64>,
168) -> Option<(i64, i16)> {
169    if !verified {
170        return Some((pid, epoch));
171    }
172    let can_increment = if recovery {
173        epoch < i16::MAX
174    } else {
175        epoch < i16::MAX - 1
176    };
177    if can_increment {
178        Some((pid, epoch + 1))
179    } else {
180        fresh.map(|fresh_pid| (fresh_pid, 0))
181    }
182}
krabka_verified/transaction/next_producer_identity.rs

krabka_verified/transaction/
next_producer_identity.rs
