//! Journaled [`ManagedDb`] and [`StateSyncDb`] implementations for QMDB //! [`immutable`](commonware_storage::qmdb::immutable) databases. //! //! Immutable databases accept new keyed values but no updates or deletions. Batch reads consult //! the database at the time of the read for keys their batch chain does not write, and return a //! `StaleRead` error once the database is at no state of that chain. use crate::stateful::db::{ BatchContext, InitError, ManagedDb, Merkleized as MerkleizedTrait, Shared, StateSyncDb, SyncEngineConfig, Unmerkleized as UnmerkleizedTrait, sync_standard_db, validate_initialization, }; use commonware_codec::{EncodeShared, Read as CodecRead}; use commonware_cryptography::Hasher; use commonware_parallel::Strategy; use commonware_runtime::Handle; use commonware_storage::{ Context, journal::contiguous::{ Mutable, fixed::Journal as FixedJournal, variable::Journal as VariableJournal, }, merkle::{Family, Location}, qmdb::{ Error, any::value::{FixedEncoding, FixedValue, ValueEncoding, VariableEncoding, VariableValue}, immutable::{ Immutable, Operation, batch::{MerkleizedBatch, UnmerkleizedBatch}, fixed, initial_root, variable, }, operation::Key, sync::{self, Target as AnySyncTarget}, }, translator::Translator, }; use commonware_utils::{ Array, channel::{mpsc, ring}, non_empty_range, }; use std::{ops::Deref, sync::Arc}; /// Shared handle to an immutable database. type ImmutableDbHandle = Shared>; /// A speculative batch of new keyed values over a shared immutable database. pub struct ImmutableUnmerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { batch: UnmerkleizedBatch, db: ImmutableDbHandle, metadata: Option, inactivity_floor: Option>, } impl Deref for ImmutableUnmerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { type Target = UnmerkleizedBatch; fn deref(&self) -> &Self::Target { &self.batch } } impl ImmutableUnmerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { /// Sets the metadata committed by [`merkleize`](UnmerkleizedTrait::merkleize). pub fn with_metadata(mut self, metadata: V::Value) -> Self { self.metadata = Some(metadata); self } /// Sets the inactivity floor committed by [`merkleize`](UnmerkleizedTrait::merkleize) /// (location 0 when unset). pub const fn with_inactivity_floor(mut self, floor: Location) -> Self { self.inactivity_floor = Some(floor); self } /// Reads a value by key, falling back to applied state. pub async fn get(&self, key: &K) -> Result, Error> { let db = self.db.read().await; self.batch.get(key, &db).await } /// Reads multiple values by key, falling back to applied state. /// /// Returns results in the same order as `keys`. pub async fn get_many(&self, keys: &[&K]) -> Result>, Error> { let db = self.db.read().await; self.batch.get_many(keys, &db).await } /// Sets `key` to `value` in the batch. pub fn set(mut self, key: K, value: V::Value) -> Self { self.batch = self.batch.set(key, value); self } } /// A sealed immutable batch with a computed root. pub struct ImmutableMerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { inner: Arc>, db: ImmutableDbHandle, } impl Clone for ImmutableMerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { fn clone(&self) -> Self { Self { inner: Arc::clone(&self.inner), db: self.db.clone(), } } } impl Deref for ImmutableMerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { type Target = MerkleizedBatch; fn deref(&self) -> &Self::Target { &self.inner } } impl ImmutableMerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { /// Reads a value by key, falling back to applied state. pub async fn get(&self, key: &K) -> Result, Error> { let db = self.db.read().await; self.inner.get(key, &db).await } /// Reads multiple values by key, falling back to applied state. /// /// Returns results in the same order as `keys`. pub async fn get_many(&self, keys: &[&K]) -> Result>, Error> { let db = self.db.read().await; self.inner.get_many(keys, &db).await } } impl UnmerkleizedTrait for ImmutableUnmerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { type Merkleized = ImmutableMerkleized; type Error = Error; async fn merkleize(self) -> Result> { let db = self.db.read().await; let merkleized = self .batch .merkleize( &db, self.metadata, self.inactivity_floor.unwrap_or_default(), ) .await?; Ok(ImmutableMerkleized { inner: merkleized, db: self.db.clone(), }) } } impl MerkleizedTrait for ImmutableMerkleized where F: Family, E: Context, K: Key, V: ValueEncoding, C: Mutable>, H: Hasher, T: Translator, S: Strategy, Operation: EncodeShared, { type Digest = H::Digest; type Unmerkleized = ImmutableUnmerkleized; fn root(&self) -> H::Digest { self.inner.root() } fn new_batch(&self) -> Self::Unmerkleized { ImmutableUnmerkleized { batch: self.inner.new_batch::(), db: self.db.clone(), metadata: None, inactivity_floor: None, } } } impl ManagedDb for fixed::Db where F: Family, E: Context, K: Array, V: FixedValue, H: Hasher, T: Translator, S: Strategy, { type Unmerkleized = ImmutableUnmerkleized< F, E, K, FixedEncoding, FixedJournal>, H, T, S, >; type Merkleized = ImmutableMerkleized< F, E, K, FixedEncoding, FixedJournal>, H, T, S, >; type Error = Error; type Config = fixed::Config; type SyncTarget = AnySyncTarget; async fn init( context: E, config: Self::Config, expected: Option, ) -> Result, Self::SyncTarget>> { let db = ::init( context, config, expected.as_ref().map(|target| target.range.end()), ) .await .map_err(InitError::Database)?; validate_initialization(db, expected) } fn initial_sync_target() -> Self::SyncTarget { AnySyncTarget::new( initial_root::, H>(), non_empty_range!(Location::new(0), Location::new(1)), ) } fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized { let (database, shared) = database.into_parts(); ImmutableUnmerkleized { batch: database.new_batch(), db: shared, metadata: None, inactivity_floor: None, } } fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool { batch.root() == target.root && *target.range.start() == batch.bounds().inactivity_floor && *target.range.end() == batch.bounds().tip.size } async fn apply(self, batch: Self::Merkleized) -> Result> { self.apply_batch(batch.inner).await } async fn finalize(self) -> Result<(Self, Handle<()>), Error> { self.start_sync().await } async fn prune(self, target: &Self::SyncTarget) -> Result> { self.prune((*target.range.start()).into()).await } fn sync_target(&self) -> Self::SyncTarget { let bounds = self.bounds(); AnySyncTarget::new( self.root(), non_empty_range!(self.sync_boundary(), bounds.end), ) } } impl ManagedDb for variable::Db where F: Family, E: Context, K: Key, V: VariableValue, H: Hasher, T: Translator, S: Strategy, { type Unmerkleized = ImmutableUnmerkleized< F, E, K, VariableEncoding, VariableJournal>, H, T, S, >; type Merkleized = ImmutableMerkleized< F, E, K, VariableEncoding, VariableJournal>, H, T, S, >; type Error = Error; type Config = variable::Config as CodecRead>::Cfg, S>; type SyncTarget = AnySyncTarget; async fn init( context: E, config: Self::Config, expected: Option, ) -> Result, Self::SyncTarget>> { let db = ::init( context, config, expected.as_ref().map(|target| target.range.end()), ) .await .map_err(InitError::Database)?; validate_initialization(db, expected) } fn initial_sync_target() -> Self::SyncTarget { AnySyncTarget::new( initial_root::, H>(), non_empty_range!(Location::new(0), Location::new(1)), ) } fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized { let (database, shared) = database.into_parts(); ImmutableUnmerkleized { batch: database.new_batch(), db: shared, metadata: None, inactivity_floor: None, } } fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool { batch.root() == target.root && *target.range.start() == batch.bounds().inactivity_floor && *target.range.end() == batch.bounds().tip.size } async fn apply(self, batch: Self::Merkleized) -> Result> { self.apply_batch(batch.inner).await } async fn finalize(self) -> Result<(Self, Handle<()>), Error> { self.start_sync().await } async fn prune(self, target: &Self::SyncTarget) -> Result> { self.prune((*target.range.start()).into()).await } fn sync_target(&self) -> Self::SyncTarget { let bounds = self.bounds(); AnySyncTarget::new( self.root(), non_empty_range!(self.sync_boundary(), bounds.end), ) } } impl StateSyncDb for fixed::Db where F: Family, E: Context, K: Array, V: FixedValue, H: Hasher, T: Translator, S: Strategy, R: sync::SourceFor, { type SyncError = sync::Error; async fn sync_db( context: E, config: Self::Config, source: R, target: Self::SyncTarget, tip_updates: ring::Receiver, finish: Option>, reached_target: Option>, sync_config: SyncEngineConfig, ) -> Result { sync_standard_db( context, config, source, target, tip_updates, finish, reached_target, sync_config, ) .await } } impl StateSyncDb for variable::Db where F: Family, E: Context, K: Key, V: VariableValue, H: Hasher, T: Translator, S: Strategy, R: sync::SourceFor, { type SyncError = sync::Error; async fn sync_db( context: E, config: Self::Config, source: R, target: Self::SyncTarget, tip_updates: ring::Receiver, finish: Option>, reached_target: Option>, sync_config: SyncEngineConfig, ) -> Result { sync_standard_db( context, config, source, target, tip_updates, finish, reached_target, sync_config, ) .await } }