sim_lib_journal/
verify.rs

1use crate::{JournalEntry, JournalHead, JournalObject, StoredState};
2use sim_kernel::ContentId;
3use std::collections::{BTreeMap, BTreeSet};
4use thiserror::Error;
5
6#[derive(Clone, Debug, PartialEq, Eq)]
7pub struct Verification {
8    pub head: Option<JournalHead>,
9    pub entries: Vec<JournalEntry>,
10    pub object_ids: BTreeSet<ContentId>,
11}
12
13#[derive(Clone, Debug, PartialEq, Eq, Error)]
14pub enum JournalError {
15    #[error("journal batch is empty")]
16    EmptyBatch,
17    #[error("journal head does not match the expected head")]
18    WrongHead,
19    #[error("writer lease is stale")]
20    StaleLease,
21    #[error("sequence is not gapless")]
22    SequenceGap,
23    #[error("entry has the wrong previous id")]
24    WrongPrevious,
25    #[error("entry identity is not canonical")]
26    CorruptEntry,
27    #[error("object {0:?} has bytes that do not match its id")]
28    CorruptObject(ContentId),
29    #[error("payload object {0:?} is missing")]
30    MissingPayload(ContentId),
31    #[error("content id was redelivered with conflicting bytes")]
32    ConflictingObject,
33    #[error("datum is not canonical")]
34    NonCanonicalDatum,
35    #[error("semantic object {0:?} is missing")]
36    MissingSemanticObject(ContentId),
37    #[error("sequence was redelivered with a conflicting entry")]
38    ConflictingDelivery,
39    #[error("backend state is corrupt: {0}")]
40    CorruptState(&'static str),
41    #[error("backend failure: {0}")]
42    Backend(String),
43    #[error("backend cannot satisfy durable journal writes: {0}")]
44    WriteRefused(&'static str),
45    #[error("journal verification exceeded its caller-supplied work bound")]
46    WorkBoundExceeded,
47    #[error("injected crash at {0}")]
48    InjectedCrash(&'static str),
49}
50
51pub(crate) fn verify_batch(
52    state: &StoredState,
53    expected: Option<&JournalHead>,
54    objects: &[JournalObject],
55    entries: &[JournalEntry],
56) -> Result<(), JournalError> {
57    let mut available: BTreeMap<ContentId, Vec<u8>> = state.objects.clone();
58    for object in objects {
59        object.verify()?;
60        if let Some(bytes) = available.get(&object.id)
61            && bytes != &object.bytes
62        {
63            return Err(JournalError::ConflictingObject);
64        }
65        available.insert(object.id.clone(), object.bytes.clone());
66    }
67    let first_sequence = expected.map_or(0, |h| h.sequence + 1);
68    let mut previous = expected.map(|h| h.entry.clone());
69    for (sequence, entry) in (first_sequence..).zip(entries) {
70        if entry.canonical_id()? != entry.id {
71            return Err(JournalError::CorruptEntry);
72        }
73        if entry.sequence != sequence {
74            return Err(JournalError::SequenceGap);
75        }
76        if entry.previous != previous {
77            return Err(JournalError::WrongPrevious);
78        }
79        for payload in &entry.payloads {
80            if !available.contains_key(payload) {
81                return Err(JournalError::MissingPayload(payload.clone()));
82            }
83        }
84        if let Some(existing) = state.entries.get(&entry.sequence)
85            && existing != entry
86        {
87            return Err(JournalError::ConflictingDelivery);
88        }
89        previous = Some(entry.id.clone());
90    }
91    Ok(())
92}
93
94pub(crate) fn verify_state(state: &StoredState) -> Result<Verification, JournalError> {
95    let mut prior = None;
96    let mut object_ids = BTreeSet::new();
97    for (expected_sequence, entry) in state.entries.values().enumerate() {
98        if entry.sequence != expected_sequence as u64 {
99            return Err(JournalError::CorruptState("sequence gap"));
100        }
101        if entry.previous != prior {
102            return Err(JournalError::CorruptState("previous id"));
103        }
104        if entry.canonical_id()? != entry.id {
105            return Err(JournalError::CorruptEntry);
106        }
107        for payload in &entry.payloads {
108            object_ids.insert(payload.clone());
109            let bytes = state
110                .objects
111                .get(payload)
112                .ok_or_else(|| JournalError::MissingPayload(payload.clone()))?;
113            let datum = state
114                .datums
115                .get(payload)
116                .ok_or_else(|| JournalError::MissingSemanticObject(payload.clone()))?;
117            let object = JournalObject::from_datum(datum.clone())?;
118            if object.id != *payload || object.bytes != *bytes {
119                return Err(JournalError::CorruptObject(payload.clone()));
120            }
121        }
122        prior = Some(entry.id.clone());
123    }
124    let computed = state.entries.last_key_value().map(|(_, e)| JournalHead {
125        sequence: e.sequence,
126        entry: e.id.clone(),
127    });
128    if computed != state.head {
129        return Err(JournalError::CorruptState("head"));
130    }
131    Ok(Verification {
132        head: computed,
133        entries: state.entries.values().cloned().collect(),
134        object_ids,
135    })
136}
sim_lib_journal/verify.rs

sim_lib_journal/
verify.rs
