commonware_storage/qmdb/sync/

compact.rs

//! Compact sync for compact-storage qmdbs.
//!
//! Compact sync does not transfer or reconstruct the full historical operation log. Instead, the
//! source serves the minimum authenticated state needed to recreate the latest committed compact db
//! state:
//!
//! - the total committed leaf count,
//! - the compact frontier's pinned nodes for that leaf count,
//! - the final commit operation, and
//! - a proof authenticating that final commit against the requested root.
//!
//! # What compact dbs store
//!
//! A compact db's only persistent state is its witness journal (`qmdb::compact::witness`), whose
//! entries each snapshot one committed state (commit operation, proof, and frontier pins).
//! The in-memory compact Merkle ([`crate::merkle::compact`]) is rebuilt from the journal tip on
//! reopen. Without the witness, a compact db could recover its root and continue appending, but
//! it could not serve compact sync to another node.
//!
//! # When compact state changes
//!
//! The servable compact state advances only on durable persistence:
//!
//! - [`sync`] verifies the final commit proof and compact frontier before database construction.
//! - [`Database::from_validated_state`] reconstructs the already-validated state without
//!   persisting it.
//! - Compact db-local commits append one witness entry during `sync`.
//! - `rewind` restores the frontier and the witness from the target journal entry.
//!
//! Unsynced in-memory mutations are therefore intentionally not servable: `target()` and
//! compact-state responses lag behind `apply_batch()` until the db's next sync.
//!
//! # Safety and invariants
//!
//! The compact path relies on these invariants:
//!
//! - the served commit proof must authenticate the final commit at `leaf_count - 1`,
//! - reopen and rewind must re-verify the persisted witness against the root recomputed from the
//!   frontier rebuilt from the same journal entry, and
//! - reconstructed state must not be persisted until the db recomputes the requested root locally.
//!
//! If those invariants are violated by missing or corrupted persisted data, compact db reopen fails
//! with `DataCorrupted` rather than silently serving or restoring mismatched state.

use crate::{
    merkle::{Family, Location, Proof},
    qmdb::{
        self,
        any::{value::ValueEncoding, FixedValue, VariableValue},
        immutable::{
            fixed::{Db as ImmutableFixedDb, Operation as ImmutableFixedOp},
            variable::{Db as ImmutableVariableDb, Operation as ImmutableVariableOp},
            CompactDb as ImmutableCompactDb, Operation as ImmutableOp,
        },
        keyless::{
            fixed::{Db as KeylessFixedDb, Operation as KeylessFixedOp},
            variable::{Db as KeylessVariableDb, Operation as KeylessVariableOp},
            CompactDb as KeylessCompactDb, Operation as KeylessOp,
        },
        operation::Key,
        sync::{EngineError, Error},
        verify_proof,
    },
    translator::Translator,
};
use commonware_codec::{
    Encode, EncodeSize, Error as CodecError, RangeCfg, Read, ReadExt as _, Write,
};
use commonware_cryptography::{Digest, Hasher};
use commonware_macros::{boxed, select};
use commonware_parallel::Strategy;
use commonware_runtime::{reschedule, Buf, BufMut, Clock, Metrics, Storage, Supervisor};
use commonware_utils::{
    channel::{mpsc, oneshot},
    sync::{AsyncRwLock, TracedAsyncRwLock},
    Array,
};
use futures::future::{pending, Either};
use std::{future::Future, num::NonZeroU64, sync::Arc};

/// Compact-sync target for a compact-storage database.
///
/// Compact sync authenticates only the final committed root and total leaf count. Unlike replay
/// sync, there is no lower replay bound here because compact sync does not transfer or reconstruct
/// historical operations.
#[derive(Debug)]
pub struct Target<F: Family, D: Digest> {
    /// Authenticated root of the committed compact state.
    pub root: D,
    /// Total committed operations/leaves in that state.
    pub leaf_count: Location<F>,
}

impl<F: Family, D: Digest> Target<F, D> {
    const INVALID_LEAF_COUNT: &'static str = "leaf_count must be in 1..=MAX_LEAVES";

    /// Create a compact-sync target.
    pub const fn new(root: D, leaf_count: Location<F>) -> Self {
        Self { root, leaf_count }
    }

    /// Validate a compact target that may have been constructed programmatically.
    pub fn validate(&self) -> Result<(), &'static str> {
        if !self.leaf_count.is_valid() || self.leaf_count == 0 {
            return Err(Self::INVALID_LEAF_COUNT);
        }
        Ok(())
    }
}

impl<F: Family, D: Digest> Clone for Target<F, D> {
    fn clone(&self) -> Self {
        Self {
            root: self.root,
            leaf_count: self.leaf_count,
        }
    }
}

impl<F: Family, D: Digest> PartialEq for Target<F, D> {
    fn eq(&self, other: &Self) -> bool {
        self.root == other.root && self.leaf_count == other.leaf_count
    }
}

impl<F: Family, D: Digest> Eq for Target<F, D> {}

impl<F: Family, D: Digest> Write for Target<F, D> {
    fn write(&self, buf: &mut impl BufMut) {
        self.root.write(buf);
        self.leaf_count.write(buf);
    }
}

impl<F: Family, D: Digest> EncodeSize for Target<F, D> {
    fn encode_size(&self) -> usize {
        self.root.encode_size() + self.leaf_count.encode_size()
    }
}

impl<F: Family, D: Digest> Read for Target<F, D> {
    type Cfg = ();

    fn read_cfg(buf: &mut impl Buf, _: &()) -> Result<Self, CodecError> {
        let root = D::read(buf)?;
        let leaf_count = Location::<F>::read(buf)?;
        let target = Self { root, leaf_count };
        target.validate().map_err(|reason| {
            CodecError::Invalid("storage::qmdb::sync::compact::Target", reason)
        })?;
        Ok(target)
    }
}

#[cfg(feature = "arbitrary")]
impl<F: Family, D: Digest> arbitrary::Arbitrary<'_> for Target<F, D>
where
    D: for<'a> arbitrary::Arbitrary<'a>,
{
    fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
        let root = u.arbitrary()?;
        let leaf_count = Location::new(u.int_in_range(1..=*F::MAX_LEAVES)?);
        Ok(Self { root, leaf_count })
    }
}

/// Authenticated state for initializing a compact-storage database at a target root.
#[derive(Clone, Debug)]
pub struct State<F: Family, Op, D: Digest> {
    /// Total number of operations/leaves in the target database.
    pub leaf_count: Location<F>,
    /// Pinned Merkle nodes for the current frontier.
    pub pinned_nodes: Vec<D>,
    /// The final commit operation at `leaf_count - 1`.
    pub last_commit_op: Op,
    /// Proof authenticating `last_commit_op` against the target root.
    pub last_commit_proof: Proof<F, D>,
}

/// Compact state that has been validated against a target root.
#[derive(Clone, Debug)]
pub struct ValidatedState<F: Family, Op, D: Digest> {
    /// The compact state fetched from a peer after validation.
    pub state: State<F, Op, D>,
    /// The target root that `state` was validated against.
    pub root: D,
}

impl<F: Family, Op, D: Digest> Write for State<F, Op, D>
where
    Op: Write,
{
    fn write(&self, buf: &mut impl BufMut) {
        self.leaf_count.write(buf);
        self.pinned_nodes.write(buf);
        self.last_commit_op.write(buf);
        self.last_commit_proof.write(buf);
    }
}

/// Result from a compact-state fetch.
pub struct FetchResult<F: Family, Op, D: Digest> {
    /// The fetched compact state.
    pub state: State<F, Op, D>,
    /// Callback used to report whether downstream validated the state.
    pub callback: Option<oneshot::Sender<bool>>,
}

impl<F: Family, Op: std::fmt::Debug, D: Digest> std::fmt::Debug for FetchResult<F, Op, D> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FetchResult")
            .field("state", &self.state)
            .field("callback", &self.callback.as_ref().map(|_| "<callback>"))
            .finish()
    }
}

impl<F: Family, Op, D: Digest> From<State<F, Op, D>> for FetchResult<F, Op, D> {
    fn from(state: State<F, Op, D>) -> Self {
        Self {
            state,
            callback: None,
        }
    }
}

impl<F: Family, Op, D: Digest> EncodeSize for State<F, Op, D>
where
    Op: EncodeSize,
{
    fn encode_size(&self) -> usize {
        self.leaf_count.encode_size()
            + self.pinned_nodes.encode_size()
            + self.last_commit_op.encode_size()
            + self.last_commit_proof.encode_size()
    }
}

impl<F: Family, Op, D: Digest> Read for State<F, Op, D>
where
    Op: Read,
{
    type Cfg = (RangeCfg<usize>, Op::Cfg, usize);

    fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result<Self, CodecError> {
        let (pinned_nodes_cfg, op_cfg, max_proof_digests) = cfg;
        Ok(Self {
            leaf_count: Location::<F>::read(buf)?,
            pinned_nodes: Vec::<D>::read_cfg(buf, &(*pinned_nodes_cfg, ()))?,
            last_commit_op: Op::read_cfg(buf, op_cfg)?,
            last_commit_proof: Proof::<F, D>::read_cfg(buf, max_proof_digests)?,
        })
    }
}

/// Resolver-side errors for compact state serving.
#[derive(Debug, thiserror::Error)]
pub enum ServeError<F: Family, D: Digest> {
    /// The source database returned an error while building compact state.
    #[error("compact source database error: {0}")]
    Database(#[from] qmdb::Error<F>),
    /// The caller requested a target that compact sync cannot serve.
    #[error("invalid compact target: {0}")]
    InvalidTarget(&'static str),
    /// The resolver wrapper did not currently hold a database.
    #[error("compact source missing")]
    MissingSource,
    /// The caller requested a target different from the source's current witness.
    #[error("stale compact target - requested {requested:?}, current {current:?}")]
    StaleTarget {
        requested: Target<F, D>,
        current: Target<F, D>,
    },
}

/// Trait for compact sync fetches from a source database.
#[allow(clippy::type_complexity)]
pub trait Resolver: Send + Sync + Clone + 'static {
    /// The merkle family backing the resolver's proofs.
    type Family: Family;

    /// The digest type used in proofs returned by the resolver.
    type Digest: Digest;

    /// The type of operations returned by the resolver.
    type Op;

    /// The error type returned by the resolver.
    type Error: std::error::Error + Send + 'static;

    /// Fetch the authenticated state for `target`.
    fn get_compact_state<'a>(
        &'a self,
        target: Target<Self::Family, Self::Digest>,
    ) -> impl Future<Output = Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error>>
           + Send
           + 'a;
}

/// Marker trait for resolvers whose associated types match a specific compact-sync database.
///
/// This is a trait-alias pattern used to avoid repeating
/// `Resolver<Family = DB::Family, Op = DB::Op, Digest = DB::Digest>`.
/// Blanket-implemented for any matching [`Resolver`], so callers never implement this directly.
pub trait CompactDbResolver<DB: Database>:
    Resolver<Family = DB::Family, Op = DB::Op, Digest = DB::Digest>
{
}

impl<DB, R> CompactDbResolver<DB> for R
where
    DB: Database,
    R: Resolver<Family = DB::Family, Op = DB::Op, Digest = DB::Digest>,
{
}

/// Database types that can be initialized directly from compact state.
pub trait Database: Sized + Send {
    type Family: Family;
    type Op: Encode + Send;
    type Config: Clone;
    type Digest: Digest;
    type Context: Storage + Clock + Metrics;
    type Hasher: Hasher<Digest = Self::Digest>;

    /// Build a database from authenticated state in memory.
    ///
    /// The caller has already validated `last_commit_proof` and the compact frontier against the
    /// requested target root and passes the derived validation artifacts with the state. This
    /// constructor must not durably persist anything; persistence happens only after the caller
    /// re-checks that `Self::root()` matches the target root.
    fn from_validated_state(
        context: Self::Context,
        config: Self::Config,
        state: ValidatedState<Self::Family, Self::Op, Self::Digest>,
    ) -> impl Future<Output = Result<Self, qmdb::Error<Self::Family>>> + Send;

    /// Return the inactivity floor if the operation is a commit.
    fn inactivity_floor(op: &Self::Op) -> Option<Location<Self::Family>>;

    /// Get the root digest for final verification.
    fn root(&self) -> Self::Digest;

    /// Persist the compact-initialized state once the caller has verified its root.
    fn persist_compact_state(
        &mut self,
    ) -> impl Future<Output = Result<(), qmdb::Error<Self::Family>>> + Send;
}

/// Configuration for compact synchronization into a compact-storage database.
pub struct Config<DB, R>
where
    DB: Database,
    R: CompactDbResolver<DB>,
{
    /// Runtime context for creating database components.
    pub context: DB::Context,
    /// Source resolver for fetching compact authenticated state.
    pub resolver: R,
    /// Sync target (root digest and total leaf count).
    pub target: Target<DB::Family, DB::Digest>,
    /// Database-specific configuration.
    pub db_config: DB::Config,
    /// Channel for receiving sync target updates. Each update supersedes the
    /// current target, cancelling any in-flight attempt against it.
    pub update_rx: Option<mpsc::Receiver<Target<DB::Family, DB::Digest>>>,
    /// Channel that requests sync completion once the current target is reached.
    ///
    /// When `None`, sync completes as soon as the target is reached.
    pub finish_rx: Option<mpsc::Receiver<()>>,
    /// Channel used to notify an observer once the current target is reached.
    /// If the receiver is dropped, sync completes with the current database.
    pub reached_target_tx: Option<mpsc::Sender<Target<DB::Family, DB::Digest>>>,
}

/// Maximum queued target updates drained per scheduling tick.
const MAX_UPDATE_DRAIN_PER_TICK: usize = 32;

/// Drain all queued target updates without blocking, returning the newest.
async fn drain_latest_target<T>(update_rx: &mut mpsc::Receiver<T>) -> Option<T> {
    let mut latest = None;
    let mut drained = 0usize;
    loop {
        match update_rx.try_recv() {
            Ok(update) => {
                latest = Some(update);
                drained += 1;
                if drained.is_multiple_of(MAX_UPDATE_DRAIN_PER_TICK) {
                    reschedule().await;
                }
            }
            Err(mpsc::error::TryRecvError::Empty | mpsc::error::TryRecvError::Disconnected) => {
                return latest;
            }
        }
    }
}

/// Create/open a compact-storage database and initialize it from compact authenticated state.
///
/// Unlike streaming sync, compact sync jumps directly to `target.leaf_count`. This path
/// authenticates the final commit and frontier state for the target root rather than replaying a
/// retained operation range.
///
/// Targets received on `update_rx` supersede the current target. When `finish_rx` is `Some(...)`,
/// reaching a target parks sync until a finish signal or another target update arrives. Each
/// reached target is reported on `reached_target_tx`.
#[boxed]
pub async fn sync<DB, R>(
    config: Config<DB, R>,
) -> Result<DB, Error<DB::Family, R::Error, DB::Digest>>
where
    DB: Database,
    R: CompactDbResolver<DB>,
{
    let Config {
        context,
        resolver,
        mut target,
        db_config,
        mut update_rx,
        mut finish_rx,
        reached_target_tx,
    } = config;
    let metrics = super::Metrics::new(&context);
    let mut attempt = 0u64;
    loop {
        // Prefer the newest queued target before starting an attempt.
        if let Some(update_rx) = update_rx.as_mut() {
            if let Some(update) = drain_latest_target(update_rx).await {
                target = update;
            }
        }
        target
            .validate()
            .map_err(|reason| Error::Engine(EngineError::InvalidCompactTarget(reason)))?;
        metrics.record_target(*target.leaf_count);

        attempt += 1;
        let update_future = update_rx.as_mut().map_or_else(
            || Either::Right(pending()),
            |update_rx| Either::Left(update_rx.recv()),
        );
        let db = select! {
            update = update_future => {
                let Some(update) = update else {
                    update_rx = None;
                    continue;
                };
                target = update;
                continue;
            },
            db = attempt_sync(&context, attempt, &resolver, &db_config, &target) => db?,
        };
        metrics.record_synced(*target.leaf_count);

        // A target queued while the attempt ran supersedes the result.
        if let Some(update_rx) = update_rx.as_mut() {
            if let Some(update) = drain_latest_target(update_rx).await {
                target = update;
                continue;
            }
        }

        if let Some(reached_target_tx) = reached_target_tx.as_ref() {
            if reached_target_tx.send(target.clone()).await.is_err() {
                return Ok(db);
            }
        }

        let Some(finish_rx) = finish_rx.as_mut() else {
            return Ok(db);
        };
        let Some(update_rx) = update_rx.as_mut() else {
            return Ok(db);
        };
        select! {
            _ = finish_rx.recv() => return Ok(db),
            update = update_rx.recv() => {
                let Some(update) = update else {
                    return Ok(db);
                };
                target = update;
            },
        }
    }
}

/// Run one compact sync attempt against `target`.
///
/// Verification order:
/// 1. Fetch the proposed compact state for `target`.
/// 2. Verify the final commit proof against `target.root`.
/// 3. Rebuild the compact frontier in memory and compare its root against `target.root`.
/// 4. Build the compact db from that already-validated state.
/// 5. Assert the db root still matches and persist the state.
///
/// A failure before the final persist leaves on-disk state untouched.
async fn attempt_sync<DB, R>(
    context: &DB::Context,
    attempt: u64,
    resolver: &R,
    db_config: &DB::Config,
    target: &Target<DB::Family, DB::Digest>,
) -> Result<DB, Error<DB::Family, R::Error, DB::Digest>>
where
    DB: Database,
    R: CompactDbResolver<DB>,
{
    // Compact sync has no request scheduler, so this loop is its retry boundary for bad peer
    // responses. Resolver errors and local construction failures remain terminal.
    loop {
        let FetchResult { state, callback } = resolver
            .get_compact_state(target.clone())
            .await
            .map_err(Error::Resolver)?;

        // Validation failures describe a bad compact response. Reject it if the resolver supplied
        // feedback, then fetch another candidate.
        let validated_state = match validate_compact_state::<DB>(target, state) {
            Ok(state) => state,
            Err(err) => {
                if let Some(callback) = callback {
                    let _ = callback.send(false);
                }
                tracing::debug!(error = ?err, "compact state failed validation, will retry");
                continue;
            }
        };

        // The peer response has already authenticated the final commit and frontier. From here,
        // construction should only fail for local database/storage reasons; a root mismatch is a
        // bug in this path.
        let mut db = DB::from_validated_state(
            context.child("compact").with_attribute("attempt", attempt),
            db_config.clone(),
            validated_state,
        )
        .await
        .map_err(Error::Database)?;
        assert_eq!(
            db.root(),
            target.root,
            "validated compact state reconstructed unexpected root",
        );

        if let Some(callback) = callback {
            let _ = callback.send(true);
        }
        db.persist_compact_state().await?;
        return Ok(db);
    }
}

/// Validate the peer-provided compact state before constructing local database storage.
fn validate_compact_state<DB>(
    target: &Target<DB::Family, DB::Digest>,
    state: State<DB::Family, DB::Op, DB::Digest>,
) -> CompactFrontierValidation<DB>
where
    DB: Database,
{
    if state.leaf_count != target.leaf_count {
        return Err(EngineError::UnexpectedLeafCount {
            expected: target.leaf_count,
            actual: state.leaf_count,
        });
    }

    let last_commit_loc = Location::new(*state.leaf_count - 1);
    if !verify_proof::<DB::Hasher, _, _>(
        &state.last_commit_proof,
        last_commit_loc,
        std::slice::from_ref(&state.last_commit_op),
        &target.root,
    ) {
        return Err(EngineError::InvalidProof);
    }

    validate_compact_frontier::<DB>(target, state)
}

/// Result of validating a peer-provided compact frontier.
type CompactFrontierValidation<DB> = Result<
    ValidatedState<<DB as Database>::Family, <DB as Database>::Op, <DB as Database>::Digest>,
    EngineError<<DB as Database>::Family, <DB as Database>::Digest>,
>;

/// Validate that a peer-provided compact frontier authenticates the requested target root.
fn validate_compact_frontier<DB>(
    target: &Target<DB::Family, DB::Digest>,
    state: State<DB::Family, DB::Op, DB::Digest>,
) -> CompactFrontierValidation<DB>
where
    DB: Database,
{
    // The final commit is the only operation carried in compact state. Its floor determines which
    // peaks are inactive when authenticating the compact frontier root.
    let last_commit_loc = Location::new(*state.leaf_count - 1);
    let Some(inactivity_floor_loc) = DB::inactivity_floor(&state.last_commit_op) else {
        return Err(EngineError::InvalidProof);
    };
    if inactivity_floor_loc > last_commit_loc {
        return Err(EngineError::InvalidProof);
    }

    // Rebuild a disposable Merkle view from the pinned frontier before opening any database
    // storage. Invalid pin counts or inactive peak layouts are treated as bad peer proofs.
    let mem = crate::merkle::mem::Mem::<DB::Family, DB::Digest>::init(crate::merkle::mem::Config {
        nodes: Vec::new(),
        pruning_boundary: state.leaf_count,
        pinned_nodes: state.pinned_nodes.clone(),
    })
    .map_err(|_| EngineError::InvalidProof)?;
    let hasher = qmdb::hasher::<DB::Hasher>();
    let inactive_peaks = DB::Family::inactive_peaks(
        DB::Family::location_to_position(state.leaf_count),
        inactivity_floor_loc,
    );
    let actual = mem
        .root(&hasher, inactive_peaks)
        .map_err(|_| EngineError::InvalidProof)?;
    if actual != target.root {
        return Err(EngineError::RootMismatch {
            expected: target.root,
            actual,
        });
    }

    Ok(ValidatedState {
        state,
        root: target.root,
    })
}

async fn fetch_state_from_full_source<F, Op, D, Current, Hist, HistFut, Pins, PinsFut>(
    target: Target<F, D>,
    current_target: Current,
    historical_proof: Hist,
    pinned_nodes_at: Pins,
) -> Result<State<F, Op, D>, ServeError<F, D>>
where
    F: Family,
    D: Digest,
    Current: FnOnce() -> Target<F, D>,
    Hist: FnOnce(Location<F>, Location<F>) -> HistFut,
    HistFut: Future<Output = Result<(Proof<F, D>, Vec<Op>), qmdb::Error<F>>>,
    Pins: FnOnce(Location<F>) -> PinsFut,
    PinsFut: Future<Output = Result<Vec<D>, qmdb::Error<F>>>,
{
    // Full sources do not cache a compact witness. Instead, derive the compact payload on demand
    // from the current tip commit plus the frontier pins at the requested tree size.
    target.validate().map_err(ServeError::InvalidTarget)?;
    let current = current_target();
    if target.root != current.root || target.leaf_count != current.leaf_count {
        return Err(ServeError::StaleTarget {
            requested: target,
            current,
        });
    }
    let leaf_count = target.leaf_count;
    let last_commit_loc = Location::new(*leaf_count - 1);
    let (last_commit_proof, mut operations) = historical_proof(leaf_count, last_commit_loc)
        .await
        .map_err(ServeError::Database)?;
    // Compact sync always authenticates exactly the final commit leaf.
    let last_commit_op =
        operations
            .pop()
            .ok_or(ServeError::Database(qmdb::Error::DataCorrupted(
                "missing last commit operation",
            )))?;
    let pinned_nodes = pinned_nodes_at(leaf_count)
        .await
        .map_err(ServeError::Database)?;
    Ok(State {
        leaf_count,
        pinned_nodes,
        last_commit_op,
        last_commit_proof,
    })
}

// Resolver impls for full keyless databases. These synthesize compact state by querying the
// historical tip proof and current frontier pins from the full source.
macro_rules! impl_compact_resolver_keyless {
    ($db:ident, $op:ident, $val_bound:ident) => {
        impl<F, E, V, H, S> Resolver for Arc<$db<F, E, V, H, S>>
        where
            F: Family,
            E: crate::Context,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                fetch_state_from_full_source(
                    target,
                    || Target::new(self.root(), self.bounds().end),
                    |leaf_count, last_commit_loc| {
                        self.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| self.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }
        impl_compact_resolver_keyless!(@locked $db, $op, $val_bound, AsyncRwLock);
        impl_compact_resolver_keyless!(@locked $db, $op, $val_bound, TracedAsyncRwLock);
    };
    (@locked $db:ident, $op:ident, $val_bound:ident, $lock:ident) => {

        impl<F, E, V, H, S> Resolver for Arc<$lock<$db<F, E, V, H, S>>>
        where
            F: Family,
            E: crate::Context,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let db = self.read().await;
                fetch_state_from_full_source(
                    target,
                    || Target::new(db.root(), db.bounds().end),
                    |leaf_count, last_commit_loc| {
                        db.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| db.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }

        impl<F, E, V, H, S> Resolver for Arc<$lock<Option<$db<F, E, V, H, S>>>>
        where
            F: Family,
            E: crate::Context,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let guard = self.read().await;
                let db = guard.as_ref().ok_or(ServeError::MissingSource)?;
                fetch_state_from_full_source(
                    target,
                    || Target::new(db.root(), db.bounds().end),
                    |leaf_count, last_commit_loc| {
                        db.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| db.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }
    };
}

// Resolver impls for full immutable databases. Same pattern as keyless, but with the extra key and
// translator parameters carried by immutable variants.
macro_rules! impl_compact_resolver_immutable {
    ($db:ident, $op:ident, $val_bound:ident, $key_bound:path) => {
        impl<F, E, K, V, H, T, S> Resolver for Arc<$db<F, E, K, V, H, T, S>>
        where
            F: Family,
            E: crate::Context,
            K: $key_bound,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            T: Translator + Send + Sync + 'static,
            T::Key: Send + Sync,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                fetch_state_from_full_source(
                    target,
                    || Target::new(self.root(), self.bounds().end),
                    |leaf_count, last_commit_loc| {
                        self.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| self.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }
        impl_compact_resolver_immutable!(@locked $db, $op, $val_bound, $key_bound, AsyncRwLock);
        impl_compact_resolver_immutable!(@locked $db, $op, $val_bound, $key_bound, TracedAsyncRwLock);
    };
    (@locked $db:ident, $op:ident, $val_bound:ident, $key_bound:path, $lock:ident) => {

        impl<F, E, K, V, H, T, S> Resolver for Arc<$lock<$db<F, E, K, V, H, T, S>>>
        where
            F: Family,
            E: crate::Context,
            K: $key_bound,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            T: Translator + Send + Sync + 'static,
            T::Key: Send + Sync,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let db = self.read().await;
                fetch_state_from_full_source(
                    target,
                    || Target::new(db.root(), db.bounds().end),
                    |leaf_count, last_commit_loc| {
                        db.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| db.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }

        impl<F, E, K, V, H, T, S> Resolver for Arc<$lock<Option<$db<F, E, K, V, H, T, S>>>>
        where
            F: Family,
            E: crate::Context,
            K: $key_bound,
            V: $val_bound + Send + Sync + 'static,
            H: Hasher,
            T: Translator + Send + Sync + 'static,
            T::Key: Send + Sync,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let guard = self.read().await;
                let db = guard.as_ref().ok_or(ServeError::MissingSource)?;
                fetch_state_from_full_source(
                    target,
                    || Target::new(db.root(), db.bounds().end),
                    |leaf_count, last_commit_loc| {
                        db.historical_proof(
                            leaf_count,
                            last_commit_loc,
                            NonZeroU64::new(1).unwrap(),
                        )
                    },
                    |leaf_count| db.pinned_nodes_at(leaf_count),
                )
                .await
                .map(Into::into)
            }
        }
    };
}

// Resolver impls for compact keyless databases. These already persist a compact witness, so serving
// is just a target check over the current witness rather than reconstructing anything from history.
macro_rules! impl_compact_resolver_compact_keyless {
    ($db:ident, $op:ident) => {
        impl<F, E, V, H, C, S> Resolver for Arc<$db<F, E, V, H, C, S>>
        where
            F: Family,
            E: crate::Context,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                self.compact_state(target).map(Into::into)
            }
        }
        impl_compact_resolver_compact_keyless!(@locked $db, $op, AsyncRwLock);
        impl_compact_resolver_compact_keyless!(@locked $db, $op, TracedAsyncRwLock);
    };
    (@locked $db:ident, $op:ident, $lock:ident) => {

        impl<F, E, V, H, C, S> Resolver for Arc<$lock<$db<F, E, V, H, C, S>>>
        where
            F: Family,
            E: crate::Context,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let db = self.read().await;
                db.compact_state(target).map(Into::into)
            }
        }

        impl<F, E, V, H, C, S> Resolver for Arc<$lock<Option<$db<F, E, V, H, C, S>>>>
        where
            F: Family,
            E: crate::Context,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let guard = self.read().await;
                let db = guard.as_ref().ok_or(ServeError::MissingSource)?;
                db.compact_state(target).map(Into::into)
            }
        }
    };
}

// Resolver impls for compact immutable databases. Like the keyless compact path, these read the
// persisted witness directly instead of rebuilding it from a full operation log.
macro_rules! impl_compact_resolver_compact_immutable {
    ($db:ident, $op:ident) => {
        impl<F, E, K, V, H, C, S> Resolver for Arc<$db<F, E, K, V, H, C, S>>
        where
            F: Family,
            E: crate::Context,
            K: Key,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, K, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                self.compact_state(target).map(Into::into)
            }
        }
        impl_compact_resolver_compact_immutable!(@locked $db, $op, AsyncRwLock);
        impl_compact_resolver_compact_immutable!(@locked $db, $op, TracedAsyncRwLock);
    };
    (@locked $db:ident, $op:ident, $lock:ident) => {

        impl<F, E, K, V, H, C, S> Resolver for Arc<$lock<$db<F, E, K, V, H, C, S>>>
        where
            F: Family,
            E: crate::Context,
            K: Key,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, K, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let db = self.read().await;
                db.compact_state(target).map(Into::into)
            }
        }

        impl<F, E, K, V, H, C, S> Resolver for Arc<$lock<Option<$db<F, E, K, V, H, C, S>>>>
        where
            F: Family,
            E: crate::Context,
            K: Key,
            V: ValueEncoding + Send + Sync + 'static,
            H: Hasher,
            $op<F, K, V>: Encode + Read<Cfg = C>,
            C: Clone + Send + Sync + 'static,
            S: Strategy,
        {
            type Family = F;
            type Digest = H::Digest;
            type Op = $op<F, K, V>;
            type Error = ServeError<F, H::Digest>;

            async fn get_compact_state(
                &self,
                target: Target<Self::Family, Self::Digest>,
            ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
                let guard = self.read().await;
                let db = guard.as_ref().ok_or(ServeError::MissingSource)?;
                db.compact_state(target).map(Into::into)
            }
        }
    };
}

impl_compact_resolver_compact_keyless!(KeylessCompactDb, KeylessOp);
impl_compact_resolver_compact_immutable!(ImmutableCompactDb, ImmutableOp);

impl_compact_resolver_keyless!(KeylessFixedDb, KeylessFixedOp, FixedValue);
impl_compact_resolver_keyless!(KeylessVariableDb, KeylessVariableOp, VariableValue);
impl_compact_resolver_immutable!(ImmutableFixedDb, ImmutableFixedOp, FixedValue, Array);
impl_compact_resolver_immutable!(ImmutableVariableDb, ImmutableVariableOp, VariableValue, Key);

#[cfg(test)]
mod tests {
    use super::{Config, Database, FetchResult, Resolver, State, Target};
    use crate::{
        merkle::{mmr, Location},
        qmdb,
    };
    use commonware_codec::{DecodeExt as _, Encode as _, RangeCfg};
    use commonware_cryptography::{sha256::Digest, Hasher as _, Sha256};
    use commonware_parallel::Rayon;
    use commonware_runtime::{deterministic, Runner as _};
    use commonware_utils::sync::AsyncRwLock;
    use std::{
        collections::VecDeque,
        convert::Infallible,
        sync::{
            atomic::{AtomicUsize, Ordering},
            Arc,
        },
    };

    macro_rules! assert_resolver_variants {
        ($db:ty) => {
            assert_resolver::<Arc<$db>>();
            assert_resolver::<Arc<AsyncRwLock<$db>>>();
            assert_resolver::<Arc<AsyncRwLock<Option<$db>>>>();
        };
    }

    fn assert_resolver<R: super::Resolver>() {}

    struct TestDb {
        root: Digest,
    }

    impl Database for TestDb {
        type Family = mmr::Family;
        type Op = u8;
        type Config = (Digest, Arc<AtomicUsize>);
        type Digest = Digest;
        type Context = deterministic::Context;
        type Hasher = Sha256;

        async fn from_validated_state(
            _context: Self::Context,
            (root, constructions): Self::Config,
            _state: super::ValidatedState<Self::Family, Self::Op, Self::Digest>,
        ) -> Result<Self, qmdb::Error<Self::Family>> {
            constructions.fetch_add(1, Ordering::SeqCst);
            Ok(Self { root })
        }

        fn inactivity_floor(_op: &Self::Op) -> Option<Location<Self::Family>> {
            Some(Location::new(0))
        }

        fn root(&self) -> Self::Digest {
            self.root
        }

        async fn persist_compact_state(&mut self) -> Result<(), qmdb::Error<Self::Family>> {
            Ok(())
        }
    }

    #[derive(Clone)]
    struct SequenceResolver {
        states: Arc<commonware_utils::sync::Mutex<VecDeque<FetchResult<mmr::Family, u8, Digest>>>>,
    }

    impl Resolver for SequenceResolver {
        type Family = mmr::Family;
        type Digest = Digest;
        type Op = u8;
        type Error = Infallible;

        async fn get_compact_state(
            &self,
            _target: Target<Self::Family, Self::Digest>,
        ) -> Result<FetchResult<Self::Family, Self::Op, Self::Digest>, Self::Error> {
            Ok(self
                .states
                .lock()
                .pop_front()
                .expect("missing compact fetch result"))
        }
    }

    fn valid_state_and_target() -> (State<mmr::Family, u8, Digest>, Target<mmr::Family, Digest>) {
        let hasher = qmdb::hasher::<Sha256>();
        let mut merkle = crate::merkle::mem::Mem::<mmr::Family, Digest>::new();
        let op = 0u8;
        let first_op = 1u8;
        let batch = merkle
            .new_batch()
            .add(&hasher, &first_op.encode())
            .add(&hasher, &op.encode());
        let batch = batch.merkleize(&merkle, &hasher);
        merkle.apply_batch(&batch).unwrap();
        let root = merkle.root(&hasher, 0).unwrap();
        let leaf_count = Location::new(2);
        let pinned_nodes = merkle
            .nodes_to_pin(leaf_count)
            .into_values()
            .collect::<Vec<_>>();
        let proof = merkle.proof(&hasher, Location::new(1), 0).unwrap();
        (
            State {
                leaf_count,
                pinned_nodes,
                last_commit_op: op,
                last_commit_proof: proof,
            },
            Target::<mmr::Family, Digest> { root, leaf_count },
        )
    }

    #[test]
    fn test_all_compact_qmdb_variants_implement_strategy_resolvers() {
        type KeylessFixedCompactDb = crate::qmdb::keyless::fixed::CompactDb<
            mmr::Family,
            deterministic::Context,
            Digest,
            commonware_cryptography::Sha256,
            Rayon,
        >;
        type KeylessVariableCompactDb = crate::qmdb::keyless::variable::CompactDb<
            mmr::Family,
            deterministic::Context,
            Vec<u8>,
            commonware_cryptography::Sha256,
            (RangeCfg<usize>, ()),
            Rayon,
        >;
        type ImmutableFixedCompactDb = crate::qmdb::immutable::fixed::CompactDb<
            mmr::Family,
            deterministic::Context,
            Digest,
            Digest,
            commonware_cryptography::Sha256,
            Rayon,
        >;
        type ImmutableVariableCompactDb = crate::qmdb::immutable::variable::CompactDb<
            mmr::Family,
            deterministic::Context,
            Digest,
            Vec<u8>,
            commonware_cryptography::Sha256,
            ((), (RangeCfg<usize>, ())),
            Rayon,
        >;

        assert_resolver_variants!(KeylessFixedCompactDb);
        assert_resolver_variants!(KeylessVariableCompactDb);
        assert_resolver_variants!(ImmutableFixedCompactDb);
        assert_resolver_variants!(ImmutableVariableCompactDb);
    }

    #[test]
    fn test_target_decode_rejects_zero_leaf_count() {
        let unused_root = commonware_cryptography::Sha256::hash(b"unused");
        let encoded = Target::<mmr::Family, Digest> {
            root: unused_root,
            leaf_count: crate::merkle::Location::new(0),
        }
        .encode();

        assert!(Target::<mmr::Family, Digest>::decode(encoded).is_err());
    }

    #[test]
    fn test_compact_sync_retries_invalid_state_without_feedback() {
        deterministic::Runner::default().start(|context| async move {
            let (good_state, target) = valid_state_and_target();
            let mut bad_state = good_state.clone();
            bad_state.pinned_nodes.push(Sha256::hash(b"extra pin"));
            let (good_tx, good_rx) = commonware_utils::channel::oneshot::channel();
            let constructions = Arc::new(AtomicUsize::new(0));

            let db = super::sync::<TestDb, _>(Config {
                context,
                resolver: SequenceResolver {
                    states: Arc::new(commonware_utils::sync::Mutex::new(VecDeque::from([
                        FetchResult {
                            state: bad_state,
                            callback: None,
                        },
                        FetchResult {
                            state: good_state,
                            callback: Some(good_tx),
                        },
                    ]))),
                },
                target: target.clone(),
                db_config: (target.root, constructions.clone()),
                update_rx: None,
                finish_rx: None,
                reached_target_tx: None,
            })
            .await
            .unwrap();

            assert!(good_rx.await.expect("valid feedback should arrive"));
            assert_eq!(constructions.load(Ordering::SeqCst), 1);
            assert_eq!(db.root(), target.root);
        });
    }
}

#[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<Target<mmr::Family, Sha256Digest>>,
        CodecConformance<Target<mmb::Family, Sha256Digest>>,
    }
}