krabka_audit/spool/
resume.rs

1//! Recovery of the hash-chain position from the records still in the spool.
2//!
3//! After a crash the writer has to continue the chain where the spooled
4//! records left off, so this module walks the spool backwards to the last
5//! chained record and recomputes the sequence number and the head hash that
6//! follow it. Checkpoints carry no chain headers and are skipped.
7
8use {
9    super::Spool,
10    crate::{
11        chain::{chain_hash, from_hex32},
12        event::AuditEventClass,
13        ids::Seq,
14        sink::{AuditError, AuditRecord, HEADER_PREV_HASH, HEADER_SEQ},
15    },
16    krabka_verified::{ChainStep, chain_step},
17    std::str,
18};
19
20/// Where the chain continues after the last chained record in the spool.
21struct ChainPosition {
22    /// The sequence number the next chained record takes.
23    next_seq: Seq,
24    /// The chain head after the last chained record.
25    head: [u8; 32],
26}
27
28/// The chain headers a chained record carries.
29struct ChainHeaders {
30    seq: u64,
31    previous: [u8; 32],
32}
33
34impl Spool {
35    /// The `(next_seq, head)` from the last chained record in the spool.
36    ///
37    /// A chained record is a record that is not a checkpoint. Returns `None` if
38    /// the spool has no chained record.
39    #[tracing::instrument(level = "debug", skip_all, fields(count = self.count.0), err)]
40    /// # Errors
41    /// Returns an error if the spool file cannot be opened or read.
42    pub fn resume_point(&self) -> Result<Option<(u64, [u8; 32])>, AuditError> {
43        let records = self.read_all()?;
44        resume_from_records(&records)
45            .map(|resume| resume.map(|ChainPosition { next_seq, head }| (next_seq.0, head)))
46    }
47}
48
49fn poisoned(reason: &str) -> AuditError {
50    AuditError::Poisoned(format!("audit spool resume: {reason}"))
51}
52
53fn resume_from_records(records: &[AuditRecord]) -> Result<Option<ChainPosition>, AuditError> {
54    let mut resume: Option<ChainPosition> = None;
55    for record in records {
56        if record.class == AuditEventClass::Checkpoint {
57            continue;
58        }
59        let ChainHeaders { seq, previous } = chain_headers(record)?;
60        let next_seq = match resume {
61            None => seq
62                .checked_add(1)
63                .filter(|next| *next < u64::MAX)
64                .ok_or_else(|| poisoned("chain sequence exhausted"))?,
65            Some(ChainPosition {
66                next_seq: expected,
67                head,
68            }) => match chain_step(expected.0, seq, previous == head) {
69                ChainStep::Continue(next) => next,
70                ChainStep::SequenceMismatch => {
71                    return Err(poisoned("noncontiguous chain sequence"));
72                }
73                ChainStep::HeadMismatch => return Err(poisoned("chain head mismatch")),
74                ChainStep::Exhausted => return Err(poisoned("chain sequence exhausted")),
75            },
76        };
77        resume = Some(ChainPosition {
78            next_seq: Seq(next_seq),
79            head: chain_hash(&previous, seq, &record.value),
80        });
81    }
82    Ok(resume)
83}
84
85fn chain_headers(record: &AuditRecord) -> Result<ChainHeaders, AuditError> {
86    let seq = record
87        .headers
88        .iter()
89        .find(|(key, _)| key == HEADER_SEQ)
90        .and_then(|(_, value)| str::from_utf8(value).ok())
91        .and_then(|value| value.parse().ok());
92    let previous = record
93        .headers
94        .iter()
95        .find(|(key, _)| key == HEADER_PREV_HASH)
96        .and_then(|(_, value)| str::from_utf8(value).ok())
97        .and_then(from_hex32);
98    seq.zip(previous)
99        .map(|(seq, previous)| ChainHeaders { seq, previous })
100        .ok_or_else(|| poisoned("missing or invalid chain headers"))
101}
102
103#[cfg(test)]
104mod tests {
105    use {
106        super::*,
107        crate::{
108            chain::{GENESIS_HEAD, to_hex},
109            spool::test_support::{ROOMY_CAP, chained_record},
110        },
111        assert2::check,
112    };
113
114    #[test]
115    fn resume_point_is_from_last_chained_record_skipping_checkpoints() {
116        let dir = tempfile::tempdir().unwrap();
117        let mut s = Spool::open(dir.path(), ROOMY_CAP).unwrap();
118        let prev0 = GENESIS_HEAD;
119        let r0 = chained_record(0, &prev0, b"v0");
120        let head0 = chain_hash(&prev0, 0, b"v0");
121        let r1 = chained_record(1, &head0, b"v1");
122        let head1 = chain_hash(&head0, 1, b"v1");
123        // a checkpoint record (no seq/prev_hash) must be skipped
124        let cp = AuditRecord {
125            class: AuditEventClass::Checkpoint,
126            value: b"{\"type\":\"checkpoint\"}".to_vec(),
127            headers: vec![("event_class".into(), b"checkpoint".to_vec())],
128        };
129        s.append(&r0).unwrap();
130        s.append(&r1).unwrap();
131        s.append(&cp).unwrap();
132        let (next_seq, head) = s.resume_point().unwrap().unwrap();
133        // after seq 1; the hex projection also checks r1's chain math.
134        check!((next_seq, head, to_hex(&head)) == (2, head1, to_hex(&head1)));
135        let _ = (HEADER_SEQ, HEADER_PREV_HASH); // used by impl
136    }
137
138    #[test]
139    fn resume_rejects_a_malformed_or_disconnected_tail() {
140        let dir = tempfile::tempdir().unwrap();
141        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
142        let record = chained_record(0, &GENESIS_HEAD, b"valid");
143        let mut malformed = chained_record(1, &chain_hash(&GENESIS_HEAD, 0, b"valid"), b"bad");
144        malformed.headers.retain(|(key, _)| key != HEADER_SEQ);
145        spool.append(&record).unwrap();
146        spool.append(&malformed).unwrap();
147        check!(matches!(spool.resume_point(), Err(AuditError::Poisoned(_))));
148
149        let dir = tempfile::tempdir().unwrap();
150        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
151        spool.append(&record).unwrap();
152        spool
153            .append(&chained_record(2, &GENESIS_HEAD, b"gap"))
154            .unwrap();
155        check!(matches!(spool.resume_point(), Err(AuditError::Poisoned(_))));
156    }
157
158    #[test]
159    fn resume_rejects_sequence_exhaustion() {
160        let dir = tempfile::tempdir().unwrap();
161        let mut spool = Spool::open(dir.path(), ROOMY_CAP).unwrap();
162        spool
163            .append(&chained_record(u64::MAX - 1, &GENESIS_HEAD, b"last"))
164            .unwrap();
165        check!(matches!(spool.resume_point(), Err(AuditError::Poisoned(_))));
166    }
167}
krabka_audit/spool/resume.rs

krabka_audit/spool/
resume.rs
