commonware_storage/qmdb/any/sync/
mod.rs

1//! Shared synchronization logic for [crate::qmdb::any] databases.
2//! Contains implementation of [crate::qmdb::sync::Database] for all [Db] variants
3//! (ordered/unordered, fixed/variable).
4//!
5//! Callers verifying `any` sync proofs directly should use [`crate::qmdb::verify_proof`].
6
7use crate::{
8    Context,
9    index::Factory as IndexFactory,
10    journal::{
11        authenticated,
12        contiguous::{Contiguous, Mutable, fixed, variable},
13    },
14    merkle::{self, Location, full},
15    qmdb::{
16        self,
17        any::{
18            FixedConfig, FixedValue, VariableConfig, VariableValue,
19            db::Db,
20            operation::{Operation, update::Update},
21            ordered::{
22                fixed::{
23                    Db as OrderedFixedDb, Operation as OrderedFixedOp, Update as OrderedFixedUpdate,
24                },
25                variable::{
26                    Db as OrderedVariableDb, Operation as OrderedVariableOp,
27                    Update as OrderedVariableUpdate,
28                },
29            },
30            unordered::{
31                fixed::{
32                    Db as UnorderedFixedDb, Operation as UnorderedFixedOp,
33                    Update as UnorderedFixedUpdate,
34                },
35                variable::{
36                    Db as UnorderedVariableDb, Operation as UnorderedVariableOp,
37                    Update as UnorderedVariableUpdate,
38                },
39            },
40        },
41        metrics::Metrics,
42        operation::Key,
43    },
44    translator::Translator,
45};
46use commonware_codec::{Codec, CodecShared, Read as CodecRead};
47use commonware_cryptography::Hasher;
48use commonware_parallel::Strategy;
49use commonware_runtime::Spawner;
50use commonware_utils::{Array, range::NonEmptyRange};
51use core::num::{NonZeroU64, NonZeroUsize};
52
53#[cfg(test)]
54pub(crate) mod tests;
55
56/// Shared helper to build a [Db] from sync components.
57#[allow(clippy::too_many_arguments)]
58async fn build_db<F, E, U, I, H, C, S>(
59    context: E,
60    merkle_config: full::Config<S>,
61    log: C,
62    translator: I::Translator,
63    pinned_nodes: Option<Vec<H::Digest>>,
64    range: NonEmptyRange<Location<F>>,
65    apply_batch_size: NonZeroU64,
66    init_concurrency: <I as crate::qmdb::SnapshotBuild<F>>::Concurrency,
67    init_buffer: NonZeroUsize,
68    cache_size: Option<NonZeroUsize>,
69) -> Result<Db<F, E, C, I, H, U, { crate::qmdb::any::BITMAP_CHUNK_BYTES }, S>, qmdb::Error<F>>
70where
71    F: merkle::Family,
72    E: Context + Spawner,
73    U: Update,
74    I: IndexFactory + crate::qmdb::SnapshotBuild<F>,
75    H: Hasher,
76    C: Mutable<Item = Operation<F, U>> + 'static,
77    S: Strategy,
78    Operation<F, U>: Codec,
79{
80    let hasher = qmdb::hasher::<H>();
81
82    let merkle = full::Merkle::<F, _, _, S>::init_sync(
83        context.child("merkle"),
84        full::SyncConfig {
85            config: merkle_config,
86            range: range.clone(),
87            pinned_nodes,
88        },
89    )
90    .await?;
91
92    let index = I::new(context.child("index"), translator);
93
94    let log = authenticated::Journal::<F, _, _, _, S>::from_components(
95        merkle,
96        log,
97        hasher,
98        apply_batch_size.get(),
99    )
100    .await?;
101    let snapshot_context = context.child("snapshot");
102    let metrics = Metrics::new(context);
103    let db = Db::init_from_log(
104        snapshot_context,
105        index,
106        log,
107        None,
108        init_concurrency,
109        init_buffer,
110        cache_size,
111        metrics,
112    )
113    .await?;
114
115    Ok(db)
116}
117
118macro_rules! impl_sync_database {
119    ($db:ident, $op:ident, $update:ident,
120     $journal:ty, $config:ty,
121     $key_bound:path, $value_bound:ident
122     $(; $($where_extra:tt)+)?) => {
123        impl<F, E, K, V, H, T, S> qmdb::sync::Database for $db<F, E, K, V, H, T, S>
124        where
125            F: merkle::Family,
126            E: Context + Spawner,
127            K: $key_bound,
128            V: $value_bound + 'static,
129            H: Hasher,
130            T: Translator,
131            S: Strategy,
132            $($($where_extra)+)?
133        {
134            type Family = F;
135            type Context = E;
136            type Op = $op<F, K, V>;
137            type Journal = $journal;
138            type Hasher = H;
139            type Config = $config;
140            type Digest = H::Digest;
141
142            async fn from_sync_result(
143                context: Self::Context,
144                config: Self::Config,
145                log: Self::Journal,
146                pinned_nodes: Option<Vec<Self::Digest>>,
147                range: NonEmptyRange<Location<F>>,
148                apply_batch_size: NonZeroU64,
149            ) -> Result<Self, qmdb::Error<F>> {
150                let merkle_config = config.merkle_config.clone();
151                let translator = config.translator.clone();
152                let cache_size = config.init_cache_size;
153                let init_buffer = config.init_buffer;
154                let init_concurrency = config.init_concurrency;
155                build_db::<F, _, $update<K, V>, _, H, _, S>(
156                    context,
157                    merkle_config,
158                    log,
159                    translator,
160                    pinned_nodes,
161                    range,
162                    apply_batch_size,
163                    init_concurrency,
164                    init_buffer,
165                    cache_size,
166                )
167                .await
168            }
169
170            async fn persist_sync_result(self) -> Result<Self, qmdb::Error<F>> {
171                Ok(self)
172            }
173
174            async fn local_pinned_nodes(
175                context: Self::Context,
176                config: &Self::Config,
177                target: &qmdb::sync::Target<Self::Family, Self::Digest>,
178                journal: &Self::Journal,
179            ) -> Result<Option<Vec<Self::Digest>>, qmdb::Error<F>> {
180                if target.range.start() == Location::new(0)
181                    || !qmdb::sync::journal_covers_range(journal.bounds(), &target.range)
182                {
183                    return Ok(None);
184                }
185
186                // The target's range starts at the inactivity floor.
187                qmdb::sync::local_pinned_nodes::<F, _, H, S>(
188                    context,
189                    config.merkle_config.clone(),
190                    target,
191                    target.range.start(),
192                )
193                .await
194            }
195
196            fn root(&self) -> Self::Digest {
197                crate::qmdb::any::db::Db::root(self)
198            }
199        }
200    };
201}
202
203impl_sync_database!(
204    UnorderedFixedDb, UnorderedFixedOp, UnorderedFixedUpdate,
205    fixed::Journal<E, Self::Op>, FixedConfig<T, S>,
206    Array, FixedValue
207);
208
209impl_sync_database!(
210    UnorderedVariableDb, UnorderedVariableOp, UnorderedVariableUpdate,
211    variable::Journal<E, Self::Op>,
212    VariableConfig<T, <UnorderedVariableOp<F, K, V> as CodecRead>::Cfg, S>,
213    Key, VariableValue;
214    UnorderedVariableOp<F, K, V>: CodecShared
215);
216
217impl_sync_database!(
218    OrderedFixedDb, OrderedFixedOp, OrderedFixedUpdate,
219    fixed::Journal<E, Self::Op>, FixedConfig<T, S>,
220    Array, FixedValue
221);
222
223impl_sync_database!(
224    OrderedVariableDb, OrderedVariableOp, OrderedVariableUpdate,
225    variable::Journal<E, Self::Op>,
226    VariableConfig<T, <OrderedVariableOp<F, K, V> as CodecRead>::Cfg, S>,
227    Key, VariableValue;
228    OrderedVariableOp<F, K, V>: CodecShared
229);
commonware_storage/qmdb/any/sync/mod.rs

commonware_storage/qmdb/any/sync/
mod.rs
