commonware_sync/databases/
current.rs

1//! Current database types and helpers for the sync example.
2//!
3//! A `current` database extends an `any` database with an activity bitmap that tracks which
4//! operations are active (i.e. represent the current state of their key) vs inactive (superseded or
5//! deleted). Its canonical root folds the ops root, a grafted merkle root (combining bitmap chunks
6//! with ops subtree roots), and an optional partial-chunk digest. See [current] module
7//! documentation for more details.
8//!
9//! For sync, the engine targets the **ops root** (not the canonical root). The operations and proof
10//! format are identical to `any`; verify proofs directly with `qmdb::verify_proof`. The bitmap is
11//! reconstructed deterministically from the operations after sync completes. See the
12//! [Root structure](commonware_storage::qmdb::current) module documentation for details.
13//!
14//! This module re-uses the same [`Operation`] type as [`super::any`] since the underlying
15//! operations log is the same.
16
17use crate::{Hasher, Key, Translator, Value};
18use commonware_codec::FixedSize;
19use commonware_cryptography::{sha256, Hasher as CryptoHasher};
20use commonware_parallel::Sequential;
21use commonware_runtime::{buffer, BufferPooler};
22use commonware_storage::{
23    journal::contiguous::fixed::Config as FConfig,
24    mmr::{self, full::Config as MmrConfig, Location, Proof},
25    qmdb::{
26        self,
27        any::unordered::{fixed::Operation as FixedOperation, Update},
28        current::{self, FixedConfig as Config},
29        operation::Committable,
30    },
31    Context,
32};
33use commonware_utils::{NZUsize, NZU16, NZU64};
34use std::{future::Future, num::NonZeroU64};
35use tracing::error;
36
37/// Bitmap chunk size in bytes. Each chunk covers `N * 8` operations' activity bits.
38const CHUNK_SIZE: usize = sha256::Digest::SIZE;
39
40/// Database type alias.
41pub type Database<E> = current::unordered::fixed::Db<
42    mmr::Family,
43    E,
44    Key,
45    Value,
46    Hasher,
47    Translator,
48    CHUNK_SIZE,
49    Sequential,
50>;
51
52/// Operation type alias. Same as the `any` operation type.
53pub type Operation = FixedOperation<mmr::Family, Key, Value>;
54
55/// Create a database configuration.
56pub fn create_config(context: &impl BufferPooler) -> Config<Translator, Sequential> {
57    let page_cache = buffer::paged::CacheRef::from_pooler(context, NZU16!(2048), NZUsize!(10));
58    Config {
59        merkle_config: MmrConfig {
60            journal_partition: "mmr-journal".into(),
61            metadata_partition: "mmr-metadata".into(),
62            items_per_blob: NZU64!(4096),
63            write_buffer: NZUsize!(4096),
64            strategy: Sequential,
65            page_cache: page_cache.clone(),
66        },
67        journal_config: FConfig {
68            partition: "log-journal".into(),
69            items_per_blob: NZU64!(4096),
70            write_buffer: NZUsize!(4096),
71            page_cache,
72        },
73        grafted_metadata_partition: "grafted-mmr-metadata".into(),
74        translator: Translator::default(),
75        init_cache_size: Some(NZUsize!(1 << 16)),
76    }
77}
78
79impl<E> super::ExampleDatabase for Database<E>
80where
81    E: Context,
82{
83    type Family = mmr::Family;
84    type Operation = Operation;
85
86    fn create_test_operations(count: usize, seed: u64, _starting_loc: u64) -> Vec<Self::Operation> {
87        let mut hasher = <Hasher as CryptoHasher>::new();
88        let mut operations = Vec::new();
89        for i in 0..count {
90            let key = {
91                hasher.update(&i.to_be_bytes());
92                hasher.update(&seed.to_be_bytes());
93                hasher.finalize()
94            };
95
96            let value = {
97                hasher.update(&key);
98                hasher.update(b"value");
99                hasher.finalize()
100            };
101
102            operations.push(Operation::Update(Update(key, value)));
103
104            if (i + 1) % 10 == 0 {
105                operations.push(Operation::CommitFloor(None, Location::from(i + 1)));
106            }
107        }
108
109        // Always end with a commit.
110        operations.push(Operation::CommitFloor(None, Location::from(count)));
111        operations
112    }
113
114    async fn add_operations(
115        &mut self,
116        operations: Vec<Self::Operation>,
117    ) -> Result<(), qmdb::Error<mmr::Family>> {
118        if operations.last().is_none() || !operations.last().unwrap().is_commit() {
119            error!("operations must end with a commit");
120            return Ok(());
121        }
122
123        let mut batch = self.new_batch();
124        for operation in operations {
125            match operation {
126                Operation::Update(Update(key, value)) => {
127                    batch = batch.write(key, Some(value));
128                }
129                Operation::Delete(key) => {
130                    batch = batch.write(key, None);
131                }
132                Operation::CommitFloor(metadata, _) => {
133                    let merkleized = batch.merkleize(self, metadata).await?;
134                    self.apply_batch(merkleized).await?;
135                    self.commit().await?;
136                    batch = self.new_batch();
137                }
138            }
139        }
140        Ok(())
141    }
142
143    fn current_floor(&self) -> u64 {
144        // `current`'s `merkleize` derives the floor internally; the `starting_loc` passed to
145        // `create_test_operations` is unused, so any value is safe.
146        0
147    }
148
149    fn root(&self) -> Key {
150        // Return the ops root (not the canonical root) because this is what the
151        // sync engine verifies against.
152        self.ops_root()
153    }
154
155    fn name() -> &'static str {
156        "current"
157    }
158}
159
160impl<E> super::Syncable for Database<E>
161where
162    E: Context,
163{
164    fn size(&self) -> Location {
165        self.bounds().end
166    }
167
168    fn sync_boundary(&self) -> Location {
169        self.sync_boundary()
170    }
171
172    fn historical_proof(
173        &self,
174        op_count: Location,
175        start_loc: Location,
176        max_ops: NonZeroU64,
177    ) -> impl Future<Output = Result<(Proof<Key>, Vec<Self::Operation>), qmdb::Error<mmr::Family>>> + Send
178    {
179        // Return ops-level proofs (not grafted proofs) for the sync engine.
180        self.ops_historical_proof(op_count, start_loc, max_ops)
181    }
182
183    fn pinned_nodes_at(
184        &self,
185        loc: Location,
186    ) -> impl Future<Output = Result<Vec<Key>, qmdb::Error<mmr::Family>>> + Send {
187        self.pinned_nodes_at(loc)
188    }
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194    use crate::databases::ExampleDatabase;
195    use commonware_runtime::deterministic;
196
197    type CurrentDb = Database<deterministic::Context>;
198
199    #[test]
200    fn test_create_test_operations() {
201        let ops = <CurrentDb as ExampleDatabase>::create_test_operations(5, 12345, 0);
202        assert_eq!(ops.len(), 6); // 5 operations + 1 commit
203
204        if let Operation::CommitFloor(_, loc) = &ops[5] {
205            assert_eq!(*loc, 5);
206        } else {
207            panic!("last operation should be a commit");
208        }
209    }
210
211    #[test]
212    fn test_deterministic_operations() {
213        let ops1 = <CurrentDb as ExampleDatabase>::create_test_operations(3, 12345, 0);
214        let ops2 = <CurrentDb as ExampleDatabase>::create_test_operations(3, 12345, 0);
215        assert_eq!(ops1, ops2);
216
217        let ops3 = <CurrentDb as ExampleDatabase>::create_test_operations(3, 54321, 0);
218        assert_ne!(ops1, ops3);
219    }
220}
commonware_sync/databases/current.rs

commonware_sync/databases/
current.rs
