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("sequence was redelivered with a conflicting entry")]
34    ConflictingDelivery,
35    #[error("backend state is corrupt: {0}")]
36    CorruptState(&'static str),
37    #[error("backend failure: {0}")]
38    Backend(String),
39    #[error("backend cannot satisfy durable journal writes: {0}")]
40    WriteRefused(&'static str),
41    #[error("journal verification exceeded its caller-supplied work bound")]
42    WorkBoundExceeded,
43    #[error("injected crash at {0}")]
44    InjectedCrash(&'static str),
45}
46
47pub(crate) fn verify_batch(
48    state: &StoredState,
49    expected: Option<&JournalHead>,
50    objects: &[JournalObject],
51    entries: &[JournalEntry],
52) -> Result<(), JournalError> {
53    let mut available: BTreeMap<ContentId, Vec<u8>> = state.objects.clone();
54    for object in objects {
55        object.verify()?;
56        if let Some(bytes) = available.get(&object.id)
57            && bytes != &object.bytes
58        {
59            return Err(JournalError::ConflictingObject);
60        }
61        available.insert(object.id.clone(), object.bytes.clone());
62    }
63    let first_sequence = expected.map_or(0, |h| h.sequence + 1);
64    let mut previous = expected.map(|h| h.entry.clone());
65    for (sequence, entry) in (first_sequence..).zip(entries) {
66        if entry.canonical_id() != entry.id {
67            return Err(JournalError::CorruptEntry);
68        }
69        if entry.sequence != sequence {
70            return Err(JournalError::SequenceGap);
71        }
72        if entry.previous != previous {
73            return Err(JournalError::WrongPrevious);
74        }
75        for payload in &entry.payloads {
76            if !available.contains_key(payload) {
77                return Err(JournalError::MissingPayload(payload.clone()));
78            }
79        }
80        if let Some(existing) = state.entries.get(&entry.sequence)
81            && existing != entry
82        {
83            return Err(JournalError::ConflictingDelivery);
84        }
85        previous = Some(entry.id.clone());
86    }
87    Ok(())
88}
89
90pub(crate) fn verify_state(state: &StoredState) -> Result<Verification, JournalError> {
91    let mut prior = None;
92    for (expected_sequence, entry) in state.entries.values().enumerate() {
93        if entry.sequence != expected_sequence as u64 {
94            return Err(JournalError::CorruptState("sequence gap"));
95        }
96        if entry.previous != prior {
97            return Err(JournalError::CorruptState("previous id"));
98        }
99        if entry.canonical_id() != entry.id {
100            return Err(JournalError::CorruptEntry);
101        }
102        for payload in &entry.payloads {
103            let bytes = state
104                .objects
105                .get(payload)
106                .ok_or_else(|| JournalError::MissingPayload(payload.clone()))?;
107            JournalObject {
108                id: payload.clone(),
109                bytes: bytes.clone(),
110            }
111            .verify()?;
112        }
113        prior = Some(entry.id.clone());
114    }
115    let computed = state.entries.last_key_value().map(|(_, e)| JournalHead {
116        sequence: e.sequence,
117        entry: e.id.clone(),
118    });
119    if computed != state.head {
120        return Err(JournalError::CorruptState("head"));
121    }
122    Ok(Verification {
123        head: computed,
124        entries: state.entries.values().cloned().collect(),
125        object_ids: state.objects.keys().cloned().collect(),
126    })
127}
sim_lib_journal/verify.rs

sim_lib_journal/
verify.rs
