use crate::{ Context, journal::{authenticated, contiguous::Contiguous}, merkle::{Family, Location, MAX_PINNED_NODES, MAX_PROOF_DIGESTS_PER_ELEMENT, Proof}, qmdb::{self, operation::Floored, sync::ServeError}, }; use bytes::{Buf, BufMut}; use commonware_codec::{ EncodeShared, EncodeSize, Error as CodecError, Read, ReadExt as _, ReadRangeExt as _, Write, }; use commonware_cryptography::{Digest, Hasher}; use commonware_parallel::Strategy; use commonware_utils::{ Span, channel::oneshot, sync::{AsyncRwLock, TracedAsyncRwLock}, }; use std::{cmp::Ordering, future::Future, num::NonZeroU64, sync::Arc}; /// A request for operations from a source's log. pub enum Request { /// Fetch the operations in `[start, start + max_ops)`. Operations { /// Prove against the root the database had at this size. size: Location, /// First operation to return. start: Location, /// Maximum number of operations to return. max_ops: NonZeroU64, }, /// Fetch the single operation at `start` plus the pinned nodes at `start`, the lowest /// location the client will retain. The proof in the response authenticates the pinned nodes, /// so there is no way to request them on their own. Boundary { /// Prove against the root the database had at this size. size: Location, /// The operation to return, which is also the location of the returned pinned nodes. start: Location, }, } impl Request { /// The size whose root the response's proof must verify against. pub const fn size(&self) -> Location { match self { Self::Operations { size, .. } | Self::Boundary { size, .. } => *size, } } /// First operation to return. pub const fn start(&self) -> Location { match self { Self::Operations { start, .. } | Self::Boundary { start, .. } => *start, } } /// Maximum number of operations to return. pub const fn max_ops(&self) -> NonZeroU64 { match self { Self::Operations { max_ops, .. } => *max_ops, Self::Boundary { .. } => NonZeroU64::MIN, } } /// Total-order key for map lookups. The final component separates the variants. fn order_key(&self) -> (u64, u64, u64, bool) { ( *self.size(), *self.start(), self.max_ops().get(), matches!(self, Self::Boundary { .. }), ) } } impl Clone for Request { fn clone(&self) -> Self { *self } } impl Copy for Request {} impl PartialEq for Request { fn eq(&self, other: &Self) -> bool { self.order_key() == other.order_key() } } impl Eq for Request {} impl PartialOrd for Request { fn partial_cmp(&self, other: &Self) -> Option { Some(self.cmp(other)) } } impl Ord for Request { fn cmp(&self, other: &Self) -> Ordering { self.order_key().cmp(&other.order_key()) } } impl std::hash::Hash for Request { fn hash(&self, state: &mut H) { self.order_key().hash(state); } } impl std::fmt::Debug for Request { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Operations { size, start, max_ops, } => f .debug_struct("Operations") .field("size", size) .field("start", start) .field("max_ops", max_ops) .finish(), Self::Boundary { size, start } => f .debug_struct("Boundary") .field("size", size) .field("start", start) .finish(), } } } impl std::fmt::Display for Request { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Operations { size, start, max_ops, } => write!(f, "Operations(size={size}, start={start}, max={max_ops})"), Self::Boundary { size, start } => write!(f, "Boundary(size={size}, start={start})"), } } } impl Write for Request { fn write(&self, buf: &mut impl BufMut) { match self { Self::Operations { size, start, max_ops, } => { 0u8.write(buf); size.write(buf); start.write(buf); max_ops.write(buf); } Self::Boundary { size, start } => { 1u8.write(buf); size.write(buf); start.write(buf); } } } } impl EncodeSize for Request { fn encode_size(&self) -> usize { 1 + match self { Self::Operations { size, start, max_ops, } => size.encode_size() + start.encode_size() + max_ops.encode_size(), Self::Boundary { size, start } => size.encode_size() + start.encode_size(), } } } impl Read for Request { type Cfg = (); fn read_cfg(buf: &mut impl Buf, _: &()) -> Result { let request = match u8::read(buf)? { 0 => Self::Operations { size: Location::::read(buf)?, start: Location::::read(buf)?, max_ops: NonZeroU64::read(buf)?, }, 1 => Self::Boundary { size: Location::::read(buf)?, start: Location::::read(buf)?, }, d => return Err(CodecError::InvalidEnum(d)), }; if request.start() >= request.size() { return Err(CodecError::Invalid("Request", "start >= size")); } Ok(request) } } impl Span for Request {} #[cfg(feature = "arbitrary")] impl arbitrary::Arbitrary<'_> for Request { fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result { let size = u.int_in_range(1..=*F::MAX_LEAVES)?; let start = u.int_in_range(0..=size - 1)?; let size = Location::new(size); let start = Location::new(start); Ok(if u.arbitrary()? { Self::Boundary { size, start } } else { Self::Operations { size, start, max_ops: u.arbitrary()?, } }) } } /// One authenticated response, shaped like the [`Request`] it answers. /// /// In a [`Response::Boundary`], the proof, the operation, and the pinned nodes are verified as a /// unit. The pinned nodes are only believable because the proof folds them into digests it already /// commits to. pub enum Response { /// Answer to a [`Request::Operations`]. Operations { /// Proof authenticating `operations` against the root at the requested size. proof: Proof, /// The operations that were fetched. operations: Vec, }, /// Answer to a [`Request::Boundary`]. Boundary { /// Proof authenticating `op` against the root at the requested size. proof: Proof, /// The operation at the requested boundary. op: Op, /// Pinned nodes at the requested location. pinned_nodes: Vec, }, } impl Response { /// The proof authenticating this response. pub const fn proof(&self) -> &Proof { match self { Self::Operations { proof, .. } | Self::Boundary { proof, .. } => proof, } } } impl Clone for Response { fn clone(&self) -> Self { match self { Self::Operations { proof, operations } => Self::Operations { proof: proof.clone(), operations: operations.clone(), }, Self::Boundary { proof, op, pinned_nodes, } => Self::Boundary { proof: proof.clone(), op: op.clone(), pinned_nodes: pinned_nodes.clone(), }, } } } impl std::fmt::Debug for Response { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Operations { proof, operations } => f .debug_struct("Operations") .field("proof", proof) .field("operations", operations) .finish(), Self::Boundary { proof, op, pinned_nodes, } => f .debug_struct("Boundary") .field("proof", proof) .field("op", op) .field("pinned_nodes", pinned_nodes) .finish(), } } } impl Write for Response { fn write(&self, buf: &mut impl BufMut) { match self { Self::Operations { proof, operations } => { 0u8.write(buf); proof.write(buf); operations.write(buf); } Self::Boundary { proof, op, pinned_nodes, } => { 1u8.write(buf); proof.write(buf); op.write(buf); pinned_nodes.write(buf); } } } } impl EncodeSize for Response { fn encode_size(&self) -> usize { 1 + match self { Self::Operations { proof, operations } => { proof.encode_size() + operations.encode_size() } Self::Boundary { proof, op, pinned_nodes, } => proof.encode_size() + op.encode_size() + pinned_nodes.encode_size(), } } } impl Read for Response { /// The `max_ops` the request asked for, and the configuration for decoding one operation. type Cfg = (usize, Op::Cfg); fn read_cfg(buf: &mut impl Buf, (max_ops, op_cfg): &Self::Cfg) -> Result { match u8::read(buf)? { 0 => { let max_proof_digests = max_ops.saturating_mul(MAX_PROOF_DIGESTS_PER_ELEMENT); let proof = Proof::::read_cfg(buf, &max_proof_digests)?; let operations = Vec::::read_cfg(buf, &((..=*max_ops).into(), op_cfg.clone()))?; Ok(Self::Operations { proof, operations }) } 1 => { let proof = Proof::::read_cfg(buf, &MAX_PROOF_DIGESTS_PER_ELEMENT)?; let op = Op::read_cfg(buf, op_cfg)?; let pinned_nodes = Vec::::read_range(buf, ..=MAX_PINNED_NODES)?; Ok(Self::Boundary { proof, op, pinned_nodes, }) } d => Err(CodecError::InvalidEnum(d)), } } } #[cfg(feature = "arbitrary")] impl arbitrary::Arbitrary<'_> for Response where Op: for<'a> arbitrary::Arbitrary<'a>, D: for<'a> arbitrary::Arbitrary<'a>, { fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result { Ok(if u.arbitrary()? { Self::Boundary { proof: u.arbitrary()?, op: u.arbitrary()?, pinned_nodes: u.arbitrary()?, } } else { Self::Operations { proof: u.arbitrary()?, operations: u.arbitrary()?, } }) } } /// Where to report whether a response verified. /// /// After verifying a response, the sync engine sends `true` if it was valid and `false` if it /// was not, letting the [`Source`] provide feedback to whoever served it. `None` means the /// source accepts no feedback and its answer is final. pub type FeedbackTx = Option>; /// A source for proofs and operations. pub trait Source: Send + Sync { /// The merkle family backing this source's proofs. type Family: Family; /// The digest type used in this source's proofs. type Digest: Digest; /// The type of operations this source yields. type Op; /// Why this source could not answer. type Error: std::error::Error + Send + 'static; /// Serve a request. #[allow(clippy::type_complexity)] fn serve<'a>( &'a self, request: Request, ) -> impl Future< Output = Result<(Response, FeedbackTx), Self::Error>, > + Send + 'a; } impl Source for Arc where T: Source + ?Sized, { type Family = T::Family; type Digest = T::Digest; type Op = T::Op; type Error = T::Error; fn serve<'a>( &'a self, request: Request, ) -> impl Future< Output = Result<(Response, FeedbackTx), Self::Error>, > + Send + 'a { T::serve(self, request) } } impl Source for Option where T: Source, ServeError: From, { type Family = T::Family; type Digest = T::Digest; type Op = T::Op; type Error = ServeError; async fn serve( &self, request: Request, ) -> Result<(Response, FeedbackTx), Self::Error> { let source = self.as_ref().ok_or(ServeError::MissingSource)?; Ok(source.serve(request).await?) } } macro_rules! impl_locked_source { ($lock:ident) => { impl Source for $lock where T: Source, { type Family = T::Family; type Digest = T::Digest; type Op = T::Op; type Error = T::Error; async fn serve( &self, request: Request, ) -> Result<(Response, FeedbackTx), Self::Error> { self.read().await.serve(request).await } } }; } impl_locked_source!(AsyncRwLock); impl_locked_source!(TracedAsyncRwLock); impl Source for authenticated::Journal where F: Family, E: Context, C: Contiguous>, H: Hasher, S: Strategy, { type Family = F; type Digest = H::Digest; type Op = C::Item; type Error = qmdb::Error; #[allow(clippy::type_complexity)] #[tracing::instrument( name = "qmdb.sync.serve", level = "info", skip_all, fields( size = *request.size(), start = *request.start(), max_ops = request.max_ops().get(), ), )] async fn serve( &self, request: Request, ) -> Result<(Response, FeedbackTx), qmdb::Error> { // Reject before the floor lookup so the error carries the requested size and the // floor read never touches out-of-range locations. if request.size() > self.size() { return Err(crate::merkle::Error::RangeOutOfBounds(request.size()).into()); } let inactive_peaks = qmdb::inactive_peaks_at::(self, request.size()).await?; let response = match request { Request::Operations { size, start, max_ops, } => { let (proof, operations) = self .historical_proof(size, start, max_ops, inactive_peaks) .await?; Response::Operations { proof, operations } } Request::Boundary { size, start } => { let (proof, mut operations) = self .historical_proof(size, start, NonZeroU64::MIN, inactive_peaks) .await?; let op = operations .pop() .ok_or(crate::merkle::Error::RangeOutOfBounds(start))?; let pinned_nodes = self.merkle.pinned_nodes_at(start).await?; Response::Boundary { proof, op, pinned_nodes, } } }; Ok((response, None)) } } impl Source for crate::qmdb::any::db::Db where F: Family, E: Context, C: crate::journal::contiguous::Mutable>, I: crate::index::Unordered>, H: Hasher, U: crate::qmdb::any::operation::update::Update, S: Strategy, crate::qmdb::any::operation::Operation: commonware_codec::Codec, { type Family = F; type Digest = H::Digest; type Op = crate::qmdb::any::operation::Operation; type Error = qmdb::Error; async fn serve( &self, request: Request, ) -> Result<(Response, FeedbackTx), Self::Error> { self.log.serve(request).await } } #[cfg(test)] pub(crate) mod tests { use super::*; use crate::{ merkle::mmr, translator::{OneCap, TwoCap}, }; use commonware_codec::{Decode as _, DecodeExt as _, Encode as _}; use commonware_cryptography::{Sha256, sha256::Digest as ShaDigest}; use commonware_parallel::Rayon; use commonware_runtime::{Runner as _, deterministic}; use commonware_utils::{ NZU64, sync::{AsyncRwLock, TracedAsyncRwLock}, }; use std::{collections::VecDeque, marker::PhantomData, sync::Arc}; macro_rules! assert_source_variants { ($db:ty) => { assert_serves::>(); assert_serves::>>(); assert_serves::>>>(); assert_serves::>>(); assert_serves::>>>(); }; } fn assert_serves() {} /// A feedback slot whose receiver is dropped. It marks a response as feedback-accepting, /// so the engine retries instead of failing. pub fn dropped_feedback() -> FeedbackTx { let (tx, _rx) = oneshot::channel(); Some(tx) } /// A source that answers each request with the next scripted response. #[derive(Clone)] pub struct SequenceSource { #[allow(clippy::type_complexity)] responses: Arc, FeedbackTx)>>>, } impl SequenceSource { pub fn new(responses: Vec<(Response, FeedbackTx)>) -> Self { Self { responses: Arc::new(commonware_utils::sync::Mutex::new(VecDeque::from( responses, ))), } } } impl Source for SequenceSource where F: Family, D: Digest, Op: Send + Sync + Clone + 'static, { type Family = F; type Digest = D; type Op = Op; type Error = qmdb::Error; async fn serve( &self, _request: Request, ) -> Result<(Response, FeedbackTx), qmdb::Error> { self.responses .lock() .pop_front() .ok_or(qmdb::Error::DataCorrupted("missing scripted response")) } } /// Fetch `target`'s final commit operation and pinned nodes from `source`. pub async fn fetch_compact_state( source: &R, target: crate::qmdb::sync::CompactTarget, ) -> Result<(Response, FeedbackTx), R::Error> { source .serve(Request::Boundary { size: target.size, start: target.size - 1, }) .await } /// A source that always fails. Not `Clone`, which the engine must not require. pub struct FailSource { _phantom: PhantomData<(F, Op, D)>, } impl Source for FailSource where F: Family, D: Digest, Op: Send + Sync + Clone + 'static, { type Family = F; type Digest = D; type Op = Op; type Error = qmdb::Error; async fn serve( &self, _request: Request, ) -> Result<(Response, FeedbackTx), qmdb::Error> { Err(qmdb::Error::KeyNotFound) // Arbitrary dummy error } } impl FailSource { pub fn new() -> Self { Self { _phantom: PhantomData, } } } #[test] fn test_all_qmdb_variants_implement_source() { type AnyOrderedFixed = crate::qmdb::any::ordered::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, OneCap, Rayon, >; type AnyOrderedVariable = crate::qmdb::any::ordered::variable::Db< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, OneCap, Rayon, >; type AnyUnorderedFixed = crate::qmdb::any::unordered::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, TwoCap, Rayon, >; type AnyUnorderedVariable = crate::qmdb::any::unordered::variable::Db< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, TwoCap, Rayon, >; type CurrentOrderedFixed = crate::qmdb::current::ordered::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, OneCap, 32, Rayon, >; type CurrentOrderedVariable = crate::qmdb::current::ordered::variable::Db< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, OneCap, 32, Rayon, >; type CurrentUnorderedFixed = crate::qmdb::current::unordered::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, TwoCap, 32, Rayon, >; type CurrentUnorderedVariable = crate::qmdb::current::unordered::variable::Db< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, TwoCap, 32, Rayon, >; type ImmutableFixed = crate::qmdb::immutable::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, TwoCap, Rayon, >; type ImmutableVariable = crate::qmdb::immutable::variable::Db< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, TwoCap, Rayon, >; type KeylessFixed = crate::qmdb::keyless::fixed::Db< mmr::Family, deterministic::Context, ShaDigest, Sha256, Rayon, >; type KeylessVariable = crate::qmdb::keyless::variable::Db< mmr::Family, deterministic::Context, Vec, Sha256, Rayon, >; assert_source_variants!(AnyOrderedFixed); assert_source_variants!(AnyOrderedVariable); assert_source_variants!(AnyUnorderedFixed); assert_source_variants!(AnyUnorderedVariable); assert_source_variants!(CurrentOrderedFixed); assert_source_variants!(CurrentOrderedVariable); assert_source_variants!(CurrentUnorderedFixed); assert_source_variants!(CurrentUnorderedVariable); assert_source_variants!(ImmutableFixed); assert_source_variants!(ImmutableVariable); assert_source_variants!(KeylessFixed); assert_source_variants!(KeylessVariable); type KeylessFixedCompactDb = crate::qmdb::keyless::fixed::CompactDb< mmr::Family, deterministic::Context, ShaDigest, Sha256, Rayon, >; type KeylessVariableCompactDb = crate::qmdb::keyless::variable::CompactDb< mmr::Family, deterministic::Context, Vec, Sha256, (commonware_codec::RangeCfg, ()), Rayon, >; type ImmutableFixedCompactDb = crate::qmdb::immutable::fixed::CompactDb< mmr::Family, deterministic::Context, ShaDigest, ShaDigest, Sha256, Rayon, >; type ImmutableVariableCompactDb = crate::qmdb::immutable::variable::CompactDb< mmr::Family, deterministic::Context, ShaDigest, Vec, Sha256, ((), (commonware_codec::RangeCfg, ())), Rayon, >; assert_source_variants!(KeylessFixedCompactDb); assert_source_variants!(KeylessVariableCompactDb); assert_source_variants!(ImmutableFixedCompactDb); assert_source_variants!(ImmutableVariableCompactDb); } /// The request codec refuses frames whose start reaches their size, and unknown tags. #[test] fn test_request_decode_rejects_malformed() { let valid = Request::::Operations { size: Location::new(10), start: Location::new(3), max_ops: NZU64!(2), }; let decoded = Request::::decode(valid.encode()).unwrap(); assert_eq!(decoded, valid); let mut malformed = Vec::new(); 1u8.write(&mut malformed); // Boundary tag Location::::new(10).write(&mut malformed); Location::::new(10).write(&mut malformed); // start == size assert!(Request::::decode(&malformed[..]).is_err()); let bad_tag = [7u8]; assert!(Request::::decode(&bad_tag[..]).is_err()); } /// Requests are map keys, so equality and ordering must separate every distinct request. /// A `Boundary` differs from a one-op `Operations` at the same coordinates only by variant. #[test] fn test_request_identity() { let operations = Request::::Operations { size: Location::new(10), start: Location::new(3), max_ops: NZU64!(1), }; let boundary = Request::::Boundary { size: Location::new(10), start: Location::new(3), }; assert_eq!(boundary.max_ops(), NZU64!(1)); assert_ne!(operations, boundary); let mut set = std::collections::BTreeSet::new(); assert!(set.insert(operations)); assert!(set.insert(boundary)); assert!(!set.insert(operations)); assert_eq!(set.len(), 2); // Ordering is by size, then start, then max_ops. let smaller_size = Request::::Operations { size: Location::new(9), start: Location::new(8), max_ops: NZU64!(5), }; let smaller_start = Request::::Operations { size: Location::new(10), start: Location::new(2), max_ops: NZU64!(5), }; let fewer_ops = Request::::Operations { size: Location::new(10), start: Location::new(3), max_ops: NZU64!(2), }; let larger_ops = Request::::Operations { size: Location::new(10), start: Location::new(3), max_ops: NZU64!(5), }; assert!(smaller_size < smaller_start); assert!(smaller_start < fewer_ops); assert!(fewer_ops < larger_ops); } /// The response codec enforces the request-derived caps and rejects unknown tags. #[test] fn test_response_decode_rejects_malformed() { type R = Response; let digest = ShaDigest::from([7u8; 32]); let proof = Proof:: { leaves: Location::new(3), inactive_peaks: 0, digests: vec![digest], }; // More operations than the request's max_ops. let response = R::Operations { proof: proof.clone(), operations: vec![1, 2, 3], }; assert!(R::decode_cfg(response.encode(), &(3, ())).is_ok()); assert!(R::decode_cfg(response.encode(), &(2, ())).is_err()); // More proof digests than the request-derived budget. let oversized = Proof:: { leaves: Location::new(3), inactive_peaks: 0, digests: vec![digest; MAX_PROOF_DIGESTS_PER_ELEMENT + 1], }; let response = R::Operations { proof: oversized, operations: vec![1], }; assert!(R::decode_cfg(response.encode(), &(1, ())).is_err()); // More pinned nodes than the codec allows. let response = R::Boundary { proof: proof.clone(), op: 1, pinned_nodes: vec![digest; MAX_PINNED_NODES + 1], }; assert!(R::decode_cfg(response.encode(), &(1, ())).is_err()); let response = R::Boundary { proof, op: 1, pinned_nodes: vec![digest; MAX_PINNED_NODES], }; assert!(R::decode_cfg(response.encode(), &(1, ())).is_ok()); // Unknown tag. assert!(R::decode_cfg(&[9u8][..], &(1, ())).is_err()); } /// A source behind a lock reaches the source and reports its error. #[test] fn test_locked_source_reaches_source() { deterministic::Runner::default().start(|_context| async move { let lock = AsyncRwLock::new(FailSource::::new()); let request = Request::Operations { size: Location::new(1), start: Location::new(0), max_ops: NZU64!(1), }; let result = lock.serve(request).await; assert!(matches!(result, Err(crate::qmdb::Error::KeyNotFound))); }); } } #[cfg(all(test, feature = "arbitrary"))] mod conformance { use super::*; use crate::merkle::{mmb, mmr}; use commonware_codec::conformance::CodecConformance; use commonware_cryptography::sha256::Digest as Sha256Digest; commonware_conformance::conformance_tests! { CodecConformance>, CodecConformance>, CodecConformance>, CodecConformance>, } }