nodedb_types/backup_envelope/
verification.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Per-collection verification records of a backup envelope.
4//!
5//! A backup records, per collection and engine part, the number of canonical
6//! rows and an order-independent digest of them. A restore recomputes both
7//! from the envelope before it writes, and from the destination after it
8//! writes.
9//!
10//! The digest is the sum, modulo 2^256, of the SHA-256 of every row:
11//!
12//! * A sum is order-independent, so rows add in any order and the partial
13//!   digests of several nodes merge by addition.
14//! * A duplicated row changes the sum. Under XOR two copies cancel.
15//! * A row's contribution subtracts out, so a restore can drop a TTL row that
16//!   expired after the backup. A sorted Merkle hash cannot, and it needs every
17//!   row hash in memory to sort.
18
19use std::fmt;
20
21/// The engine part of a collection a verification record covers. A collection
22/// with graph edges has an `Edges` record next to its `Documents` record.
23#[derive(
24    Debug,
25    Clone,
26    Copy,
27    PartialEq,
28    Eq,
29    PartialOrd,
30    Ord,
31    Hash,
32    zerompk::ToMessagePack,
33    zerompk::FromMessagePack,
34)]
35pub enum VerifiedPart {
36    Documents,
37    Edges,
38    KeyValue,
39    Vectors,
40    Timeseries,
41    Columnar,
42    Crdt,
43    /// Array cell versions.
44    Array,
45}
46
47impl VerifiedPart {
48    /// The byte that separates the parts' row hashes.
49    pub fn tag(self) -> u8 {
50        match self {
51            Self::Documents => 1,
52            Self::Edges => 2,
53            Self::KeyValue => 3,
54            Self::Vectors => 4,
55            Self::Timeseries => 5,
56            Self::Columnar => 6,
57            Self::Crdt => 7,
58            Self::Array => 8,
59        }
60    }
61
62    pub fn as_str(self) -> &'static str {
63        match self {
64            Self::Documents => "documents",
65            Self::Edges => "edges",
66            Self::KeyValue => "kv",
67            Self::Vectors => "vectors",
68            Self::Timeseries => "timeseries",
69            Self::Columnar => "columnar",
70            Self::Crdt => "crdt",
71            Self::Array => "array",
72        }
73    }
74}
75
76impl fmt::Display for VerifiedPart {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        f.write_str(self.as_str())
79    }
80}
81
82/// A row count and the sum of the rows' SHA-256, a little-endian 256-bit
83/// integer.
84#[derive(
85    Debug, Clone, Copy, PartialEq, Eq, Default, zerompk::ToMessagePack, zerompk::FromMessagePack,
86)]
87pub struct VerifiedTally {
88    pub count: u64,
89    pub digest: [u8; 32],
90}
91
92impl VerifiedTally {
93    /// Add one row by its SHA-256.
94    pub fn add(&mut self, row: &[u8; 32]) {
95        self.count = self.count.wrapping_add(1);
96        let mut carry = 0u16;
97        for (sum, byte) in self.digest.iter_mut().zip(row) {
98            let total = u16::from(*sum) + u16::from(*byte) + carry;
99            *sum = total as u8;
100            carry = total >> 8;
101        }
102    }
103
104    /// Remove one row [`Self::add`] added.
105    pub fn remove(&mut self, row: &[u8; 32]) {
106        self.count = self.count.wrapping_sub(1);
107        let mut borrow = 0i16;
108        for (sum, byte) in self.digest.iter_mut().zip(row) {
109            let total = i16::from(*sum) - i16::from(*byte) - borrow;
110            borrow = i16::from(total < 0);
111            *sum = total.rem_euclid(256) as u8;
112        }
113    }
114}
115
116impl fmt::Display for VerifiedTally {
117    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
118        let tail: Vec<u8> = self.digest.iter().rev().take(8).copied().collect();
119        write!(f, "{} rows, digest {}", self.count, hex::encode(tail))
120    }
121}
122
123/// The body of a `SECTION_ORIGIN_VERIFICATION` section is a
124/// msgpack-encoded `Vec<CollectionVerification>`: one record per collection
125/// part that holds at least one row.
126#[derive(Debug, Clone, PartialEq, Eq, zerompk::ToMessagePack, zerompk::FromMessagePack)]
127pub struct CollectionVerification {
128    /// Source id of the database the collection lives in.
129    pub database_id: u64,
130    /// Bare catalog name of the collection.
131    pub collection: String,
132    pub part: VerifiedPart,
133    pub tally: VerifiedTally,
134}
135
136/// Which check a verification ran as.
137#[derive(Debug, Clone, Copy, PartialEq, Eq)]
138pub enum VerificationPhase {
139    /// The envelope's rows against the tallies it records. Runs before the
140    /// first write.
141    Envelope,
142    /// The destination's rows against the envelope's. Runs after the last
143    /// re-issue.
144    Destination,
145    /// A MOVE TENANT target's rows against the source capture. Runs after the
146    /// re-issue and before the catalog moves.
147    Move,
148}
149
150/// One collection part whose recomputed tally differs from the expected one.
151#[derive(Debug, Clone, PartialEq, Eq)]
152pub struct VerificationMismatch {
153    /// The database the collection lives in: the source id in the envelope
154    /// phase, the destination id in the destination phase.
155    pub database_id: u64,
156    pub collection: String,
157    pub part: VerifiedPart,
158    pub expected: VerifiedTally,
159    pub found: VerifiedTally,
160}
161
162impl fmt::Display for VerificationMismatch {
163    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
164        write!(
165            f,
166            "collection '{}' in database {} ({}): expected {}, found {}",
167            self.collection, self.database_id, self.part, self.expected, self.found
168        )
169    }
170}
171
172/// The message of a failed verification. It names every mismatched
173/// collection and states what the restore or move left behind.
174pub fn verification_failure_message(
175    phase: &VerificationPhase,
176    mismatches: &[VerificationMismatch],
177) -> String {
178    let list = mismatches
179        .iter()
180        .map(ToString::to_string)
181        .collect::<Vec<_>>()
182        .join("; ");
183    match phase {
184        VerificationPhase::Envelope => format!(
185            "restore verification failed: the backup's rows do not match the counts and \
186             digests it records for {list}; nothing was restored"
187        ),
188        VerificationPhase::Destination => format!(
189            "restore verification failed: the destination does not hold the backed-up rows \
190             of {list}; the restore is failed and not rolled back, and the restored data \
191             stays in place for inspection"
192        ),
193        VerificationPhase::Move => format!(
194            "move verification failed: the target does not hold the moved rows of {list}; \
195             the move was not applied and the source is untouched"
196        ),
197    }
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203
204    fn row(seed: u8) -> [u8; 32] {
205        let mut bytes = [0u8; 32];
206        for (i, b) in bytes.iter_mut().enumerate() {
207            *b = seed
208                .wrapping_mul(31)
209                .wrapping_add(i as u8)
210                .wrapping_mul(0x9D);
211        }
212        bytes
213    }
214
215    #[test]
216    fn the_sum_ignores_row_order() {
217        let mut forward = VerifiedTally::default();
218        let mut backward = VerifiedTally::default();
219        for seed in 0..50 {
220            forward.add(&row(seed));
221        }
222        for seed in (0..50).rev() {
223            backward.add(&row(seed));
224        }
225        assert_eq!(forward, backward);
226        assert_eq!(forward.count, 50);
227    }
228
229    #[test]
230    fn a_duplicated_row_changes_the_sum() {
231        let mut once = VerifiedTally::default();
232        once.add(&row(1));
233        once.add(&row(2));
234        let mut twice = once;
235        twice.add(&row(2));
236        twice.remove(&row(1));
237        assert_eq!(once.count, twice.count);
238        assert_ne!(once.digest, twice.digest);
239    }
240
241    #[test]
242    fn remove_undoes_add_across_carries() {
243        let full = [0xFFu8; 32];
244        let mut tally = VerifiedTally::default();
245        tally.add(&row(7));
246        let before = tally;
247        tally.add(&full);
248        tally.add(&full);
249        tally.remove(&full);
250        tally.remove(&full);
251        assert_eq!(tally, before);
252    }
253
254    #[test]
255    fn a_record_round_trips() {
256        let mut tally = VerifiedTally::default();
257        tally.add(&row(3));
258        let records = vec![CollectionVerification {
259            database_id: 1025,
260            collection: "orders".into(),
261            part: VerifiedPart::Edges,
262            tally,
263        }];
264        let bytes = zerompk::to_msgpack_vec(&records).expect("encode");
265        let decoded: Vec<CollectionVerification> = zerompk::from_msgpack(&bytes).expect("decode");
266        assert_eq!(decoded, records);
267    }
268
269    #[test]
270    fn the_message_names_every_collection_and_the_outcome() {
271        let mismatch = |collection: &str| VerificationMismatch {
272            database_id: 0,
273            collection: collection.into(),
274            part: VerifiedPart::Documents,
275            expected: VerifiedTally::default(),
276            found: VerifiedTally::default(),
277        };
278        let both = [mismatch("orders"), mismatch("users")];
279        let message = verification_failure_message(&VerificationPhase::Destination, &both);
280        assert!(message.contains("'orders'") && message.contains("'users'"));
281        assert!(message.contains("not rolled back"));
282        let message = verification_failure_message(&VerificationPhase::Envelope, &both);
283        assert!(message.contains("nothing was restored"));
284        let message = verification_failure_message(&VerificationPhase::Move, &both);
285        assert!(message.starts_with("move verification failed"));
286        assert!(message.contains("'orders'") && message.contains("'users'"));
287        assert!(message.contains("the move was not applied and the source is untouched"));
288    }
289}
nodedb_types/backup_envelope/verification.rs

nodedb_types/backup_envelope/
verification.rs
