commonware_storage/qmdb/
batch_chain.rs

1//! Shared validation for QMDB batch chains.
2//!
3//! A batch chain is a linked sequence of in-memory batches built on top of a DB state. Each batch
4//! records its state via [`Bounds`] (where its operations sit in the log, and the root at each
5//! applicable boundary) and the inactivity floor declared by its commit. Older batches in the chain
6//! are tracked as [`AncestorBounds`] in newest-first order. Some may already be applied to the
7//! database while others may not.
8//!
9//! Before applying a batch to the DB, the internal validation checks two things shared across QMDB
10//! variants (any, immutable, keyless):
11//!
12//! - The batch is not stale: the current DB state must match either the batch's recorded DB state
13//!   or one of its ancestor states.
14//! - Commit floors are monotonically non-decreasing along the chain, and no floor exceeds
15//!   its own commit location. Ancestors already applied to the database are skipped because their
16//!   floors were validated when they were first applied. The rest of the chain and the tip are
17//!   checked.
18//!
19//! Internal helpers walk the chain via weak parent references and snapshot ancestor bounds into a
20//! `Vec` for storage on a merkleized batch.
21
22use crate::{
23    merkle::{Family, Location},
24    qmdb::Error,
25};
26use commonware_cryptography::Digest;
27use core::iter;
28use std::sync::{Arc, Weak};
29
30/// Identifies a QMDB state by its operation `size` and authenticated `root`.
31#[derive(Clone, Copy, Debug)]
32pub struct Commitment<F: Family, D: Digest> {
33    /// Number of operations in the state.
34    pub size: Location<F>,
35    /// Root committing to those operations.
36    pub root: D,
37}
38
39impl<F: Family, D: Digest> Commitment<F, D> {
40    /// Create a [`Commitment`] from an operation `size` and its committing `root` digest.
41    pub(crate) const fn new(size: Location<F>, root: D) -> Self {
42        Self { size, root }
43    }
44}
45
46// `Family` is not `PartialEq`, so deriving would demand `F: PartialEq` at every call site.
47// Compare the fields directly, which needs only `Location<F>: PartialEq` and `D: Eq`.
48impl<F: Family, D: Digest> PartialEq for Commitment<F, D> {
49    fn eq(&self, other: &Self) -> bool {
50        self.size == other.size && self.root == other.root
51    }
52}
53
54impl<F: Family, D: Digest> Eq for Commitment<F, D> {}
55
56/// Bounds declared by an ancestor batch's commit.
57#[derive(Clone)]
58pub struct AncestorBounds<F: Family, D: Digest> {
59    /// Inactivity floor declared by the ancestor commit.
60    pub floor: Location<F>,
61    /// [`Commitment`] after the ancestor batch.
62    pub state: Commitment<F, D>,
63}
64
65/// Position and inactivity-floor state for a merkleized QMDB batch.
66#[derive(Clone)]
67pub struct Bounds<F: Family, D: Digest> {
68    /// [`Commitment`] immediately before this batch's own operations.
69    pub base: Commitment<F, D>,
70    /// [`Commitment`] at the boundary between applied database operations and operations kept in
71    /// this batch chain.
72    ///
73    /// Usually this is the database state when the batch chain was created.
74    pub db: Commitment<F, D>,
75    /// This batch's tip [`Commitment`]: the state after all its operations.
76    pub tip: Commitment<F, D>,
77    /// Ancestor bounds in newest-first order.
78    pub ancestors: Vec<AncestorBounds<F, D>>,
79    /// Inactivity floor declared by this batch's commit.
80    pub inactivity_floor: Location<F>,
81}
82
83impl<F: Family, D: Digest> Bounds<F, D> {
84    /// Create initial bounds for a batch built directly from the current database state.
85    ///
86    /// The base, DB boundary, and tip all coincide at `state`, with no ancestors.
87    pub(crate) const fn from_db(state: Commitment<F, D>, inactivity_floor: Location<F>) -> Self {
88        Self {
89            base: state,
90            db: state,
91            tip: state,
92            ancestors: Vec::new(),
93            inactivity_floor,
94        }
95    }
96
97    /// Validate that this batch can be applied to the current database state.
98    pub(crate) fn validate_apply_to(
99        &self,
100        current: Commitment<F, D>,
101        current_floor: Location<F>,
102    ) -> Result<(), Error<F>> {
103        validate_batch_applicable(current, self.db, &self.ancestors)?;
104        validate_commit_floors(
105            current_floor,
106            current.size,
107            &self.ancestors,
108            self.inactivity_floor,
109            self.tip
110                .size
111                .checked_sub(1)
112                .expect("merkleized batch includes a commit"),
113        )
114    }
115}
116
117/// Iterate over a batch's live ancestors, starting at `parent`.
118///
119/// Iteration stops when a weak parent reference cannot be upgraded.
120pub(crate) fn ancestors<T, P>(
121    parent: Option<Weak<T>>,
122    mut parent_of: P,
123) -> impl Iterator<Item = Arc<T>>
124where
125    P: for<'a> FnMut(&'a T) -> Option<&'a Weak<T>>,
126{
127    let mut next = parent.as_ref().and_then(Weak::upgrade);
128    iter::from_fn(move || {
129        let batch = next.take()?;
130        next = parent_of(&batch).and_then(Weak::upgrade);
131        Some(batch)
132    })
133}
134
135/// Iterate over a strong parent followed by its live ancestors.
136pub(crate) fn parent_and_ancestors<T, P, I>(
137    parent: Option<&Arc<T>>,
138    mut ancestors_of: P,
139) -> impl Iterator<Item = Arc<T>> + use<T, P, I>
140where
141    P: FnMut(&Arc<T>) -> I,
142    I: IntoIterator<Item = Arc<T>>,
143{
144    parent.cloned().into_iter().flat_map(move |parent| {
145        let ancestors = ancestors_of(&parent);
146        iter::once(parent).chain(ancestors)
147    })
148}
149
150/// Collect ancestor bounds in newest-first order.
151pub(crate) fn collect_ancestor_bounds<T, F, D, I, L, C>(
152    ancestors: I,
153    floor: L,
154    state: C,
155) -> Vec<AncestorBounds<F, D>>
156where
157    F: Family,
158    D: Digest,
159    I: IntoIterator<Item = Arc<T>>,
160    L: Fn(&T) -> Location<F>,
161    C: Fn(&T) -> Commitment<F, D>,
162{
163    ancestors
164        .into_iter()
165        .map(|batch| AncestorBounds {
166            floor: floor(&batch),
167            state: state(&batch),
168        })
169        .collect()
170}
171
172/// Advance the inherited DB boundary past applied ancestors no longer reachable
173/// through the weak parent chain.
174pub(crate) fn effective_boundary<F: Family, D: Digest>(
175    inherited: Commitment<F, D>,
176    oldest_live_base: Option<Commitment<F, D>>,
177) -> Commitment<F, D> {
178    oldest_live_base
179        .filter(|base| base.size > inherited.size)
180        .unwrap_or(inherited)
181}
182
183/// Validate that a batch can be applied to the database at the given [`Commitment`].
184///
185/// A batch is applicable if the database has not advanced since the batch was created, if all
186/// ancestors are already applied, or if the database has advanced to one of the batch's ancestor
187/// [`Commitment`]s.
188pub(crate) fn validate_batch_applicable<F: Family, D: Digest>(
189    current: Commitment<F, D>,
190    batch_db: Commitment<F, D>,
191    ancestors: &[AncestorBounds<F, D>],
192) -> Result<(), Error<F>> {
193    // A separate base check is unnecessary: a direct batch's base is `batch_db`, while a child
194    // batch's base is its first ancestor.
195    if current == batch_db || ancestors.iter().any(|ancestor| ancestor.state == current) {
196        return Ok(());
197    }
198
199    Err(Error::StaleBatch)
200}
201
202/// Validate commit-floor monotonicity for a batch chain.
203///
204/// Ancestors are stored newest-first. Validation walks them in reverse so unapplied ancestors are
205/// checked oldest-to-newest, then checks the tip. Ancestors at or below `db_size` are already
206/// applied locally and are skipped.
207pub(crate) fn validate_commit_floors<F: Family, D: Digest>(
208    starting_floor: Location<F>,
209    db_size: Location<F>,
210    ancestors: &[AncestorBounds<F, D>],
211    tip_floor: Location<F>,
212    tip_commit_loc: Location<F>,
213) -> Result<(), Error<F>> {
214    let mut prev_floor = starting_floor;
215    for ancestor in ancestors.iter().rev() {
216        if ancestor.state.size <= db_size {
217            continue;
218        }
219
220        let ancestor_commit_loc = ancestor.state.size - 1;
221        if ancestor.floor < prev_floor {
222            return Err(Error::FloorRegressed(ancestor.floor, prev_floor));
223        }
224        if ancestor.floor > ancestor_commit_loc {
225            return Err(Error::FloorBeyondSize(ancestor.floor, ancestor_commit_loc));
226        }
227        prev_floor = ancestor.floor;
228    }
229
230    if tip_floor < prev_floor {
231        return Err(Error::FloorRegressed(tip_floor, prev_floor));
232    }
233    if tip_floor > tip_commit_loc {
234        return Err(Error::FloorBeyondSize(tip_floor, tip_commit_loc));
235    }
236    Ok(())
237}
238
239#[cfg(test)]
240mod tests {
241    use super::*;
242    use crate::merkle::mmr;
243    use commonware_cryptography::sha256;
244    use std::sync::{Arc, Weak};
245
246    type F = mmr::Family;
247    type D = sha256::Digest;
248
249    struct TestBatch {
250        id: u8,
251        bounds: Bounds<F, D>,
252        parent: Option<Weak<Self>>,
253    }
254
255    const fn loc(n: u64) -> Location<F> {
256        Location::new(n)
257    }
258
259    fn state(size: u64, marker: u8) -> Commitment<F, D> {
260        Commitment::new(Location::new(size), D::from([marker; 32]))
261    }
262
263    fn ancestor(floor: Location<F>, end: u64, marker: u8) -> AncestorBounds<F, D> {
264        AncestorBounds {
265            floor,
266            state: state(end, marker),
267        }
268    }
269
270    #[test]
271    fn validate_batch_applicable_accepts_valid_boundaries() {
272        let ancestors = vec![ancestor(loc(10), 12, 12), ancestor(loc(14), 16, 16)];
273        // Current matches the recorded DB state.
274        assert!(validate_batch_applicable::<F, D>(state(10, 1), state(10, 1), &ancestors).is_ok());
275        // Current matches one of the ancestor states.
276        assert!(validate_batch_applicable::<F, D>(state(16, 16), state(10, 1), &ancestors).is_ok());
277    }
278
279    #[test]
280    fn validate_batch_applicable_rejects_stale_batch() {
281        let ancestors = vec![ancestor(loc(10), 12, 12), ancestor(loc(14), 16, 16)];
282        let result = validate_batch_applicable::<F, D>(state(18, 18), state(10, 1), &ancestors);
283        assert!(matches!(result, Err(Error::StaleBatch)));
284    }
285
286    #[test]
287    fn validate_batch_applicable_rejects_equal_size_sibling() {
288        let ancestors = vec![ancestor(loc(14), 16, 16)];
289        let result = validate_batch_applicable::<F, D>(state(16, 99), state(10, 1), &ancestors);
290        assert!(matches!(result, Err(Error::StaleBatch)));
291    }
292
293    #[test]
294    fn ancestors_iterates_parent_first() {
295        let grandparent = Arc::new(TestBatch {
296            id: 1,
297            bounds: Bounds {
298                base: state(0, 0),
299                db: state(0, 0),
300                tip: state(5, 5),
301                ancestors: Vec::new(),
302                inactivity_floor: loc(3),
303            },
304            parent: None,
305        });
306        let parent = Arc::new(TestBatch {
307            id: 2,
308            bounds: Bounds {
309                base: state(5, 5),
310                db: state(0, 0),
311                tip: state(7, 7),
312                ancestors: vec![ancestor(loc(3), 5, 5)],
313                inactivity_floor: loc(6),
314            },
315            parent: Some(Arc::downgrade(&grandparent)),
316        });
317
318        let ids: Vec<_> = ancestors(Some(Arc::downgrade(&parent)), |batch| batch.parent.as_ref())
319            .map(|batch| batch.id)
320            .collect();
321
322        assert_eq!(ids, vec![2, 1]);
323    }
324
325    #[test]
326    fn collect_ancestor_bounds_preserves_pairing_and_order() {
327        let parent = Arc::new(TestBatch {
328            id: 1,
329            bounds: Bounds {
330                base: state(0, 0),
331                db: state(0, 0),
332                tip: state(12, 12),
333                ancestors: Vec::new(),
334                inactivity_floor: loc(10),
335            },
336            parent: None,
337        });
338        let grandparent = Arc::new(TestBatch {
339            id: 2,
340            bounds: Bounds {
341                base: state(0, 0),
342                db: state(0, 0),
343                tip: state(8, 8),
344                ancestors: Vec::new(),
345                inactivity_floor: loc(6),
346            },
347            parent: None,
348        });
349
350        let bounds = collect_ancestor_bounds(
351            vec![Arc::clone(&parent), Arc::clone(&grandparent)],
352            |batch| batch.bounds.inactivity_floor,
353            |batch| state(*batch.bounds.tip.size, *batch.bounds.tip.size as u8),
354        );
355
356        assert_eq!(bounds.len(), 2);
357        assert_eq!(bounds[0].floor, loc(10));
358        assert_eq!(bounds[0].state, state(12, 12));
359        assert_eq!(bounds[1].floor, loc(6));
360        assert_eq!(bounds[1].state, state(8, 8));
361    }
362
363    #[test]
364    fn bounds_validates_apply_to_current_state() {
365        let bounds = Bounds::<F, D> {
366            base: state(10, 1),
367            db: state(10, 1),
368            tip: state(14, 14),
369            ancestors: vec![ancestor(loc(10), 12, 12)],
370            inactivity_floor: loc(11),
371        };
372        assert!(bounds.validate_apply_to(state(10, 1), loc(9)).is_ok());
373
374        let result = bounds.validate_apply_to(state(11, 11), loc(9));
375        assert!(matches!(result, Err(Error::StaleBatch)));
376    }
377
378    #[test]
379    fn validate_commit_floors_accepts_monotonic_chain() {
380        let ancestors = vec![ancestor(loc(6), 7, 7), ancestor(loc(4), 5, 5)];
381        assert!(
382            validate_commit_floors::<F, D>(loc(2), loc(1), &ancestors, loc(8), loc(9),).is_ok()
383        );
384    }
385
386    #[test]
387    fn validate_commit_floors_skips_committed_ancestors() {
388        let ancestors = vec![ancestor(loc(1), 7, 7), ancestor(loc(1), 5, 5)];
389        assert!(
390            validate_commit_floors::<F, D>(loc(6), loc(7), &ancestors, loc(8), loc(9),).is_ok()
391        );
392    }
393
394    #[test]
395    fn validate_commit_floors_rejects_ancestor_regression() {
396        let ancestors = vec![ancestor(loc(6), 7, 7), ancestor(loc(3), 5, 5)];
397        let result = validate_commit_floors::<F, D>(loc(4), loc(1), &ancestors, loc(8), loc(9));
398        assert!(matches!(
399            result,
400            Err(Error::FloorRegressed(floor, previous)) if floor == loc(3) && previous == loc(4)
401        ));
402    }
403
404    #[test]
405    fn validate_commit_floors_rejects_ancestor_floor_beyond_commit() {
406        let ancestors = vec![ancestor(loc(8), 7, 7), ancestor(loc(4), 5, 5)];
407        let result = validate_commit_floors::<F, D>(loc(2), loc(1), &ancestors, loc(9), loc(9));
408        assert!(matches!(
409            result,
410            Err(Error::FloorBeyondSize(floor, commit)) if floor == loc(8) && commit == loc(6)
411        ));
412    }
413
414    #[test]
415    fn validate_commit_floors_rejects_tip_regression() {
416        let ancestors = vec![ancestor(loc(4), 5, 5)];
417        let result = validate_commit_floors::<F, D>(loc(2), loc(1), &ancestors, loc(3), loc(9));
418        assert!(matches!(
419            result,
420            Err(Error::FloorRegressed(floor, previous)) if floor == loc(3) && previous == loc(4)
421        ));
422    }
423
424    #[test]
425    fn validate_commit_floors_rejects_tip_floor_beyond_commit() {
426        let ancestors = vec![ancestor(loc(4), 5, 5)];
427        let result = validate_commit_floors::<F, D>(loc(2), loc(1), &ancestors, loc(10), loc(9));
428        assert!(matches!(
429            result,
430            Err(Error::FloorBeyondSize(floor, commit)) if floor == loc(10) && commit == loc(9)
431        ));
432    }
433}
commonware_storage/qmdb/batch_chain.rs

commonware_storage/qmdb/
batch_chain.rs
