krabka_verified/compaction/
retain_decision.rs

1use {
2    super::{BatchMeta, CompactionDecodeStep, RecordMeta, RetainDecision, TxnDataState},
3    creusot_std::prelude::*,
4};
5
6/// Classify one compaction batch-stream decode step.
7///
8/// A decode failure can never be mistaken for end-of-stream: completion is
9/// admitted only when no input remains, and a successful decode must consume
10/// at least one byte.
11#[ensures((result == CompactionDecodeStep::Done) == (remaining_before@ == 0))]
12#[ensures((result == CompactionDecodeStep::Continue) == (remaining_before@ > 0
13    && decode_succeeded
14    && remaining_after@ < remaining_before@))]
15#[ensures((result == CompactionDecodeStep::Corrupt) == (remaining_before@ > 0
16    && (!decode_succeeded || remaining_after@ >= remaining_before@)))]
17#[must_use]
18pub const fn compaction_decode_step(
19    remaining_before: usize,
20    decode_succeeded: bool,
21    remaining_after: usize,
22) -> CompactionDecodeStep {
23    if remaining_before == 0 {
24        CompactionDecodeStep::Done
25    } else if decode_succeeded && remaining_after < remaining_before {
26        CompactionDecodeStep::Continue
27    } else {
28        CompactionDecodeStep::Corrupt
29    }
30}
31
32/// The delete horizon timestamp as an unbounded integer: `now +
33/// delete.retention.ms`, clamped to the `i64` range.
34// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
35#[cfg(creusot)]
36#[cfg_attr(test, mutants::skip)]
37#[logic]
38pub fn compute_horizon_model(now_ms: i64, delete_retention_ms: i64) -> Int {
39    pearlite! {
40        if now_ms@ + delete_retention_ms@ > 9223372036854775807 {
41            9223372036854775807
42        } else if now_ms@ + delete_retention_ms@ < -9223372036854775807 - 1 {
43            -9223372036854775807 - 1
44        } else {
45            now_ms@ + delete_retention_ms@
46        }
47    }
48}
49
50/// Compute the delete horizon timestamp: `now + delete.retention.ms`,
51/// saturated at the `i64` bounds.
52///
53/// The tombstone or marker is retained until the wall clock reaches this
54/// value.
55#[ensures(result@ == compute_horizon_model(now_ms, delete_retention_ms))]
56#[must_use]
57pub const fn compute_horizon(now_ms: i64, delete_retention_ms: i64) -> i64 {
58    now_ms.saturating_add(delete_retention_ms)
59}
60
61// cargo-mutants: #[cfg(creusot)] spec function; not compiled outside Creusot, so no test can tell.
62#[cfg(creusot)]
63#[cfg_attr(test, mutants::skip)]
64#[logic(open)]
65fn horizon_decision_matches(
66    existing: Option<i64>,
67    now: i64,
68    retention: i64,
69    decision: RetainDecision,
70) -> bool {
71    pearlite! { match existing {
72        None => exists<h: i64> decision == RetainDecision::SetHorizon(h)
73            && h@ == compute_horizon_model(now, retention),
74        Some(h) => decision == if now@ >= h@ { RetainDecision::Delete } else { RetainDecision::Keep },
75    } }
76}
77
78#[ensures(horizon_decision_matches(existing, now_ms, delete_retention_ms, result))]
79const fn horizon_retention(
80    existing: Option<i64>,
81    now_ms: i64,
82    delete_retention_ms: i64,
83) -> RetainDecision {
84    match existing {
85        Some(h) if now_ms >= h => RetainDecision::Delete,
86        Some(_) => RetainDecision::Keep,
87        None => RetainDecision::SetHorizon(compute_horizon(now_ms, delete_retention_ms)),
88    }
89}
90
91/// The single per-record KIP-534 retain decision.
92///
93/// Control batches, that is transaction commit and abort markers, are retained
94/// as long as the pass reads data of their transaction. Once the pass finds no
95/// data of it, the marker ages out through the delete horizon. Data records
96/// dedup newest-wins. A tombstone, that is a record with a null value, ages out
97/// through the delete horizon after it becomes the newest entry for its key.
98#[ensures(batch.is_control && (txn == TxnDataState::DataSurvives || txn == TxnDataState::NotTransactional)
99    ==> result == RetainDecision::Keep)]
100#[ensures(batch.is_control && txn == TxnDataState::DataFullyGone
101    ==> horizon_decision_matches(batch.existing_horizon, now_ms, delete_retention_ms, result))]
102#[ensures(!batch.is_control && !rec.has_key ==> result == RetainDecision::Delete)]
103#[ensures(!batch.is_control && rec.has_key && !is_newest_for_key ==> result == RetainDecision::Delete)]
104#[ensures(!batch.is_control && rec.has_key && is_newest_for_key && rec.has_value
105    ==> result == RetainDecision::Keep)]
106#[ensures(!batch.is_control && rec.has_key && is_newest_for_key && !rec.has_value
107    ==> horizon_decision_matches(batch.existing_horizon, now_ms, delete_retention_ms, result))]
108#[must_use]
109pub const fn retain_decision(
110    rec: RecordMeta,
111    batch: BatchMeta,
112    is_newest_for_key: bool,
113    txn: TxnDataState,
114    now_ms: i64,
115    delete_retention_ms: i64,
116) -> RetainDecision {
117    if batch.is_control {
118        return match txn {
119            TxnDataState::DataSurvives | TxnDataState::NotTransactional => RetainDecision::Keep,
120            TxnDataState::DataFullyGone => {
121                horizon_retention(batch.existing_horizon, now_ms, delete_retention_ms)
122            }
123        };
124    }
125    if !rec.has_key {
126        return RetainDecision::Delete;
127    }
128    if !is_newest_for_key {
129        return RetainDecision::Delete;
130    }
131    if rec.has_value {
132        return RetainDecision::Keep;
133    }
134    // Newest-for-key tombstone: age out via the delete horizon.
135    horizon_retention(batch.existing_horizon, now_ms, delete_retention_ms)
136}
krabka_verified/compaction/retain_decision.rs

krabka_verified/compaction/
retain_decision.rs
