krabka_verified/recovery/
replay_cursor_decision.rs

1use {
2    super::{
3        BarrierRecoveryFinalizeDecision, BarrierRecoveryFoldAction, BarrierRecoveryRecordKind,
4        ReplayCursorDecision, ReplayRecordDecision,
5    },
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 record is replayed iff its absolute offset `batch_base + record_delta`
14/// lies in the half-open replay window `[from, end)` and its batch is of the
15/// kind this pass replays: data batches for the metadata image, control
16/// batches for the voter and quorum state.
17pub fn replay_record_admitted(
18    batch_base: i64,
19    record_delta: i32,
20    from: i64,
21    end: i64,
22    control_batch: bool,
23    replay_control: bool,
24) -> bool {
25    pearlite! {
26        record_delta@ >= 0
27            && control_batch == replay_control
28            && from@ <= batch_base@ + record_delta@
29            && batch_base@ + record_delta@ < end@
30    }
31}
32
33/// Decide whether controller recovery replays one decoded metadata record,
34/// and at which absolute offset.
35///
36/// A negative delta is malformed and skipped. Because an admitted offset is
37/// below `end`, it is always representable, so no overflow case needs a rule
38/// of its own.
39#[ensures(match result {
40    ReplayRecordDecision::Apply(offset) =>
41        replay_record_admitted(batch_base, record_delta, from, end, control_batch, replay_control)
42            && offset@ == batch_base@ + record_delta@,
43    ReplayRecordDecision::Skip =>
44        !replay_record_admitted(batch_base, record_delta, from, end, control_batch, replay_control),
45})]
46#[must_use]
47pub fn replay_record_decision(
48    batch_base: i64,
49    record_delta: i32,
50    from: i64,
51    end: i64,
52    control_batch: bool,
53    replay_control: bool,
54) -> ReplayRecordDecision {
55    if record_delta < 0 || batch_base >= end || control_batch != replay_control {
56        return ReplayRecordDecision::Skip;
57    }
58    let delta = i64::from(record_delta);
59    if batch_base > i64::MAX - delta {
60        return ReplayRecordDecision::Skip;
61    }
62    let offset = batch_base + delta;
63    if offset < from || offset >= end {
64        ReplayRecordDecision::Skip
65    } else {
66        ReplayRecordDecision::Apply(offset)
67    }
68}
69
70/// Select the exact mutation for one decoded barrier-state record.
71///
72/// The host applies this result to the entry named by the decoded group key,
73/// in log order. A consumed epoch suppresses a later retry of its
74/// injection-start, and only a matching epoch may clear the pending injection.
75#[ensures(match result {
76    BarrierRecoveryFoldAction::DefineGroup => {
77        kind == BarrierRecoveryRecordKind::Group && value_present
78    }
79    BarrierRecoveryFoldAction::RemoveGroup => {
80        kind == BarrierRecoveryRecordKind::Group && !value_present
81    }
82    BarrierRecoveryFoldAction::SetPending => {
83            kind == BarrierRecoveryRecordKind::InjectionStart
84            && value_present
85            && !epoch_already_consumed
86    }
87    BarrierRecoveryFoldAction::KeepPending => {
88        kind == BarrierRecoveryRecordKind::InjectionStart
89            && ((value_present && epoch_already_consumed)
90                || (!value_present && pending_epoch != Some(record_epoch)))
91    }
92    BarrierRecoveryFoldAction::ClearPending => {
93        kind == BarrierRecoveryRecordKind::InjectionStart
94            && !value_present
95            && pending_epoch == Some(record_epoch)
96    }
97    BarrierRecoveryFoldAction::UpsertCut { retire_pending } => {
98        kind == BarrierRecoveryRecordKind::Cut
99            && value_present
100            && retire_pending == (pending_epoch == Some(record_epoch))
101    }
102    BarrierRecoveryFoldAction::RemoveCut => {
103        kind == BarrierRecoveryRecordKind::Cut && !value_present
104    }
105})]
106#[must_use]
107pub fn barrier_recovery_fold_action(
108    kind: BarrierRecoveryRecordKind,
109    value_present: bool,
110    record_epoch: i64,
111    pending_epoch: Option<i64>,
112    epoch_already_consumed: bool,
113) -> BarrierRecoveryFoldAction {
114    match kind {
115        BarrierRecoveryRecordKind::Group => {
116            if value_present {
117                BarrierRecoveryFoldAction::DefineGroup
118            } else {
119                BarrierRecoveryFoldAction::RemoveGroup
120            }
121        }
122        BarrierRecoveryRecordKind::InjectionStart => {
123            if value_present {
124                if epoch_already_consumed {
125                    BarrierRecoveryFoldAction::KeepPending
126                } else {
127                    BarrierRecoveryFoldAction::SetPending
128                }
129            } else if pending_epoch == Some(record_epoch) {
130                BarrierRecoveryFoldAction::ClearPending
131            } else {
132                BarrierRecoveryFoldAction::KeepPending
133            }
134        }
135        BarrierRecoveryRecordKind::Cut => {
136            if value_present {
137                BarrierRecoveryFoldAction::UpsertCut {
138                    retire_pending: pending_epoch == Some(record_epoch),
139                }
140            } else {
141                BarrierRecoveryFoldAction::RemoveCut
142            }
143        }
144    }
145}
146
147/// Decide whether an interrupted injection can be finalized conservatively.
148///
149/// A valid finalization has a current coordinator at or above the frozen
150/// coordinator epoch and at least one valid target partition. The recovery
151/// adapter supplies no observed marker offsets, so this decision can only
152/// authorize a partial cut.
153#[ensures((result == BarrierRecoveryFinalizeDecision::NoPending) == !has_pending)]
154#[ensures((result == BarrierRecoveryFinalizeDecision::MalformedPending)
155    == (has_pending && (frozen_coordinator_epoch@ < 0 || !targets_valid)))]
156#[ensures((result == BarrierRecoveryFinalizeDecision::UnknownCoordinator)
157    == (has_pending
158        && frozen_coordinator_epoch@ >= 0
159        && targets_valid
160        && current_coordinator_epoch == None))]
161#[ensures((result == BarrierRecoveryFinalizeDecision::FencedCoordinator)
162    == (has_pending
163        && frozen_coordinator_epoch@ >= 0
164        && targets_valid
165        && match current_coordinator_epoch {
166            Some(current) => current@ < frozen_coordinator_epoch@,
167            None => false,
168        }))]
169#[ensures((result == BarrierRecoveryFinalizeDecision::FinalizePartial)
170    == (has_pending
171        && frozen_coordinator_epoch@ >= 0
172        && targets_valid
173        && match current_coordinator_epoch {
174            Some(current) => current@ >= frozen_coordinator_epoch@,
175            None => false,
176        }))]
177#[must_use]
178pub fn barrier_recovery_finalize_decision(
179    has_pending: bool,
180    current_coordinator_epoch: Option<i32>,
181    frozen_coordinator_epoch: i32,
182    targets_valid: bool,
183) -> BarrierRecoveryFinalizeDecision {
184    if !has_pending {
185        return BarrierRecoveryFinalizeDecision::NoPending;
186    }
187    if frozen_coordinator_epoch < 0 || !targets_valid {
188        return BarrierRecoveryFinalizeDecision::MalformedPending;
189    }
190    let Some(current) = current_coordinator_epoch else {
191        return BarrierRecoveryFinalizeDecision::UnknownCoordinator;
192    };
193    if current < frozen_coordinator_epoch {
194        BarrierRecoveryFinalizeDecision::FencedCoordinator
195    } else {
196        BarrierRecoveryFinalizeDecision::FinalizePartial
197    }
198}
199
200/// Advance a replay cursor to the next batch offset the reader reported, and
201/// stop when there is none or it would not move the cursor forward, so a
202/// replay loop always terminates.
203#[ensures(match result {
204    ReplayCursorDecision::Advance(next_offset) => next == Some(next_offset)
205        && next_offset@ > cursor@,
206    ReplayCursorDecision::Stop => match next {
207        None => true,
208        Some(next_offset) => next_offset@ <= cursor@,
209    },
210})]
211#[must_use]
212pub fn replay_cursor_decision(cursor: i64, next: Option<i64>) -> ReplayCursorDecision {
213    match next {
214        Some(next_offset) if next_offset > cursor => ReplayCursorDecision::Advance(next_offset),
215        Some(_) | None => ReplayCursorDecision::Stop,
216    }
217}
krabka_verified/recovery/replay_cursor_decision.rs

krabka_verified/recovery/
replay_cursor_decision.rs
