commonware_storage/qmdb/immutable/
variable.rs

1//! An immutable authenticated database with variable-size values.
2//!
3//! For fixed-size values, use [super::fixed] instead.
4
5use super::{Config as BaseConfig, Immutable, operation::Operation as BaseOperation};
6use crate::{
7    Context,
8    journal::{
9        authenticated,
10        contiguous::variable::{self, Config as JournalConfig},
11    },
12    merkle::Family,
13    qmdb::{
14        Error, ROOT_BAGGING,
15        any::{VariableValue, value::VariableEncoding},
16        operation::Key,
17    },
18    translator::Translator,
19};
20use commonware_codec::Read;
21use commonware_cryptography::Hasher;
22use commonware_parallel::Strategy;
23
24/// Type alias for a variable-size operation.
25pub type Operation<F, K, V> = BaseOperation<F, K, VariableEncoding<V>>;
26
27/// Type alias for the variable-size immutable database.
28pub type Db<F, E, K, V, H, T, S> =
29    Immutable<F, E, K, VariableEncoding<V>, variable::Journal<E, Operation<F, K, V>>, H, T, S>;
30
31/// Type alias for the variable-size compact immutable db.
32pub type CompactDb<F, E, K, V, H, C, S> = super::CompactDb<F, E, K, VariableEncoding<V>, H, C, S>;
33
34type Journal<F, E, K, V, H, S> =
35    authenticated::Journal<F, E, variable::Journal<E, Operation<F, K, V>>, H, S>;
36
37/// Configuration for a variable-size immutable authenticated db.
38pub type Config<T, C, S> = BaseConfig<T, JournalConfig<C>, S>;
39
40/// Configuration for a variable-size compact immutable db.
41pub type CompactConfig<C, S> = super::CompactConfig<C, S>;
42
43impl<F: Family, E: Context, K: Key, V: VariableValue, H: Hasher, T: Translator, S: Strategy>
44    Db<F, E, K, V, H, T, S>
45{
46    /// Returns a [Db] initialized from `cfg`. Any uncommitted log operations will be
47    /// discarded and the state of the db will be as of the last committed operation.
48    pub async fn init(
49        context: E,
50        cfg: Config<T, <Operation<F, K, V> as Read>::Cfg, S>,
51    ) -> Result<Self, Error<F>> {
52        let journal: Journal<F, E, K, V, H, S> = Journal::new(
53            context.child("journal"),
54            cfg.merkle_config,
55            cfg.log,
56            Operation::<F, K, V>::is_commit,
57            ROOT_BAGGING,
58        )
59        .await?;
60        Self::init_from_journal(journal, context, cfg.translator, cfg.init_buffer).await
61    }
62}
63
64impl<
65    F: Family,
66    E: Context,
67    K: Key,
68    V: VariableValue,
69    H: Hasher,
70    C: Clone + Send + Sync + 'static,
71    S: Strategy,
72> CompactDb<F, E, K, V, H, C, S>
73where
74    Operation<F, K, V>: Read<Cfg = C>,
75{
76    /// Returns a [CompactDb] initialized from `cfg`.
77    pub async fn init(context: E, cfg: CompactConfig<C, S>) -> Result<Self, Error<F>> {
78        let merkle = crate::merkle::compact::Merkle::new(cfg.strategy);
79        Self::init_from_merkle(
80            merkle,
81            context.child("witness"),
82            cfg.witness,
83            cfg.commit_codec_config,
84        )
85        .await
86    }
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92    use crate::{
93        journal::contiguous::variable::Config as JournalConfig,
94        merkle::{full::Config as MmrConfig, mmb, mmr},
95        qmdb::immutable::tests::{self, immutable_tests},
96        translator::TwoCap,
97    };
98    use commonware_cryptography::{Sha256, sha256::Digest};
99    use commonware_macros::{boxed, test_traced};
100    use commonware_parallel::Sequential;
101    use commonware_runtime::{
102        BufferPooler, Runner as _, Supervisor as _, buffer::paged::CacheRef, deterministic,
103    };
104    use commonware_utils::{NZU16, NZU64, NZUsize};
105    use core::{future::Future, pin::Pin};
106    use std::num::{NonZeroU16, NonZeroUsize};
107
108    const PAGE_SIZE: NonZeroU16 = NZU16!(77);
109    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(9);
110
111    fn config(suffix: &str, pooler: &impl BufferPooler) -> Config<TwoCap, ((), ()), Sequential> {
112        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
113        super::BaseConfig {
114            merkle_config: MmrConfig {
115                journal_partition: format!("journal-{suffix}"),
116                metadata_partition: format!("metadata-{suffix}"),
117                items_per_blob: NZU64!(11),
118                write_buffer: NZUsize!(1024),
119                replay_buffer: NZUsize!(1024),
120                strategy: Sequential,
121                page_cache: page_cache.clone(),
122            },
123            log: JournalConfig {
124                partition: format!("log-{suffix}"),
125                items_per_section: NZU64!(5),
126                compression: None,
127                codec_config: ((), ()),
128                page_cache,
129                write_buffer: NZUsize!(1024),
130                replay_buffer: NZUsize!(1024),
131            },
132            translator: TwoCap,
133            init_buffer: NZUsize!(1 << 21),
134        }
135    }
136
137    async fn open_db<F: Family>(
138        context: deterministic::Context,
139    ) -> Db<F, deterministic::Context, Digest, Digest, Sha256, TwoCap, Sequential> {
140        let cfg = config("partition", &context);
141        Db::init(context, cfg).await.unwrap()
142    }
143
144    async fn open_compact<F: Family>(
145        context: deterministic::Context,
146    ) -> CompactDb<F, deterministic::Context, Digest, Digest, Sha256, ((), ()), Sequential> {
147        let cfg = CompactConfig {
148            strategy: Sequential,
149            witness: crate::journal::contiguous::variable::Config {
150                partition: "compact-immutable-variable-witness".into(),
151                items_per_section: NZU64!(64),
152                compression: None,
153                codec_config: (),
154                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
155                write_buffer: NZUsize!(1024),
156                replay_buffer: NZUsize!(1024),
157            },
158            commit_codec_config: ((), ()),
159        };
160        CompactDb::init(context, cfg).await.unwrap()
161    }
162
163    #[allow(clippy::type_complexity)]
164    fn open<F: Family>(
165        ctx: deterministic::Context,
166    ) -> Pin<
167        Box<
168            dyn Future<
169                    Output = Db<
170                        F,
171                        deterministic::Context,
172                        Digest,
173                        Digest,
174                        Sha256,
175                        TwoCap,
176                        Sequential,
177                    >,
178                > + Send,
179        >,
180    > {
181        Box::pin(open_db::<F>(ctx))
182    }
183
184    fn is_send<T: Send>(_: T) {}
185
186    #[allow(dead_code)]
187    fn assert_db_futures_are_send(
188        db: Db<mmr::Family, deterministic::Context, Digest, Digest, Sha256, TwoCap, Sequential>,
189        key: Digest,
190        loc: crate::merkle::mmr::Location,
191    ) {
192        is_send(db.get(&key));
193        is_send(db.get_metadata());
194        is_send(db.proof(loc, NZU64!(1)));
195        is_send(db.sync());
196    }
197
198    #[allow(dead_code)]
199    fn assert_rewind_is_send(
200        db: Db<mmr::Family, deterministic::Context, Digest, Digest, Sha256, TwoCap, Sequential>,
201        loc: crate::merkle::mmr::Location,
202    ) {
203        is_send(db.rewind(loc));
204    }
205
206    fn small_sections_config(
207        suffix: &str,
208        pooler: &impl BufferPooler,
209    ) -> Config<TwoCap, ((), ()), Sequential> {
210        let mut cfg = config(suffix, pooler);
211        cfg.log.items_per_section = NZU64!(1);
212        cfg
213    }
214
215    async fn open_small_sections_db<F: Family>(
216        context: deterministic::Context,
217    ) -> Db<F, deterministic::Context, Digest, Digest, Sha256, TwoCap, Sequential> {
218        let cfg = small_sections_config("partition", &context);
219        Db::init(context, cfg).await.unwrap()
220    }
221
222    #[allow(clippy::type_complexity)]
223    fn open_small_sections<F: Family>(
224        ctx: deterministic::Context,
225    ) -> Pin<
226        Box<
227            dyn Future<
228                    Output = Db<
229                        F,
230                        deterministic::Context,
231                        Digest,
232                        Digest,
233                        Sha256,
234                        TwoCap,
235                        Sequential,
236                    >,
237                > + Send,
238        >,
239    > {
240        Box::pin(open_small_sections_db::<F>(ctx))
241    }
242
243    immutable_tests! {
244        test_variable_empty => run_empty, open;
245        test_variable_build_basic => run_build_basic, open;
246        test_variable_proof_verify => run_proof_verify, open;
247        test_variable_prune => run_prune, open;
248        test_variable_batch_chain => run_batch_chain, open;
249        test_variable_operations_match_applied_log => run_operations_match_applied_log, open;
250        test_variable_build_and_authenticate => run_build_and_authenticate, open;
251        test_variable_recovery_from_failed_merkle_sync => run_recovery_from_failed_merkle_sync, open;
252        test_variable_recovery_from_failed_log_sync => run_recovery_from_failed_log_sync, open;
253        test_variable_pruning => run_pruning, open;
254        test_variable_prune_beyond_floor => run_prune_beyond_floor, open;
255        test_variable_batch_get_read_through => run_batch_get_read_through, open;
256        test_variable_batch_stacked_get => run_batch_stacked_get, open;
257        test_variable_batch_stacked_apply => run_batch_stacked_apply, open;
258        test_variable_batch_speculative_root => run_batch_speculative_root, open;
259        test_variable_merkleized_batch_get => run_merkleized_batch_get, open;
260        test_variable_batch_sequential_apply => run_batch_sequential_apply, open;
261        test_variable_batch_many_sequential => run_batch_many_sequential, open;
262        test_variable_batch_empty_batch => run_batch_empty_batch, open;
263        test_variable_batch_chained_merkleized_get => run_batch_chained_merkleized_get, open;
264        test_variable_batch_large => run_batch_large, open;
265        test_variable_batch_chained_key_override => run_batch_chained_key_override, open;
266        test_variable_batch_sequential_key_override => run_batch_sequential_key_override, open_small_sections;
267        test_variable_batch_metadata => run_batch_metadata, open;
268        test_variable_stale_batch_rejected => run_stale_batch_rejected, open;
269        test_variable_stale_batch_chained => run_stale_batch_chained, open;
270        test_variable_sequential_commit_parent_then_child => run_sequential_commit_parent_then_child, open;
271        test_variable_stale_batch_child_applied_before_parent => run_stale_batch_child_applied_before_parent, open;
272        test_variable_child_root_matches_pending_and_committed => run_child_root_matches_pending_and_committed, open;
273        test_variable_to_batch => run_to_batch, open;
274        test_variable_rewind_recovery => run_rewind_recovery, open;
275        test_variable_rewind_pruned_target_errors => run_rewind_pruned_target_errors, open_small_sections;
276        test_variable_inactivity_floor_tracking => run_inactivity_floor_tracking, open;
277        test_variable_floor_monotonicity => run_floor_monotonicity, open;
278        test_variable_floor_monotonicity_violation => run_floor_monotonicity_violation, open;
279        test_variable_floor_beyond_size => run_floor_beyond_size, open;
280        test_variable_chained_ancestor_floor_regression => run_chained_ancestor_floor_regression, open;
281        test_variable_chained_ancestor_floor_beyond_size => run_chained_ancestor_floor_beyond_size, open;
282        test_variable_rewind_restores_floor => run_rewind_restores_floor, open;
283        test_variable_single_commit_live_set => run_single_commit_live_set, open;
284        test_variable_rewind_after_reopen_with_floor_change => run_rewind_after_reopen_with_floor_change, open;
285        test_variable_rewind_after_reopen_partial_floor_gap => run_rewind_after_reopen_partial_floor_gap, open;
286        test_variable_commit_after_sync_recovery => run_commit_after_sync_recovery, open;
287        test_variable_partial_ancestor_commit => run_partial_ancestor_commit, open;
288        test_variable_delayed_merkleize_after_ancestor_apply => run_delayed_merkleize_after_ancestor_apply, open;
289        test_variable_get_many => run_get_many, open;
290        test_variable_get_many_unexpected_data => run_get_many_unexpected_data, open;
291        test_variable_apply_after_ancestor_dropped => run_apply_after_ancestor_dropped, open;
292        test_variable_rewind_preserves_collision_bucket => run_rewind_preserves_collision_bucket, open;
293        test_variable_rewind_after_reopen_repeated_key_gap => run_rewind_after_reopen_repeated_key_gap, open;
294        test_variable_rewind_after_reopen_mixed_gap_retained => run_rewind_after_reopen_mixed_gap_retained, open;
295        test_variable_rewind_repeated_key_live => run_rewind_repeated_key_live, open;
296        test_variable_rewind_after_reopen_repeated_key_retained => run_rewind_after_reopen_repeated_key_retained, open;
297    }
298
299    #[boxed]
300    async fn assert_compact_root_compatibility<F: Family>(ctx: deterministic::Context) {
301        let db = open_db::<F>(ctx.child("db")).await;
302        let compact = open_compact::<F>(ctx.child("compact")).await;
303        assert_eq!(db.root(), compact.root());
304
305        let k1 = Sha256::fill(1u8);
306        let v1 = Sha256::fill(11u8);
307        let k2 = Sha256::fill(2u8);
308        let v2 = Sha256::fill(22u8);
309        let metadata = Sha256::fill(99u8);
310
311        let floor = db.inactivity_floor_loc();
312        let retained = db
313            .new_batch()
314            .set(k1, v1)
315            .set(k2, v2)
316            .merkleize(&db, Some(metadata), floor)
317            .await;
318        let compact_batch = compact
319            .new_batch()
320            .set(k1, v1)
321            .set(k2, v2)
322            .merkleize(&compact, Some(metadata), floor)
323            .await;
324
325        assert_eq!(retained.root(), compact_batch.root());
326
327        let (db, _) = db.apply_batch(retained).await.unwrap();
328        let (compact, _) = compact.apply_batch(compact_batch).await.unwrap();
329        let db = db.commit().await.unwrap();
330        let compact = compact.sync().await.unwrap();
331
332        assert_eq!(db.root(), compact.root());
333        assert_eq!(compact.get_metadata(), Some(metadata));
334
335        drop(compact);
336        let reopened = open_compact::<F>(ctx.child("reopen")).await;
337        assert_eq!(db.root(), reopened.root());
338        assert_eq!(reopened.get_metadata(), Some(metadata));
339
340        reopened.destroy().await.unwrap();
341        db.destroy().await.unwrap();
342    }
343
344    #[test_traced("INFO")]
345    fn test_variable_compact_root_compatibility() {
346        let executor = deterministic::Runner::default();
347        executor.start(|ctx| async move {
348            assert_compact_root_compatibility::<mmr::Family>(ctx).await;
349        });
350    }
351
352    #[test_traced("INFO")]
353    fn test_variable_compact_root_compatibility_mmb() {
354        let executor = deterministic::Runner::default();
355        executor.start(|ctx| async move {
356            assert_compact_root_compatibility::<mmb::Family>(ctx).await;
357        });
358    }
359}
commonware_storage/qmdb/immutable/variable.rs

commonware_storage/qmdb/immutable/
variable.rs
