use crate::{ journal::{ authenticated, contiguous::{Contiguous as _, Mutable}, }, merkle::{ full::{self, Merkle}, Family, Location, }, qmdb::{ self, any::value::ValueEncoding, keyless::{operation::Codec, CompactDb, Keyless, Metrics, Operation}, sync, }, Context, }; use commonware_codec::{EncodeShared, Read}; use commonware_cryptography::Hasher; use commonware_parallel::Strategy; use commonware_utils::range::NonEmptyRange; impl sync::Database for Keyless where F: Family, E: Context, V: ValueEncoding + Codec, C: Mutable> + sync::Journal>, C::Config: Clone + Send, H: Hasher, S: Strategy, Operation: EncodeShared, { type Family = F; type Op = Operation; type Journal = C; type Hasher = H; type Config = super::Config; type Digest = H::Digest; type Context = E; /// Returns a [Keyless] db initialized from data collected in the sync process. /// /// # Behavior /// /// This method handles different initialization scenarios based on existing data: /// - If the Merkle journal is empty or the last item is before the range start, it creates /// a fresh Merkle structure from the provided `pinned_nodes` /// - If the Merkle journal has data but is incomplete (has length < range end), missing /// operations from the log are applied to bring it up to the target state /// - If the Merkle journal has data beyond the range end, it is rewound to match the sync /// target /// /// # Returns /// /// A [Keyless] db populated with the state from the given range. async fn from_sync_result( context: Self::Context, config: Self::Config, log: Self::Journal, pinned_nodes: Option>, range: NonEmptyRange>, apply_batch_size: usize, ) -> Result> { let hasher = qmdb::hasher::(); let merkle = Merkle::::init_sync( context.child("merkle"), full::SyncConfig { config: config.merkle.clone(), range: range.clone(), pinned_nodes, }, ) .await?; let journal = authenticated::Journal::::from_components( merkle, log, hasher, apply_batch_size as u64, ) .await?; let (last_commit_loc, inactivity_floor_loc) = { let bounds = journal.bounds(); let loc = bounds .end .checked_sub(1) .ok_or(qmdb::Error::HistoricalFloorPruned(Location::new( bounds.end, )))?; let floor = qmdb::find_inactivity_floor_at::(&journal, Location::new(bounds.end), |op| { op.has_floor() }) .await?; (Location::new(loc), floor) }; let inactive_peaks = F::inactive_peaks( F::location_to_position(Location::new(*last_commit_loc + 1)), inactivity_floor_loc, ); let root = journal.root(inactive_peaks)?; let metrics = Metrics::new(context); let mut db = Self { journal, root, last_commit_loc, inactivity_floor_loc, metrics, }; db.update_metrics(); db.sync().await?; Ok(db) } async fn local_boundary_nodes( context: Self::Context, config: &Self::Config, target: &sync::Target, journal: &Self::Journal, ) -> Result>, qmdb::Error> { if target.range.start() == Location::new(0) || !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(), |op| { op.has_floor() }) .await?; sync::local_boundary_nodes::( context, config.merkle.clone(), target, inactivity_floor, ) .await } fn root(&self) -> Self::Digest { self.root() } } impl sync::compact::Database for CompactDb where F: Family, E: Context, V: ValueEncoding + Codec, H: Hasher, S: Strategy, Operation: EncodeShared, Operation: Read, Cfg: Clone + Send + Sync + 'static, { type Family = F; type Op = Operation; type Config = super::CompactConfig; type Digest = H::Digest; type Context = E; type Hasher = H; async fn from_validated_state( context: Self::Context, config: Self::Config, state: sync::compact::ValidatedState, ) -> Result> { let journal: crate::qmdb::compact::witness::Journal = crate::journal::contiguous::variable::Journal::init( context.child("witness"), config.witness, ) .await?; Self::init_from_validated_state(config.strategy, journal, config.commit_codec_config, state) } fn inactivity_floor(op: &Self::Op) -> Option> { op.has_floor() } fn root(&self) -> Self::Digest { self.root() } async fn persist_compact_state(&mut self) -> Result<(), qmdb::Error> { self.sync().await } } #[cfg(test)] mod tests;