commonware_glue/stateful/db/
current.rs

1//! [`ManagedDb`] implementation for QMDB [`current`](commonware_storage::qmdb::current) databases.
2//!
3//! The QMDB batch API passes `&db` to `get()` and `merkleize()` for
4//! read-through to applied state. This module provides wrapper types
5//! that capture a [`Shared`] database handle alongside the raw batch so the
6//! [`Unmerkleized`](super::Unmerkleized) and [`Merkleized`](super::Merkleized)
7//! traits can be implemented without a DB parameter.
8
9use crate::stateful::db::{
10    BatchContext, ManagedDb, Merkleized as MerkleizedTrait, Shared, StateSyncDb, SyncEngineConfig,
11    Unmerkleized as UnmerkleizedTrait, sync_standard_db,
12};
13use commonware_codec::{Codec, Read as CodecRead};
14use commonware_cryptography::Hasher;
15use commonware_parallel::Strategy;
16use commonware_runtime::{Handle, Spawner};
17use commonware_storage::{
18    Context,
19    index::{
20        Ordered as OrderedIndex, Unordered as UnorderedIndex, ordered::Index as OrderedIdx,
21        unordered::Index as UnorderedIdx,
22    },
23    journal::contiguous::{
24        Contiguous, Mutable, fixed::Journal as FixedJournal, variable::Journal as VariableJournal,
25    },
26    merkle::{Graftable, Location},
27    qmdb::{
28        Error,
29        any::{
30            initial_root,
31            operation::{Operation, Update},
32            ordered, unordered,
33            value::{self, FixedEncoding, ValueEncoding, VariableEncoding},
34        },
35        current::{
36            FixedConfig, VariableConfig,
37            batch::{MerkleizedBatch, Staged, UnmerkleizedBatch},
38            db::Db,
39        },
40        operation::Key,
41        sync::{self, Target as CurrentSyncTarget},
42    },
43    translator::Translator,
44};
45use commonware_utils::{Array, channel::mpsc, non_empty_range};
46use std::{
47    ops::{Deref, Range},
48    sync::Arc,
49};
50
51/// Wraps a QMDB [`UnmerkleizedBatch`] with a reference to the parent
52/// database, implementing the [`Unmerkleized`](super::Unmerkleized) trait.
53pub struct CurrentUnmerkleized<F, E, C, I, H, U, const N: usize, S>
54where
55    F: Graftable,
56    E: Context,
57    U: Update,
58    C: Contiguous<Item = Operation<F, U>>,
59    I: UnorderedIndex<Value = Location<F>>,
60    H: Hasher,
61    S: Strategy,
62    Operation<F, U>: Codec,
63{
64    batch: UnmerkleizedBatch<F, H, U, N, S>,
65    db: Shared<Db<F, E, C, I, H, U, N, S>>,
66    metadata: Option<U::Value>,
67}
68
69/// Staged batch returned by [`CurrentUnmerkleized::stage`], wrapping a QMDB [`Staged`] with a
70/// reference to the parent database.
71///
72/// Like any speculative batch, this handle is a branch-scoped view of the shared database: it
73/// stays valid only while every batch finalized on the database is an ancestor of this batch
74/// (see [`MerkleizedBatch`]'s branch-validity contract).
75pub struct CurrentStaged<F, E, C, I, H, U, const N: usize, S>
76where
77    F: Graftable,
78    E: Context,
79    U: Update,
80    C: Contiguous<Item = Operation<F, U>>,
81    I: UnorderedIndex<Value = Location<F>>,
82    H: Hasher,
83    S: Strategy,
84    Operation<F, U>: Codec,
85{
86    staged: Staged<F, H, U, N, S>,
87    db: Shared<Db<F, E, C, I, H, U, N, S>>,
88    metadata: Option<U::Value>,
89}
90
91/// Key-value operations shared by both `current` update kinds.
92impl<F, E, C, I, H, U, const N: usize, S> CurrentUnmerkleized<F, E, C, I, H, U, N, S>
93where
94    F: Graftable,
95    E: Context,
96    U: Update,
97    C: Contiguous<Item = Operation<F, U>>,
98    I: UnorderedIndex<Value = Location<F>> + 'static,
99    H: Hasher,
100    S: Strategy,
101    Operation<F, U>: Codec,
102{
103    /// Set commit metadata included in the next
104    /// [`merkleize`](UnmerkleizedTrait::merkleize) call.
105    pub fn with_metadata(mut self, metadata: U::Value) -> Self {
106        self.metadata = Some(metadata);
107        self
108    }
109
110    /// Read a value by key, falling back to applied state.
111    pub async fn get(&self, key: &U::Key) -> Result<Option<U::Value>, Error<F>> {
112        let db = self.db.read().await;
113        self.batch.get(key, &db).await
114    }
115
116    /// Read multiple values by key, falling back to applied state.
117    ///
118    /// Returns results in the same order as the input keys.
119    pub async fn get_many(&self, keys: &[&U::Key]) -> Result<Vec<Option<U::Value>>, Error<F>> {
120        let db = self.db.read().await;
121        self.batch.get_many(keys, &db).await
122    }
123
124    /// Read multiple values and return a staged batch for the same keys.
125    ///
126    /// Returns results in the same order as the input keys.
127    pub async fn stage(
128        self,
129        keys: &[&U::Key],
130    ) -> Result<(Vec<Option<U::Value>>, CurrentStaged<F, E, C, I, H, U, N, S>), Error<F>> {
131        let Self {
132            batch,
133            db,
134            metadata,
135        } = self;
136        let (values, staged) = {
137            let guard = db.read().await;
138            batch.stage(keys, &guard).await?
139        };
140        Ok((
141            values,
142            CurrentStaged {
143                staged,
144                db,
145                metadata,
146            },
147        ))
148    }
149
150    /// Record a mutation. `Some(value)` for upsert, `None` for delete.
151    pub fn write(mut self, key: U::Key, value: Option<U::Value>) -> Self {
152        self.batch = self.batch.write(key, value);
153        self
154    }
155}
156
157/// Wraps a QMDB [`MerkleizedBatch`] with a reference to the parent
158/// database, implementing the [`Merkleized`](super::Merkleized) trait.
159pub struct CurrentMerkleized<F, E, C, I, H, U, const N: usize, S>
160where
161    F: Graftable,
162    E: Context,
163    U: Update,
164    C: Contiguous<Item = Operation<F, U>>,
165    I: UnorderedIndex<Value = Location<F>>,
166    H: Hasher,
167    S: Strategy,
168    Operation<F, U>: Codec,
169{
170    inner: Arc<MerkleizedBatch<F, H::Digest, U, N, S>>,
171    db: Shared<Db<F, E, C, I, H, U, N, S>>,
172}
173
174impl<F, E, C, I, H, U, const N: usize, S> Clone for CurrentMerkleized<F, E, C, I, H, U, N, S>
175where
176    F: Graftable,
177    E: Context,
178    U: Update,
179    C: Contiguous<Item = Operation<F, U>>,
180    I: UnorderedIndex<Value = Location<F>>,
181    H: Hasher,
182    S: Strategy,
183    Operation<F, U>: Codec,
184{
185    fn clone(&self) -> Self {
186        Self {
187            inner: Arc::clone(&self.inner),
188            db: self.db.clone(),
189        }
190    }
191}
192
193impl<F, E, C, I, H, U, const N: usize, S> Deref for CurrentUnmerkleized<F, E, C, I, H, U, N, S>
194where
195    F: Graftable,
196    E: Context,
197    U: Update,
198    C: Contiguous<Item = Operation<F, U>>,
199    I: UnorderedIndex<Value = Location<F>>,
200    H: Hasher,
201    S: Strategy,
202    Operation<F, U>: Codec,
203{
204    type Target = UnmerkleizedBatch<F, H, U, N, S>;
205
206    fn deref(&self) -> &Self::Target {
207        &self.batch
208    }
209}
210
211impl<F, E, C, I, H, U, const N: usize, S> Deref for CurrentMerkleized<F, E, C, I, H, U, N, S>
212where
213    F: Graftable,
214    E: Context,
215    U: Update,
216    C: Contiguous<Item = Operation<F, U>>,
217    I: UnorderedIndex<Value = Location<F>>,
218    H: Hasher,
219    S: Strategy,
220    Operation<F, U>: Codec,
221{
222    type Target = MerkleizedBatch<F, H::Digest, U, N, S>;
223
224    fn deref(&self) -> &Self::Target {
225        &self.inner
226    }
227}
228
229/// Read-expansion operations for the `current` staged batch.
230impl<F, E, C, I, H, U, const N: usize, S> CurrentStaged<F, E, C, I, H, U, N, S>
231where
232    F: Graftable,
233    E: Context,
234    U: Update,
235    C: Contiguous<Item = Operation<F, U>>,
236    I: UnorderedIndex<Value = Location<F>> + 'static,
237    H: Hasher,
238    S: Strategy,
239    Operation<F, U>: Codec,
240{
241    /// Set commit metadata included in the [`merkleize`](Self::merkleize) call, replacing any
242    /// metadata set before staging.
243    pub fn with_metadata(mut self, metadata: U::Value) -> Self {
244        self.metadata = Some(metadata);
245        self
246    }
247
248    /// Expand this staged batch with more reads.
249    ///
250    /// Existing read indices remain stable. Newly read keys are appended to the staged read set and
251    /// assigned the returned range. Expansion does not deduplicate against previously staged keys
252    /// and does not observe values computed for earlier staged slots but not yet passed to
253    /// `merkleize`.
254    pub async fn expand(
255        self,
256        keys: &[&U::Key],
257    ) -> Result<(Range<usize>, Vec<Option<U::Value>>, Self), Error<F>> {
258        let Self {
259            staged,
260            db,
261            metadata,
262        } = self;
263        let (range, values, staged) = {
264            let guard = db.read().await;
265            staged.expand(keys, &guard).await?
266        };
267        Ok((
268            range,
269            values,
270            Self {
271                staged,
272                db,
273                metadata,
274            },
275        ))
276    }
277}
278
279/// Staged merkleize for the `current` unordered update kind.
280impl<F, E, C, I, H, K, V, const N: usize, S>
281    CurrentStaged<F, E, C, I, H, unordered::Update<K, V>, N, S>
282where
283    F: Graftable,
284    E: Context,
285    K: Key,
286    V: ValueEncoding + 'static,
287    C: Mutable<Item = Operation<F, unordered::Update<K, V>>>,
288    I: UnorderedIndex<Value = Location<F>> + 'static,
289    H: Hasher,
290    S: Strategy,
291    Operation<F, unordered::Update<K, V>>: Codec,
292{
293    /// Record updates for staged reads and upserts for unread keys, then merkleize.
294    ///
295    /// Consumes the staged handle and write vectors. Call [`expand`](CurrentStaged::expand)
296    /// before this method if more keys must be read into the staged index space.
297    ///
298    /// A `Some` value is an upsert. `None` is a delete. Update indices refer to the staged read
299    /// set: the initial `stage` input followed by any [`expand`](CurrentStaged::expand) ranges.
300    /// Metadata set via [`with_metadata`](CurrentStaged::with_metadata) (or before staging) is
301    /// committed with the returned batch.
302    ///
303    /// # Panics
304    ///
305    /// Panics if any update's `read_index` is out of the staged read range.
306    pub async fn merkleize(
307        self,
308        updates: Vec<(usize, Option<V::Value>)>,
309        upserts: Vec<(K, Option<V::Value>)>,
310    ) -> Result<CurrentMerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>, Error<F>> {
311        let Self {
312            staged,
313            db,
314            metadata,
315        } = self;
316        let inner = {
317            let guard = db.read().await;
318            staged.merkleize(updates, upserts, metadata, &guard).await?
319        };
320        Ok(CurrentMerkleized { inner, db })
321    }
322}
323
324/// Staged merkleize for the `current` ordered update kind.
325impl<F, E, C, I, H, K, V, const N: usize, S>
326    CurrentStaged<F, E, C, I, H, ordered::Update<K, V>, N, S>
327where
328    F: Graftable,
329    E: Context,
330    K: Key,
331    V: ValueEncoding + 'static,
332    C: Mutable<Item = Operation<F, ordered::Update<K, V>>>,
333    I: OrderedIndex<Value = Location<F>> + 'static,
334    H: Hasher,
335    S: Strategy,
336    Operation<F, ordered::Update<K, V>>: Codec,
337{
338    /// Record updates for staged reads and upserts for unread keys, then merkleize.
339    ///
340    /// Consumes the staged handle and write vectors. Call [`expand`](CurrentStaged::expand)
341    /// before this method if more keys must be read into the staged index space.
342    ///
343    /// A `Some` value is an upsert. `None` is a delete. Update indices refer to the staged read
344    /// set: the initial `stage` input followed by any [`expand`](CurrentStaged::expand) ranges.
345    /// Metadata set via [`with_metadata`](CurrentStaged::with_metadata) (or before staging) is
346    /// committed with the returned batch.
347    ///
348    /// # Panics
349    ///
350    /// Panics if any update's `read_index` is out of the staged read range.
351    pub async fn merkleize(
352        self,
353        updates: Vec<(usize, Option<V::Value>)>,
354        upserts: Vec<(K, Option<V::Value>)>,
355    ) -> Result<CurrentMerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>, Error<F>> {
356        let Self {
357            staged,
358            db,
359            metadata,
360        } = self;
361        let inner = {
362            let guard = db.read().await;
363            staged.merkleize(updates, upserts, metadata, &guard).await?
364        };
365        Ok(CurrentMerkleized { inner, db })
366    }
367}
368
369/// Read-through operations for the `current` merkleized batch.
370impl<F, E, C, I, H, U, const N: usize, S> CurrentMerkleized<F, E, C, I, H, U, N, S>
371where
372    F: Graftable,
373    E: Context,
374    U: Update,
375    C: Contiguous<Item = Operation<F, U>>,
376    I: UnorderedIndex<Value = Location<F>> + 'static,
377    H: Hasher,
378    S: Strategy,
379    Operation<F, U>: Codec,
380{
381    /// Read a value by key, falling back to applied state.
382    pub async fn get(&self, key: &U::Key) -> Result<Option<U::Value>, Error<F>> {
383        let db = self.db.read().await;
384        self.inner.get(key, &db).await
385    }
386
387    /// Read multiple values by key, falling back to applied state.
388    ///
389    /// Returns results in the same order as the input keys.
390    pub async fn get_many(&self, keys: &[&U::Key]) -> Result<Vec<Option<U::Value>>, Error<F>> {
391        let db = self.db.read().await;
392        self.inner.get_many(keys, &db).await
393    }
394}
395
396/// Implement [`Unmerkleized`](UnmerkleizedTrait) for the `current` unordered update kind.
397impl<F, E, C, I, H, K, V, const N: usize, S> UnmerkleizedTrait
398    for CurrentUnmerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>
399where
400    F: Graftable,
401    E: Context,
402    K: Key,
403    V: ValueEncoding + 'static,
404    C: Mutable<Item = Operation<F, unordered::Update<K, V>>>,
405    I: UnorderedIndex<Value = Location<F>> + 'static,
406    H: Hasher,
407    S: Strategy,
408    Operation<F, unordered::Update<K, V>>: Codec,
409{
410    type Merkleized = CurrentMerkleized<F, E, C, I, H, unordered::Update<K, V>, N, S>;
411    type Error = Error<F>;
412
413    async fn merkleize(self) -> Result<Self::Merkleized, Error<F>> {
414        let db = self.db.read().await;
415        let merkleized = self.batch.merkleize(&db, self.metadata).await?;
416        Ok(CurrentMerkleized {
417            inner: merkleized,
418            db: self.db.clone(),
419        })
420    }
421}
422
423/// Implement [`Unmerkleized`](UnmerkleizedTrait) for the `current` ordered update kind.
424impl<F, E, C, I, H, K, V, const N: usize, S> UnmerkleizedTrait
425    for CurrentUnmerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>
426where
427    F: Graftable,
428    E: Context,
429    K: Key,
430    V: ValueEncoding + 'static,
431    C: Mutable<Item = Operation<F, ordered::Update<K, V>>>,
432    I: OrderedIndex<Value = Location<F>> + 'static,
433    H: Hasher,
434    S: Strategy,
435    Operation<F, ordered::Update<K, V>>: Codec,
436{
437    type Merkleized = CurrentMerkleized<F, E, C, I, H, ordered::Update<K, V>, N, S>;
438    type Error = Error<F>;
439
440    async fn merkleize(self) -> Result<Self::Merkleized, Error<F>> {
441        let db = self.db.read().await;
442        let merkleized = self.batch.merkleize(&db, self.metadata).await?;
443        Ok(CurrentMerkleized {
444            inner: merkleized,
445            db: self.db.clone(),
446        })
447    }
448}
449
450/// Implement [`Merkleized`](MerkleizedTrait) for all supported `current` update kinds.
451impl<F, E, C, I, H, U, const N: usize, S> MerkleizedTrait
452    for CurrentMerkleized<F, E, C, I, H, U, N, S>
453where
454    F: Graftable,
455    E: Context,
456    U: Update,
457    C: Mutable<Item = Operation<F, U>>,
458    I: UnorderedIndex<Value = Location<F>> + 'static,
459    H: Hasher,
460    S: Strategy,
461    Operation<F, U>: Codec,
462    CurrentUnmerkleized<F, E, C, I, H, U, N, S>: UnmerkleizedTrait,
463{
464    type Digest = H::Digest;
465    type Unmerkleized = CurrentUnmerkleized<F, E, C, I, H, U, N, S>;
466
467    fn root(&self) -> H::Digest {
468        self.inner.root()
469    }
470
471    fn new_batch(&self) -> Self::Unmerkleized {
472        CurrentUnmerkleized {
473            batch: self.inner.new_batch::<H>(),
474            db: self.db.clone(),
475            metadata: None,
476        }
477    }
478}
479
480/// Implement [`ManagedDb`] for unordered current QMDB databases with fixed-size values.
481impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
482    for Db<
483        F,
484        E,
485        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
486        UnorderedIdx<T, Location<F>>,
487        H,
488        unordered::Update<K, FixedEncoding<V>>,
489        N,
490        S,
491    >
492where
493    F: Graftable,
494    E: Context + Spawner,
495    K: Array,
496    V: value::FixedValue + 'static,
497    H: Hasher + 'static,
498    T: Translator,
499    S: Strategy,
500{
501    type Unmerkleized = CurrentUnmerkleized<
502        F,
503        E,
504        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
505        UnorderedIdx<T, Location<F>>,
506        H,
507        unordered::Update<K, FixedEncoding<V>>,
508        N,
509        S,
510    >;
511    type Merkleized = CurrentMerkleized<
512        F,
513        E,
514        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
515        UnorderedIdx<T, Location<F>>,
516        H,
517        unordered::Update<K, FixedEncoding<V>>,
518        N,
519        S,
520    >;
521    type Error = Error<F>;
522    type Config = FixedConfig<T, S>;
523    type SyncTarget = CurrentSyncTarget<F, H::Digest>;
524
525    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
526        <Self>::init(context, config).await
527    }
528
529    fn initial_sync_target() -> Self::SyncTarget {
530        CurrentSyncTarget::new(
531            initial_root::<F, unordered::Update<K, FixedEncoding<V>>, H>(),
532            non_empty_range!(Location::new(0), Location::new(1)),
533        )
534    }
535
536    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
537        let (database, shared) = database.into_parts();
538        CurrentUnmerkleized {
539            batch: database.new_batch(),
540            db: shared,
541            metadata: None,
542        }
543    }
544
545    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
546        batch.ops_root() == target.root
547            && *target.range.start() == batch.sync_boundary()
548            && *target.range.end() == batch.bounds().tip.size
549    }
550
551    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
552        let (db, _) = self.apply_batch(batch.inner).await?;
553        Ok(db)
554    }
555
556    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
557        self.start_sync().await
558    }
559
560    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
561        self.prune((*target.range.start()).into()).await
562    }
563
564    fn sync_target(&self) -> Self::SyncTarget {
565        let bounds = self.bounds();
566        CurrentSyncTarget::new(
567            self.ops_root(),
568            non_empty_range!(self.sync_boundary(), bounds.end),
569        )
570    }
571
572    async fn rewind_to_target(self, target: Self::SyncTarget) -> Result<Self, Error<F>> {
573        let db = self.rewind(target.range.end()).await?;
574        let db = db.sync().await?;
575
576        let rewound_target = db.sync_target();
577        assert_eq!(
578            rewound_target, target,
579            "rewound database target mismatch after rewind",
580        );
581        Ok(db)
582    }
583}
584
585/// Implement [`ManagedDb`] for ordered current QMDB databases with fixed-size values.
586impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
587    for Db<
588        F,
589        E,
590        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
591        OrderedIdx<T, Location<F>>,
592        H,
593        ordered::Update<K, FixedEncoding<V>>,
594        N,
595        S,
596    >
597where
598    F: Graftable,
599    E: Context + Spawner,
600    K: Array,
601    V: value::FixedValue + 'static,
602    H: Hasher + 'static,
603    T: Translator,
604    S: Strategy,
605{
606    type Unmerkleized = CurrentUnmerkleized<
607        F,
608        E,
609        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
610        OrderedIdx<T, Location<F>>,
611        H,
612        ordered::Update<K, FixedEncoding<V>>,
613        N,
614        S,
615    >;
616    type Merkleized = CurrentMerkleized<
617        F,
618        E,
619        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
620        OrderedIdx<T, Location<F>>,
621        H,
622        ordered::Update<K, FixedEncoding<V>>,
623        N,
624        S,
625    >;
626    type Error = Error<F>;
627    type Config = FixedConfig<T, S>;
628    type SyncTarget = CurrentSyncTarget<F, H::Digest>;
629
630    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
631        <Self>::init(context, config).await
632    }
633
634    fn initial_sync_target() -> Self::SyncTarget {
635        CurrentSyncTarget::new(
636            initial_root::<F, ordered::Update<K, FixedEncoding<V>>, H>(),
637            non_empty_range!(Location::new(0), Location::new(1)),
638        )
639    }
640
641    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
642        let (database, shared) = database.into_parts();
643        CurrentUnmerkleized {
644            batch: database.new_batch(),
645            db: shared,
646            metadata: None,
647        }
648    }
649
650    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
651        batch.ops_root() == target.root
652            && *target.range.start() == batch.sync_boundary()
653            && *target.range.end() == batch.bounds().tip.size
654    }
655
656    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
657        let (db, _) = self.apply_batch(batch.inner).await?;
658        Ok(db)
659    }
660
661    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
662        self.start_sync().await
663    }
664
665    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
666        self.prune((*target.range.start()).into()).await
667    }
668
669    fn sync_target(&self) -> Self::SyncTarget {
670        let bounds = self.bounds();
671        CurrentSyncTarget::new(
672            self.ops_root(),
673            non_empty_range!(self.sync_boundary(), bounds.end),
674        )
675    }
676
677    async fn rewind_to_target(self, target: Self::SyncTarget) -> Result<Self, Error<F>> {
678        let db = self.rewind(target.range.end()).await?;
679        let db = db.sync().await?;
680
681        let rewound_target = db.sync_target();
682        assert_eq!(
683            rewound_target, target,
684            "rewound database target mismatch after rewind",
685        );
686        Ok(db)
687    }
688}
689
690/// Workaround for <https://github.com/rust-lang/rust/issues/115188>.
691///
692/// Inside a `ManagedDb` trait impl, `<Self>::init(...)` in a non-async `fn`
693/// resolves to the *trait* method (infinite recursion), while in an
694/// `async fn` it resolves correctly to the inherent method but the compiler
695/// cannot verify the RPITIT future is `Send`. By placing the call in this
696/// module -- which does not import `ManagedDb` -- the compiler
697/// unambiguously picks the inherent `Db::init`.
698mod open {
699    use commonware_codec::{Codec, Read};
700    use commonware_cryptography::Hasher;
701    use commonware_parallel::Strategy;
702    use commonware_runtime::Spawner;
703    use commonware_storage::{
704        Context,
705        merkle::Graftable,
706        qmdb::{
707            Error,
708            any::{
709                operation::Operation,
710                ordered, unordered,
711                value::{VariableEncoding, VariableValue},
712            },
713            current::{
714                VariableConfig, ordered::variable::Db as OrderedVariableDb, unordered::variable::Db,
715            },
716        },
717    };
718    type VConfig<T, F, K, V, S> = VariableConfig<
719        T,
720        <Operation<F, unordered::Update<K, VariableEncoding<V>>> as Read>::Cfg,
721        S,
722    >;
723    type OrderedVConfig<T, F, K, V, S> =
724        VariableConfig<T, <Operation<F, ordered::Update<K, VariableEncoding<V>>> as Read>::Cfg, S>;
725
726    pub(super) async fn variable<F, E, K, V, H, T, const N: usize, S>(
727        context: E,
728        config: VConfig<T, F, K, V, S>,
729    ) -> Result<Db<F, E, K, V, H, T, N, S>, Error<F>>
730    where
731        F: Graftable,
732        E: Context + Spawner,
733        K: commonware_storage::qmdb::operation::Key,
734        V: VariableValue + 'static,
735        H: Hasher,
736        T: commonware_storage::translator::Translator,
737        S: Strategy,
738        Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
739    {
740        Db::init(context, config).await
741    }
742
743    pub(super) async fn ordered_variable<F, E, K, V, H, T, const N: usize, S>(
744        context: E,
745        config: OrderedVConfig<T, F, K, V, S>,
746    ) -> Result<OrderedVariableDb<F, E, K, V, H, T, N, S>, Error<F>>
747    where
748        F: Graftable,
749        E: Context + Spawner,
750        K: commonware_storage::qmdb::operation::Key,
751        V: VariableValue + 'static,
752        H: Hasher,
753        T: commonware_storage::translator::Translator,
754        S: Strategy,
755        Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
756    {
757        OrderedVariableDb::init(context, config).await
758    }
759}
760
761/// Implement [`ManagedDb`] for unordered current QMDB databases with variable-size values.
762impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
763    for Db<
764        F,
765        E,
766        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
767        UnorderedIdx<T, Location<F>>,
768        H,
769        unordered::Update<K, VariableEncoding<V>>,
770        N,
771        S,
772    >
773where
774    F: Graftable,
775    E: Context + Spawner,
776    K: Key,
777    V: value::VariableValue + 'static,
778    H: Hasher,
779    T: Translator,
780    S: Strategy,
781    Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
782{
783    type Unmerkleized = CurrentUnmerkleized<
784        F,
785        E,
786        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
787        UnorderedIdx<T, Location<F>>,
788        H,
789        unordered::Update<K, VariableEncoding<V>>,
790        N,
791        S,
792    >;
793    type Merkleized = CurrentMerkleized<
794        F,
795        E,
796        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
797        UnorderedIdx<T, Location<F>>,
798        H,
799        unordered::Update<K, VariableEncoding<V>>,
800        N,
801        S,
802    >;
803    type Error = Error<F>;
804    type Config = VariableConfig<
805        T,
806        <Operation<F, unordered::Update<K, VariableEncoding<V>>> as CodecRead>::Cfg,
807        S,
808    >;
809    type SyncTarget = CurrentSyncTarget<F, H::Digest>;
810
811    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
812        open::variable(context, config).await
813    }
814
815    fn initial_sync_target() -> Self::SyncTarget {
816        CurrentSyncTarget::new(
817            initial_root::<F, unordered::Update<K, VariableEncoding<V>>, H>(),
818            non_empty_range!(Location::new(0), Location::new(1)),
819        )
820    }
821
822    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
823        let (database, shared) = database.into_parts();
824        CurrentUnmerkleized {
825            batch: database.new_batch(),
826            db: shared,
827            metadata: None,
828        }
829    }
830
831    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
832        batch.ops_root() == target.root
833            && *target.range.start() == batch.sync_boundary()
834            && *target.range.end() == batch.bounds().tip.size
835    }
836
837    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
838        let (db, _) = self.apply_batch(batch.inner).await?;
839        Ok(db)
840    }
841
842    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
843        self.start_sync().await
844    }
845
846    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
847        self.prune((*target.range.start()).into()).await
848    }
849
850    fn sync_target(&self) -> Self::SyncTarget {
851        let bounds = self.bounds();
852        CurrentSyncTarget::new(
853            self.ops_root(),
854            non_empty_range!(self.sync_boundary(), bounds.end),
855        )
856    }
857
858    async fn rewind_to_target(self, target: Self::SyncTarget) -> Result<Self, Error<F>> {
859        let db = self.rewind(target.range.end()).await?;
860        let db = db.sync().await?;
861
862        let rewound_target = db.sync_target();
863        assert_eq!(
864            rewound_target, target,
865            "rewound database target mismatch after rewind",
866        );
867        Ok(db)
868    }
869}
870
871/// Implement [`ManagedDb`] for ordered current QMDB databases with variable-size values.
872impl<F, E, K, V, H, T, const N: usize, S> ManagedDb<E>
873    for Db<
874        F,
875        E,
876        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
877        OrderedIdx<T, Location<F>>,
878        H,
879        ordered::Update<K, VariableEncoding<V>>,
880        N,
881        S,
882    >
883where
884    F: Graftable,
885    E: Context + Spawner,
886    K: Key,
887    V: value::VariableValue + 'static,
888    H: Hasher,
889    T: Translator,
890    S: Strategy,
891    Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
892{
893    type Unmerkleized = CurrentUnmerkleized<
894        F,
895        E,
896        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
897        OrderedIdx<T, Location<F>>,
898        H,
899        ordered::Update<K, VariableEncoding<V>>,
900        N,
901        S,
902    >;
903    type Merkleized = CurrentMerkleized<
904        F,
905        E,
906        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
907        OrderedIdx<T, Location<F>>,
908        H,
909        ordered::Update<K, VariableEncoding<V>>,
910        N,
911        S,
912    >;
913    type Error = Error<F>;
914    type Config = VariableConfig<
915        T,
916        <Operation<F, ordered::Update<K, VariableEncoding<V>>> as CodecRead>::Cfg,
917        S,
918    >;
919    type SyncTarget = CurrentSyncTarget<F, H::Digest>;
920
921    async fn init(context: E, config: Self::Config) -> Result<Self, Error<F>> {
922        open::ordered_variable(context, config).await
923    }
924
925    fn initial_sync_target() -> Self::SyncTarget {
926        CurrentSyncTarget::new(
927            initial_root::<F, ordered::Update<K, VariableEncoding<V>>, H>(),
928            non_empty_range!(Location::new(0), Location::new(1)),
929        )
930    }
931
932    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized {
933        let (database, shared) = database.into_parts();
934        CurrentUnmerkleized {
935            batch: database.new_batch(),
936            db: shared,
937            metadata: None,
938        }
939    }
940
941    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool {
942        batch.ops_root() == target.root
943            && *target.range.start() == batch.sync_boundary()
944            && *target.range.end() == batch.bounds().tip.size
945    }
946
947    async fn apply(self, batch: Self::Merkleized) -> Result<Self, Error<F>> {
948        let (db, _) = self.apply_batch(batch.inner).await?;
949        Ok(db)
950    }
951
952    async fn finalize(self) -> Result<(Self, Handle<()>), Error<F>> {
953        self.start_sync().await
954    }
955
956    async fn prune(self, target: &Self::SyncTarget) -> Result<Self, Error<F>> {
957        self.prune((*target.range.start()).into()).await
958    }
959
960    fn sync_target(&self) -> Self::SyncTarget {
961        let bounds = self.bounds();
962        CurrentSyncTarget::new(
963            self.ops_root(),
964            non_empty_range!(self.sync_boundary(), bounds.end),
965        )
966    }
967
968    async fn rewind_to_target(self, target: Self::SyncTarget) -> Result<Self, Error<F>> {
969        let db = self.rewind(target.range.end()).await?;
970        let db = db.sync().await?;
971
972        let rewound_target = db.sync_target();
973        assert_eq!(
974            rewound_target, target,
975            "rewound database target mismatch after rewind",
976        );
977        Ok(db)
978    }
979}
980
981/// Implement [`StateSyncDb`] for unordered current QMDB databases with fixed-size values.
982impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
983    for Db<
984        F,
985        E,
986        FixedJournal<E, Operation<F, unordered::Update<K, FixedEncoding<V>>>>,
987        UnorderedIdx<T, Location<F>>,
988        H,
989        unordered::Update<K, FixedEncoding<V>>,
990        N,
991        S,
992    >
993where
994    F: Graftable,
995    E: Context + Spawner,
996    K: Array,
997    V: value::FixedValue + 'static,
998    H: Hasher,
999    T: Translator,
1000    S: Strategy,
1001    R: sync::SourceFor<Self>,
1002{
1003    type SyncError = sync::Error<F, R::Error, H::Digest>;
1004
1005    async fn sync_db(
1006        context: E,
1007        config: Self::Config,
1008        source: R,
1009        target: Self::SyncTarget,
1010        tip_updates: mpsc::Receiver<Self::SyncTarget>,
1011        finish: Option<mpsc::Receiver<()>>,
1012        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
1013        sync_config: SyncEngineConfig,
1014    ) -> Result<Self, Self::SyncError> {
1015        sync_standard_db(
1016            context,
1017            config,
1018            source,
1019            target,
1020            tip_updates,
1021            finish,
1022            reached_target,
1023            sync_config,
1024        )
1025        .await
1026    }
1027}
1028
1029/// Implement [`StateSyncDb`] for ordered current QMDB databases with fixed-size values.
1030impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
1031    for Db<
1032        F,
1033        E,
1034        FixedJournal<E, Operation<F, ordered::Update<K, FixedEncoding<V>>>>,
1035        OrderedIdx<T, Location<F>>,
1036        H,
1037        ordered::Update<K, FixedEncoding<V>>,
1038        N,
1039        S,
1040    >
1041where
1042    F: Graftable,
1043    E: Context + Spawner,
1044    K: Array,
1045    V: value::FixedValue + 'static,
1046    H: Hasher,
1047    T: Translator,
1048    S: Strategy,
1049    R: sync::SourceFor<Self>,
1050{
1051    type SyncError = sync::Error<F, R::Error, H::Digest>;
1052
1053    async fn sync_db(
1054        context: E,
1055        config: Self::Config,
1056        source: R,
1057        target: Self::SyncTarget,
1058        tip_updates: mpsc::Receiver<Self::SyncTarget>,
1059        finish: Option<mpsc::Receiver<()>>,
1060        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
1061        sync_config: SyncEngineConfig,
1062    ) -> Result<Self, Self::SyncError> {
1063        sync_standard_db(
1064            context,
1065            config,
1066            source,
1067            target,
1068            tip_updates,
1069            finish,
1070            reached_target,
1071            sync_config,
1072        )
1073        .await
1074    }
1075}
1076
1077/// Implement [`StateSyncDb`] for unordered current QMDB databases with variable-size values.
1078impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
1079    for Db<
1080        F,
1081        E,
1082        VariableJournal<E, Operation<F, unordered::Update<K, VariableEncoding<V>>>>,
1083        UnorderedIdx<T, Location<F>>,
1084        H,
1085        unordered::Update<K, VariableEncoding<V>>,
1086        N,
1087        S,
1088    >
1089where
1090    F: Graftable,
1091    E: Context + Spawner,
1092    K: Key,
1093    V: value::VariableValue + 'static,
1094    H: Hasher,
1095    T: Translator,
1096    S: Strategy,
1097    Operation<F, unordered::Update<K, VariableEncoding<V>>>: Codec,
1098    R: sync::SourceFor<Self>,
1099{
1100    type SyncError = sync::Error<F, R::Error, H::Digest>;
1101
1102    async fn sync_db(
1103        context: E,
1104        config: Self::Config,
1105        source: R,
1106        target: Self::SyncTarget,
1107        tip_updates: mpsc::Receiver<Self::SyncTarget>,
1108        finish: Option<mpsc::Receiver<()>>,
1109        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
1110        sync_config: SyncEngineConfig,
1111    ) -> Result<Self, Self::SyncError> {
1112        sync_standard_db(
1113            context,
1114            config,
1115            source,
1116            target,
1117            tip_updates,
1118            finish,
1119            reached_target,
1120            sync_config,
1121        )
1122        .await
1123    }
1124}
1125
1126/// Implement [`StateSyncDb`] for ordered current QMDB databases with variable-size values.
1127impl<F, E, K, V, H, T, R, const N: usize, S> StateSyncDb<E, R>
1128    for Db<
1129        F,
1130        E,
1131        VariableJournal<E, Operation<F, ordered::Update<K, VariableEncoding<V>>>>,
1132        OrderedIdx<T, Location<F>>,
1133        H,
1134        ordered::Update<K, VariableEncoding<V>>,
1135        N,
1136        S,
1137    >
1138where
1139    F: Graftable,
1140    E: Context + Spawner,
1141    K: Key,
1142    V: value::VariableValue + 'static,
1143    H: Hasher,
1144    T: Translator,
1145    S: Strategy,
1146    Operation<F, ordered::Update<K, VariableEncoding<V>>>: Codec,
1147    R: sync::SourceFor<Self>,
1148{
1149    type SyncError = sync::Error<F, R::Error, H::Digest>;
1150
1151    async fn sync_db(
1152        context: E,
1153        config: Self::Config,
1154        source: R,
1155        target: Self::SyncTarget,
1156        tip_updates: mpsc::Receiver<Self::SyncTarget>,
1157        finish: Option<mpsc::Receiver<()>>,
1158        reached_target: Option<mpsc::Sender<Self::SyncTarget>>,
1159        sync_config: SyncEngineConfig,
1160    ) -> Result<Self, Self::SyncError> {
1161        sync_standard_db(
1162            context,
1163            config,
1164            source,
1165            target,
1166            tip_updates,
1167            finish,
1168            reached_target,
1169            sync_config,
1170        )
1171        .await
1172    }
1173}
1174
1175#[cfg(test)]
1176mod tests {
1177    use super::*;
1178    use commonware_cryptography::{Sha256, sha256::Digest};
1179    use commonware_macros::boxed;
1180    use commonware_parallel::Sequential;
1181    use commonware_runtime::{
1182        BufferPooler, Runner as _, Supervisor as _, buffer::paged::CacheRef, deterministic,
1183    };
1184    use commonware_storage::{
1185        journal::contiguous::{
1186            fixed::Config as FixedJournalConfig, variable::Config as VariableJournalConfig,
1187        },
1188        merkle::{full::Config as MerkleConfig, mmr},
1189        qmdb::current::{
1190            ordered::{fixed as ordered_fixed, variable as ordered_variable},
1191            unordered::{fixed, variable},
1192        },
1193        translator::TwoCap,
1194    };
1195    use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range};
1196    use std::num::{NonZeroU16, NonZeroUsize};
1197
1198    #[boxed]
1199    async fn apply_and_finalize<D: ManagedDb<deterministic::Context>>(
1200        db: D,
1201        batch: D::Merkleized,
1202    ) -> D {
1203        let db = D::apply(db, batch).await.unwrap();
1204        let (db, sync) = D::finalize(db).await.unwrap();
1205        sync.await.expect("database sync failed");
1206        db
1207    }
1208
1209    type FixedDb = fixed::Db<
1210        mmr::Family,
1211        deterministic::Context,
1212        Digest,
1213        Digest,
1214        Sha256,
1215        TwoCap,
1216        64,
1217        Sequential,
1218    >;
1219    type OrderedFixedDb = ordered_fixed::Db<
1220        mmr::Family,
1221        deterministic::Context,
1222        Digest,
1223        Digest,
1224        Sha256,
1225        TwoCap,
1226        64,
1227        Sequential,
1228    >;
1229    type OrderedVariableDb = ordered_variable::Db<
1230        mmr::Family,
1231        deterministic::Context,
1232        Digest,
1233        Digest,
1234        Sha256,
1235        TwoCap,
1236        64,
1237        Sequential,
1238    >;
1239
1240    /// The unordered variable wrapper accepts variable-length keys.
1241    type VariableDb = variable::Db<
1242        mmr::Family,
1243        deterministic::Context,
1244        Vec<u8>,
1245        Digest,
1246        Sha256,
1247        TwoCap,
1248        64,
1249        Sequential,
1250    >;
1251
1252    const PAGE_SIZE: NonZeroU16 = NZU16!(101);
1253    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(11);
1254
1255    fn fixed_config(suffix: &str, pooler: &impl BufferPooler) -> FixedConfig<TwoCap, Sequential> {
1256        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
1257        FixedConfig {
1258            merkle_config: MerkleConfig {
1259                journal_partition: format!("stateful-current-journal-{suffix}"),
1260                metadata_partition: format!("stateful-current-metadata-{suffix}"),
1261                items_per_blob: NZU64!(11),
1262                write_buffer: NZUsize!(1024),
1263                replay_buffer: NZUsize!(1024),
1264                strategy: Sequential,
1265                page_cache: page_cache.clone(),
1266            },
1267            journal_config: FixedJournalConfig {
1268                partition: format!("stateful-current-log-{suffix}"),
1269                items_per_blob: NZU64!(7),
1270                page_cache,
1271                write_buffer: NZUsize!(1024),
1272                replay_buffer: NZUsize!(1024),
1273            },
1274            grafted_metadata_partition: format!("stateful-current-grafted-{suffix}"),
1275            translator: TwoCap,
1276            init_cache_size: Some(NZUsize!(1024)),
1277            init_buffer: NZUsize!(1 << 21),
1278            init_concurrency: (),
1279        }
1280    }
1281
1282    fn variable_config(
1283        suffix: &str,
1284        pooler: &impl BufferPooler,
1285    ) -> VariableConfig<TwoCap, ((), ()), Sequential> {
1286        let page_cache = CacheRef::from_pooler(pooler, PAGE_SIZE, PAGE_CACHE_SIZE);
1287        VariableConfig {
1288            merkle_config: MerkleConfig {
1289                journal_partition: format!("stateful-current-journal-{suffix}"),
1290                metadata_partition: format!("stateful-current-metadata-{suffix}"),
1291                items_per_blob: NZU64!(11),
1292                write_buffer: NZUsize!(1024),
1293                replay_buffer: NZUsize!(1024),
1294                strategy: Sequential,
1295                page_cache: page_cache.clone(),
1296            },
1297            journal_config: VariableJournalConfig {
1298                partition: format!("stateful-current-log-{suffix}"),
1299                items_per_section: NZU64!(7),
1300                compression: None,
1301                codec_config: ((), ()),
1302                page_cache,
1303                write_buffer: NZUsize!(1024),
1304                replay_buffer: NZUsize!(1024),
1305            },
1306            grafted_metadata_partition: format!("stateful-current-grafted-{suffix}"),
1307            translator: TwoCap,
1308            init_cache_size: Some(NZUsize!(1024)),
1309            init_buffer: NZUsize!(1 << 21),
1310            init_concurrency: (),
1311        }
1312    }
1313
1314    fn assert_managed_db<T: ManagedDb<deterministic::Context>>() {}
1315
1316    fn assert_state_sync_db<T, R>()
1317    where
1318        T: StateSyncDb<deterministic::Context, R>,
1319    {
1320    }
1321
1322    fn assert_database_set<T: crate::stateful::db::DatabaseSet<deterministic::Context>>() {}
1323
1324    #[test]
1325    fn ordered_current_db_trait_impls_compile() {
1326        assert_managed_db::<OrderedFixedDb>();
1327        assert_managed_db::<OrderedVariableDb>();
1328        assert_state_sync_db::<OrderedFixedDb, Arc<OrderedFixedDb>>();
1329        assert_state_sync_db::<OrderedVariableDb, Arc<OrderedVariableDb>>();
1330        assert_database_set::<Shared<OrderedFixedDb>>();
1331        assert_database_set::<Shared<OrderedVariableDb>>();
1332    }
1333
1334    #[test]
1335    fn variable_current_db_trait_impls_compile() {
1336        assert_managed_db::<VariableDb>();
1337        assert_state_sync_db::<VariableDb, Arc<VariableDb>>();
1338        assert_database_set::<Shared<VariableDb>>();
1339    }
1340
1341    #[test]
1342    fn ordered_fixed_managed_db_applies_batch_and_proves_exclusion() {
1343        deterministic::Runner::default().start(|context| async move {
1344            let config = fixed_config("ordered-fixed-managed-db", &context);
1345            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
1346                .await
1347                .unwrap();
1348            let db = Shared::new("test", db);
1349            let key = Sha256::hash(&[b"key"]);
1350            let value = Sha256::hash(&[b"value"]);
1351            let metadata = Sha256::hash(&[b"metadata"]);
1352            let missing = Sha256::hash(&[b"missing"]);
1353
1354            let batch = db
1355                .new_batch_for_test::<_>()
1356                .await
1357                .write(key, Some(value))
1358                .with_metadata(metadata);
1359            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
1360                .await
1361                .unwrap();
1362            let expected_root = merkleized.root();
1363
1364            {
1365                let (slot, database) = db.write().await;
1366                slot.put(apply_and_finalize::<OrderedFixedDb>(database, merkleized).await);
1367            }
1368
1369            let guard = db.read().await;
1370            assert_eq!(guard.root(), expected_root);
1371            assert_eq!(guard.get(&key).await.unwrap(), Some(value));
1372
1373            let proof = guard.exclusion_proof(&missing).await.unwrap();
1374            assert!(OrderedFixedDb::verify_exclusion_proof(
1375                &missing,
1376                &proof,
1377                &guard.root(),
1378            ));
1379        });
1380    }
1381
1382    /// The glue staged wrapper (`CurrentUnmerkleized::stage` -> `CurrentStaged::expand` ->
1383    /// `CurrentStaged::merkleize`) must return the same values and root as an explicit `get_many` +
1384    /// `write` + `merkleize`, including a staged delete, an upsert, and metadata flow (both set
1385    /// on the staged handle via `with_metadata` and carried from before staging). This guards
1386    /// metadata flow and db-handle pairing through the wrapper.
1387    #[test]
1388    fn ordered_fixed_staged_merkleize_matches_explicit_writes() {
1389        deterministic::Runner::default().start(|context| async move {
1390            let config = fixed_config("ordered-fixed-glue-staged", &context);
1391            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
1392                .await
1393                .unwrap();
1394            let db = Shared::new("test", db);
1395
1396            let key = |i: u64| Sha256::hash(&[&i.to_be_bytes()]);
1397            let val = |i: u64| Sha256::hash(&[&(i + 10_000).to_be_bytes()]);
1398            let metadata = Sha256::hash(&[b"metadata"]);
1399
1400            // Seed keys 0..50 and persist them.
1401            let mut seed = db.new_batch_for_test::<_>().await;
1402            for i in 0..50u64 {
1403                seed = seed.write(key(i), Some(val(i)));
1404            }
1405            let merkleized = crate::stateful::db::Unmerkleized::merkleize(seed)
1406                .await
1407                .unwrap();
1408            {
1409                let (slot, database) = db.write().await;
1410                slot.put(apply_and_finalize::<OrderedFixedDb>(database, merkleized).await);
1411            }
1412
1413            // Read set: key(1) updated, key(2) deleted, key(999) missing -> created.
1414            let read_keys = [key(1), key(2), key(999)];
1415            let keys: Vec<&Digest> = read_keys.iter().collect();
1416            let indexed_updates = vec![(0, Some(val(1_000))), (1, None), (2, Some(val(1_001)))];
1417            let upserts = vec![(key(3), Some(val(1_002)))];
1418
1419            // Explicit path.
1420            let mut explicit = db.new_batch_for_test::<_>().await;
1421            let explicit_values = explicit.get_many(&keys).await.unwrap();
1422            for (slot, value) in &indexed_updates {
1423                explicit = explicit.write(read_keys[*slot], *value);
1424            }
1425            for (k, v) in &upserts {
1426                explicit = explicit.write(*k, *v);
1427            }
1428            let explicit_root =
1429                crate::stateful::db::Unmerkleized::merkleize(explicit.with_metadata(metadata))
1430                    .await
1431                    .unwrap()
1432                    .root();
1433
1434            // Staged path, with metadata set on the staged handle.
1435            let staged_batch = db.new_batch_for_test::<_>().await;
1436            let split = 2;
1437            let (mut staged_values, staged) = staged_batch.stage(&keys[..split]).await.unwrap();
1438            let (range, suffix_values, staged) = staged.expand(&keys[split..]).await.unwrap();
1439            assert_eq!(range, split..keys.len());
1440            staged_values.extend(suffix_values);
1441            let staged_root = staged
1442                .with_metadata(metadata)
1443                .merkleize(indexed_updates.clone(), upserts.clone())
1444                .await
1445                .unwrap()
1446                .root();
1447
1448            assert_eq!(explicit_values, staged_values);
1449            assert_eq!(explicit_root, staged_root);
1450
1451            // Metadata set before staging must be carried through to staged merkleize.
1452            let carried_batch = db.new_batch_for_test::<_>().await.with_metadata(metadata);
1453            let (carried_values, staged) = carried_batch.stage(&keys).await.unwrap();
1454            let carried_root = staged
1455                .merkleize(indexed_updates.clone(), upserts.clone())
1456                .await
1457                .unwrap()
1458                .root();
1459            assert_eq!(explicit_values, carried_values);
1460            assert_eq!(explicit_root, carried_root);
1461        });
1462    }
1463
1464    #[test]
1465    fn ordered_variable_managed_db_applies_batch_and_proves_exclusion() {
1466        deterministic::Runner::default().start(|context| async move {
1467            let config = variable_config("ordered-variable-managed-db", &context);
1468            let db = <OrderedVariableDb as ManagedDb<_>>::init(context.child("db"), config)
1469                .await
1470                .unwrap();
1471            let db = Shared::new("test", db);
1472            let key = Sha256::hash(&[b"key"]);
1473            let value = Sha256::hash(&[b"value"]);
1474            let metadata = Sha256::hash(&[b"metadata"]);
1475            let missing = Sha256::hash(&[b"missing"]);
1476
1477            let batch = db
1478                .new_batch_for_test::<_>()
1479                .await
1480                .write(key, Some(value))
1481                .with_metadata(metadata);
1482            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
1483                .await
1484                .unwrap();
1485            let expected_root = merkleized.root();
1486
1487            {
1488                let (slot, database) = db.write().await;
1489                slot.put(apply_and_finalize::<OrderedVariableDb>(database, merkleized).await);
1490            }
1491
1492            let guard = db.read().await;
1493            assert_eq!(guard.root(), expected_root);
1494            assert_eq!(guard.get(&key).await.unwrap(), Some(value));
1495
1496            let proof = guard.exclusion_proof(&missing).await.unwrap();
1497            assert!(OrderedVariableDb::verify_exclusion_proof(
1498                &missing,
1499                &proof,
1500                &guard.root(),
1501            ));
1502        });
1503    }
1504
1505    #[test]
1506    fn ordered_managed_db_matches_sync_target_rejects_wrong_ops_root_and_range() {
1507        deterministic::Runner::default().start(|context| async move {
1508            let config = fixed_config("ordered-matches-sync-target", &context);
1509            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config.clone())
1510                .await
1511                .unwrap();
1512            let db = Shared::new("test", db);
1513
1514            let key = Sha256::hash(&[b"key"]);
1515            let value = Sha256::hash(&[b"value"]);
1516            let metadata = Sha256::hash(&[b"metadata"]);
1517
1518            let batch = db
1519                .new_batch_for_test::<_>()
1520                .await
1521                .write(key, Some(value))
1522                .with_metadata(metadata);
1523            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
1524                .await
1525                .unwrap();
1526
1527            let verification_db =
1528                <OrderedFixedDb as ManagedDb<_>>::init(context.child("verification_db"), config)
1529                    .await
1530                    .unwrap();
1531            let (verification_db, _) = verification_db
1532                .apply_batch(merkleized.inner.clone())
1533                .await
1534                .unwrap();
1535            let verification_db = verification_db.sync().await.unwrap();
1536
1537            let valid_target = <OrderedFixedDb as ManagedDb<_>>::sync_target(&verification_db);
1538            assert!(<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
1539                &merkleized,
1540                &valid_target,
1541            ));
1542
1543            let mut wrong_root = valid_target.clone();
1544            wrong_root.root = Sha256::hash(&[b"wrong ops root"]);
1545            assert!(!<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
1546                &merkleized,
1547                &wrong_root,
1548            ));
1549
1550            let mut wrong_range = valid_target.clone();
1551            wrong_range.range =
1552                non_empty_range!(valid_target.range.start(), valid_target.range.end() + 1);
1553            assert!(!<OrderedFixedDb as ManagedDb<_>>::matches_sync_target(
1554                &merkleized,
1555                &wrong_range,
1556            ));
1557        });
1558    }
1559
1560    #[test]
1561    fn ordered_managed_db_rewind_to_target_round_trips() {
1562        deterministic::Runner::default().start(|context| async move {
1563            let config = fixed_config("ordered-rewind-round-trip", &context);
1564            let db = <OrderedFixedDb as ManagedDb<_>>::init(context.child("db"), config)
1565                .await
1566                .unwrap();
1567            let db = Shared::new("test", db);
1568
1569            let key1 = Sha256::hash(&[b"key1"]);
1570            let value1 = Sha256::hash(&[b"value1"]);
1571            let metadata1 = Sha256::hash(&[b"metadata1"]);
1572            let batch1 = db
1573                .new_batch_for_test::<_>()
1574                .await
1575                .write(key1, Some(value1))
1576                .with_metadata(metadata1);
1577            let merkleized1 = crate::stateful::db::Unmerkleized::merkleize(batch1)
1578                .await
1579                .unwrap();
1580            {
1581                let (slot, database) = db.write().await;
1582                slot.put(apply_and_finalize::<OrderedFixedDb>(database, merkleized1).await);
1583            }
1584            let target_after_first = {
1585                let guard = db.read().await;
1586                <OrderedFixedDb as ManagedDb<_>>::sync_target(&guard)
1587            };
1588
1589            let key2 = Sha256::hash(&[b"key2"]);
1590            let value2 = Sha256::hash(&[b"value2"]);
1591            let metadata2 = Sha256::hash(&[b"metadata2"]);
1592            let batch2 = db
1593                .new_batch_for_test::<_>()
1594                .await
1595                .write(key2, Some(value2))
1596                .with_metadata(metadata2);
1597            let merkleized2 = crate::stateful::db::Unmerkleized::merkleize(batch2)
1598                .await
1599                .unwrap();
1600            {
1601                let (slot, database) = db.write().await;
1602                slot.put(apply_and_finalize::<OrderedFixedDb>(database, merkleized2).await);
1603            }
1604
1605            {
1606                let (slot, database) = db.write().await;
1607                slot.put(
1608                    <OrderedFixedDb as ManagedDb<_>>::rewind_to_target(
1609                        database,
1610                        target_after_first.clone(),
1611                    )
1612                    .await
1613                    .unwrap(),
1614                );
1615            }
1616            let target_after_rewind = {
1617                let guard = db.read().await;
1618                <OrderedFixedDb as ManagedDb<_>>::sync_target(&guard)
1619            };
1620            assert_eq!(target_after_rewind, target_after_first);
1621        });
1622    }
1623
1624    #[test]
1625    fn managed_db_matches_sync_target_rejects_wrong_ops_root_and_range() {
1626        deterministic::Runner::default().start(|context| async move {
1627            let config = fixed_config("matches-sync-target", &context);
1628            let db = FixedDb::init(context.child("db"), config.clone())
1629                .await
1630                .unwrap();
1631            let db = Shared::new("test", db);
1632
1633            let key = Sha256::hash(&[b"key"]);
1634            let value = Sha256::hash(&[b"value"]);
1635            let metadata = Sha256::hash(&[b"metadata"]);
1636
1637            let batch = db
1638                .new_batch_for_test::<_>()
1639                .await
1640                .write(key, Some(value))
1641                .with_metadata(metadata);
1642            let merkleized = crate::stateful::db::Unmerkleized::merkleize(batch)
1643                .await
1644                .unwrap();
1645
1646            let verification_db = FixedDb::init(context.child("verification_db"), config)
1647                .await
1648                .unwrap();
1649            let (verification_db, _) = verification_db
1650                .apply_batch(merkleized.inner.clone())
1651                .await
1652                .unwrap();
1653            let verification_db = verification_db.sync().await.unwrap();
1654
1655            let valid_target = <FixedDb as ManagedDb<_>>::sync_target(&verification_db);
1656            assert!(<FixedDb as ManagedDb<_>>::matches_sync_target(
1657                &merkleized,
1658                &valid_target,
1659            ));
1660
1661            let mut wrong_root = valid_target.clone();
1662            wrong_root.root = Sha256::hash(&[b"wrong ops root"]);
1663            assert!(!<FixedDb as ManagedDb<_>>::matches_sync_target(
1664                &merkleized,
1665                &wrong_root,
1666            ));
1667
1668            let mut wrong_range = valid_target.clone();
1669            wrong_range.range =
1670                non_empty_range!(valid_target.range.start(), valid_target.range.end() + 1);
1671            assert!(!<FixedDb as ManagedDb<_>>::matches_sync_target(
1672                &merkleized,
1673                &wrong_range,
1674            ));
1675        });
1676    }
1677}
commonware_glue/stateful/db/current.rs

commonware_glue/stateful/db/
current.rs
