commonware_storage/qmdb/keyless/
fixed.rs

1//! A keyless authenticated database for fixed-size data.
2//!
3//! For variable-size values, use [super::variable].
4
5use crate::{
6    Context,
7    journal::{
8        authenticated,
9        contiguous::fixed::{self, Config as JournalConfig},
10    },
11    merkle::Family,
12    qmdb::{
13        Error, ROOT_BAGGING,
14        any::value::{FixedEncoding, FixedValue},
15        keyless::operation::Operation as BaseOperation,
16        operation::Committable,
17    },
18};
19use commonware_cryptography::Hasher;
20use commonware_parallel::Strategy;
21
22/// Keyless operation for fixed-size values.
23pub type Operation<F, V> = BaseOperation<F, FixedEncoding<V>>;
24
25/// A keyless authenticated database for fixed-size data.
26pub type Db<F, E, V, H, S> =
27    super::Keyless<F, E, FixedEncoding<V>, fixed::Journal<E, Operation<F, V>>, H, S>;
28
29/// A compact keyless authenticated db for fixed-size data.
30pub type CompactDb<F, E, V, H, S> = super::CompactDb<F, E, FixedEncoding<V>, H, (), S>;
31
32type Journal<F, E, V, H, S> =
33    authenticated::Journal<F, E, fixed::Journal<E, Operation<F, V>>, H, S>;
34
35/// Configuration for a fixed-size [keyless](super) authenticated db.
36pub type Config<S> = super::Config<JournalConfig, S>;
37
38/// Configuration for a fixed-size [keyless](super) compact db.
39pub type CompactConfig<S> = super::CompactConfig<(), S>;
40
41impl<F: Family, E: Context, V: FixedValue, H: Hasher, S: Strategy> Db<F, E, V, H, S> {
42    /// Returns a [Db] initialized from `cfg`. Any uncommitted operations will be
43    /// discarded and the state of the db will be as of the last committed operation.
44    pub async fn init(context: E, cfg: Config<S>) -> Result<Self, Error<F>> {
45        let journal: Journal<F, E, V, H, S> = Journal::new(
46            context.child("journal"),
47            cfg.merkle,
48            cfg.log,
49            Operation::<F, V>::is_commit,
50            ROOT_BAGGING,
51        )
52        .await?;
53        Self::init_from_journal(journal, context).await
54    }
55}
56
57impl<F: Family, E: Context, V: FixedValue, H: Hasher, S: Strategy> CompactDb<F, E, V, H, S> {
58    /// Returns a [CompactDb] initialized from `cfg`.
59    pub async fn init(context: E, cfg: CompactConfig<S>) -> Result<Self, Error<F>> {
60        let merkle = crate::merkle::compact::Merkle::new(cfg.strategy);
61        Self::init_from_merkle(merkle, context.child("witness"), cfg.witness, ()).await
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68    use crate::{
69        merkle::{Location, mmb, mmr},
70        qmdb::keyless::tests::{self, keyless_tests},
71    };
72    use commonware_cryptography::Sha256;
73    use commonware_macros::{boxed, test_traced};
74    use commonware_parallel::{Rayon, Sequential, Strategy};
75    use commonware_runtime::{
76        BufferPooler, Metrics as _, Runner as _, Spawner as _, Strategizer as _, Supervisor as _,
77        buffer::paged::CacheRef,
78        deterministic,
79        mocks::{DelayedSyncContext, PendingSyncs, drive_pending_syncs},
80        reschedule,
81    };
82    use commonware_utils::{NZU16, NZU64, NZUsize, sequence::U64};
83    use core::future::Future;
84    use futures::FutureExt as _;
85    use std::num::{NonZeroU16, NonZeroUsize};
86
87    const PAGE_SIZE: NonZeroU16 = NZU16!(101);
88    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(11);
89
90    fn db_config<S: Strategy>(suffix: &str, pooler: &impl BufferPooler, strategy: S) -> Config<S> {
91        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
92        Config {
93            merkle: crate::merkle::full::Config {
94                journal_partition: format!("fixed-journal-{suffix}"),
95                metadata_partition: format!("fixed-metadata-{suffix}"),
96                items_per_blob: NZU64!(11),
97                write_buffer: NZUsize!(1024),
98                replay_buffer: NZUsize!(1024),
99                strategy,
100                page_cache: page_cache.clone(),
101            },
102            log: JournalConfig {
103                partition: format!("fixed-log-journal-{suffix}"),
104                items_per_blob: NZU64!(7),
105                page_cache,
106                write_buffer: NZUsize!(1024),
107                replay_buffer: NZUsize!(1024),
108            },
109        }
110    }
111
112    type TestDb<F> =
113        Db<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Sequential>;
114    type TestRayonDb<F> =
115        Db<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Rayon>;
116    type TestCompactDb<F> =
117        CompactDb<F, deterministic::Context, commonware_utils::sequence::U64, Sha256, Sequential>;
118
119    async fn open_db<F: Family>(context: deterministic::Context) -> TestDb<F> {
120        open_db_with_suffix("partition", context).await
121    }
122
123    async fn open_db_with_suffix<F: Family>(
124        suffix: &str,
125        context: deterministic::Context,
126    ) -> TestDb<F> {
127        let cfg = db_config(suffix, &context, Sequential);
128        TestDb::init(context, cfg).await.unwrap()
129    }
130
131    async fn open_rayon_db<F: Family>(context: deterministic::Context) -> TestRayonDb<F> {
132        let strategy = context.strategy(NZUsize!(2));
133        let cfg = db_config("rayon", &context, strategy);
134        TestRayonDb::init(context, cfg).await.unwrap()
135    }
136
137    async fn open_compact<F: crate::merkle::Family>(
138        context: deterministic::Context,
139    ) -> TestCompactDb<F> {
140        let cfg = CompactConfig {
141            strategy: Sequential,
142            witness: crate::journal::contiguous::variable::Config {
143                partition: "compact-keyless-fixed-witness".into(),
144                items_per_section: NZU64!(64),
145                compression: None,
146                codec_config: (),
147                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
148                write_buffer: NZUsize!(1024),
149                replay_buffer: NZUsize!(1024),
150            },
151            commit_codec_config: (),
152        };
153        TestCompactDb::init(context, cfg).await.unwrap()
154    }
155
156    fn reopen<F: Family>() -> tests::Reopen<TestDb<F>> {
157        Box::new(|ctx| Box::pin(open_db(ctx)))
158    }
159
160    /// A keyless db over a delayed-sync storage backend.
161    type DelayedDb =
162        Db<mmr::Family, DelayedSyncContext<deterministic::Context>, U64, Sha256, Sequential>;
163
164    /// Open a [DelayedDb] whose blob syncs park on `pending`.
165    ///
166    /// Init durably persists the recovered database, so while syncs park the returned future
167    /// must be driven with [drive_pending_syncs] (or the mock unblocked first). The journal
168    /// uses large pages and blobs: an apply that fills the write buffer or rolls the blob over
169    /// waits for the in-flight sync, so mid-sync applies must stay clear of both.
170    fn open_delayed_db(
171        context: &deterministic::Context,
172        label: &'static str,
173        suffix: &str,
174        pending: &PendingSyncs,
175    ) -> impl Future<Output = Result<DelayedDb, Error<mmr::Family>>> {
176        let mut cfg = db_config(suffix, context, Sequential);
177        let page_cache = CacheRef::from_pooler(context, NZU16!(1024), NZUsize!(8));
178        cfg.log.items_per_blob = NZU64!(1000);
179        cfg.log.page_cache = page_cache.clone();
180        cfg.merkle.items_per_blob = NZU64!(1000);
181        cfg.merkle.page_cache = page_cache;
182        DelayedDb::init(
183            DelayedSyncContext {
184                inner: context.child(label),
185                pending: pending.clone(),
186            },
187            cfg,
188        )
189    }
190
191    /// Apply a single-append batch with inactivity floor `floor`, returning the appended
192    /// value's location.
193    async fn apply_append(
194        db: DelayedDb,
195        value: U64,
196        floor: Location<mmr::Family>,
197    ) -> (DelayedDb, Location<mmr::Family>) {
198        let batch = db
199            .new_batch()
200            .append(value)
201            .merkleize(&db, None, floor)
202            .await;
203        let (db, range) = db.apply_batch(batch).await.unwrap();
204        (db, range.start)
205    }
206
207    /// A sync handle must not block database use while the backend sync is pending.
208    #[test_traced]
209    fn test_keyless_fixed_start_sync_overlaps_work() {
210        deterministic::Runner::default().start(|ctx| async move {
211            let pending = PendingSyncs::default();
212            let open = open_delayed_db(&ctx, "delayed", "start-sync-overlap", &pending);
213            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
214            let value0 = U64::new(1);
215            let loc0;
216            let floor = db.inactivity_floor_loc();
217            (db, loc0) = apply_append(db, value0.clone(), floor).await;
218
219            let starts_before = pending.starts();
220            let entered_before = pending.entered();
221            let completions_before = pending.completions();
222            let handle;
223            (db, handle) = db.start_sync().await.unwrap();
224            assert!(pending.starts() > starts_before);
225            assert_eq!(pending.completions(), completions_before);
226
227            // Observe the sync while the database keeps working.
228            let waiter = ctx
229                .child("await_sync")
230                .spawn(|_| async move { handle.await.unwrap() });
231            while pending.entered() == entered_before {
232                reschedule().await;
233            }
234
235            // Reads and applies complete before the sync does.
236            assert_eq!(db.get(loc0).await.unwrap(), Some(value0));
237            let value1 = U64::new(2);
238            let loc1;
239            let floor = db.inactivity_floor_loc();
240            (db, loc1) = apply_append(db, value1.clone(), floor).await;
241            assert_eq!(
242                pending.completions(),
243                completions_before,
244                "the database made progress while the sync was still in flight"
245            );
246
247            pending.unblock();
248            waiter.await.unwrap();
249
250            // The mid-sync batch is durable after the next start_sync completes.
251            let handle;
252            (db, handle) = db.start_sync().await.unwrap();
253            handle.await.unwrap();
254            let root = db.root();
255            drop(db);
256
257            let db = open_delayed_db(&ctx, "reopen", "start-sync-overlap", &pending)
258                .await
259                .unwrap();
260            assert_eq!(db.root(), root);
261            assert_eq!(db.get(loc1).await.unwrap(), Some(value1));
262            db.destroy().await.unwrap();
263        });
264    }
265
266    /// A sync begun by `start_sync` that fails in flight surfaces the error through both the
267    /// returned handle and the next durability operation.
268    #[test_traced]
269    fn test_keyless_fixed_start_sync_failure_propagates() {
270        deterministic::Runner::default().start(|ctx| async move {
271            // Pass syncs through so opening the database doesn't park.
272            let pending = PendingSyncs::default();
273            pending.unblock();
274            let mut db = open_delayed_db(&ctx, "delayed", "start-sync-fail", &pending)
275                .await
276                .unwrap();
277            let floor = db.inactivity_floor_loc();
278            (db, _) = apply_append(db, U64::new(1), floor).await;
279
280            // Arm all future syncs to resolve to an injected error.
281            pending.arm_fail();
282
283            let handle;
284            (db, handle) = db.start_sync().await.unwrap();
285            assert!(
286                handle.await.is_err(),
287                "the sync handle surfaces the failure"
288            );
289            let starts_before = pending.starts();
290            // A failed mutable method consumes the database per the failures-are-fatal contract.
291            assert!(
292                db.commit().await.is_err(),
293                "the next durability op surfaces the failed in-flight sync"
294            );
295            assert_eq!(
296                pending.starts(),
297                starts_before,
298                "the surfaced error is the retained failure, not a fresh sync's"
299            );
300        });
301    }
302
303    /// State persisted via an awaited start_sync handle is recovered on reopen.
304    #[test_traced]
305    fn test_keyless_fixed_start_sync_recovery() {
306        deterministic::Runner::default().start(|ctx| async move {
307            let pending = PendingSyncs::default();
308            pending.unblock();
309            let mut db = open_delayed_db(&ctx, "delayed", "start-sync-recovery", &pending)
310                .await
311                .unwrap();
312            let value = U64::new(1);
313            let loc;
314            let floor = db.inactivity_floor_loc();
315            (db, loc) = apply_append(db, value.clone(), floor).await;
316
317            let handle;
318            (db, handle) = db.start_sync().await.unwrap();
319            handle.await.unwrap();
320            let root = db.root();
321            drop(db);
322
323            let db = open_delayed_db(&ctx, "reopen", "start-sync-recovery", &pending)
324                .await
325                .unwrap();
326            assert_eq!(db.root(), root);
327            assert_eq!(db.get(loc).await.unwrap(), Some(value));
328            db.destroy().await.unwrap();
329        });
330    }
331
332    /// Pruning drains the in-flight sync before mutating storage.
333    #[test_traced]
334    fn test_keyless_fixed_start_sync_prune_waits() {
335        deterministic::Runner::default().start(|ctx| async move {
336            let pending = PendingSyncs::default();
337            let open = open_delayed_db(&ctx, "delayed", "start-sync-prune", &pending);
338            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
339            // Two batches: the second declares floor 2 so the prune below is non-trivial.
340            (db, _) = apply_append(db, U64::new(1), Location::new(0)).await;
341            (db, _) = apply_append(db, U64::new(2), Location::new(2)).await;
342
343            let starts_before = pending.starts();
344            let handle;
345            (db, handle) = db.start_sync().await.unwrap();
346            assert!(pending.starts() > starts_before);
347
348            let floor = db.inactivity_floor_loc();
349            assert!(*floor > 0);
350            let db = {
351                let mut prune = std::pin::pin!(db.prune(floor));
352                assert!(
353                    prune.as_mut().now_or_never().is_none(),
354                    "prune proceeded while the started sync was pending"
355                );
356                pending.unblock();
357                prune.await.unwrap()
358            };
359            handle.await.unwrap();
360            db.destroy().await.unwrap();
361        });
362    }
363
364    /// Rewinding drains the in-flight sync before mutating storage.
365    #[test_traced]
366    fn test_keyless_fixed_start_sync_rewind_waits() {
367        deterministic::Runner::default().start(|ctx| async move {
368            let pending = PendingSyncs::default();
369            let open = open_delayed_db(&ctx, "delayed", "start-sync-rewind", &pending);
370            let mut db = drive_pending_syncs(&pending, open).await.unwrap();
371            (db, _) = apply_append(db, U64::new(1), Location::new(0)).await;
372            db = drive_pending_syncs(&pending, db.commit()).await.unwrap();
373            let committed_root = db.root();
374            let committed_size = db.bounds().end;
375            (db, _) = apply_append(db, U64::new(2), Location::new(0)).await;
376
377            let handle;
378            (db, handle) = db.start_sync().await.unwrap();
379
380            let db = {
381                let mut rewind = std::pin::pin!(db.rewind(committed_size));
382                assert!(
383                    rewind.as_mut().now_or_never().is_none(),
384                    "rewind proceeded while the started sync was pending"
385                );
386                pending.unblock();
387                rewind.await.unwrap()
388            };
389            handle.await.unwrap();
390            assert_eq!(db.root(), committed_root);
391            db.destroy().await.unwrap();
392        });
393    }
394
395    #[test_traced("INFO")]
396    fn test_keyless_fixed_metrics() {
397        deterministic::Runner::default().start(|ctx| async move {
398            let db = open_db::<mmr::Family>(ctx.child("db")).await;
399            let value = commonware_utils::sequence::U64::new(7);
400            let floor = db.inactivity_floor_loc();
401            let batch = db
402                .new_batch()
403                .append(value.clone())
404                .merkleize(&db, None, floor)
405                .await;
406            let (db, range) = db.apply_batch(batch).await.unwrap();
407            assert_eq!(db.get(range.start).await.unwrap(), Some(value.clone()));
408            assert_eq!(
409                db.get_many(&[range.start]).await.unwrap(),
410                vec![Some(value)]
411            );
412            let db = db.commit().await.unwrap();
413            let db = db.sync().await.unwrap();
414            let (db, handle) = db.start_sync().await.unwrap();
415            handle.await.unwrap();
416            let _db = db.prune(crate::merkle::Location::new(0)).await.unwrap();
417
418            let metrics = ctx.encode();
419            for expected in [
420                "db_size 3",
421                "db_pruning_boundary 0",
422                "db_retained 3",
423                "db_inactivity_floor 0",
424                "db_last_commit 2",
425                "db_get_calls_total 1",
426                "db_get_many_calls_total 1",
427                "db_lookups_requested_total 2",
428                "db_apply_batch_calls_total 1",
429                "db_operations_applied_total 2",
430                "db_commit_calls_total 1",
431                "db_sync_calls_total 1",
432                "db_start_sync_calls_total 1",
433                "db_prune_calls_total 1",
434                "db_get_duration_count 1",
435                "db_get_many_duration_count 1",
436                "db_apply_batch_duration_count 1",
437                "db_commit_duration_count 1",
438                "db_sync_duration_count 1",
439                "db_prune_duration_count 1",
440            ] {
441                assert!(metrics.contains(expected), "missing {expected}\n{metrics}");
442            }
443        });
444    }
445
446    keyless_tests! {
447        test_keyless_fixed_empty => run_empty, reopen_indexed;
448        test_keyless_fixed_build_basic => run_build_basic, reopen_indexed;
449        test_keyless_fixed_recovery => run_recovery, reopen_indexed;
450        test_keyless_fixed_non_empty_recovery => run_non_empty_recovery, reopen_indexed;
451        test_keyless_fixed_proof => run_proof, db;
452        test_keyless_fixed_proof_comprehensive => run_proof_comprehensive, db;
453        test_keyless_fixed_proof_with_pruning => run_proof_with_pruning, reopen_indexed;
454        test_keyless_fixed_empty_db_recovery => run_empty_db_recovery, reopen_indexed;
455        test_keyless_fixed_replay_with_trailing_appends => run_replay_with_trailing_appends, reopen_indexed;
456        test_keyless_fixed_get_out_of_bounds => run_get_out_of_bounds, db;
457        test_keyless_fixed_metadata => run_metadata, db;
458        test_keyless_fixed_pruning => run_pruning, reopen;
459        test_keyless_fixed_batch_get => run_batch_get, db;
460        test_keyless_fixed_batch_stacked_get => run_batch_stacked_get, db;
461        test_keyless_fixed_batch_speculative_root => run_batch_speculative_root, db;
462        test_keyless_fixed_merkleized_batch_get => run_merkleized_batch_get, db;
463        test_keyless_fixed_batch_chained => run_batch_chained, db;
464        test_keyless_fixed_operations_match_applied_log => run_operations_match_applied_log, db;
465        test_keyless_fixed_batch_chained_apply_sequential => run_batch_chained_apply_sequential, db;
466        test_keyless_fixed_batch_many_sequential => run_batch_many_sequential, db;
467        test_keyless_fixed_batch_empty => run_batch_empty, db;
468        test_keyless_fixed_batch_chained_merkleized_get => run_batch_chained_merkleized_get, db;
469        test_keyless_fixed_batch_large => run_batch_large, db;
470        test_keyless_fixed_stale_batch => run_stale_batch, reopen;
471        test_keyless_fixed_stale_batch_chained => run_stale_batch_chained, db;
472        test_keyless_fixed_sequential_commit_parent_then_child => run_sequential_commit_parent_then_child, db;
473        test_keyless_fixed_stale_batch_child_before_parent => run_stale_batch_child_before_parent, db;
474        test_keyless_fixed_to_batch => run_to_batch, db;
475        test_keyless_fixed_child_root_matches_pending_and_committed => run_child_root_matches_pending_and_committed, db;
476        test_keyless_fixed_rewind_recovery => run_rewind_recovery, reopen;
477        test_keyless_fixed_rewind_pruned_target_errors => run_rewind_pruned_target_errors, reopen;
478        test_keyless_fixed_floor_tracking => run_floor_tracking, reopen_indexed;
479        test_keyless_fixed_floor_regression_rejected => run_floor_regression_rejected, reopen;
480        test_keyless_fixed_floor_beyond_commit_loc_rejected => run_floor_beyond_commit_loc_rejected, reopen;
481        test_keyless_fixed_rewind_restores_floor => run_rewind_restores_floor, db;
482        test_keyless_fixed_floor_at_commit_loc_accepted => run_floor_at_commit_loc_accepted, db;
483        test_keyless_fixed_rewind_after_reopen_with_floor => run_rewind_after_reopen_with_floor, reopen_indexed;
484        test_keyless_fixed_ancestor_floor_regression_rejected => run_ancestor_floor_regression_rejected, reopen;
485        test_keyless_fixed_ancestor_floor_beyond_commit_loc_rejected => run_ancestor_floor_beyond_commit_loc_rejected, db;
486        test_keyless_fixed_chained_apply_with_valid_floors_succeeds => run_chained_apply_with_valid_floors_succeeds, db;
487        test_keyless_fixed_single_commit_live_set => run_single_commit_live_set, reopen_indexed;
488        test_keyless_fixed_commit_after_sync_recovery => run_commit_after_sync_recovery, reopen_indexed;
489        test_keyless_fixed_get_many => run_get_many, db;
490    }
491
492    #[test_traced("INFO")]
493    fn test_keyless_fixed_shared_helper_accepts_rayon_strategy() {
494        deterministic::Runner::default().start(|ctx| async move {
495            let db = open_rayon_db::<mmr::Family>(ctx.child("db").with_attribute("index", 1)).await;
496            tests::run_metadata(db).await;
497        });
498    }
499
500    #[boxed]
501    async fn assert_compact_root_compatibility<F: crate::merkle::Family>(
502        ctx: deterministic::Context,
503    ) {
504        let db = open_db::<F>(ctx.child("db")).await;
505        let compact = open_compact::<F>(ctx.child("compact")).await;
506        assert_eq!(db.root(), compact.root());
507
508        let v1 = commonware_utils::sequence::U64::new(1);
509        let v2 = commonware_utils::sequence::U64::new(2);
510        let metadata = commonware_utils::sequence::U64::new(99);
511
512        let floor = db.inactivity_floor_loc();
513        let retained = db
514            .new_batch()
515            .append(v1.clone())
516            .append(v2.clone())
517            .merkleize(&db, Some(metadata.clone()), floor)
518            .await;
519        let compact_batch = compact
520            .new_batch()
521            .append(v1)
522            .append(v2)
523            .merkleize(&compact, Some(metadata.clone()), floor)
524            .await;
525
526        assert_eq!(retained.root(), compact_batch.root());
527
528        let (db, _) = db.apply_batch(retained).await.unwrap();
529        let (compact, _) = compact.apply_batch(compact_batch).await.unwrap();
530        let db = db.commit().await.unwrap();
531        let compact = compact.sync().await.unwrap();
532
533        assert_eq!(db.root(), compact.root());
534        assert_eq!(compact.get_metadata(), Some(metadata.clone()));
535
536        drop(compact);
537        let reopened = open_compact::<F>(ctx.child("reopen")).await;
538        assert_eq!(db.root(), reopened.root());
539        assert_eq!(reopened.get_metadata(), Some(metadata));
540
541        reopened.destroy().await.unwrap();
542        db.destroy().await.unwrap();
543    }
544
545    #[test_traced("INFO")]
546    fn test_keyless_fixed_compact_root_compatibility() {
547        deterministic::Runner::default().start(|ctx| async move {
548            assert_compact_root_compatibility::<mmr::Family>(ctx).await;
549        });
550    }
551
552    #[test_traced("INFO")]
553    fn test_keyless_fixed_compact_root_compatibility_mmb() {
554        deterministic::Runner::default().start(|ctx| async move {
555            assert_compact_root_compatibility::<mmb::Family>(ctx).await;
556        });
557    }
558
559    #[test_traced("INFO")]
560    fn test_keyless_fixed_floor_changes_root() {
561        deterministic::Runner::default().start(|ctx| async move {
562            let db_a = open_db_with_suffix::<mmr::Family>("root-a", ctx.child("a")).await;
563            let db_b = open_db_with_suffix::<mmr::Family>("root-b", ctx.child("b")).await;
564            tests::run_floor_changes_root(db_a, db_b).await;
565        });
566    }
567
568    #[test_traced("INFO")]
569    fn test_keyless_fixed_floor_changes_root_mmb() {
570        deterministic::Runner::default().start(|ctx| async move {
571            let db_a = open_db_with_suffix::<mmb::Family>("root-a", ctx.child("a")).await;
572            let db_b = open_db_with_suffix::<mmb::Family>("root-b", ctx.child("b")).await;
573            tests::run_floor_changes_root(db_a, db_b).await;
574        });
575    }
576
577    /// Smoke test: verify the sync engine works end-to-end with a fixed-size keyless database.
578    /// The full sync test suite runs against the variable variant via the harness in
579    /// [`super::super::sync::tests`]; this test covers the fixed-size code path.
580    #[test_traced("WARN")]
581    fn test_keyless_fixed_sync() {
582        use crate::{
583            merkle::Location,
584            qmdb::sync::{self, Target, engine::Config},
585        };
586        use commonware_utils::{non_empty_range, sequence::U64};
587        use std::sync::Arc;
588
589        deterministic::Runner::default().start(|ctx| async move {
590            let target_config = db_config("sync-target", &ctx, Sequential);
591            let target_db: TestDb<mmr::Family> = TestDb::init(ctx.child("target"), target_config)
592                .await
593                .unwrap();
594
595            let mut batch = target_db.new_batch();
596            for i in 0..20u64 {
597                batch = batch.append(U64::new(i * 10 + 1));
598            }
599            let floor = target_db.inactivity_floor_loc();
600            let merkleized = batch.merkleize(&target_db, None, floor).await;
601            let (target_db, _) = target_db.apply_batch(merkleized).await.unwrap();
602
603            let target_root = target_db.root();
604            let bounds = target_db.bounds();
605            let lower_bound = bounds.start;
606            let upper_bound = bounds.end;
607
608            let client_config = db_config("sync-client", &ctx, Sequential);
609            let target_db = Arc::new(target_db);
610            let config = Config {
611                db_config: client_config,
612                fetch_batch_size: NZU64!(5),
613                target: Target {
614                    root: target_root,
615                    range: non_empty_range!(lower_bound, upper_bound),
616                },
617                context: ctx.child("client"),
618                source: target_db.clone(),
619                apply_batch_size: NZU64!(1024),
620                max_outstanding_requests: 1,
621                update_rx: None,
622                finish_rx: None,
623                reached_target_tx: None,
624                max_retained_roots: 8,
625            };
626            let synced_db: TestDb<mmr::Family> = sync::sync(config).await.unwrap();
627
628            assert_eq!(synced_db.root(), target_root);
629            let bounds = synced_db.bounds();
630            assert_eq!(bounds.end, upper_bound);
631            assert_eq!(bounds.start, lower_bound);
632
633            for i in 0..20u64 {
634                let got = synced_db.get(Location::new(i + 1)).await.unwrap();
635                assert_eq!(got, Some(U64::new(i * 10 + 1)));
636            }
637
638            synced_db.destroy().await.unwrap();
639            let target_db =
640                Arc::try_unwrap(target_db).unwrap_or_else(|_| panic!("failed to unwrap Arc"));
641            target_db.destroy().await.unwrap();
642        });
643    }
644}
commonware_storage/qmdb/keyless/fixed.rs

commonware_storage/qmdb/keyless/
fixed.rs
