krabka_verified/restore/
rewritten_record.rs

1#[cfg(creusot)]
2use super::{kafka_record_timestamp, restore_record_placeable};
3use {
4    super::{
5        RestoreBatchFrame, RestoreLayout, RestoreProducer, RestoreRecordDeltas,
6        restore_record_coordinates,
7    },
8    creusot_std::prelude::*,
9};
10
11// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
12#[cfg(creusot)]
13#[cfg_attr(test, mutants::skip)]
14#[logic(open)]
15/// Kafka's legal producer identities (`RecordBatch.NO_PRODUCER_ID`,
16/// `NO_PRODUCER_EPOCH`, `NO_SEQUENCE` are all `-1`).
17///
18/// A data batch is either non-idempotent, with every identity field at its
19/// `-1` sentinel and no transactional bit, or idempotent, with a nonnegative
20/// producer id, epoch and base sequence. A transaction marker is a
21/// transactional control batch with a nonnegative producer and epoch and the
22/// `-1` sequence (`EndTransactionMarker` batches carry no sequence). A
23/// non-transactional control batch carries the non-idempotent sentinels.
24pub fn legal_producer(producer: RestoreProducer) -> bool {
25    pearlite! {
26        if producer.control {
27            non_idempotent_producer(producer)
28                || (producer.transactional
29                    && producer.producer_id@ >= 0
30                    && producer.producer_epoch@ >= 0
31                    && producer.base_sequence@ == -1)
32        } else {
33            non_idempotent_producer(producer)
34                || (producer.producer_id@ >= 0
35                    && producer.producer_epoch@ >= 0
36                    && producer.base_sequence@ >= 0)
37        }
38    }
39}
40
41// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
42#[cfg(creusot)]
43#[cfg_attr(test, mutants::skip)]
44#[logic(open)]
45/// Every identity field at its `-1` sentinel and no transactional bit.
46pub fn non_idempotent_producer(producer: RestoreProducer) -> bool {
47    pearlite! {
48        !producer.transactional
49            && producer.producer_id@ == -1
50            && producer.producer_epoch@ == -1
51            && producer.base_sequence@ == -1
52    }
53}
54
55// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
56#[cfg(creusot)]
57#[cfg_attr(test, mutants::skip)]
58#[logic(open)]
59/// A batch header's offset layout is legal: nonnegative base, span and
60/// record count, an exclusive end that fits `i64`, and for a control batch
61/// Kafka's single-offset span holding at most its one marker. A control batch
62/// may hold zero records because Kafka's `LogCleaner` (and krabka's
63/// compaction) keeps a producer's last batch as an empty header
64/// (`BatchRetention.RETAIN_EMPTY`) once the marker record itself is discarded.
65pub fn legal_header(layout: RestoreLayout, control: bool) -> bool {
66    pearlite! {
67        layout.base_offset@ >= 0
68            && layout.last_offset_delta@ >= 0
69            && layout.records_count@ >= 0
70            && (control ==> layout.last_offset_delta@ == 0 && layout.records_count@ <= 1)
71            && layout.base_offset@ + layout.last_offset_delta@ + 1 <= i64::MAX@
72    }
73}
74
75/// Admit a batch header synthesized or re-encoded by restore and return its
76/// exclusive offset frontier.
77#[ensures(match result {
78    Some(frontier) => legal_header(layout, producer.control)
79        && legal_producer(producer)
80        && frontier@ == layout.base_offset@ + layout.last_offset_delta@ + 1,
81    None => !(legal_header(layout, producer.control) && legal_producer(producer)),
82})]
83#[must_use]
84pub fn restore_rewritten_batch_header(
85    layout: RestoreLayout,
86    producer: RestoreProducer,
87) -> Option<i64> {
88    let non_idempotent = !producer.transactional
89        && producer.producer_id == -1
90        && producer.producer_epoch == -1
91        && producer.base_sequence == -1;
92    let producer_legal = if producer.control {
93        non_idempotent
94            || (producer.transactional
95                && producer.producer_id >= 0
96                && producer.producer_epoch >= 0
97                && producer.base_sequence == -1)
98    } else {
99        non_idempotent
100            || (producer.producer_id >= 0
101                && producer.producer_epoch >= 0
102                && producer.base_sequence >= 0)
103    };
104    if !producer_legal
105        || layout.base_offset < 0
106        || layout.last_offset_delta < 0
107        || layout.records_count < 0
108        || (producer.control && (layout.last_offset_delta != 0 || layout.records_count > 1))
109    {
110        return None;
111    }
112
113    layout
114        .base_offset
115        .checked_add(i64::from(layout.last_offset_delta))?
116        .checked_add(1)
117}
118
119// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
120#[cfg(creusot)]
121#[cfg_attr(test, mutants::skip)]
122#[logic(open)]
123/// One retained record is legal in a rewritten batch: it is placeable, its
124/// offset delta strictly follows the previous retained record's, and its
125/// Kafka timestamp does not exceed the preserved archived `max_timestamp`.
126/// The bound may be loose when the record that set it was filtered out;
127/// under `LogAppendTime` it holds with equality.
128pub fn restore_rewrite_record_legal(
129    previous_offset_delta: Option<i32>,
130    frame: RestoreBatchFrame,
131    record: RestoreRecordDeltas,
132) -> bool {
133    pearlite! {
134        restore_record_placeable(frame, record)
135            && match previous_offset_delta {
136                Some(previous) => previous@ < record.offset_delta@,
137                None => true,
138            }
139            && kafka_record_timestamp(frame, record) <= frame.max_timestamp@
140    }
141}
142
143/// Validate one retained record against the synthesized header and return its
144/// absolute offset and Kafka timestamp.
145#[ensures(match result {
146    Some((offset, timestamp)) => restore_rewrite_record_legal(previous_offset_delta, frame, record)
147        && offset@ == frame.base_offset@ + record.offset_delta@
148        && timestamp@ == kafka_record_timestamp(frame, record),
149    None => !restore_rewrite_record_legal(previous_offset_delta, frame, record),
150})]
151#[must_use]
152pub fn restore_rewritten_record(
153    previous_offset_delta: Option<i32>,
154    frame: RestoreBatchFrame,
155    record: RestoreRecordDeltas,
156) -> Option<(i64, i64)> {
157    if previous_offset_delta.is_some_and(|previous| previous >= record.offset_delta) {
158        return None;
159    }
160    let coordinates = restore_record_coordinates(frame, record)?;
161    if coordinates.1 > frame.max_timestamp {
162        None
163    } else {
164        Some(coordinates)
165    }
166}
krabka_verified/restore/rewritten_record.rs

krabka_verified/restore/
rewritten_record.rs
