//! Shared synchronization logic for [crate::qmdb::any] databases. //! Contains the implementation of [crate::qmdb::sync::Database] for [Db], covering every //! variant (ordered/unordered, fixed/variable, any snapshot index). //! //! Callers verifying `any` sync proofs directly should use [`crate::qmdb::verify_proof`]. use crate::{ Context, index::{Factory as IndexFactory, Unordered}, journal::{authenticated, contiguous::Mutable}, merkle::{self, Location, full}, qmdb::{ self, SnapshotBuild, any::{ Config, db::Db, operation::{Operation, update::Update}, }, metrics::Metrics, sync, }, }; use commonware_codec::Codec; use commonware_cryptography::Hasher; use commonware_parallel::Strategy; use commonware_runtime::Spawner; use commonware_utils::range::NonEmptyRange; use core::num::{NonZeroU64, NonZeroUsize}; #[cfg(test)] pub(crate) mod tests; /// Shared helper to build a [Db] from sync components. #[allow(clippy::too_many_arguments)] async fn build_db( context: E, merkle_config: full::Config, log: C, translator: I::Translator, pinned_nodes: Option>, range: NonEmptyRange>, apply_batch_size: NonZeroU64, init_concurrency: >::Concurrency, init_buffer: NonZeroUsize, cache_size: Option, ) -> Result, qmdb::Error> where F: merkle::Family, E: Context + Spawner, U: Update, I: IndexFactory + SnapshotBuild, H: Hasher, C: Mutable> + 'static, S: Strategy, Operation: Codec, { let hasher = qmdb::hasher::(); let merkle = full::Merkle::::init_sync( context.child("merkle"), full::SyncConfig { config: merkle_config, range: range.clone(), pinned_nodes, }, ) .await?; let index = I::new(context.child("index"), translator); let log = authenticated::Journal::::from_components( merkle, log, hasher, apply_batch_size, ) .await?; let snapshot_context = context.child("snapshot"); let metrics = Metrics::new(context); let db = Db::init_from_log( snapshot_context, index, log, None, init_concurrency, init_buffer, cache_size, metrics, ) .await?; Ok(db) } impl sync::Database for Db where F: merkle::Family, E: Context + Spawner, C: Mutable> + sync::Journal> + 'static, I: IndexFactory + SnapshotBuild + Unordered>, H: Hasher, U: Update, S: Strategy, Operation: Codec, { type Family = F; type Context = E; type Op = Operation; type Journal = C; type Hasher = H; type Config = Config< I::Translator, >::Config, S, >::Concurrency, >; type Digest = H::Digest; async fn from_sync_result( context: Self::Context, config: Self::Config, log: Self::Journal, pinned_nodes: Option>, range: NonEmptyRange>, apply_batch_size: NonZeroU64, ) -> Result> { build_db::( context, config.merkle_config, log, config.translator, pinned_nodes, range, apply_batch_size, config.init_concurrency, config.init_buffer, config.init_cache, ) .await } async fn persist_sync_result(self) -> Result> { Ok(self) } async fn local_pinned_nodes( context: Self::Context, config: &Self::Config, target: &qmdb::sync::Target, journal: &Self::Journal, ) -> Result>, qmdb::Error> { if target.range.start() == Location::new(0) || !qmdb::sync::journal_covers_range(journal.bounds(), &target.range) { return Ok(None); } // The inactivity floor is carried by the last commit operation rather than being // the target range's start. let inactivity_floor = qmdb::find_inactivity_floor_at::(journal, target.range.end()).await?; qmdb::sync::local_pinned_nodes::( context, config.merkle_config.clone(), target, inactivity_floor, ) .await } fn root(&self) -> Self::Digest { Self::root(self) } }