krabka_verified/restore_sidecar/
restore_producer_ids_strict.rs

1#[cfg(creusot)]
2use super::{Int, invariant};
3use {
4    super::{RestoreAbortedTxn, RestoreSegmentExtent},
5    crate::log_index::{IndexEntry, OffsetEntry, TimeEntry},
6    creusot_std::prelude::*,
7};
8
9/// Compute the largest relative offset an archived sidecar may reference.
10///
11/// Kafka's sparse indexes store offsets relative to the segment base as a
12/// 32-bit value (`AbstractIndex.relativeOffset`), so a segment whose span does
13/// not fit `u32` has no valid index. The frontier is returned as `i64` so the
14/// entry validators compare widened `u32` offsets against it and no narrowing
15/// conversion is needed.
16#[ensures(match result {
17    Some(frontier) => segment_base@ >= 0
18        && segment_end@ >= segment_base@
19        && frontier@ == segment_end@ - segment_base@
20        && frontier@ <= u32::MAX@,
21    None => segment_base@ < 0
22        || segment_end@ < segment_base@
23        || segment_end@ - segment_base@ > u32::MAX@,
24})]
25#[must_use]
26pub fn restore_index_frontier(segment_base: i64, segment_end: i64) -> Option<i64> {
27    if segment_base < 0 || segment_end < segment_base {
28        return None;
29    }
30    let relative = segment_end - segment_base;
31    if relative > i64::from(u32::MAX) {
32        None
33    } else {
34        Some(relative)
35    }
36}
37
38/// Validate one decoded offset-index entry against its predecessor and the
39/// verified log extent.
40#[ensures(result == (relative_offset@ <= max_relative@
41    && position@ < log_bytes@
42    && match previous {
43        Some(IndexEntry { key: previous_relative, coordinate: previous_position }) =>
44            previous_relative@ < relative_offset@ && previous_position@ < position@,
45        None => true,
46    }))]
47#[must_use]
48pub fn restore_offset_index_entry_valid(
49    previous: Option<OffsetEntry>,
50    relative_offset: u32,
51    position: u32,
52    max_relative: i64,
53    log_bytes: u64,
54) -> bool {
55    i64::from(relative_offset) <= max_relative
56        && u64::from(position) < log_bytes
57        && match previous {
58            Some(IndexEntry {
59                key: previous_relative,
60                coordinate: previous_position,
61            }) => previous_relative < relative_offset && previous_position < position,
62            None => true,
63        }
64}
65
66/// Validate one decoded time-index entry. Relative offsets strictly advance;
67/// timestamps may repeat but may not decrease.
68///
69/// The segment writer emits strictly newer running maxima, at the last offset
70/// of the batch that established each maximum. Archive validation also admits
71/// repeated maxima: non-decreasing timestamps with advancing offsets suffice
72/// for the binary lookup and strict-start kernels. Structural validity cannot
73/// establish truthful record prefix bounds; those require faithful construction
74/// or a separate semantic check against complete decoded records.
75#[ensures(result == (relative_offset@ <= max_relative@
76    && match previous {
77        Some(IndexEntry { key: previous_timestamp, coordinate: previous_relative }) =>
78            previous_timestamp@ <= timestamp@ && previous_relative@ < relative_offset@,
79        None => true,
80    }))]
81#[must_use]
82pub fn restore_time_index_entry_valid(
83    previous: Option<TimeEntry>,
84    timestamp: i64,
85    relative_offset: u32,
86    max_relative: i64,
87) -> bool {
88    i64::from(relative_offset) <= max_relative
89        && match previous {
90            Some(IndexEntry {
91                key: previous_timestamp,
92                coordinate: previous_relative,
93            }) => previous_timestamp <= timestamp && previous_relative < relative_offset,
94            None => true,
95        }
96}
97
98/// Validate one decoded aborted-transaction index entry against the previous
99/// entry's `last_offset`.
100///
101/// An aborted transaction is indexed in the segment that holds its abort
102/// marker, in marker order, so `last_offset` (the marker) lies inside the
103/// segment and strictly increases across entries (Kafka's
104/// `TransactionIndex.append` rejects a non-increasing `lastOffset`, and
105/// `TransactionIndex.sanityCheck` requires `lastOffset` at or past the index's
106/// start offset). `start_offset` is neither bounded below by the segment base
107/// (a transaction may span a segment roll) nor ordered across entries
108/// (transactions interleave: one opened later can abort first).
109#[ensures(result == (entry.producer_id@ >= 0
110    && segment.base_offset@ >= 0
111    && entry.start_offset@ >= 0
112    && entry.start_offset@ <= entry.last_offset@
113    && segment.base_offset@ <= entry.last_offset@
114    && entry.last_offset@ <= segment.last_offset@
115    && match previous_last {
116        Some(previous) => previous@ < entry.last_offset@,
117        None => true,
118    }))]
119#[must_use]
120pub fn restore_txn_index_entry_valid(
121    previous_last: Option<i64>,
122    entry: RestoreAbortedTxn,
123    segment: RestoreSegmentExtent,
124) -> bool {
125    entry.producer_id >= 0
126        && segment.base_offset >= 0
127        && entry.start_offset >= 0
128        && entry.start_offset <= entry.last_offset
129        && segment.base_offset <= entry.last_offset
130        && entry.last_offset <= segment.last_offset
131        && !matches!(previous_last, Some(previous) if previous >= entry.last_offset)
132}
133
134/// Validate one segment-scoped leader-epoch checkpoint row.
135#[ensures(result == (epoch@ >= 0
136    && segment_base@ >= 0
137    && segment_base@ <= start_offset@
138    && start_offset@ <= segment_end@
139    && match previous {
140        Some((previous_epoch, previous_start)) =>
141            previous_epoch@ < epoch@ && previous_start@ < start_offset@,
142        None => true,
143    }))]
144#[must_use]
145pub fn restore_leader_epoch_entry_valid(
146    previous: Option<(i32, i64)>,
147    epoch: i32,
148    start_offset: i64,
149    segment_base: i64,
150    segment_end: i64,
151) -> bool {
152    epoch >= 0
153        && segment_base >= 0
154        && segment_base <= start_offset
155        && start_offset <= segment_end
156        && match previous {
157            Some((previous_epoch, previous_start)) => {
158                previous_epoch < epoch && previous_start < start_offset
159            }
160            None => true,
161        }
162}
163
164/// Accept exactly the canonical strictly increasing producer-ID order. Strict
165/// order also proves every ID is unique while keeping validation linear.
166#[ensures(result == (forall<i: Int> 1 <= i && i < [email protected]()
167    ==> producer_ids@[i - 1]@ < producer_ids@[i]@))]
168#[must_use]
169pub fn restore_producer_ids_strict(producer_ids: &[i64]) -> bool {
170    let mut index = 0usize;
171    #[invariant(index@ <= [email protected]())]
172    #[invariant(forall<i: Int> 1 <= i && i < index@
173        ==> producer_ids@[i - 1]@ < producer_ids@[i]@)]
174    #[variant([email protected]() - index@)]
175    while index < producer_ids.len() {
176        if index > 0
177            && let (Some(&previous), Some(&current)) =
178                (producer_ids.get(index - 1), producer_ids.get(index))
179            && previous >= current
180        {
181            return false;
182        }
183        index += 1;
184    }
185    true
186}
krabka_verified/restore_sidecar/restore_producer_ids_strict.rs

krabka_verified/restore_sidecar/
restore_producer_ids_strict.rs
