commonware_sync/databases/
immutable.rs

1//! Immutable database types and helpers for the sync example.
2
3use crate::{Hasher, Key, Translator, Value};
4use commonware_cryptography::{Hasher as CryptoHasher, Sha256};
5use commonware_parallel::Sequential;
6use commonware_runtime::BufferPooler;
7use commonware_storage::{
8    journal::contiguous::fixed::Config as FConfig,
9    merkle::{
10        full::Config as MmrConfig,
11        mmr::{self, Location, Proof},
12    },
13    qmdb::{
14        self,
15        immutable::{fixed, Config},
16        sync::compact,
17    },
18    Context,
19};
20use commonware_utils::{NZUsize, NZU16, NZU64};
21use std::{future::Future, num::NonZeroU64};
22use tracing::error;
23
24/// Database type alias.
25pub type Database<E> = fixed::Db<mmr::Family, E, Key, Value, Hasher, Translator, Sequential>;
26
27/// Operation type alias.
28pub type Operation = fixed::Operation<mmr::Family, Key, Value>;
29
30/// Create a database configuration with appropriate partitioning for Immutable.
31pub fn create_config(context: &impl BufferPooler) -> Config<Translator, FConfig, Sequential> {
32    let page_cache = commonware_runtime::buffer::paged::CacheRef::from_pooler(
33        context,
34        NZU16!(2048),
35        NZUsize!(10),
36    );
37    Config {
38        merkle_config: MmrConfig {
39            journal_partition: "mmr-journal".into(),
40            metadata_partition: "mmr-metadata".into(),
41            items_per_blob: NZU64!(4096),
42            write_buffer: NZUsize!(4096),
43            strategy: Sequential,
44            page_cache: page_cache.clone(),
45        },
46        log: FConfig {
47            partition: "log".into(),
48            items_per_blob: NZU64!(4096),
49            write_buffer: NZUsize!(4096),
50            page_cache,
51        },
52        translator: commonware_storage::translator::EightCap,
53        init_cache_size: Some(NZUsize!(1 << 16)),
54    }
55}
56
57/// Create deterministic test operations for demonstration purposes.
58///
59/// Generates Set operations and periodic Commit operations. Every commit in the stream
60/// carries `starting_loc` as its inactivity floor. Pass `0` for a fresh db; for growth, pass
61/// the live db's [`super::ExampleDatabase::current_floor`] so floors stay monotonic.
62pub fn create_test_operations(count: usize, seed: u64, starting_loc: u64) -> Vec<Operation> {
63    let mut operations = Vec::new();
64    let mut hasher = <Hasher as CryptoHasher>::new();
65    let floor = Location::new(starting_loc);
66
67    for i in 0..count {
68        let key = {
69            hasher.update(&i.to_be_bytes());
70            hasher.update(&seed.to_be_bytes());
71            hasher.finalize()
72        };
73
74        let value = {
75            hasher.update(&key);
76            hasher.update(b"value");
77            hasher.finalize()
78        };
79
80        operations.push(Operation::Set(key, value));
81
82        if (i + 1) % 10 == 0 {
83            operations.push(Operation::Commit(None, floor));
84        }
85    }
86
87    // Always end with a commit
88    operations.push(Operation::Commit(Some(Sha256::fill(1)), floor));
89    operations
90}
91
92impl<E> super::ExampleDatabase for Database<E>
93where
94    E: Context,
95{
96    type Family = mmr::Family;
97    type Operation = Operation;
98
99    fn create_test_operations(count: usize, seed: u64, starting_loc: u64) -> Vec<Self::Operation> {
100        create_test_operations(count, seed, starting_loc)
101    }
102
103    async fn add_operations(
104        &mut self,
105        operations: Vec<Self::Operation>,
106    ) -> Result<(), commonware_storage::qmdb::Error<mmr::Family>> {
107        if operations.last().is_none() || !operations.last().unwrap().is_commit() {
108            // Ignore bad inputs rather than return errors.
109            error!("operations must end with a commit");
110            return Ok(());
111        }
112
113        let mut batch = self.new_batch();
114        for operation in operations {
115            match operation {
116                Operation::Set(key, value) => {
117                    batch = batch.set(key, value);
118                }
119                Operation::Commit(metadata, floor) => {
120                    let merkleized = batch.merkleize(self, metadata, floor).await;
121                    self.apply_batch(merkleized).await?;
122                    self.commit().await?;
123                    batch = self.new_batch();
124                }
125            }
126        }
127        Ok(())
128    }
129
130    fn current_floor(&self) -> u64 {
131        *self.inactivity_floor_loc()
132    }
133
134    fn root(&self) -> Key {
135        self.root()
136    }
137
138    fn name() -> &'static str {
139        "immutable"
140    }
141}
142
143impl<E> super::Syncable for Database<E>
144where
145    E: Context,
146{
147    fn size(&self) -> Location {
148        self.bounds().end
149    }
150
151    fn sync_boundary(&self) -> Location {
152        self.sync_boundary()
153    }
154
155    fn historical_proof(
156        &self,
157        op_count: Location,
158        start_loc: Location,
159        max_ops: NonZeroU64,
160    ) -> impl Future<Output = Result<(Proof<Key>, Vec<Self::Operation>), qmdb::Error<mmr::Family>>> + Send
161    {
162        self.historical_proof(op_count, start_loc, max_ops)
163    }
164
165    fn pinned_nodes_at(
166        &self,
167        loc: Location,
168    ) -> impl Future<Output = Result<Vec<Key>, qmdb::Error<mmr::Family>>> + Send {
169        self.pinned_nodes_at(loc)
170    }
171}
172
173impl<E> super::CompactSyncable for Database<E>
174where
175    E: Context,
176{
177    fn target(&self) -> compact::Target<Self::Family, Key> {
178        compact::Target::new(self.root(), self.bounds().end)
179    }
180}
commonware_sync/databases/immutable.rs

commonware_sync/databases/
immutable.rs
