krabka_verified/local_recovery/
index_frontier.rs

1#[cfg(creusot)]
2use super::{Int, invariant, logic};
3use {
4    super::{LocalRecoveryStep, LocalRecoverySwapAction, LocalRecoverySwapFacts},
5    crate::restore::restore_batch_step,
6    creusot_std::prelude::*,
7};
8
9// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
10#[cfg(creusot)]
11#[cfg_attr(test, mutants::skip)]
12#[logic(open)]
13/// A `.log.swap` is authoritative once it exists without its `.log.cleaned`.
14///
15/// Kafka's `LocalLog.replaceSegments` renames `.cleaned` to `.swap` before it
16/// deletes a single old segment, and `LogLoader.load` completes every such
17/// swap rather than discarding it, because the old segments may be gone.
18pub fn swap_committed(facts: LocalRecoverySwapFacts) -> bool {
19    pearlite! { facts.log_swap_exists && !facts.log_cleaned_exists }
20}
21
22/// Classify one swap set: abort it only while it is uncommitted, and complete
23/// it whenever it is committed, whether or not the original log still exists.
24///
25/// Aborting is safe only because the host deletes no replaced segment before
26/// the `.cleaned`-to-`.swap` rename of the log is durable (host
27/// responsibility: `atomic_swap` fsyncs the directory between the two steps).
28#[ensures((result == LocalRecoverySwapAction::AbortSwap) == facts.log_cleaned_exists)]
29#[ensures((result == LocalRecoverySwapAction::CompleteSwap) == swap_committed(facts))]
30#[ensures((result == LocalRecoverySwapAction::PromoteSidecars)
31    == (!facts.log_cleaned_exists && !facts.log_swap_exists && facts.final_log_exists))]
32#[ensures((result == LocalRecoverySwapAction::Reject)
33    == (!facts.log_cleaned_exists && !facts.log_swap_exists && !facts.final_log_exists))]
34#[must_use]
35pub const fn local_recovery_swap_action(facts: LocalRecoverySwapFacts) -> LocalRecoverySwapAction {
36    if facts.log_cleaned_exists {
37        LocalRecoverySwapAction::AbortSwap
38    } else if facts.log_swap_exists {
39        LocalRecoverySwapAction::CompleteSwap
40    } else if facts.final_log_exists {
41        LocalRecoverySwapAction::PromoteSidecars
42    } else {
43        LocalRecoverySwapAction::Reject
44    }
45}
46
47/// Whether completing a swap based at `swap_base`, whose records end before
48/// `swap_next`, replaces the segment based at `segment_base`.
49///
50/// This is `LogLoader.load`'s second pass: a segment whose base lies in
51/// `[swap_base, swap_next)` was compacted into the swap and is deleted. The
52/// swap's own base is always replaced, even by an empty swap, because the
53/// swap is renamed onto it.
54#[ensures(result == (segment_base@ == swap_base@
55    || (swap_base@ < segment_base@ && segment_base@ < swap_next@)))]
56#[must_use]
57pub const fn local_recovery_swap_replaces(
58    swap_base: i64,
59    swap_next: i64,
60    segment_base: i64,
61) -> bool {
62    segment_base == swap_base || (swap_base < segment_base && segment_base < swap_next)
63}
64
65/// Validate the discovered segment bases as one strictly ordered,
66/// nonnegative chain.
67#[ensures(result == (forall<i: Int> 0 <= i && i < [email protected]() ==>
68    bases@[i]@ >= 0
69        && (i == 0 || bases@[i - 1]@ < bases@[i]@)))]
70#[must_use]
71pub fn local_recovery_segment_chain(bases: &[i64]) -> bool {
72    let mut i = 0usize;
73    #[invariant(i@ <= [email protected]())]
74    #[invariant(forall<j: Int> 0 <= j && j < i@ ==>
75        bases@[j]@ >= 0
76            && (j == 0 || bases@[j - 1]@ < bases@[j]@))]
77    #[variant([email protected]() - i@)]
78    while i < bases.len() {
79        let Some(&base) = bases.get(i) else {
80            return false;
81        };
82        if base < 0 || (i > 0 && matches!(bases.get(i - 1), Some(&previous) if previous >= base)) {
83            return false;
84        }
85        i += 1;
86    }
87    true
88}
89
90/// Close a sealed segment exactly one offset before the next segment base.
91#[ensures(match result {
92    Some(last) => base@ >= 0 && next_base@ > base@ && last@ == next_base@ - 1,
93    None => base@ < 0 || next_base@ <= base@,
94})]
95#[must_use]
96pub fn local_recovery_sealed_last(base: i64, next_base: i64) -> Option<i64> {
97    if base < 0 || next_base <= base {
98        None
99    } else {
100        Some(next_base - 1)
101    }
102}
103
104/// Admit one completely decoded batch into the maximal valid tail prefix.
105///
106/// Compaction gaps are allowed, but overlap, empty byte progress, file-bound
107/// overflow, malformed offsets, and signed offset overflow stop the prefix.
108#[ensures(match result {
109    Some(step) => position@ <= file_end@
110        && encoded_len@ > 0
111        && step.valid_end@ == position@ + encoded_len@
112        && step.valid_end@ <= file_end@
113        && batch_base@ >= expected_offset@
114        && last_offset_delta@ >= 0
115        && step.last_offset@ == batch_base@ + last_offset_delta@
116        && step.next_offset@ == step.last_offset@ + 1
117        && step.next_offset@ > expected_offset@,
118    None => position@ > file_end@
119        || encoded_len@ == 0
120        || position@ + encoded_len@ > u64::MAX@
121        || position@ + encoded_len@ > file_end@
122        || batch_base@ < expected_offset@
123        || last_offset_delta@ < 0
124        || batch_base@ + last_offset_delta@ > i64::MAX@
125        || batch_base@ + last_offset_delta@ + 1 > i64::MAX@,
126})]
127#[must_use]
128pub fn local_recovery_batch_step(
129    position: u64,
130    file_end: u64,
131    expected_offset: i64,
132    batch_base: i64,
133    last_offset_delta: i32,
134    encoded_len: u64,
135) -> Option<LocalRecoveryStep> {
136    if position > file_end || encoded_len == 0 {
137        return None;
138    }
139    let valid_end = position.checked_add(encoded_len)?;
140    if valid_end > file_end {
141        return None;
142    }
143    let next_offset = restore_batch_step(expected_offset, batch_base, last_offset_delta)?;
144    Some(LocalRecoveryStep {
145        valid_end,
146        last_offset: next_offset - 1,
147        next_offset,
148    })
149}
150
151/// The exclusive frontier of a recovered segment, relative to its base, when
152/// it fits Kafka's 32-bit relative-offset index.
153///
154/// The host uses the result as an admission check on tail recovery: a segment
155/// whose recovered records reach past the `u32` index range is corrupt. The
156/// frontier is returned as `i64` so that one executable body is the one
157/// Creusot proves; its range is what the contract pins.
158#[ensures(match result {
159    Some(frontier) => segment_base@ >= 0
160        && last_offset@ >= segment_base@ - 1
161        && frontier@ == last_offset@ + 1 - segment_base@
162        && 0 <= frontier@
163        && frontier@ <= u32::MAX@,
164    None => segment_base@ < 0
165        || last_offset@ < segment_base@ - 1
166        || last_offset@ + 1 - segment_base@ > u32::MAX@,
167})]
168#[must_use]
169pub fn local_recovery_index_frontier(segment_base: i64, last_offset: i64) -> Option<i64> {
170    if segment_base < 0 || last_offset < segment_base - 1 {
171        return None;
172    }
173    // `segment_base >= 0`, so the subtraction cannot overflow, and the
174    // increment happens only once the result is inside the `u32` range.
175    let relative_last = last_offset - segment_base;
176    if relative_last >= i64::from(u32::MAX) {
177        None
178    } else {
179        Some(relative_last + 1)
180    }
181}
krabka_verified/local_recovery/index_frontier.rs

krabka_verified/local_recovery/
index_frontier.rs
