krabka_verified/
share.rs

1//! Share-group state pruning and offset-mutation decisions.
2
3use creusot_std::prelude::ensures;
4#[cfg(creusot)]
5use creusot_std::prelude::{DeepModel, Int, invariant};
6
7/// Admission result for an administrative share-offset mutation.
8#[krabka_macros::model_type(runtime(Debug, Clone, Copy, PartialEq, Eq))]
9pub enum ShareOffsetMutationDecision {
10    NotCoordinator,
11    NonEmptyGroup,
12    Unrequested,
13    FencedLeaderEpoch,
14    StateEpochOverflow,
15    ExactRetry,
16    Apply { next_state_epoch: i32 },
17}
18
19/// The ordered coordinator, membership, request, and retry gate established
20/// by the host before the epoch checks.
21#[krabka_macros::model_type(runtime(Debug, Clone, Copy, PartialEq, Eq))]
22pub enum ShareOffsetMutationGate {
23    NotCoordinator,
24    NonEmptyGroup,
25    Unrequested,
26    Admissible { exact_retry: bool },
27}
28
29/// Admit a requested share-offset mutation and choose its next state epoch.
30///
31/// The caller supplies the data-partition leader epoch captured while it
32/// planned the request and a fresh value read immediately before mutation.
33/// An exact retry is a no-op. Every other admitted mutation advances the
34/// durable share-state epoch by exactly one, or fails closed at `i32::MAX`.
35#[must_use]
36#[ensures(match result {
37    ShareOffsetMutationDecision::Apply { next_state_epoch } => {
38        gate == ShareOffsetMutationGate::Admissible { exact_retry: false }
39            && observed_leader_epoch@ == current_leader_epoch@
40            && state_epoch@ < i32::MAX@
41            && next_state_epoch@ == state_epoch@ + 1
42    }
43    ShareOffsetMutationDecision::NotCoordinator => {
44        gate == ShareOffsetMutationGate::NotCoordinator
45    }
46    ShareOffsetMutationDecision::NonEmptyGroup => {
47        gate == ShareOffsetMutationGate::NonEmptyGroup
48    }
49    ShareOffsetMutationDecision::Unrequested => gate == ShareOffsetMutationGate::Unrequested,
50    ShareOffsetMutationDecision::FencedLeaderEpoch => {
51        (exists<retry: bool> gate == ShareOffsetMutationGate::Admissible { exact_retry: retry })
52            && observed_leader_epoch@ != current_leader_epoch@
53    }
54    ShareOffsetMutationDecision::StateEpochOverflow => {
55        gate == ShareOffsetMutationGate::Admissible { exact_retry: false }
56            && observed_leader_epoch@ == current_leader_epoch@
57            && state_epoch@ == i32::MAX@
58    }
59    ShareOffsetMutationDecision::ExactRetry => {
60        gate == ShareOffsetMutationGate::Admissible { exact_retry: true }
61            && observed_leader_epoch@ == current_leader_epoch@
62    }
63})]
64pub const fn share_offset_mutation_decision(
65    gate: ShareOffsetMutationGate,
66    observed_leader_epoch: i32,
67    current_leader_epoch: i32,
68    state_epoch: i32,
69) -> ShareOffsetMutationDecision {
70    let exact_retry = match gate {
71        ShareOffsetMutationGate::NotCoordinator => {
72            return ShareOffsetMutationDecision::NotCoordinator;
73        }
74        ShareOffsetMutationGate::NonEmptyGroup => {
75            return ShareOffsetMutationDecision::NonEmptyGroup;
76        }
77        ShareOffsetMutationGate::Unrequested => {
78            return ShareOffsetMutationDecision::Unrequested;
79        }
80        ShareOffsetMutationGate::Admissible { exact_retry } => exact_retry,
81    };
82    if observed_leader_epoch != current_leader_epoch {
83        return ShareOffsetMutationDecision::FencedLeaderEpoch;
84    }
85    if exact_retry {
86        return ShareOffsetMutationDecision::ExactRetry;
87    }
88    match state_epoch.checked_add(1) {
89        Some(next_state_epoch) => ShareOffsetMutationDecision::Apply { next_state_epoch },
90        None => ShareOffsetMutationDecision::StateEpochOverflow,
91    }
92}
93
94/// Return the minimum latest-snapshot offset across every live share key.
95#[ensures(match result {
96    Some(frontier) => [email protected]() > 0
97        && (forall<i: Int> 0 <= i && i < [email protected]()
98            ==> frontier@ <= offsets@[i]@)
99        && (exists<i: Int> 0 <= i && i < [email protected]()
100            && frontier@ == offsets@[i]@),
101    None => [email protected]() == 0,
102})]
103#[must_use]
104pub fn share_prune_frontier(offsets: &[i64]) -> Option<i64> {
105    let &first = offsets.first()?;
106    let mut frontier = first;
107    let mut i = 1usize;
108    #[invariant(1 <= i@ && i@ <= [email protected]())]
109    #[invariant(forall<k: Int> 0 <= k && k < i@ ==> frontier@ <= offsets@[k]@)]
110    #[invariant(exists<k: Int> 0 <= k && k < i@ && frontier@ == offsets@[k]@)]
111    #[variant([email protected]() - i@)]
112    while i < offsets.len() {
113        if let Some(&offset) = offsets.get(i)
114            && offset < frontier
115        {
116            frontier = offset;
117        }
118        i += 1;
119    }
120    Some(frontier)
121}
122
123#[cfg(test)]
124mod tests {
125    use super::{
126        ShareOffsetMutationDecision, ShareOffsetMutationGate, share_offset_mutation_decision,
127        share_prune_frontier,
128    };
129
130    #[yare::parameterized(
131        not_coordinator = {
132            ShareOffsetMutationGate::NotCoordinator,
133            3,
134            3,
135            7,
136            ShareOffsetMutationDecision::NotCoordinator,
137        },
138        nonempty_group = {
139            ShareOffsetMutationGate::NonEmptyGroup,
140            3,
141            3,
142            7,
143            ShareOffsetMutationDecision::NonEmptyGroup,
144        },
145        unrequested = {
146            ShareOffsetMutationGate::Unrequested,
147            3,
148            3,
149            7,
150            ShareOffsetMutationDecision::Unrequested,
151        },
152        fenced_epoch = {
153            ShareOffsetMutationGate::Admissible { exact_retry: false },
154            2,
155            3,
156            7,
157            ShareOffsetMutationDecision::FencedLeaderEpoch,
158        },
159        state_epoch_overflow = {
160            ShareOffsetMutationGate::Admissible { exact_retry: false },
161            3,
162            3,
163            i32::MAX,
164            ShareOffsetMutationDecision::StateEpochOverflow,
165        },
166        exact_retry = {
167            ShareOffsetMutationGate::Admissible { exact_retry: true },
168            3,
169            3,
170            7,
171            ShareOffsetMutationDecision::ExactRetry,
172        },
173        apply = {
174            ShareOffsetMutationGate::Admissible { exact_retry: false },
175            3,
176            3,
177            7,
178            ShareOffsetMutationDecision::Apply { next_state_epoch: 8 },
179        },
180    )]
181    fn offset_mutation_fails_closed_and_advances_exactly_once(
182        gate: ShareOffsetMutationGate,
183        observed: i32,
184        current: i32,
185        state: i32,
186        expected: ShareOffsetMutationDecision,
187    ) {
188        assert2::check!(share_offset_mutation_decision(gate, observed, current, state) == expected);
189    }
190
191    #[yare::parameterized(
192        empty = { &[][..], None },
193        single = { &[42][..], Some(42) },
194        duplicate_minimum = { &[100, 30, 75, 30, 200][..], Some(30) },
195        minimum_at_start = { &[0, 5, 9][..], Some(0) },
196        minimum_integer = { &[i64::MAX, i64::MIN, 0][..], Some(i64::MIN) },
197    )]
198    fn prune_frontier_covers_empty_single_ties_and_missing_snapshots(
199        offsets: &[i64],
200        expected: Option<i64>,
201    ) {
202        assert2::check!(share_prune_frontier(offsets) == expected);
203    }
204}
krabka_verified/share.rs

krabka_verified/
share.rs
