uqa_storage/mvcc/serializable/checkpoint/
records.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Stable checkpoint records let providers publish only inserted, replaced and removed state.
8
9mod decode;
10mod encode;
11mod key;
12#[cfg(test)]
13mod tests;
14
15pub use encode::SerializableCheckpointRecord;
16pub use key::SerializableCheckpointKey;
17
18use super::{io::invalid, SerializableGraph, VersionResult};
19use crate::read_control::StorageReadControl;
20
21const MAGIC: &[u8; 8] = b"UQASREC1";
22
23impl SerializableGraph {
24    /// Whether keyed checkpoint state needs publication. Graphs restored from a complete legacy checkpoint need initial record publication even without a logical change. Only a validated keyed-record restore establishes the comparison baseline.
25    pub fn checkpoint_records_changed(&self) -> bool {
26        self.checkpoint_changed || self.checkpoint_records.is_empty()
27    }
28
29    /// Stream the exact difference from the keyed checkpoint loaded under this admission. A missing value deletes the old key; a value inserts a new record or replaces the header. Providers apply the complete stream in one durable transaction, including changes preceding an operation error. This borrows existing graph data and allocates no deletion journal or second encoded graph. It does not acknowledge durability or change the baseline.
30    pub fn write_checkpoint_changes<'a>(
31        &'a self,
32        control: &StorageReadControl,
33        mut update: impl FnMut(
34            SerializableCheckpointKey,
35            Option<SerializableCheckpointRecord<'a>>,
36        ) -> VersionResult<()>,
37    ) -> VersionResult<()> {
38        control.check()?;
39        if !self.checkpoint_records_changed() {
40            return Ok(());
41        }
42        let current = [
43            self.transactions.len(),
44            self.outgoing.len(),
45            self.predicates.reads.len(),
46            self.predicates.writes.len(),
47        ]
48        .into_iter()
49        .try_fold(1_usize, |total, count| {
50            total.checked_add(count).ok_or_else(invalid)
51        })?;
52        let additional = current
53            .saturating_sub(self.checkpoint_records.capacity())
54            .checked_mul(std::mem::size_of::<SerializableCheckpointKey>())
55            .ok_or_else(invalid)?;
56        // A successful publication must remain decodable under this allowance. Existing baseline capacity already covers unchanged and removed records; reserve only the additional future key storage, without allocating another index.
57        let _future_keys = self.transactions.budget().reserve(additional)?;
58        let mut old = self.checkpoint_records.iter().copied().peekable();
59        let mut previous = None;
60        self.visit_checkpoint_records(|record| {
61            control.check()?;
62            let key = record.key(control)?;
63            if previous.is_some_and(|previous| previous >= key) {
64                return Err(invalid());
65            }
66            previous = Some(key);
67            while old.peek().is_some_and(|old| *old < key) {
68                update(old.next().expect("peeked checkpoint key"), None)?;
69            }
70            let present = old.peek() == Some(&key);
71            if present {
72                old.next();
73            }
74            if !present || key == SerializableCheckpointKey::HEADER {
75                update(key, Some(record))?;
76            }
77            Ok(())
78        })?;
79        for key in old {
80            control.check()?;
81            update(key, None)?;
82        }
83        Ok(())
84    }
85
86    fn visit_checkpoint_records<'a>(
87        &'a self,
88        mut visit: impl FnMut(SerializableCheckpointRecord<'a>) -> VersionResult<()>,
89    ) -> VersionResult<()> {
90        visit(SerializableCheckpointRecord::header(self))?;
91        for &entry in &*self.transactions {
92            visit(SerializableCheckpointRecord::transaction(self, entry))?;
93        }
94        for &edge in &*self.outgoing {
95            visit(SerializableCheckpointRecord::edge(self, edge))?;
96        }
97        for (writing, entries) in [
98            (false, &self.predicates.reads),
99            (true, &self.predicates.writes),
100        ] {
101            for entry in &**entries {
102                visit(SerializableCheckpointRecord::predicate(
103                    self, entry, writing,
104                ))?;
105            }
106        }
107        Ok(())
108    }
109}
uqa_storage/mvcc/serializable/checkpoint/records.rs

uqa_storage/mvcc/serializable/checkpoint/
records.rs
