//! Database batch lifecycle and state sync for [`Stateful`](super::Stateful).
//!
//! `db` defines the traits a storage backend implements to be driven by
//! [`Stateful`](super::Stateful) and implements them for QMDB databases ([`any`], [`current`],
//! [`immutable`], [`keyless`]). [`p2p`] fetches and serves state sync data over the network.
//!
//! # Batch Lifecycle
//!
//! A block's state changes pass through three stages:
//! 1. [`Unmerkleized`]: a mutable batch (concrete types expose reads and writes).
//! 2. [`Merkleized`]: a sealed batch with a computed state root.
//! 3. Applied: [`ManagedDb::apply`] exposes the batch as a recoverable checkpoint.
//!    [`ManagedDb::finalize`] then starts persisting it, and a [`Barrier`] observes completion.
//!    A barrier covers the state applied before it was requested. Later batches may be applied
//!    while it is pending.
//!
//! [`DatabaseSet`] groups one or more [`ManagedDb`] instances into one unit for execution and
//! commit. [`Shared`] implements it for one database, and tuples of up to eight [`Shared`]
//! databases implement it for several.
//!
//! # State Sync
//!
//! [`StateSyncDb`] builds one database by syncing it from peers. [`StateSyncSet`] syncs every
//! database in a set to the targets carried by a finalized block, following newer tips until the
//! databases converge.
//!
//! ## Anchors
//!
//! Each set of sync targets is paired with an [`Anchor`], the finalized block that carries them. A
//! running sync handles each tip update as follows:
//!
//! - Upon a tip at or below the height of the most recently adopted anchor (initially the one
//!   passed to [`StateSyncSet::sync`]): ignore it.
//! - Upon a tip at a greater height: adopt its anchor. If its targets differ from the current
//!   targets, send them to the databases (tuple sets follow the
//!   [convergence rules](#convergence-tuple-sets)).
//!
//! Queued tips are coalesced: a tip superseded by a newer one before the sync dispatches it may
//! never reach the databases, and a target already sent to a database can likewise be replaced
//! before the database takes it. [`StateSyncSet::sync`] returns an anchor whose targets every
//! database reached. Tips delivered after convergence are not observed, so the returned anchor can
//! trail the latest tip sent.
//!
//! ## Convergence (tuple sets)
//!
//! A tuple set assigns a _generation_ each time it dispatches a tip to its databases, and tracks
//! the generation each database is assigned and whether it has reached that generation's target.
//!
//! - Upon a tip while some database is still seeking its target: start a new generation and send
//!   its targets only to the seeking databases. Databases that already reached their target stay
//!   frozen at their generation, so they cannot run ahead to a newer anchor. A reached database
//!   whose target is unchanged in the new generation counts as reached for it.
//! - Upon every database reaching the same generation with a tip pending: start a new generation
//!   for all databases.
//! - Upon every database reaching the same generation with no tip pending: finish at that
//!   generation's anchor.
//! - Upon every database reaching its target, at different generations: _regroup_. Send the
//!   highest generation's targets to the databases behind it and mark them seeking again.
//!
//! The coordinator retains state only for generations assigned to some database, so its memory
//! is bounded by the number of databases however long sync runs.
//!
//! ### Chasing a moving tip
//!
//! ```text
//! time ------------------------------------------------------------------------------->
//!
//! tips:          A0      A1                A2 A3
//! generation:    g0      g1                g2 = A3 (A2 is superseded before dispatch)
//!
//! db0 (slow):    g0 ---- g1 -------------- g2 ------- reached g2
//! db1 (fast):    g0 ---- g1 -- reached g1 -- frozen ------------ regroup -- reached g2
//! db2 (fast):    g0 ---- g1 -- reached g1 -- frozen ------------ regroup -- reached g2
//!
//! finish at A3 only when:
//! - every database has reported the same generation
//! - no newer tip update is pending
//! ```
//!
//! # Failures
//!
//! Database failures are fatal. [`DatabaseSet`] implementations panic when a database fails to
//! open, apply, finalize, or prune, and [`Barrier::durable`] panics when a deferred sync fails. A
//! mutation that is cancelled also loses its database until restart (see [`Shared`]). A database
//! that fails state sync is reported through the error returned by [`StateSyncSet::sync`].

use commonware_actor::mailbox::{self, Policy};
use commonware_consensus::{
    CertifiableBlock, Epochable, Heightable, Roundable, Viewable,
    types::{Height, Round},
};
use commonware_cryptography::{Digest, Digested, Digestible};
use commonware_macros::select;
use commonware_runtime::{Error as RuntimeError, Handle, Metrics, Spawner};
use commonware_storage::qmdb::sync::{self, Request, Source, source};
use commonware_utils::{
    NZUsize,
    channel::{fallible::AsyncFallibleExt, mpsc, oneshot, ring},
    futures::try_join_all,
    sync::{AsyncRwLockReadGuard, AsyncRwLockWriteGuard, TracedAsyncRwLock},
};
use futures::{
    future::{Either, pending},
    join,
};
use std::{
    collections::{BTreeMap, VecDeque},
    fmt::Debug,
    future::Future,
    num::{NonZeroU64, NonZeroUsize},
    ops::Deref,
    sync::Arc,
};
use tracing::debug;

pub mod any;
pub mod current;
pub mod immutable;
pub mod keyless;
pub mod p2p;

/// A database shared across tasks.
///
/// By-value mutations take the database out with [`Self::write`] and restore it with
/// [`WriteSlot::put`]. A mutation that fails, panics, or is cancelled before the put leaves the
/// database _lost_: every later [`Self::read`] or [`Self::write`] panics, and only a restart
/// recovers it.
/// [`Source::serve`] on a lost database returns an error instead of panicking.
pub struct Shared<DB>(Inner<DB>);

/// The lock behind [`Shared`], for which storage implements [`Source`].
type Inner<DB> = Arc<TracedAsyncRwLock<Option<DB>>>;

impl<DB> Clone for Shared<DB> {
    fn clone(&self) -> Self {
        Self(self.0.clone())
    }
}

const DB_LOST_MSG: &str =
    "database was lost by an earlier failed or interrupted operation; restart to recover";

impl<DB> Shared<DB> {
    /// Creates a shared database identified by `label` in lock traces.
    pub fn new(label: &'static str, db: DB) -> Self {
        Self(Arc::new(TracedAsyncRwLock::new(label, Some(db))))
    }

    /// Acquires shared read access to the database.
    ///
    /// The lock is write-preferring: once a writer is queued, new readers wait
    /// behind it. Holding a guard across an await that acquires this cell
    /// again therefore deadlocks once a writer arrives in between.
    ///
    /// # Panics
    ///
    /// Panics if the database was lost by an earlier failed or interrupted mutation.
    pub async fn read(&self) -> ReadGuard<'_, DB> {
        ReadGuard(AsyncRwLockReadGuard::map(self.0.read().await, |db| {
            db.as_ref().expect(DB_LOST_MSG)
        }))
    }

    /// Takes the database out for a by-value mutation.
    ///
    /// The returned [`WriteSlot`] holds the cell locked and empty until
    /// [`WriteSlot::put`] restores the database. Dropping the slot without a put
    /// leaves the database lost.
    ///
    /// # Panics
    ///
    /// Panics if the database was lost by an earlier failed or interrupted mutation.
    pub async fn write(&self) -> (WriteSlot<'_, DB>, DB) {
        let mut guard = self.0.write().await;
        let db = guard.take().expect(DB_LOST_MSG);
        (WriteSlot(guard), db)
    }

    async fn read_locked(&self) -> ReadLocked<'_, DB> {
        ReadLocked {
            database: self.0.read().await,
            shared: self,
        }
    }

    #[cfg(test)]
    async fn new_batch_for_test<E>(&self) -> <DB as ManagedDb<E>>::Unmerkleized
    where
        DB: ManagedDb<E>,
    {
        let database = self.read_locked().await;
        DB::new_batch(database.batch_context())
    }
}

/// Read-only access to a database managed by [`Stateful`](super::Stateful).
///
/// Unlike [`Shared`], this handle cannot acquire a write slot, construct or
/// apply batches, finalize database state, or prune. Applications receive readers in
/// [`Application::capture`](super::Application::capture) and
/// [`Application::finalized`](super::Application::finalized) so observing
/// finalized state cannot invalidate concurrent speculative batches.
///
/// ```compile_fail
/// use commonware_glue::stateful::db::Reader;
///
/// async fn mutate<DB>(reader: Reader<DB>) {
///     let _ = reader.write().await;
/// }
/// ```
pub struct Reader<DB>(Shared<DB>);

impl<DB> Reader<DB> {
    /// Acquires shared read access to the database.
    ///
    /// The guard follows the same write-preferring lock discipline as
    /// [`Shared::read`].
    ///
    /// # Panics
    ///
    /// Panics if the database was lost by an earlier failed or interrupted mutation.
    pub async fn read(&self) -> ReadGuard<'_, DB> {
        self.0.read().await
    }
}

/// Shared read access to a [`Shared`] database.
pub struct ReadGuard<'a, DB>(AsyncRwLockReadGuard<'a, DB>);

impl<DB> Deref for ReadGuard<'_, DB> {
    type Target = DB;

    fn deref(&self) -> &DB {
        &self.0
    }
}

/// An exclusively locked [`Shared`] cell whose database has been taken out.
pub struct WriteSlot<'a, DB>(AsyncRwLockWriteGuard<'a, Option<DB>>);

impl<DB> WriteSlot<'_, DB> {
    /// Restores the database and releases the write lock.
    pub fn put(mut self, db: DB) {
        *self.0 = Some(db);
    }
}

/// Origin-bound read access used to construct a database batch.
struct ReadLocked<'a, DB> {
    database: AsyncRwLockReadGuard<'a, Option<DB>>,
    shared: &'a Shared<DB>,
}

impl<DB> ReadLocked<'_, DB> {
    fn batch_context(&self) -> BatchContext<'_, DB> {
        BatchContext {
            database: self.database.as_ref().expect(DB_LOST_MSG),
            shared: Shared::clone(self.shared),
        }
    }
}

/// Origin-bound database access for synchronous batch construction.
///
/// Only the [`DatabaseSet`] implementations in this module construct it. Its borrow prevents a
/// [`ManagedDb`] implementation from retaining the set's read lock in the returned batch.
pub struct BatchContext<'a, DB> {
    database: &'a DB,
    shared: Shared<DB>,
}

impl<'a, DB> BatchContext<'a, DB> {
    /// Splits the capability into the read-locked database and its [`Shared`] handle.
    pub fn into_parts(self) -> (&'a DB, Shared<DB>) {
        (self.database, self.shared)
    }
}

impl<DB> Source for Shared<DB>
where
    DB: Send + Sync + 'static,
    Inner<DB>: Source,
{
    type Family = <Inner<DB> as Source>::Family;
    type Digest = <Inner<DB> as Source>::Digest;
    type Op = <Inner<DB> as Source>::Op;
    type Error = <Inner<DB> as Source>::Error;

    fn serve(
        &self,
        request: Request<Self::Family>,
    ) -> impl Future<Output = source::Result<Self>> + Send {
        self.0.serve(request)
    }
}

/// A batch of speculative mutations that has not been merkleized.
///
/// Concrete types expose reads and writes (`get`, `write`, `set`, `append`, and so on) as
/// inherent methods.
pub trait Unmerkleized: Sized + Send {
    /// The sealed batch returned by [`Self::merkleize`].
    type Merkleized: Merkleized;

    /// Error returned by [`Self::merkleize`].
    type Error: Send;

    /// Computes the state root over every mutation and seals the batch.
    fn merkleize(self) -> impl Future<Output = Result<Self::Merkleized, Self::Error>> + Send;
}

/// A sealed batch with a computed state root.
pub trait Merkleized: Sized + Send + Sync {
    /// Digest type of the state root returned by [`Self::root`].
    type Digest: Digest;

    /// The child batch returned by [`Self::new_batch`].
    type Unmerkleized: Unmerkleized;

    /// Returns the state root of the database with this batch applied.
    fn root(&self) -> Self::Digest;

    /// Creates a child batch whose reads see this batch's pending changes before the applied
    /// database state.
    fn new_batch(&self) -> Self::Unmerkleized;
}

/// A database whose batches [`Stateful`](super::Stateful) builds, applies, persists, and prunes.
///
/// Applying a batch and persisting it are separate steps: [`Self::apply`] exposes a batch as a
/// recoverable checkpoint, and [`Self::finalize`] starts making applied checkpoints durable.
///
/// # Ownership
///
/// Mutating methods take the database by value and return it on success. If a mutating method
/// returns an error or its future is dropped, the instance is lost. Durable state remains
/// recoverable on restart. State that was not yet durable may or may not be recovered.
pub trait ManagedDb<E>: Send + Sync + Sized {
    /// A batch of mutations that has not been merkleized.
    type Unmerkleized: Unmerkleized;

    /// A merkleized batch that has not been applied.
    ///
    /// Cloning must preserve the same sealed branch state and should be cheap.
    type Merkleized: Clone + Merkleized<Unmerkleized = Self::Unmerkleized>;

    /// Error returned by [`Self::apply`], [`Self::finalize`], and [`Self::prune`], and carried by
    /// [`InitError::Database`] from [`Self::init`].
    type Error: Debug + Send;

    /// Configuration passed to [`Self::init`].
    type Config: Send;

    /// Recovery and state sync target for this database.
    ///
    /// Typically a database-specific state commitment plus the operation range needed to reach it.
    type SyncTarget: Clone + Debug + PartialEq + Send + Sync;

    /// Opens the database at `expected`, or at its latest checkpoint when `expected` is `None`.
    ///
    /// State beyond the selected checkpoint must be durably discarded before this returns.
    /// Returns [`InitError::TargetMismatch`] if the recovered target is not `expected`.
    fn init(
        context: E,
        config: Self::Config,
        expected: Option<Self::SyncTarget>,
    ) -> impl Future<Output = Result<Self, InitError<Self::Error, Self::SyncTarget>>> + Send;

    /// Returns the sync target of a new, empty database.
    ///
    /// It must equal [`Self::sync_target`] after [`Self::init`] opens an empty partition.
    fn initial_sync_target() -> Self::SyncTarget;

    /// Creates a batch over the applied state of `database`.
    ///
    /// The batch may keep the [`Shared`] handle from [`BatchContext::into_parts`] for later reads.
    /// It cannot keep the borrowed database, so the batch never holds the set's read lock.
    fn new_batch(database: BatchContext<'_, Self>) -> Self::Unmerkleized;

    /// Returns whether applying `batch` yields a database whose [`Self::sync_target`] is
    /// `target`.
    fn matches_sync_target(batch: &Self::Merkleized, target: &Self::SyncTarget) -> bool;

    /// Applies `batch` to the database.
    ///
    /// The returned database must expose `batch` as an independently recoverable checkpoint. The
    /// checkpoint need not be durable until the handle returned by [`Self::finalize`] resolves.
    fn apply(
        self,
        batch: Self::Merkleized,
    ) -> impl Future<Output = Result<Self, Self::Error>> + Send;

    /// Starts persisting every checkpoint applied before this call.
    ///
    /// The returned handle resolves once that state is durable. Batches applied while it is
    /// pending are not covered and need a later finalization. Callers must await the handle before
    /// finalizing again.
    fn finalize(self) -> impl Future<Output = Result<(Self, Handle<()>), Self::Error>> + Send;

    /// Prunes the database to a previously finalized sync target.
    ///
    /// Callers must await every handle returned by [`Self::finalize`] before pruning. Pruning
    /// effects must be durable before this returns. The default implementation does nothing, for
    /// databases without prunable history.
    fn prune(
        self,
        _target: &Self::SyncTarget,
    ) -> impl Future<Output = Result<Self, Self::Error>> + Send {
        async { Ok(self) }
    }

    /// Returns the target of the latest applied checkpoint (which need not be durable yet).
    fn sync_target(&self) -> Self::SyncTarget;
}

/// A durability barrier returned by [`DatabaseSet::finalize`].
///
/// Deferred sync failures surface only through [`Self::durable`], so every barrier must be
/// awaited. A barrier must resolve before [`DatabaseSet::prune`] runs.
///
/// # Examples
///
/// ```
/// use commonware_glue::stateful::db::Barrier;
/// use commonware_runtime::Handle;
///
/// struct CustomDatabaseSet;
///
/// # async fn example() {
/// let barrier = Barrier::from_handles::<CustomDatabaseSet>([
///     Handle::ready(Ok(())),
/// ]);
/// assert!(barrier.durable().await);
/// # }
/// ```
#[must_use = "await `durable` to surface deferred sync failures"]
pub struct Barrier {
    syncs: Vec<(&'static str, Option<usize>, Handle<()>)>,
}

impl Barrier {
    /// Builds a barrier from deferred sync handles owned by `T`.
    ///
    /// Failures identify `T` and the handle's zero-based position in the
    /// provided iteration. An empty barrier is immediately durable.
    pub fn from_handles<T: ?Sized>(handles: impl IntoIterator<Item = Handle<()>>) -> Self {
        let db_type = std::any::type_name::<T>();
        Self {
            syncs: handles
                .into_iter()
                .enumerate()
                .map(|(index, handle)| (db_type, Some(index), handle))
                .collect(),
        }
    }

    /// Resolves `true` once every deferred sync completes, or `false` if runtime shutdown aborts
    /// or closes a sync handle.
    ///
    /// # Panics
    ///
    /// Panics if a sync fails, because the database has already advanced past the state that
    /// failed to persist.
    pub async fn durable(self) -> bool {
        let syncs = self
            .syncs
            .into_iter()
            .map(|(db_type, index, handle)| async move {
                match handle.await {
                    Ok(()) => Ok(true),
                    Err(RuntimeError::Closed | RuntimeError::Aborted) => {
                        debug!(db_type, "runtime shutdown before database sync completed");
                        Ok(false)
                    }
                    Err(err) => Err((db_type, index, err)),
                }
            });

        match try_join_all(syncs).await {
            Ok(results) => results.into_iter().all(|durable| durable),
            Err((db_type, index, err)) => {
                let index = index.map_or(String::new(), |i| format!("index {i}, "));
                panic!("database sync failed ({index}type {db_type}): {err}");
            }
        }
    }
}

/// A group of [`ManagedDb`] instances executed and committed as one unit.
///
/// Read access may span several members at once, so a mutation must not hold one member's write
/// access while waiting for another's.
///
/// # Mutation Safety
///
/// Calls to [`Self::apply`], [`Self::finalize`], and [`Self::prune`] must not overlap.
/// Implementations panic if a database mutation fails (see
/// [Failures](crate::stateful::db#failures)).
pub trait DatabaseSet<E>: Clone + Send + Sync + 'static {
    /// One [`ManagedDb::Unmerkleized`] per database in the set.
    type Unmerkleized: Send;

    /// One [`ManagedDb::Merkleized`] per database in the set.
    ///
    /// Cloning must preserve the same sealed branch state and should be cheap.
    type Merkleized: Clone + Send + Sync;

    /// Read-only handles for observing the applied database state.
    ///
    /// Implementations must not expose mutation capabilities through this
    /// type. In particular, readers must not construct or apply batches,
    /// finalize database state, or prune.
    type Readers: Send;

    /// Configuration needed to construct every database in the set.
    ///
    /// - Single database sets use that database's [`ManagedDb::Config`].
    /// - Multi-database tuple sets use a tuple of per-database configs
    ///   `(Db1::Config, Db2::Config, ...)`.
    type Config: Send;

    /// Per-database sync targets carried by a finalized block.
    ///
    /// For a single-database set this is one target. For multi-database sets it is a tuple of
    /// targets, one per database.
    type SyncTargets: Clone + PartialEq + Send + Sync;

    /// Opens every database in the set, each at its target in `expected` when supplied.
    ///
    /// # Panics
    ///
    /// Panics if a database fails to open, including when it does not match its expected target.
    fn init(
        context: E,
        config: Self::Config,
        expected: Option<Self::SyncTargets>,
    ) -> impl Future<Output = Self> + Send;

    /// Returns the sync targets of a new, empty set.
    fn initial_sync_targets() -> Self::SyncTargets;

    /// Creates a batch over each database's applied state.
    ///
    /// Implementations must release every read lock before returning.
    fn new_batches(&self) -> impl Future<Output = Self::Unmerkleized> + Send;

    /// Creates child batches of a pending merkleized parent.
    ///
    /// Construction takes no lock. Reads see `parent`'s pending changes before falling back to the
    /// applied state.
    fn fork_batches(parent: &Self::Merkleized) -> Self::Unmerkleized;

    /// Returns whether applying `batches` yields `targets` (see
    /// [`ManagedDb::matches_sync_target`]).
    fn matches_sync_targets(batches: &Self::Merkleized, targets: &Self::SyncTargets) -> bool;

    /// Returns read-only handles for every database in the set.
    fn readers(&self) -> Self::Readers;

    /// Applies each batch to its database.
    ///
    /// Returns once every database exposes its batch as an independently recoverable checkpoint.
    /// Durability starts separately with [`Self::finalize`].
    fn apply(&self, batches: Self::Merkleized) -> impl Future<Output = ()> + Send;

    /// Starts persisting every checkpoint applied before this call.
    ///
    /// The returned [`Barrier`] resolves once that state is durable in every database. Batches
    /// applied while it is pending are not covered and need a later finalization. The barrier
    /// must be awaited before finalizing again.
    fn finalize(&self) -> impl Future<Output = Barrier> + Send;

    /// Prunes each database to its target in `targets`.
    ///
    /// The state represented by `targets` must already be durable, and no barrier may be pending.
    /// Pruning effects must be durable before this returns.
    fn prune(&self, targets: &Self::SyncTargets) -> impl Future<Output = ()> + Send;

    /// Returns the targets of the latest applied checkpoints (which need not be durable yet).
    fn committed_targets(&self) -> impl Future<Output = Self::SyncTargets> + Send;
}

/// Parameters for a one-time state-sync pass.
#[derive(Clone, Copy, Debug)]
pub struct SyncEngineConfig {
    /// Maximum operations fetched per source request.
    ///
    /// Peers ignore requests larger than their `max_serve_ops` (see [`p2p::Config`]), so a larger
    /// value stalls sync. Keep this at or below the `max_serve_ops` every peer uses.
    pub fetch_batch_size: NonZeroU64,

    /// Number of operations applied per local apply step.
    pub apply_batch_size: NonZeroU64,

    /// Maximum number of outstanding source requests. The request for the pinned nodes counts
    /// toward it.
    pub max_outstanding_requests: NonZeroUsize,
}

/// A [`ManagedDb`] that can be built by syncing it from peers.
pub trait StateSyncDb<E, R>: ManagedDb<E> {
    /// Error returned by [`Self::sync_db`].
    type SyncError: Debug + Send;

    /// Syncs a new database from `source` to `target` and returns it.
    ///
    /// The caller sends valid targets of the same database on `tip_updates` in advancing order; a
    /// target may be superseded by a newer one before it is read.
    ///
    /// Implementations must follow these rules:
    /// - Adopt a target from `tip_updates` only if it advances the current target.
    /// - When `finish` is `Some`, complete only once it has signaled and the current target is
    ///   reached. When `finish` is `None`, complete as soon as the current target is reached.
    /// - When `reached_target` is `Some`, report each reached target on it at most once, in the
    ///   order reached. Reports never wait, so an observer that falls behind skips intermediate
    ///   reports.
    #[allow(clippy::too_many_arguments)]
    fn sync_db(
        context: E,
        config: Self::Config,
        source: R,
        target: Self::SyncTarget,
        tip_updates: ring::Receiver<Self::SyncTarget>,
        finish: Option<mpsc::Receiver<()>>,
        reached_target: Option<ring::Sender<Self::SyncTarget>>,
        sync_config: SyncEngineConfig,
    ) -> impl Future<Output = Result<Self, Self::SyncError>> + Send;
}

/// The finalized block that carries a set of sync targets.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Anchor<D: Digest> {
    /// Height of the block.
    pub height: Height,
    /// Consensus round of the block.
    pub round: Round,
    /// Digest of the block.
    pub digest: D,
}

impl<D: Digest> Anchor<D> {
    /// Builds the anchor of `block`.
    pub fn from_block<B>(block: &Digested<B>) -> Self
    where
        B: CertifiableBlock<Digest = D>,
        B::Context: Epochable + Viewable,
    {
        Self {
            height: block.height(),
            round: block.context().round(),
            digest: block.digest(),
        }
    }
}

/// A finalized tip delivered to a running [`StateSyncSet::sync`].
///
/// See [Anchors](crate::stateful::db#anchors) for how a sync handles each update.
pub struct TipUpdate<D: Digest, T> {
    anchor: Anchor<D>,
    targets: T,
    observed: Option<oneshot::Sender<()>>,
}

impl<D: Digest, T> TipUpdate<D, T> {
    /// Creates an update for the block identified by `anchor`, which carries `targets`.
    pub const fn new(anchor: Anchor<D>, targets: T) -> Self {
        Self {
            anchor,
            targets,
            observed: None,
        }
    }

    /// Creates an update and a receiver that resolves once a sync has handled it, whether or not
    /// the sync adopted it.
    ///
    /// The receiver errors if the update is dropped unhandled.
    pub(crate) fn with_observation(anchor: Anchor<D>, targets: T) -> (Self, oneshot::Receiver<()>) {
        let (observed, receiver) = oneshot::channel();
        (
            Self {
                anchor,
                targets,
                observed: Some(observed),
            },
            receiver,
        )
    }

    /// Passes the update to `record`, then resolves its observer.
    pub(crate) fn record<R>(self, record: impl FnOnce(Anchor<D>, T) -> R) -> R {
        let result = record(self.anchor, self.targets);
        if let Some(observed) = self.observed {
            let _ = observed.send(());
        }
        result
    }
}

/// A [`DatabaseSet`] that can be built by one-time state sync.
pub trait StateSyncSet<E, R>: DatabaseSet<E> {
    /// Error returned if any database in the set fails state sync.
    type Error: Debug + Send;

    /// Syncs every database from `sources` to `targets`, carried by `anchor`, and returns the
    /// synced set with an anchor whose targets every database reached.
    ///
    /// Updates on `tip_updates` follow the [module rules](crate::stateful::db#anchors). The
    /// returned anchor, not the latest tip delivered, identifies the synced state.
    ///
    /// `D` is the block digest type of each [`Anchor`].
    fn sync<D: Digest>(
        context: E,
        config: Self::Config,
        sources: R,
        anchor: Anchor<D>,
        targets: Self::SyncTargets,
        tip_updates: ring::Receiver<TipUpdate<D, Self::SyncTargets>>,
        sync_config: SyncEngineConfig,
    ) -> impl Future<Output = Result<(Self, Anchor<D>), Self::Error>> + Send;
}

/// An error opening a [`ManagedDb`].
#[derive(Debug, thiserror::Error)]
pub enum InitError<E: Debug, T: Debug> {
    /// The database failed to open or recover.
    #[error("database initialization failed: {0:?}")]
    Database(E),
    /// The opened database's [`ManagedDb::sync_target`] is not the expected target.
    #[error("database target mismatch: expected {expected:?}, recovered {recovered:?}")]
    TargetMismatch {
        /// The target requested by the caller.
        expected: T,
        /// The target recovered from storage.
        recovered: T,
    },
}

/// Validates the requested sync target before returning the database.
fn validate_initialization<E, T>(
    db: T,
    expected: Option<T::SyncTarget>,
) -> Result<T, InitError<T::Error, T::SyncTarget>>
where
    T: ManagedDb<E>,
{
    let Some(expected) = expected else {
        return Ok(db);
    };
    let recovered = db.sync_target();
    if recovered != expected {
        return Err(InitError::TargetMismatch {
            expected,
            recovered,
        });
    }
    Ok(db)
}

impl<E: Send + Sync, T: ManagedDb<E> + 'static> DatabaseSet<E> for Shared<T> {
    type Unmerkleized = T::Unmerkleized;
    type Merkleized = T::Merkleized;
    type Readers = Reader<T>;
    type Config = T::Config;
    type SyncTargets = T::SyncTarget;

    async fn init(context: E, config: Self::Config, expected: Option<Self::SyncTargets>) -> Self {
        let db = T::init(context, config, expected)
            .await
            .expect("database init failed");
        Self::new("stateful.db", db)
    }

    fn initial_sync_targets() -> Self::SyncTargets {
        T::initial_sync_target()
    }

    async fn new_batches(&self) -> Self::Unmerkleized {
        let database = self.read_locked().await;
        T::new_batch(database.batch_context())
    }

    fn fork_batches(parent: &Self::Merkleized) -> Self::Unmerkleized {
        parent.new_batch()
    }

    fn matches_sync_targets(batches: &Self::Merkleized, targets: &Self::SyncTargets) -> bool {
        T::matches_sync_target(batches, targets)
    }

    fn readers(&self) -> Self::Readers {
        Reader(self.clone())
    }

    async fn apply(&self, batches: Self::Merkleized) {
        apply_shared::<E, T>(self, batches, None).await;
    }

    async fn finalize(&self) -> Barrier {
        let handle = finalize_shared::<E, T>(self, None).await;
        Barrier {
            syncs: vec![(core::any::type_name::<T>(), None, handle)],
        }
    }

    async fn prune(&self, target: &Self::SyncTargets) {
        prune_shared::<E, T>(self, target, None).await;
    }

    async fn committed_targets(&self) -> Self::SyncTargets {
        let database = self.read().await;
        T::sync_target(&database)
    }
}

impl<E, T, R> StateSyncSet<E, R> for Shared<T>
where
    E: Metrics,
    T: StateSyncDb<E, R> + 'static,
    R: Send + 'static,
{
    type Error = T::SyncError;

    async fn sync<D: Digest>(
        context: E,
        config: Self::Config,
        source: R,
        anchor: Anchor<D>,
        target: Self::SyncTargets,
        tip_updates: ring::Receiver<TipUpdate<D, Self::SyncTargets>>,
        sync_config: SyncEngineConfig,
    ) -> Result<(Self, Anchor<D>), Self::Error> {
        let (target_tx, target_rx) = ring::channel(NZUsize!(1));
        let (finish_tx, finish_rx) = mpsc::channel(1);
        let (reached_tx, mut reached_rx) = ring::channel(NZUsize!(1));
        let mut current_target = target.clone();
        let sync = T::sync_db(
            context,
            config,
            source,
            target,
            target_rx,
            Some(finish_rx),
            Some(reached_tx),
            sync_config,
        );

        let coordinator = async {
            let mut current_anchor = anchor;
            let mut tip_updates = Some(tip_updates);
            let mut last_reached = None;
            loop {
                if !dispatch_queued_tips(
                    &mut tip_updates,
                    &target_tx,
                    &mut current_anchor,
                    &mut current_target,
                ) {
                    return (current_anchor, current_target);
                }

                // Finish once the database has reached the newest adopted target. Queued tips are
                // adopted first, so a report for a target they replaced cannot finish the sync.
                if last_reached.as_ref() == Some(&current_target) {
                    let _ = finish_tx.send_lossy(()).await;
                    return (current_anchor, current_target);
                }

                // Wait for a progress report or a newer tip. Once the tip channel closes, only the
                // database can end the wait.
                let update_future = tip_updates.as_mut().map_or_else(
                    || Either::Right(pending()),
                    |updates| Either::Left(updates.recv()),
                );
                select! {
                    reached = reached_rx.recv() => {
                        let Some(reached) = reached else {
                            return (current_anchor, current_target);
                        };
                        last_reached = Some(reached);
                    },
                    update = update_future => {
                        let Some(update) = update else {
                            tip_updates = None;
                            continue;
                        };
                        let target = update.record(|new_anchor, new_target| {
                            if new_anchor.height <= current_anchor.height {
                                return None;
                            }
                            current_anchor = new_anchor;
                            if new_target == current_target {
                                return None;
                            }
                            current_target = new_target.clone();
                            Some(new_target)
                        });
                        let Some(new_target) = target else {
                            continue;
                        };
                        if !target_tx.send_lossy(new_target) {
                            return (current_anchor, current_target);
                        }
                    },
                }
            }
        };

        let (db_result, (converged_anchor, converged_target)) = join!(sync, coordinator);
        let database = db_result?;
        assert!(
            T::sync_target(&database) == converged_target,
            "state sync database target does not match the coordinator target",
        );
        Ok((Self::new("stateful.db", database), converged_anchor))
    }
}

/// Records every queued tip, then sends the newest adopted target to the database if it changed.
///
/// Returns `false` if the database has dropped its target receiver.
fn dispatch_queued_tips<D, T>(
    tip_updates: &mut Option<ring::Receiver<TipUpdate<D, T>>>,
    target_tx: &ring::Sender<T>,
    current_anchor: &mut Anchor<D>,
    current_target: &mut T,
) -> bool
where
    D: Digest,
    T: Clone + PartialEq,
{
    let mut latest = None;
    loop {
        let update = match tip_updates.as_mut().map(ring::Receiver::try_recv) {
            Some(Ok(update)) => update,
            Some(Err(ring::TryRecvError::Empty)) => break,
            Some(Err(ring::TryRecvError::Disconnected)) => {
                *tip_updates = None;
                break;
            }
            None => break,
        };

        update.record(|new_anchor, new_target| {
            let latest_height = latest
                .as_ref()
                .map_or(current_anchor.height, |(anchor, _): &(Anchor<D>, T)| {
                    anchor.height
                });
            if new_anchor.height > latest_height {
                latest = Some((new_anchor, new_target));
            }
        });
    }

    let Some((new_anchor, new_target)) = latest else {
        return true;
    };
    *current_anchor = new_anchor;
    if new_target == *current_target {
        return true;
    }
    *current_target = new_target.clone();
    target_tx.send_lossy(new_target)
}

macro_rules! impl_database_set {
    ($($T:ident : $idx:tt),+) => {
        impl<E: Send + Sync + Metrics, $($T: ManagedDb<E> + 'static),+> DatabaseSet<E>
            for ($(Shared<$T>,)+)
        {
            type Unmerkleized = ($($T::Unmerkleized,)+);
            type Merkleized = ($($T::Merkleized,)+);
            type Readers = ($(Reader<$T>,)+);
            type Config = ($($T::Config,)+);
            type SyncTargets = ($($T::SyncTarget,)+);

            async fn init(
                context: E,
                config: Self::Config,
                expected: Option<Self::SyncTargets>,
            ) -> Self {
                let result = join!($(
                    async {
                        let db = $T::init(
                                context.child(concat!("db_", stringify!($idx))),
                                config.$idx,
                                expected.as_ref().map(|targets| targets.$idx.clone()),
                            )
                            .await
                            .expect(concat!(
                                "database init failed (index ",
                                stringify!($idx),
                                ", type ",
                                stringify!($T),
                                ")",
                            ));
                        Shared::new(concat!("stateful.db.", stringify!($idx)), db)
                    },
                )+);
                result
            }

            fn initial_sync_targets() -> Self::SyncTargets {
                ($($T::initial_sync_target(),)+)
            }

            async fn new_batches(&self) -> Self::Unmerkleized {
                let databases = join!($(self.$idx.read_locked(),)+);
                ($($T::new_batch(databases.$idx.batch_context()),)+)
            }

            fn fork_batches(parent: &Self::Merkleized) -> Self::Unmerkleized {
                ($(parent.$idx.new_batch(),)+)
            }

            fn matches_sync_targets(batches: &Self::Merkleized, targets: &Self::SyncTargets) -> bool {
                $($T::matches_sync_target(&batches.$idx, &targets.$idx))&&+
            }

            fn readers(&self) -> Self::Readers {
                ($(Reader(self.$idx.clone()),)+)
            }

            async fn apply(
                &self,
                batches: Self::Merkleized,
            ) {
                // Each member completes its own write-lock lifecycle. Holding
                // a partial tuple of writers can deadlock cross-database reads.
                join!($(apply_shared::<E, $T>(
                    &self.$idx,
                    batches.$idx,
                    Some($idx),
                ),)+);
            }

            async fn finalize(&self) -> Barrier {
                let handles = join!($(finalize_shared::<E, $T>(
                    &self.$idx,
                    Some($idx),
                ),)+);
                Barrier {
                    syncs: vec![$((
                        core::any::type_name::<$T>(),
                        Some($idx),
                        handles.$idx,
                    ),)+],
                }
            }

            async fn prune(
                &self,
                targets: &Self::SyncTargets,
            ) {
                join!($(prune_shared::<E, $T>(
                    &self.$idx,
                    &targets.$idx,
                    Some($idx),
                ),)+);
            }

            async fn committed_targets(&self) -> Self::SyncTargets {
                let databases = join!($(self.$idx.read(),)+);
                ($($T::sync_target(&databases.$idx),)+)
            }

        }
    };
}

impl_database_set!(DB1: 0);
impl_database_set!(DB1: 0, DB2: 1);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2, DB4: 3);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2, DB4: 3, DB5: 4);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2, DB4: 3, DB5: 4, DB6: 5);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2, DB4: 3, DB5: 4, DB6: 5, DB7: 6);
impl_database_set!(DB1: 0, DB2: 1, DB3: 2, DB4: 3, DB5: 4, DB6: 5, DB7: 6, DB8: 7);

struct DbSyncChannels<T> {
    target_tx: ring::Sender<T>,
    target_rx: ring::Receiver<T>,
    finish_tx: mpsc::Sender<()>,
    finish_rx: mpsc::Receiver<()>,
    generation_tx: ring::Sender<(usize, T)>,
    generation_rx: ring::Receiver<(usize, T)>,
    reached_tx: ring::Sender<T>,
    reached_rx: ring::Receiver<T>,
}

impl<T> DbSyncChannels<T> {
    fn new() -> Self {
        let (target_tx, target_rx) = ring::channel(NZUsize!(1));
        let (finish_tx, finish_rx) = mpsc::channel(1);
        let (generation_tx, generation_rx) = ring::channel(NZUsize!(1));
        let (reached_tx, reached_rx) = ring::channel(NZUsize!(1));
        Self {
            target_tx,
            target_rx,
            finish_tx,
            finish_rx,
            generation_tx,
            generation_rx,
            reached_tx,
            reached_rx,
        }
    }
}

struct CoordinatorSyncSenders<T> {
    target_tx: ring::Sender<T>,
    finish_tx: mpsc::Sender<()>,
    generation_tx: ring::Sender<(usize, T)>,
}

/// Reports that database `index` reached the targets of `generation`.
struct ReachedEvent {
    index: usize,
    generation: usize,
}

/// Keeps the newest pending event for each database once the ready queue is full, so one
/// database's reports cannot evict another database's only event. A database reports
/// non-decreasing generations, so its newest event is also its highest.
impl Policy for ReachedEvent {
    type Overflow = VecDeque<Self>;

    fn handle(overflow: &mut Self::Overflow, message: Self) {
        overflow.retain(|event| event.index != message.index);
        overflow.push_back(message);
    }
}

macro_rules! impl_state_sync_set {
    ($($T:ident : $R:ident : $idx:tt),+) => {
        impl<E, $($T, $R),+> StateSyncSet<E, ($($R,)+)> for ($(Shared<$T>,)+)
        where
            E: Send + Sync + Spawner + Metrics + 'static,
            $(
                $T: StateSyncDb<E, $R> + 'static,
                $R: Send + 'static,
            )+
        {
            type Error = String;

            async fn sync<D: Digest>(
                context: E,
                config: Self::Config,
                sources: ($($R,)+),
                anchor: Anchor<D>,
                targets: Self::SyncTargets,
                tip_updates: ring::Receiver<TipUpdate<D, Self::SyncTargets>>,
                sync_config: SyncEngineConfig,
            ) -> Result<(Self, Anchor<D>), Self::Error> {
                let db_channels = ($(
                    DbSyncChannels::<<$T as ManagedDb<E>>::SyncTarget>::new(),
                )+);
                let coordinator_senders = ($(
                    CoordinatorSyncSenders {
                        target_tx: db_channels.$idx.target_tx,
                        finish_tx: db_channels.$idx.finish_tx,
                        generation_tx: db_channels.$idx.generation_tx,
                    },
                )+);
                let db_count = [$($idx,)+].len();
                let (reached_event_tx, mut reached_event_rx) = mailbox::new::<ReachedEvent>(
                    context.child("reached_events"),
                    NonZeroUsize::new(db_count).expect("tuple sets hold at least two databases"),
                );
                let (completion_tx, mut completion_rx) = ring::channel(NZUsize!(1));
                let coordinator_targets = targets.clone();
                let initial_targets = targets.clone();
                let first_db_error: Arc<commonware_utils::sync::Mutex<Option<String>>> =
                    Arc::new(commonware_utils::sync::Mutex::new(None));

                // The coordinator applies the convergence rules and dispatches each generation.
                let coordinator_handle = context.child("coordinator").spawn({
                    move |_context| async move {
                        // Move the senders into the body so they drop when it returns.
                        let coordinator_senders = coordinator_senders;
                        let mut tip_updates = Some(tip_updates);
                        let mut state = CoordinatorState::new(db_count, anchor, coordinator_targets);
                        let mut last_dispatched_targets = initial_targets;

                        loop {
                            loop {
                                match reached_event_rx.try_recv() {
                                    Ok(ReachedEvent { index, generation }) => {
                                        state.record_reached(index, generation)
                                    }
                                    Err(std::sync::mpsc::TryRecvError::Empty) => break,
                                    Err(std::sync::mpsc::TryRecvError::Disconnected) => return None,
                                }
                            }

                            if let Some(updates) = tip_updates.as_mut() {
                                loop {
                                    match updates.try_recv() {
                                        Ok(update) => {
                                            update.record(|anchor, targets| {
                                                state.record_tip_update(anchor, targets);
                                            });
                                        }
                                        Err(ring::TryRecvError::Empty) => break,
                                        Err(ring::TryRecvError::Disconnected) => {
                                            tip_updates = None;
                                            break;
                                        }
                                    }
                                }
                            }

                            match state.next_action() {
                                CoordinatorAction::Converged { anchor, targets } => {
                                    $(
                                        let _ = coordinator_senders.$idx.finish_tx.send_lossy(()).await;
                                    )+
                                    return Some((anchor, targets));
                                }
                                CoordinatorAction::Dispatch {
                                    generation,
                                    targets: dispatch_targets,
                                } => {
                                    $(
                                        let dispatch_target = dispatch_targets.$idx.clone();
                                        if !coordinator_senders.$idx
                                            .generation_tx
                                            .send_lossy((generation, dispatch_target.clone()))
                                        {
                                            return None;
                                        }
                                        if state.should_dispatch($idx) {
                                            if dispatch_target != last_dispatched_targets.$idx {
                                                if !coordinator_senders.$idx
                                                    .target_tx
                                                    .send_lossy(dispatch_target.clone())
                                                {
                                                    return None;
                                                }
                                                last_dispatched_targets.$idx = dispatch_target;
                                            }
                                        } else if dispatch_target == last_dispatched_targets.$idx {
                                            state.mark_reached_same_target($idx, generation);
                                        }
                                    )+
                                    continue;
                                }
                                CoordinatorAction::Wait => {}
                            }

                            let update_future = tip_updates.as_mut().map_or_else(
                                || Either::Right(pending()),
                                |updates| Either::Left(updates.recv()),
                            );
                            select! {
                                reached_event = reached_event_rx.recv() => {
                                    let ReachedEvent { index, generation } = reached_event?;
                                    state.record_reached(index, generation);
                                },
                                _ = completion_rx.recv() => {
                                    return None;
                                },
                                update = update_future => {
                                    let Some(update) = update else {
                                        tip_updates = None;
                                        continue;
                                    };
                                    update.record(|anchor, targets| {
                                        state.record_tip_update(anchor, targets);
                                    });
                                },
                            };
                        }
                    }
                });

                // Each database task runs its sync and reports the assigned generations whose
                // targets it has reached.
                let db_handles = (
                    $(
                        context.child(concat!("db_", stringify!($idx))).spawn({
                            let first_db_error = first_db_error.clone();
                            let mut reached_target_rx = db_channels.$idx.reached_rx;
                            let mut generation_rx = Some(db_channels.$idx.generation_rx);
                            let mut current_generation = 0usize;
                            let mut current_target = targets.$idx.clone();
                            let mut last_reached_target = None;
                            let mut last_reported_generation = None;
                            let reached_event_sender = reached_event_tx.clone();
                            let completion_signal = completion_tx.clone();
                            let config = config.$idx;
                            let source = sources.$idx;
                            let target = targets.$idx;
                            let target_rx = db_channels.$idx.target_rx;
                            let finish_rx = db_channels.$idx.finish_rx;
                            let reached_tx = db_channels.$idx.reached_tx;
                            move |context| async move {
                                let sync = $T::sync_db(
                                    context,
                                    config,
                                    source,
                                    target,
                                    target_rx,
                                    Some(finish_rx),
                                    Some(reached_tx),
                                    sync_config,
                                );
                                let forward_reached = async move {
                                    loop {
                                        drain_generation_updates(
                                            &mut generation_rx,
                                            &mut current_generation,
                                            &mut current_target,
                                            &last_reached_target,
                                            &mut last_reported_generation,
                                            &reached_event_sender,
                                            $idx,
                                        );

                                        let update_future = generation_rx.as_mut().map_or_else(
                                            || Either::Right(pending()),
                                            |updates| Either::Left(updates.recv()),
                                        );
                                        select! {
                                            reached_target = reached_target_rx.recv() => {
                                                let Some(reached_target) = reached_target else {
                                                    return;
                                                };

                                                last_reached_target = Some(reached_target.clone());
                                                drain_generation_updates(
                                                    &mut generation_rx,
                                                    &mut current_generation,
                                                    &mut current_target,
                                                    &last_reached_target,
                                                    &mut last_reported_generation,
                                                    &reached_event_sender,
                                                    $idx,
                                                );

                                                if reached_target != current_target {
                                                    continue;
                                                }

                                                if last_reported_generation != Some(current_generation) {
                                                    let event = ReachedEvent {
                                                        index: $idx,
                                                        generation: current_generation,
                                                    };
                                                    if !reached_event_sender
                                                        .enqueue(event)
                                                        .accepted()
                                                    {
                                                        return;
                                                    }
                                                    last_reported_generation = Some(current_generation);
                                                }
                                            },
                                            update = update_future => {
                                                let Some((generation, target)) = update else {
                                                    generation_rx = None;
                                                    continue;
                                                };
                                                current_generation = generation;
                                                current_target = target;
                                                if last_reached_target.as_ref() == Some(&current_target)
                                                    && last_reported_generation != Some(current_generation)
                                                {
                                                    let event = ReachedEvent {
                                                        index: $idx,
                                                        generation: current_generation,
                                                    };
                                                    if !reached_event_sender
                                                        .enqueue(event)
                                                        .accepted()
                                                    {
                                                        return;
                                                    }
                                                    last_reported_generation = Some(current_generation);
                                                }
                                            },
                                        };
                                    }
                                };
                                let (sync_result, _) = join!(sync, forward_reached);
                                let result = sync_result
                                    .map(|database| {
                                        Shared::new(
                                            concat!("stateful.db.", stringify!($idx)),
                                            database,
                                        )
                                    })
                                    .map_err(|err| {
                                        format!(
                                            "state sync failed (index {}, db {}): {err:?}",
                                            $idx,
                                            core::any::type_name::<$T>(),
                                        )
                                    });
                                if let Err(err) = &result {
                                    let mut first = first_db_error.lock();
                                    if first.is_none() {
                                        *first = Some(err.clone());
                                    }
                                }
                                let _ = completion_signal.send_lossy(());
                                result
                            }
                        }),
                    )+
                );

                let synced = join!(
                    $(
                        async {
                            db_handles.$idx
                                .await
                                .expect("state sync database task exited")
                        },
                    )+
                );
                let converged_anchor = coordinator_handle
                    .await
                    .expect("state sync coordinator task exited");

                if let Some(err) = first_db_error.lock().take() {
                    return Err(err);
                }

                let synced = ($(synced.$idx?,)+);
                let Some((converged_anchor, converged_targets)) = converged_anchor else {
                    return Err("state sync coordinator did not report a converged anchor".into());
                };
                let committed_targets =
                    <Self as DatabaseSet<E>>::committed_targets(&synced).await;
                if committed_targets != converged_targets {
                    return Err(
                        "state sync database targets do not match the coordinator target set"
                            .into(),
                    );
                }

                Ok((synced, converged_anchor))
            }
        }
    };
}

impl_state_sync_set!(DB1: R1: 0, DB2: R2: 1);
impl_state_sync_set!(DB1: R1: 0, DB2: R2: 1, DB3: R3: 2);
impl_state_sync_set!(DB1: R1: 0, DB2: R2: 1, DB3: R3: 2, DB4: R4: 3);
impl_state_sync_set!(DB1: R1: 0, DB2: R2: 1, DB3: R3: 2, DB4: R4: 3, DB5: R5: 4);
impl_state_sync_set!(DB1: R1: 0, DB2: R2: 1, DB3: R3: 2, DB4: R4: 3, DB5: R5: 4, DB6: R6: 5);
impl_state_sync_set!(
    DB1: R1: 0,
    DB2: R2: 1,
    DB3: R3: 2,
    DB4: R4: 3,
    DB5: R5: 4,
    DB6: R6: 5,
    DB7: R7: 6
);
impl_state_sync_set!(
    DB1: R1: 0,
    DB2: R2: 1,
    DB3: R3: 2,
    DB4: R4: 3,
    DB5: R5: 4,
    DB6: R6: 5,
    DB7: R7: 6,
    DB8: R8: 7
);

/// Applies the generation assignments waiting for database `idx`, reporting each one whose target
/// the database has already reached. An assignment replaced before it is read is skipped. The
/// coordinator only accepts reports for a database's current generation, so a skipped assignment
/// needs no report.
fn drain_generation_updates<T>(
    generation_rx: &mut Option<ring::Receiver<(usize, T)>>,
    current_generation: &mut usize,
    current_target: &mut T,
    last_reached_target: &Option<T>,
    last_reported_generation: &mut Option<usize>,
    reached_event_sender: &mailbox::Sender<ReachedEvent>,
    idx: usize,
) where
    T: Clone + PartialEq,
{
    if let Some(updates) = generation_rx.as_mut() {
        loop {
            match updates.try_recv() {
                Ok((generation, target)) => {
                    *current_generation = generation;
                    *current_target = target;

                    if last_reached_target.as_ref() == Some(current_target)
                        && *last_reported_generation != Some(*current_generation)
                    {
                        let event = ReachedEvent {
                            index: idx,
                            generation: *current_generation,
                        };
                        if !reached_event_sender.enqueue(event).accepted() {
                            return;
                        }
                        *last_reported_generation = Some(*current_generation);
                    }
                }
                Err(ring::TryRecvError::Empty) => break,
                Err(ring::TryRecvError::Disconnected) => {
                    *generation_rx = None;
                    break;
                }
            }
        }
    }
}

/// Per-database sync tracking state.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DbSyncState {
    /// Database is still syncing toward its assigned generation's targets.
    Seeking { generation: usize },
    /// Database reported it reached its assigned generation's targets.
    Reached { generation: usize },
}

impl DbSyncState {
    const fn generation(self) -> usize {
        match self {
            Self::Seeking { generation } | Self::Reached { generation } => generation,
        }
    }

    const fn is_reached(self) -> bool {
        matches!(self, Self::Reached { .. })
    }
}

/// What the coordinator should do after processing events.
enum CoordinatorAction<D: Digest, T> {
    /// Nothing to do until the next event.
    Wait,
    /// Dispatch `targets` as `generation` to the databases still seeking.
    Dispatch { generation: usize, targets: T },
    /// Every database reached the targets of one generation, carried by `anchor`.
    Converged { anchor: Anchor<D>, targets: T },
}

/// State machine for tuple-set sync convergence (see the
/// [module docs](crate::stateful::db#convergence-tuple-sets)).
///
/// Tracks each database's assigned generation and whether it has reached it, and decides when to
/// dispatch, regroup, or finish.
struct CoordinatorState<D: Digest, T> {
    dbs: Vec<DbSyncState>,
    generation_state: BTreeMap<usize, (Anchor<D>, T)>,
    current_generation: usize,
    latest_tip: Option<(Anchor<D>, T)>,
    last_dispatched_anchor: Anchor<D>,
}

impl<D: Digest, T: Clone> CoordinatorState<D, T> {
    fn new(db_count: usize, anchor: Anchor<D>, targets: T) -> Self {
        let dbs = vec![DbSyncState::Seeking { generation: 0 }; db_count];
        let mut generation_state = BTreeMap::new();
        generation_state.insert(0, (anchor, targets));
        Self {
            dbs,
            generation_state,
            current_generation: 0,
            latest_tip: None,
            last_dispatched_anchor: anchor,
        }
    }

    /// Records that database `idx` reached `generation`.
    ///
    /// Reached events can arrive late. An event for a generation the database is no longer
    /// assigned is ignored.
    fn record_reached(&mut self, idx: usize, generation: usize) {
        if self.dbs[idx].generation() != generation {
            return;
        }
        if self.dbs[idx].is_reached() {
            return;
        }
        self.dbs[idx] = DbSyncState::Reached { generation };
    }

    /// Records a tip as the pending dispatch, replacing any earlier pending tip.
    ///
    /// A tip at or below the height of the pending or last dispatched anchor is ignored. Targets
    /// are not compared.
    fn record_tip_update(&mut self, anchor: Anchor<D>, targets: T) {
        let current_height = self
            .latest_tip
            .as_ref()
            .map_or(self.last_dispatched_anchor.height, |(latest_anchor, _)| {
                latest_anchor.height
            });
        if anchor.height <= current_height {
            return;
        }
        self.latest_tip = Some((anchor, targets));
    }

    /// Returns the next coordinator action, updating assignments for a `Dispatch`.
    ///
    /// Returns `Converged` when every database reached the same generation and no tip is pending.
    /// Returns `Dispatch` for a pending tip (a new generation) or a regroup (every database
    /// reached, at different generations), and `Wait` otherwise. After a `Dispatch`,
    /// [`Self::should_dispatch`] identifies the databases that receive the targets.
    fn next_action(&mut self) -> CoordinatorAction<D, T> {
        let all_reached = self.dbs.iter().all(|db| db.is_reached());

        if all_reached {
            let min_gen = self.dbs.iter().map(|db| db.generation()).min().unwrap();
            let max_gen = self.dbs.iter().map(|db| db.generation()).max().unwrap();

            if min_gen == max_gen {
                if let Some((anchor, targets)) = self.latest_tip.take() {
                    let generation = self.current_generation + 1;
                    self.current_generation = generation;
                    self.dbs.fill(DbSyncState::Seeking { generation });
                    self.generation_state
                        .insert(generation, (anchor, targets.clone()));
                    self.last_dispatched_anchor = anchor;
                    self.prune_generations();
                    return CoordinatorAction::Dispatch {
                        generation,
                        targets,
                    };
                }

                let (anchor, targets) = self
                    .generation_state
                    .get(&min_gen)
                    .expect("missing state for converged generation")
                    .clone();
                return CoordinatorAction::Converged { anchor, targets };
            }

            // Regroup: reset behind databases to seek the highest generation.
            let (_anchor, targets) = self
                .generation_state
                .get(&max_gen)
                .expect("missing state for regroup generation")
                .clone();
            for db in &mut self.dbs {
                if db.generation() != max_gen {
                    *db = DbSyncState::Seeking {
                        generation: max_gen,
                    };
                }
            }
            self.prune_generations();
            return CoordinatorAction::Dispatch {
                generation: max_gen,
                targets,
            };
        }

        // With some database still seeking, a pending tip starts a new generation for the
        // seeking databases only.
        let Some((anchor, targets)) = self.latest_tip.take() else {
            return CoordinatorAction::Wait;
        };

        let generation = self.current_generation + 1;
        self.current_generation = generation;
        for db in &mut self.dbs {
            if !db.is_reached() {
                *db = DbSyncState::Seeking { generation };
            }
        }
        self.generation_state
            .insert(generation, (anchor, targets.clone()));
        self.last_dispatched_anchor = anchor;

        self.prune_generations();
        CoordinatorAction::Dispatch {
            generation,
            targets,
        }
    }

    /// Drops the state of generations no database is assigned.
    fn prune_generations(&mut self) {
        self.generation_state
            .retain(|r#gen, _| self.dbs.iter().any(|db| db.generation() == *r#gen));
    }

    /// Returns whether database `idx` receives a dispatch's targets (it has not reached its
    /// generation).
    fn should_dispatch(&self, idx: usize) -> bool {
        !self.dbs[idx].is_reached()
    }

    /// Marks a reached database as reached for `generation` (no effect on a seeking database).
    ///
    /// Callers must only use this when the database's target is unchanged in `generation`.
    fn mark_reached_same_target(&mut self, idx: usize, generation: usize) {
        if !self.dbs[idx].is_reached() {
            return;
        }
        self.dbs[idx] = DbSyncState::Reached { generation };
    }
}

/// Syncs a database that durably persists its operation log.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn sync_standard_db<E, DB, S>(
    context: E,
    config: DB::Config,
    source: S,
    target: sync::Target<DB::Family, DB::Digest>,
    tip_updates: ring::Receiver<sync::Target<DB::Family, DB::Digest>>,
    finish: Option<mpsc::Receiver<()>>,
    reached_target: Option<ring::Sender<sync::Target<DB::Family, DB::Digest>>>,
    sync_config: SyncEngineConfig,
) -> Result<DB, sync::Error<DB::Family, S::Error, DB::Digest>>
where
    DB: sync::Database<Context = E>,
    S: sync::SourceFor<DB>,
{
    sync::sync(sync::engine::Config {
        context,
        source,
        target,
        max_outstanding_requests: sync_config.max_outstanding_requests,
        fetch_batch_size: sync_config.fetch_batch_size,
        apply_batch_size: sync_config.apply_batch_size,
        db_config: config,
        update_rx: Some(tip_updates),
        finish_rx: finish,
        reached_target_tx: reached_target,
    })
    .await
}

/// Aborts an adapter task when its owning sync future completes or is cancelled.
struct Forwarder(Handle<()>);

impl Drop for Forwarder {
    fn drop(&mut self) {
        self.0.abort();
    }
}

/// Syncs a compact database, which does not persist its operation log.
///
/// A compact target that does not convert to an engine target fails the sync when passed as
/// `target` and is ignored when it arrives on `tip_updates`.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn sync_compact_db<E, DB, S>(
    context: E,
    config: DB::Config,
    source: S,
    target: sync::CompactTarget<DB::Family, DB::Digest>,
    mut tip_updates: ring::Receiver<sync::CompactTarget<DB::Family, DB::Digest>>,
    finish: Option<mpsc::Receiver<()>>,
    reached_target: Option<ring::Sender<sync::CompactTarget<DB::Family, DB::Digest>>>,
    sync_config: SyncEngineConfig,
) -> Result<DB, sync::Error<DB::Family, S::Error, DB::Digest>>
where
    E: Metrics + Spawner,
    DB: sync::Database<Context = E>,
    S: sync::SourceFor<DB>,
{
    let mut initial = sync::Target::try_from(&target).map_err(sync::Error::Engine)?;
    // Start at the newest target already queued.
    while let Ok(update) = tip_updates.try_recv() {
        let Ok(update) = sync::Target::try_from(&update) else {
            continue;
        };
        if update.advances(&initial) {
            initial = update;
        }
    }

    let (update_tx, update_rx) = ring::channel(NZUsize!(1));
    let update_forwarder = Forwarder(context.child("compact_updates").spawn(move |_| async move {
        while let Some(update) = tip_updates.recv().await {
            let Ok(update) = sync::Target::try_from(&update) else {
                continue;
            };
            if !update_tx.send_lossy(update) {
                break;
            }
        }
    }));

    let reached_target_tx = reached_target.map(|reached| {
        let (tx, mut rx) = ring::channel::<sync::Target<DB::Family, DB::Digest>>(NZUsize!(1));
        context.child("compact_reached").spawn(move |_| async move {
            while let Some(reached_engine_target) = rx.recv().await {
                let target = sync::CompactTarget {
                    root: reached_engine_target.root,
                    size: reached_engine_target.range.end(),
                };
                if !reached.send_lossy(target) {
                    break;
                }
            }
        });
        tx
    });

    let result = sync::sync(sync::engine::Config {
        context,
        source,
        target: initial,
        db_config: config,
        fetch_batch_size: sync_config.fetch_batch_size,
        apply_batch_size: sync_config.apply_batch_size,
        max_outstanding_requests: sync_config.max_outstanding_requests,
        update_rx: Some(update_rx),
        finish_rx: finish,
        reached_target_tx,
    })
    .await;
    drop(update_forwarder);
    result
}

#[tracing::instrument(name = "stateful.db.apply", level = "info", skip_all, fields(index = index))]
async fn apply<E, T: ManagedDb<E>>(database: T, batch: T::Merkleized, index: Option<usize>) -> T {
    // Mutable apply failures are fatal because the batch may already have been
    // applied to other databases in the same set, leaving partially applied state.
    match database.apply(batch).await {
        Ok(result) => result,
        Err(err) => {
            let index = index.map_or(String::new(), |i| format!("index {i}, "));
            panic!(
                "database apply failed ({index}type {}): {err:?}",
                core::any::type_name::<T>(),
            );
        }
    }
}

/// Applies `batch` to one database while holding only that database's write lock.
async fn apply_shared<E, T: ManagedDb<E>>(
    shared: &Shared<T>,
    batch: T::Merkleized,
    index: Option<usize>,
) {
    let (slot, database) = shared.write().await;
    slot.put(apply(database, batch, index).await);
}

#[tracing::instrument(name = "stateful.db.finalize", level = "info", skip_all, fields(index = index))]
async fn finalize<E, T: ManagedDb<E>>(database: T, index: Option<usize>) -> (T, Handle<()>) {
    match database.finalize().await {
        Ok(result) => result,
        Err(err) => {
            let index = index.map_or(String::new(), |i| format!("index {i}, "));
            panic!(
                "database finalize failed ({index}type {}): {err:?}",
                core::any::type_name::<T>(),
            );
        }
    }
}

/// Starts persisting one database and returns the handle that resolves once it is durable.
async fn finalize_shared<E, T: ManagedDb<E>>(
    shared: &Shared<T>,
    index: Option<usize>,
) -> Handle<()> {
    let (slot, database) = shared.write().await;
    let (database, handle) = finalize(database, index).await;
    slot.put(database);
    handle
}

async fn prune_shared<E, T: ManagedDb<E>>(
    shared: &Shared<T>,
    target: &T::SyncTarget,
    index: Option<usize>,
) {
    let (slot, database) = shared.write().await;
    slot.put(prune(database, target, index).await);
}

#[tracing::instrument(name = "stateful.db.prune", level = "info", skip_all, fields(index = index))]
async fn prune<E, T: ManagedDb<E>>(database: T, target: &T::SyncTarget, index: Option<usize>) -> T {
    // Prune failures are fatal because pruning may already have discarded part
    // of the retained history before the error surfaced.
    match database.prune(target).await {
        Ok(database) => database,
        Err(err) => {
            let index = index.map_or(String::new(), |i| format!("index {i}, "));
            panic!(
                "database prune failed ({index}type {}): {err:?}",
                core::any::type_name::<T>(),
            );
        }
    }
}

/// A resolver that serves sync requests from a database attached after startup.
pub trait AttachableResolver<DB>: Clone + Send + Sync + 'static {
    /// Attaches `db` for serving incoming requests.
    fn attach_database(&self, db: Shared<DB>) -> impl Future<Output = ()> + Send;
}

/// A set of resolvers with the same shape as a database set.
pub trait AttachableResolverSet<DBs>: Clone + Send + Sync + 'static {
    /// Attaches each database to its resolver.
    fn attach_databases(&self, databases: DBs) -> impl Future<Output = ()> + Send;
}

impl<R, DB> AttachableResolverSet<Shared<DB>> for R
where
    R: AttachableResolver<DB>,
    DB: Send + Sync + 'static,
{
    async fn attach_databases(&self, db: Shared<DB>) {
        self.attach_database(db).await;
    }
}

macro_rules! impl_attachable_resolver_set {
    ($($R:ident : $DB:ident : $idx:tt),+) => {
        impl<$($R, $DB),+> AttachableResolverSet<($(Shared<$DB>,)+)> for ($($R,)+)
        where
            $(
                $R: AttachableResolver<$DB>,
                $DB: Send + Sync + 'static,
            )+
        {
            async fn attach_databases(&self, databases: ($(Shared<$DB>,)+)) {
                futures::join!($(
                    self.$idx.attach_database(databases.$idx),
                )+);
            }
        }
    };
}

impl_attachable_resolver_set!(R1: DB1: 0, R2: DB2: 1);
impl_attachable_resolver_set!(R1: DB1: 0, R2: DB2: 1, R3: DB3: 2);
impl_attachable_resolver_set!(R1: DB1: 0, R2: DB2: 1, R3: DB3: 2, R4: DB4: 3);
impl_attachable_resolver_set!(R1: DB1: 0, R2: DB2: 1, R3: DB3: 2, R4: DB4: 3, R5: DB5: 4);
impl_attachable_resolver_set!(
    R1: DB1: 0,
    R2: DB2: 1,
    R3: DB3: 2,
    R4: DB4: 3,
    R5: DB5: 4,
    R6: DB6: 5
);
impl_attachable_resolver_set!(
    R1: DB1: 0,
    R2: DB2: 1,
    R3: DB3: 2,
    R4: DB4: 3,
    R5: DB5: 4,
    R6: DB6: 5,
    R7: DB7: 6
);
impl_attachable_resolver_set!(
    R1: DB1: 0,
    R2: DB2: 1,
    R3: DB3: 2,
    R4: DB4: 3,
    R5: DB5: 4,
    R6: DB6: 5,
    R7: DB7: 6,
    R8: DB8: 7
);

#[cfg(test)]
mod tests {
    use super::{
        Anchor, AttachableResolver, AttachableResolverSet, Barrier, BatchContext,
        CoordinatorAction, CoordinatorState, DatabaseSet, InitError, ManagedDb, ReachedEvent,
        Shared, StateSyncDb, StateSyncSet, SyncEngineConfig, TipUpdate, dispatch_queued_tips,
        drain_generation_updates, mailbox,
    };
    use crate::stateful::tests::mocks::{TestMerkleized, TestUnmerkleized, anchor as mock_anchor};
    use commonware_cryptography::sha256;
    use commonware_macros::select;
    use commonware_runtime::{
        Clock, Error as RuntimeError, Handle, Runner as _, Spawner as _, Supervisor as _,
        deterministic,
    };
    use commonware_utils::{
        NZU64, NZUsize,
        channel::{mpsc, oneshot, ring},
    };
    use futures::{FutureExt, SinkExt, pin_mut};
    use std::{
        convert::Infallible,
        num::{NonZeroU64, NonZeroUsize},
        ops::RangeInclusive,
        panic::AssertUnwindSafe,
        sync::{
            Arc,
            atomic::{AtomicU64, AtomicUsize, Ordering},
        },
        time::Duration,
    };

    mod managed_db_lifecycle {
        use super::{ManagedDb, Shared};
        use crate::stateful::db::Unmerkleized;
        use commonware_cryptography::{Sha256, sha256::Digest};
        use commonware_parallel::Sequential;
        use commonware_runtime::{
            Runner as _, Supervisor as _, buffer::paged::CacheRef, deterministic,
        };
        use commonware_storage::{
            journal::contiguous::{
                fixed::Config as FixedJournalConfig, variable::Config as VariableJournalConfig,
            },
            merkle::{full::Config as MerkleConfig, mmr},
            qmdb::{
                any as storage_any, current as storage_current, immutable as storage_immutable,
                keyless as storage_keyless,
            },
            translator::TwoCap,
        };
        use commonware_utils::{NZU16, NZU64, NZUsize, sequence::U64};
        use rstest::rstest;
        use std::{fmt::Debug, marker::PhantomData};

        type Context = deterministic::Context;

        type AnyFixed = storage_any::unordered::fixed::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            Sequential,
        >;
        type AnyVariable = storage_any::unordered::variable::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            Sequential,
        >;

        type CurrentUnorderedFixed = storage_current::unordered::fixed::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            64,
            Sequential,
        >;
        type CurrentOrderedFixed = storage_current::ordered::fixed::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            64,
            Sequential,
        >;
        type CurrentUnorderedVariable = storage_current::unordered::variable::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            64,
            Sequential,
        >;
        type CurrentOrderedVariable = storage_current::ordered::variable::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            64,
            Sequential,
        >;

        type ImmutableFixed = storage_immutable::fixed::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            Sequential,
        >;
        type ImmutableVariable = storage_immutable::variable::Db<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            TwoCap,
            Sequential,
        >;
        type ImmutableCompactFixed = storage_immutable::fixed::CompactDb<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            Sequential,
        >;
        type ImmutableCompactVariable = storage_immutable::variable::CompactDb<
            mmr::Family,
            Context,
            Digest,
            U64,
            Sha256,
            Sequential,
        >;

        type KeylessFixed =
            storage_keyless::fixed::Db<mmr::Family, Context, U64, Sha256, Sequential>;
        type KeylessVariable =
            storage_keyless::variable::Db<mmr::Family, Context, U64, Sha256, Sequential>;
        type KeylessCompactFixed =
            storage_keyless::fixed::CompactDb<mmr::Family, Context, U64, Sha256, Sequential>;
        type KeylessCompactVariable =
            storage_keyless::variable::CompactDb<mmr::Family, Context, U64, Sha256, Sequential>;

        fn page_cache() -> CacheRef {
            CacheRef::new(NZU16!(101), NZUsize!(11))
        }

        fn merkle_config(suffix: &str) -> MerkleConfig<Sequential> {
            MerkleConfig {
                journal_partition: format!("initial-target-{suffix}-merkle-journal"),
                metadata_partition: format!("initial-target-{suffix}-merkle-metadata"),
                items_per_blob: NZU64!(11),
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
                strategy: Sequential,
                page_cache: page_cache(),
            }
        }

        fn fixed_journal_config(suffix: &str) -> FixedJournalConfig {
            FixedJournalConfig {
                partition: format!("initial-target-{suffix}-log"),
                items_per_blob: NZU64!(7),
                page_cache: page_cache(),
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
            }
        }

        fn variable_journal_config<C>(suffix: &str, codec_config: C) -> VariableJournalConfig<C> {
            VariableJournalConfig {
                partition: format!("initial-target-{suffix}-log"),
                items_per_section: NZU64!(7),
                compression: None,
                codec_config,
                page_cache: page_cache(),
                write_buffer: NZUsize!(1024),
                replay_buffer: NZUsize!(1024),
            }
        }

        fn any_fixed_config(suffix: &str) -> storage_any::FixedConfig<TwoCap, Sequential> {
            storage_any::Config {
                merkle_config: merkle_config(suffix),
                journal_config: fixed_journal_config(suffix),
                translator: TwoCap,
                init_cache: Some(NZUsize!(1024)),
                init_buffer: NZUsize!(1 << 21),
                init_concurrency: (),
            }
        }

        fn any_variable_config(
            suffix: &str,
        ) -> storage_any::VariableConfig<TwoCap, ((), ()), Sequential> {
            storage_any::Config {
                merkle_config: merkle_config(suffix),
                journal_config: variable_journal_config(suffix, ((), ())),
                translator: TwoCap,
                init_cache: Some(NZUsize!(1024)),
                init_buffer: NZUsize!(1 << 21),
                init_concurrency: (),
            }
        }

        fn current_fixed_config(suffix: &str) -> storage_current::FixedConfig<TwoCap, Sequential> {
            storage_current::Config {
                merkle_config: merkle_config(suffix),
                journal_config: fixed_journal_config(suffix),
                grafted_metadata_partition: format!("initial-target-{suffix}-grafted-metadata"),
                translator: TwoCap,
                init_cache: Some(NZUsize!(1024)),
                init_buffer: NZUsize!(1 << 21),
                init_concurrency: (),
            }
        }

        fn current_variable_config(
            suffix: &str,
        ) -> storage_current::VariableConfig<TwoCap, ((), ()), Sequential> {
            storage_current::Config {
                merkle_config: merkle_config(suffix),
                journal_config: variable_journal_config(suffix, ((), ())),
                grafted_metadata_partition: format!("initial-target-{suffix}-grafted-metadata"),
                translator: TwoCap,
                init_cache: Some(NZUsize!(1024)),
                init_buffer: NZUsize!(1 << 21),
                init_concurrency: (),
            }
        }

        fn immutable_fixed_config(
            suffix: &str,
        ) -> storage_immutable::fixed::Config<TwoCap, Sequential> {
            storage_immutable::Config {
                merkle_config: merkle_config(suffix),
                log: fixed_journal_config(suffix),
                translator: TwoCap,
                init_buffer: NZUsize!(1 << 21),
            }
        }

        fn immutable_variable_config(
            suffix: &str,
        ) -> storage_immutable::variable::Config<TwoCap, ((), ()), Sequential> {
            storage_immutable::Config {
                merkle_config: merkle_config(suffix),
                log: variable_journal_config(suffix, ((), ())),
                translator: TwoCap,
                init_buffer: NZUsize!(1 << 21),
            }
        }

        fn keyless_fixed_config(suffix: &str) -> storage_keyless::fixed::Config<Sequential> {
            storage_keyless::Config {
                merkle: merkle_config(suffix),
                log: fixed_journal_config(suffix),
            }
        }

        fn keyless_variable_config(
            suffix: &str,
        ) -> storage_keyless::variable::Config<(), Sequential> {
            storage_keyless::Config {
                merkle: merkle_config(suffix),
                log: variable_journal_config(suffix, ()),
            }
        }

        fn immutable_compact_fixed_config(
            suffix: &str,
        ) -> storage_immutable::fixed::CompactConfig<Sequential> {
            storage_immutable::CompactConfig {
                strategy: Sequential,
                witness: variable_journal_config(suffix, ()),
                commit_codec_config: (),
            }
        }

        fn immutable_compact_variable_config(
            suffix: &str,
        ) -> storage_immutable::variable::CompactConfig<((), ()), Sequential> {
            storage_immutable::CompactConfig {
                strategy: Sequential,
                witness: variable_journal_config(suffix, ()),
                commit_codec_config: ((), ()),
            }
        }

        fn keyless_compact_fixed_config(
            suffix: &str,
        ) -> storage_keyless::fixed::CompactConfig<Sequential> {
            storage_keyless::CompactConfig {
                strategy: Sequential,
                witness: variable_journal_config(suffix, ()),
                commit_codec_config: (),
            }
        }

        fn keyless_compact_variable_config(
            suffix: &str,
        ) -> storage_keyless::variable::CompactConfig<(), Sequential> {
            storage_keyless::CompactConfig {
                strategy: Sequential,
                witness: variable_journal_config(suffix, ()),
                commit_codec_config: (),
            }
        }

        async fn assert_initial_sync_target_and_apply<T>(context: Context, config: T::Config)
        where
            T: ManagedDb<Context> + 'static,
            T::Unmerkleized: Unmerkleized<Merkleized = T::Merkleized>,
            <T::Unmerkleized as Unmerkleized>::Error: Debug,
        {
            let initial = T::initial_sync_target();
            let db = T::init(context, config, None).await.unwrap();
            assert_eq!(initial, db.sync_target());
            let db = Shared::new("test", db);
            let batch = db
                .new_batch_for_test::<Context>()
                .await
                .merkleize()
                .await
                .expect("empty batch must merkleize");
            let (slot, database) = db.write().await;
            let database = T::apply(database, batch).await.unwrap();
            let (database, sync) = T::finalize(database).await.unwrap();
            slot.put(database);
            sync.await.expect("empty batch database sync failed");
        }

        #[rstest]
        #[case::any(PhantomData::<AnyFixed>, any_fixed_config)]
        #[case::current(PhantomData::<CurrentUnorderedFixed>, current_fixed_config)]
        #[case::immutable(PhantomData::<ImmutableFixed>, immutable_fixed_config)]
        #[case::immutable_compact(
            PhantomData::<ImmutableCompactFixed>, immutable_compact_fixed_config
        )]
        #[case::keyless(PhantomData::<KeylessFixed>, keyless_fixed_config)]
        #[case::keyless_compact(
            PhantomData::<KeylessCompactFixed>, keyless_compact_fixed_config
        )]
        fn merkleize_stale_batch_returns_error<T>(
            #[case] _db: PhantomData<T>,
            #[case] config: fn(&str) -> T::Config,
        ) where
            T: ManagedDb<Context> + 'static,
            T::Unmerkleized: Unmerkleized<
                    Merkleized = T::Merkleized,
                    Error = commonware_storage::qmdb::Error<mmr::Family>,
                >,
            T::SyncTarget: Debug,
        {
            deterministic::Runner::default().start(|context| async move {
                let config = config("db");
                let database = T::init(context.child("db"), config, None).await.unwrap();
                let db = Shared::new("test", database);
                let stale = db.new_batch_for_test::<Context>().await;
                let winner = db
                    .new_batch_for_test::<Context>()
                    .await
                    .merkleize()
                    .await
                    .unwrap();

                // Batch creation releases the read lock, permitting a sibling to be applied.
                let (slot, database) = db.write().await;
                let database = T::apply(database, winner).await.unwrap();
                slot.put(database);

                assert!(matches!(
                    stale.merkleize().await,
                    Err(commonware_storage::qmdb::Error::StaleBatch)
                ));
                assert!(
                    db.new_batch_for_test::<Context>()
                        .await
                        .merkleize()
                        .await
                        .is_ok()
                );
            });
        }

        #[rstest]
        #[case::any_fixed(PhantomData::<AnyFixed>, any_fixed_config)]
        #[case::any_variable(PhantomData::<AnyVariable>, any_variable_config)]
        #[case::current_unordered_fixed(
            PhantomData::<CurrentUnorderedFixed>,
            current_fixed_config
        )]
        #[case::current_ordered_fixed(
            PhantomData::<CurrentOrderedFixed>,
            current_fixed_config
        )]
        #[case::current_unordered_variable(
            PhantomData::<CurrentUnorderedVariable>,
            current_variable_config
        )]
        #[case::current_ordered_variable(
            PhantomData::<CurrentOrderedVariable>,
            current_variable_config
        )]
        #[case::immutable_fixed(PhantomData::<ImmutableFixed>, immutable_fixed_config)]
        #[case::immutable_variable(
            PhantomData::<ImmutableVariable>,
            immutable_variable_config
        )]
        #[case::immutable_compact_fixed(
            PhantomData::<ImmutableCompactFixed>,
            immutable_compact_fixed_config
        )]
        #[case::immutable_compact_variable(
            PhantomData::<ImmutableCompactVariable>,
            immutable_compact_variable_config
        )]
        #[case::keyless_fixed(PhantomData::<KeylessFixed>, keyless_fixed_config)]
        #[case::keyless_variable(PhantomData::<KeylessVariable>, keyless_variable_config)]
        #[case::keyless_compact_fixed(
            PhantomData::<KeylessCompactFixed>,
            keyless_compact_fixed_config
        )]
        #[case::keyless_compact_variable(
            PhantomData::<KeylessCompactVariable>,
            keyless_compact_variable_config
        )]
        fn initial_sync_target_and_empty_apply_match_initialized_database<T>(
            #[case] _db: PhantomData<T>,
            #[case] config: fn(&str) -> T::Config,
        ) where
            T: ManagedDb<Context> + 'static,
            T::Unmerkleized: Unmerkleized<Merkleized = T::Merkleized>,
            <T::Unmerkleized as Unmerkleized>::Error: Debug,
            T::SyncTarget: Debug,
        {
            deterministic::Runner::default().start(|context| async move {
                let config = config("db");
                assert_initial_sync_target_and_apply::<T>(context.child("db"), config).await;
            });
        }
    }

    macro_rules! ready_apply {
        () => {
            async fn apply(self, _batch: Self::Merkleized) -> Result<Self, Self::Error> {
                Ok(self)
            }

            async fn finalize(self) -> Result<(Self, Handle<()>), Self::Error> {
                Ok((self, Handle::ready(Ok(()))))
            }
        };
    }

    #[derive(Default)]
    struct TestDb;

    struct InitializationDb {
        /// Restart target selected during initialization.
        current_target: u64,
    }

    struct PruneCountingDb {
        prune_count: Arc<AtomicUsize>,
    }

    impl<E: Send> ManagedDb<E> for TestDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = ();

        fn initial_sync_target() -> Self::SyncTarget {}

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            Ok(Self)
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {}
    }

    impl<E: Send> ManagedDb<E> for InitializationDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = u64;
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            0
        }

        async fn init(
            _context: E,
            config: Self::Config,
            expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            let target = expected.unwrap_or(config);
            assert!(
                target <= config,
                "database is behind its initialization target"
            );
            Ok(Self {
                current_target: target,
            })
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.current_target
        }
    }

    impl<E: Send> ManagedDb<E> for PruneCountingDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = Arc<AtomicUsize>;
        type SyncTarget = ();

        fn initial_sync_target() -> Self::SyncTarget {}

        async fn init(
            _context: E,
            prune_count: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            Ok(Self { prune_count })
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        async fn prune(self, _target: &Self::SyncTarget) -> Result<Self, Self::Error> {
            self.prune_count.fetch_add(1, Ordering::SeqCst);
            Ok(self)
        }

        fn sync_target(&self) -> Self::SyncTarget {}
    }

    struct BlockingApplyDb {
        started: Option<oneshot::Sender<()>>,
        release: Option<oneshot::Receiver<()>>,
    }

    impl BlockingApplyDb {
        fn new(started: oneshot::Sender<()>, release: oneshot::Receiver<()>) -> Self {
            Self {
                started: Some(started),
                release: Some(release),
            }
        }
    }

    #[derive(Debug)]
    struct TestApplyError;

    struct FailingApplyDb;

    struct SlowSyncDb {
        final_target: u64,
    }

    struct RejectDuplicateTargetSyncDb {
        final_target: u64,
    }

    struct StaleReachedSyncDb {
        final_target: u64,
    }

    struct FastSyncDb {
        final_target: u64,
    }

    struct ImmediateStateSyncDb;

    struct FailingStateSyncDb;

    struct MismatchedTargetSyncDb {
        final_target: u64,
    }

    struct FinishClosedSyncDb {
        final_target: u64,
    }

    struct ObservedSlowSyncDb {
        final_target: u64,
    }

    struct ObservedFastSyncDb {
        final_target: u64,
    }

    struct DistinctObservedFastSyncDb {
        final_target: u64,
    }

    /// Observes an [`ObservedFastSyncDb`]: `ready` resolves once the database is syncing, and
    /// `update_count` counts the targets it takes.
    struct FastSyncObserver {
        ready: oneshot::Sender<()>,
        update_count: Arc<AtomicUsize>,
    }

    impl<E: Send> ManagedDb<E> for FailingApplyDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = TestApplyError;
        type Config = ();
        type SyncTarget = ();

        fn initial_sync_target() -> Self::SyncTarget {}

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            Ok(Self)
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        async fn apply(self, _batch: Self::Merkleized) -> Result<Self, Self::Error> {
            Err(TestApplyError)
        }

        async fn finalize(self) -> Result<(Self, Handle<()>), Self::Error> {
            Ok((self, Handle::ready(Ok(()))))
        }

        fn sync_target(&self) -> Self::SyncTarget {}
    }

    /// Database mock that can fail after recording its selected restart target.
    struct RecoverableStartupDb(u64);

    impl<E: Send> ManagedDb<E> for RecoverableStartupDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = &'static str;
        type Config = (Arc<AtomicU64>, bool);
        type SyncTarget = u64;

        fn initial_sync_target() -> u64 {
            0
        }

        async fn init(
            _context: E,
            (state, fail): Self::Config,
            expected: Option<u64>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            let retained = state.load(Ordering::Relaxed);
            let selected = expected.unwrap_or(retained);
            if selected > retained {
                return Err(InitError::Database("database is behind"));
            }
            state.store(selected, Ordering::Relaxed);
            if fail {
                return Err(InitError::Database(
                    "initialization interrupted after durable repair",
                ));
            }
            Ok(Self(selected))
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &u64) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> u64 {
            self.0
        }
    }

    #[test]
    fn tuple_initialization_retries_partial_failure_and_rejects_behind() {
        deterministic::Runner::default().start(|context| async move {
            type DbSet = (Shared<RecoverableStartupDb>, Shared<RecoverableStartupDb>);
            let left = Arc::new(AtomicU64::new(3));
            let right = Arc::new(AtomicU64::new(2));
            let result = std::panic::AssertUnwindSafe(<DbSet as DatabaseSet<_>>::init(
                context.child("interrupted"),
                ((left.clone(), false), (right.clone(), true)),
                Some((1, 1)),
            ))
            .catch_unwind()
            .await;
            assert!(result.is_err());
            assert_eq!(left.load(Ordering::Relaxed), 1);
            assert_eq!(right.load(Ordering::Relaxed), 1);
            let recovered = <DbSet as DatabaseSet<_>>::init(
                context.child("retry"),
                ((left.clone(), false), (right.clone(), false)),
                Some((1, 1)),
            )
            .await;
            assert_eq!((*recovered.0.read().await).0, 1);
            assert_eq!((*recovered.1.read().await).0, 1);
            drop(recovered);
            let result = std::panic::AssertUnwindSafe(<DbSet as DatabaseSet<_>>::init(
                context.child("behind"),
                ((left, false), (right, false)),
                Some((1, 2)),
            ))
            .catch_unwind()
            .await;
            assert!(result.is_err());
        });
    }

    #[test]
    fn tuple_initialization_validates_ahead_and_aligned_databases() {
        deterministic::Runner::default().start(|context| async move {
            type DbSet = (Shared<InitializationDb>, Shared<InitializationDb>);
            let (left, right) =
                <DbSet as DatabaseSet<_>>::init(context, (2, 1), Some((1, 1))).await;
            for database in [left, right] {
                let db = database.read().await;
                assert_eq!(db.current_target, 1);
            }
        });
    }

    #[test]
    fn database_set_prune_calls_managed_db_prune() {
        deterministic::Runner::default().start(|_context| async move {
            let prune_count = Arc::new(AtomicUsize::new(0));
            let database = Shared::new(
                "test",
                PruneCountingDb {
                    prune_count: prune_count.clone(),
                },
            );

            <Shared<PruneCountingDb> as DatabaseSet<deterministic::Context>>::prune(&database, &())
                .await;

            assert_eq!(prune_count.load(Ordering::SeqCst), 1);
        });
    }

    impl<E: Send> ManagedDb<E> for BlockingApplyDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = ();

        fn initial_sync_target() -> Self::SyncTarget {}

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("BlockingApplyDb is constructed directly in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        async fn apply(mut self, _batch: Self::Merkleized) -> Result<Self, Self::Error> {
            if let Some(started) = self.started.take() {
                let _ = started.send(());
            }
            if let Some(release) = self.release.take() {
                let _ = release.await;
            }
            Ok(self)
        }

        async fn finalize(self) -> Result<(Self, Handle<()>), Self::Error> {
            Ok((self, Handle::ready(Ok(()))))
        }

        fn sync_target(&self) -> Self::SyncTarget {}
    }

    impl<E: Send> ManagedDb<E> for SlowSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("SlowSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("SlowSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for RejectDuplicateTargetSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!(
                "RejectDuplicateTargetSyncDb is only constructed through state sync in tests"
            )
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!(
                "RejectDuplicateTargetSyncDb is only constructed through state sync in tests"
            )
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for FastSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("FastSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("FastSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for FailingStateSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("FailingStateSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("FailingStateSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            0
        }
    }

    impl<E: Send> ManagedDb<E> for MismatchedTargetSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("MismatchedTargetSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("MismatchedTargetSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for ImmediateStateSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("ImmediateStateSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("ImmediateStateSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            0
        }
    }

    impl<E: Send> ManagedDb<E> for FinishClosedSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("FinishClosedSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("FinishClosedSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for ObservedSlowSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("ObservedSlowSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("ObservedSlowSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for ObservedFastSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("ObservedFastSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("ObservedFastSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> ManagedDb<E> for DistinctObservedFastSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!(
                "DistinctObservedFastSyncDb is only constructed through state sync in tests"
            )
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!(
                "DistinctObservedFastSyncDb is only constructed through state sync in tests"
            )
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> StateSyncDb<E, oneshot::Receiver<()>> for SlowSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            release: oneshot::Receiver<()>,
            target: Self::SyncTarget,
            tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let _ = release.await;
            let mut final_target = target;
            let mut tip_updates = Some(tip_updates);

            loop {
                if let Some(reached_target) = reached_target.as_ref()
                    && !reached_target.send_lossy(final_target)
                {
                    break;
                }

                if finish.is_none() && tip_updates.is_none() {
                    break;
                }

                let finish_signal = finish.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |finish_rx| futures::future::Either::Left(finish_rx.recv()),
                );
                let update_signal = tip_updates.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |update_rx| futures::future::Either::Left(update_rx.recv()),
                );

                select! {
                    _ = finish_signal => {
                        break;
                    },
                    update = update_signal => match update {
                        Some(update) => {
                            final_target = update;
                        }
                        None => {
                            tip_updates = None;
                            if finish.is_none() {
                                break;
                            }
                        }
                    },
                }
            }

            Ok(Self { final_target })
        }
    }

    impl<E: Send> StateSyncDb<E, oneshot::Receiver<()>> for RejectDuplicateTargetSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            mut release: oneshot::Receiver<()>,
            target: Self::SyncTarget,
            mut tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let mut final_target = target;
            loop {
                select! {
                    _ = &mut release => break,
                    update = tip_updates.recv() => {
                        let Some(update) = update else {
                            break;
                        };
                        assert_ne!(
                            update, final_target,
                            "state sync must not send duplicate target updates"
                        );
                        final_target = update;
                    },
                }
            }

            // Targets queued before the release are taken now; a repeat of the current target is
            // a coordinator bug.
            while let Ok(update) = tip_updates.try_recv() {
                assert_ne!(
                    update, final_target,
                    "state sync must not send duplicate target updates"
                );
                final_target = update;
            }

            if let Some(reached_target) = reached_target.as_ref() {
                let _ = reached_target.send_lossy(final_target);
            }
            if let Some(finish_rx) = finish.as_mut() {
                let _ = finish_rx.recv().await;
            }

            Ok(Self { final_target })
        }
    }

    impl<E: Send> ManagedDb<E> for StaleReachedSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("StaleReachedSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("StaleReachedSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E> StateSyncDb<E, ()> for StaleReachedSyncDb
    where
        E: Clock,
    {
        type SyncError = Infallible;

        async fn sync_db(
            context: E,
            _config: Self::Config,
            _resolver: (),
            target: Self::SyncTarget,
            mut tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let update = tip_updates.recv().await.expect("expected forwarded tip");
            if let Some(reached_target) = reached_target.as_ref() {
                let _ = reached_target.send_lossy(target);
            }

            // Hold at the stale target for a settle window. A coordinator that finishes on the
            // stale report sends finish within it, and the database then completes at a target
            // the coordinator did not converge on. A correct coordinator ignores the report, so
            // only the window ends the hold.
            let finish_signal = finish.as_mut().map_or_else(
                || futures::future::Either::Right(futures::future::pending()),
                |finish_rx| futures::future::Either::Left(finish_rx.recv()),
            );
            select! {
                _ = finish_signal => Ok(Self {
                    final_target: target
                }),
                _ = context.sleep(Duration::from_millis(10)) => {
                    if let Some(reached_target) = reached_target.as_ref() {
                        let _ = reached_target.send_lossy(update);
                    }
                    if let Some(finish_rx) = finish.as_mut() {
                        let _ = finish_rx.recv().await;
                    }
                    Ok(Self {
                        final_target: update,
                    })
                },
            }
        }
    }

    impl<E: Send> StateSyncDb<E, oneshot::Sender<()>> for FastSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            done: oneshot::Sender<()>,
            target: Self::SyncTarget,
            tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let _ = done.send(());
            let mut final_target = target;
            let mut tip_updates = Some(tip_updates);

            loop {
                if let Some(reached_target) = reached_target.as_ref()
                    && !reached_target.send_lossy(final_target)
                {
                    break;
                }

                if finish.is_none() && tip_updates.is_none() {
                    break;
                }

                let finish_signal = finish.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |finish_rx| futures::future::Either::Left(finish_rx.recv()),
                );
                let update_signal = tip_updates.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |update_rx| futures::future::Either::Left(update_rx.recv()),
                );

                select! {
                    _ = finish_signal => {
                        break;
                    },
                    update = update_signal => match update {
                        Some(update) => {
                            final_target = update;
                        }
                        None => {
                            tip_updates = None;
                            if finish.is_none() {
                                break;
                            }
                        }
                    },
                }
            }

            Ok(Self { final_target })
        }
    }

    #[derive(Debug)]
    struct TestSyncError;

    #[derive(Debug)]
    struct FinishClosedSyncError;

    impl<E: Send> StateSyncDb<E, ()> for FailingStateSyncDb {
        type SyncError = TestSyncError;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            _resolver: (),
            _target: Self::SyncTarget,
            _tip_updates: ring::Receiver<Self::SyncTarget>,
            _finish: Option<mpsc::Receiver<()>>,
            _reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            Err(TestSyncError)
        }
    }

    impl<E: Send> StateSyncDb<E, ()> for ImmediateStateSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            _resolver: (),
            _target: Self::SyncTarget,
            _tip_updates: ring::Receiver<Self::SyncTarget>,
            _finish: Option<mpsc::Receiver<()>>,
            _reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            Ok(Self)
        }
    }

    impl<E: Send> StateSyncDb<E, ()> for MismatchedTargetSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            _resolver: (),
            target: Self::SyncTarget,
            _tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            if let Some(reached_target) = reached_target.as_ref() {
                let _ = reached_target.send_lossy(target);
            }
            if let Some(finish_rx) = finish.as_mut() {
                let _ = finish_rx.recv().await;
            }
            Ok(Self {
                final_target: target + 1,
            })
        }
    }

    impl<E: Send> StateSyncDb<E, ()> for FinishClosedSyncDb {
        type SyncError = FinishClosedSyncError;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            _resolver: (),
            target: Self::SyncTarget,
            _tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            _reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let Some(finish_rx) = finish.as_mut() else {
                panic!("finish receiver should be provided");
            };
            match finish_rx.recv().await {
                Some(()) => Ok(Self {
                    final_target: target,
                }),
                None => Err(FinishClosedSyncError),
            }
        }
    }

    impl<E: Send> StateSyncDb<E, ()> for ObservedSlowSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            _resolver: (),
            target: Self::SyncTarget,
            tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let mut final_target = target;
            let mut tip_updates = Some(tip_updates);
            let mut reported_target = None;
            let mut observed_update = false;
            loop {
                if let Some(update_rx) = tip_updates.as_mut() {
                    loop {
                        match update_rx.try_recv() {
                            Ok(update) => {
                                final_target = update;
                                observed_update = true;
                                reported_target = None;
                            }
                            Err(ring::TryRecvError::Empty) => {
                                break;
                            }
                            Err(ring::TryRecvError::Disconnected) => {
                                tip_updates = None;
                                break;
                            }
                        }
                    }
                }

                if observed_update && reported_target != Some(final_target) {
                    if let Some(reached_target) = reached_target.as_ref()
                        && !reached_target.send_lossy(final_target)
                    {
                        break;
                    }
                    reported_target = Some(final_target);
                }

                if finish.is_none() && tip_updates.is_none() {
                    break;
                }

                let finish_signal = finish.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |finish_rx| futures::future::Either::Left(finish_rx.recv()),
                );
                let update_signal = tip_updates.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |update_rx| futures::future::Either::Left(update_rx.recv()),
                );

                select! {
                    _ = finish_signal => {
                        break;
                    },
                    update = update_signal => match update {
                        Some(update) => {
                            final_target = update;
                            observed_update = true;
                            reported_target = None;
                        }
                        None => {
                            tip_updates = None;
                            if finish.is_none() {
                                break;
                            }
                        }
                    },
                }
            }

            Ok(Self { final_target })
        }
    }

    impl<E: Send> StateSyncDb<E, FastSyncObserver> for ObservedFastSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            observer: FastSyncObserver,
            target: Self::SyncTarget,
            tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let mut final_target = target;
            let mut tip_updates = Some(tip_updates);
            let mut reported_target = None;
            let _ = observer.ready.send(());

            loop {
                if reported_target != Some(final_target) {
                    if let Some(reached_target) = reached_target.as_ref()
                        && !reached_target.send_lossy(final_target)
                    {
                        break;
                    }
                    reported_target = Some(final_target);
                }

                if finish.is_none() && tip_updates.is_none() {
                    break;
                }

                let finish_signal = finish.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |finish_rx| futures::future::Either::Left(finish_rx.recv()),
                );
                let update_signal = tip_updates.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |update_rx| futures::future::Either::Left(update_rx.recv()),
                );

                select! {
                    _ = finish_signal => {
                        break;
                    },
                    update = update_signal => match update {
                        Some(update) => {
                            observer.update_count.fetch_add(1, Ordering::SeqCst);
                            final_target = update;
                            reported_target = None;
                        }
                        None => {
                            tip_updates = None;
                            if finish.is_none() {
                                break;
                            }
                        }
                    },
                }
            }

            Ok(Self { final_target })
        }
    }

    impl<E: Send> StateSyncDb<E, FastSyncObserver> for DistinctObservedFastSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            observer: FastSyncObserver,
            target: Self::SyncTarget,
            tip_updates: ring::Receiver<Self::SyncTarget>,
            mut finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let mut final_target = target;
            let mut tip_updates = Some(tip_updates);
            let mut reported_target = None;
            let _ = observer.ready.send(());

            loop {
                if reported_target != Some(final_target) {
                    if let Some(reached_target) = reached_target.as_ref()
                        && !reached_target.send_lossy(final_target)
                    {
                        break;
                    }
                    reported_target = Some(final_target);
                }

                if finish.is_none() && tip_updates.is_none() {
                    break;
                }

                let finish_signal = finish.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |finish_rx| futures::future::Either::Left(finish_rx.recv()),
                );
                let update_signal = tip_updates.as_mut().map_or_else(
                    || futures::future::Either::Right(futures::future::pending()),
                    |update_rx| futures::future::Either::Left(update_rx.recv()),
                );

                select! {
                    _ = finish_signal => {
                        break;
                    },
                    update = update_signal => match update {
                        Some(update) => {
                            observer.update_count.fetch_add(1, Ordering::SeqCst);
                            if update != final_target {
                                final_target = update;
                                reported_target = None;
                            }
                        }
                        None => {
                            tip_updates = None;
                            if finish.is_none() {
                                break;
                            }
                        }
                    },
                }
            }

            Ok(Self { final_target })
        }
    }

    #[test]
    fn tuple_new_batches_queues_reads_concurrently() {
        deterministic::Runner::default().start(|_context| async move {
            let db1 = Shared::new("test", TestDb);
            let db2 = Shared::new("test", TestDb);
            let databases = (db1.clone(), db2.clone());

            let (slot1, taken1) = db1.write().await;
            let (slot2, taken2) = db2.write().await;

            let new_batches = <(Shared<TestDb>, Shared<TestDb>) as DatabaseSet<
                deterministic::Context,
            >>::new_batches(&databases);
            pin_mut!(new_batches);
            assert!(new_batches.as_mut().now_or_never().is_none());

            slot2.put(taken2);
            {
                let writer2_again = db2.write();
                pin_mut!(writer2_again);
                assert!(
                    writer2_again.as_mut().now_or_never().is_none(),
                    "tuple new_batches should queue reads for all databases concurrently"
                );
            }

            slot1.put(taken1);
            let _ = new_batches.await;
        });
    }

    #[test]
    fn new_batches_releases_read_lock_before_returning() {
        deterministic::Runner::default().start(|_context| async move {
            let database = Shared::new("test", TestDb);
            let _ = <Shared<TestDb> as DatabaseSet<deterministic::Context>>::new_batches(&database)
                .await;

            let writer = database.write();
            pin_mut!(writer);
            assert!(
                writer.as_mut().now_or_never().is_some(),
                "batch construction must release its read lock before returning",
            );
        });
    }

    #[test]
    fn tuple_apply_does_not_hold_ready_writer_while_waiting_for_reader() {
        deterministic::Runner::default().start(|_context| async move {
            type DbSet = (Shared<TestDb>, Shared<TestDb>);

            let db1 = Shared::new("test", TestDb);
            let db2 = Shared::new("test", TestDb);
            let databases = (db1.clone(), db2.clone());
            let reader1 = db1.read().await;

            let apply = async {
                <DbSet as DatabaseSet<deterministic::Context>>::apply(
                    &databases,
                    (TestMerkleized, TestMerkleized),
                )
                .await
            };
            pin_mut!(apply);
            assert!(apply.as_mut().now_or_never().is_none());

            let reader2 = db2.read();
            pin_mut!(reader2);
            assert!(
                reader2.as_mut().now_or_never().is_some(),
                "tuple apply must not hold one writer while waiting for another database's reader",
            );

            drop(reader1);
            apply.await;
        });
    }

    #[test]
    fn tuple_apply_runs_databases_in_parallel() {
        deterministic::Runner::default().start(|_context| async move {
            let (started1_tx, started1_rx) = oneshot::channel();
            let (started2_tx, started2_rx) = oneshot::channel();
            let (release1_tx, release1_rx) = oneshot::channel();
            let (release2_tx, release2_rx) = oneshot::channel();

            let databases = (
                Shared::new("test", BlockingApplyDb::new(started1_tx, release1_rx)),
                Shared::new("test", BlockingApplyDb::new(started2_tx, release2_rx)),
            );

            let apply = <(Shared<BlockingApplyDb>, Shared<BlockingApplyDb>) as DatabaseSet<
                deterministic::Context,
            >>::apply(&databases, (TestMerkleized, TestMerkleized));
            pin_mut!(apply);
            assert!(apply.as_mut().now_or_never().is_none());

            let started1 = started1_rx;
            let started2 = started2_rx;
            pin_mut!(started1);
            pin_mut!(started2);
            assert!(matches!(started1.as_mut().now_or_never(), Some(Ok(()))));
            assert!(
                matches!(started2.as_mut().now_or_never(), Some(Ok(()))),
                "tuple apply should start on all databases concurrently"
            );

            let _ = release1_tx.send(());
            let _ = release2_tx.send(());
            apply.await;
        });
    }

    #[test]
    #[should_panic(
        expected = "database apply failed (index 1, type commonware_glue::stateful::db::tests::FailingApplyDb)"
    )]
    fn tuple_apply_panic_identifies_failing_database() {
        deterministic::Runner::default().start(|_context| async move {
            let databases = (
                Shared::new("test", TestDb),
                Shared::new("test", FailingApplyDb),
            );
            let _ = <(Shared<TestDb>, Shared<FailingApplyDb>) as DatabaseSet<
                deterministic::Context,
            >>::apply(&databases, (TestMerkleized, TestMerkleized))
            .await;
        });
    }

    #[test]
    #[should_panic(
        expected = "database sync failed (index 1, type commonware_glue::stateful::db::tests::TestDb)"
    )]
    fn barrier_panics_on_flush_failure() {
        deterministic::Runner::default().start(|_context| async move {
            let barrier = Barrier::from_handles::<TestDb>([
                Handle::ready(Ok(())),
                Handle::ready(Err(RuntimeError::WriteFailed)),
            ]);
            let _ = barrier.durable().await;
        });
    }

    #[rstest::rstest]
    #[case(30)]
    #[case(31)]
    #[case(64)]
    fn barrier_panics_on_later_failure(
        #[case] count: usize,
        #[values(false, true)] shutdown: bool,
    ) {
        deterministic::Runner::default().start(|context| async move {
            // The first handle stays pending and the last one fails. With `shutdown`, the second
            // handle also reports `Closed`, and the barrier must still panic on the failure.
            let (pending_tx, pending_rx) = oneshot::channel();
            let mut handles = vec![Handle::from_receiver(pending_rx)];
            handles.extend((1..count).map(|index| {
                Handle::ready(if index == count - 1 {
                    Err(RuntimeError::WriteFailed)
                } else if shutdown && index == 1 {
                    Err(RuntimeError::Closed)
                } else {
                    Ok(())
                })
            }));

            // Catching the unwind lets the test inspect the panic message. The timeout fails the
            // test if the failure is held behind the pending handle.
            let barrier = Barrier::from_handles::<TestDb>(handles);
            let mut durable = Box::pin(AssertUnwindSafe(barrier.durable()).catch_unwind());
            let result = select! {
                result = &mut durable => result,
                _ = context.sleep(Duration::from_secs(1)) => {
                    panic!("barrier must panic without waiting for the pending sync")
                }
            };
            let panic = result
                .expect_err("sync failure must panic even when another sync reports shutdown");
            let message = panic
                .downcast_ref::<String>()
                .expect("panic payload must be a String");
            assert!(message.contains(&format!(
                "database sync failed (index {}, type {}):",
                count - 1,
                std::any::type_name::<TestDb>(),
            )));

            // The pending handle is dropped by the time the panic surfaces, before the outer
            // future is dropped.
            assert!(pending_tx.send(Ok(())).is_err());
            drop(durable);
        });
    }

    #[rstest::rstest]
    #[case(1)]
    #[case(30)]
    #[case(31)]
    #[case(64)]
    fn barrier_waits_for_every_handle(#[case] count: usize, #[values(false, true)] shutdown: bool) {
        deterministic::Runner::default().start(|context| async move {
            // The first handle stays pending, so the barrier must not resolve before it completes.
            let (pending_tx, pending_rx) = oneshot::channel();
            let handles = std::iter::once(Handle::from_receiver(pending_rx))
                .chain((1..count).map(|_| Handle::ready(Ok(()))));
            let mut durable = Box::pin(Barrier::from_handles::<TestDb>(handles).durable());
            assert!(durable.as_mut().now_or_never().is_none());

            // Once the pending handle completes, the barrier must resolve. It is durable unless
            // that handle reports `Aborted`.
            pending_tx
                .send(if shutdown {
                    Err(RuntimeError::Aborted)
                } else {
                    Ok(())
                })
                .unwrap();
            let result = select! {
                result = durable => result,
                _ = context.sleep(Duration::from_secs(1)) => {
                    panic!("barrier must resolve once every handle completes")
                }
            };
            assert_eq!(result, !shutdown);
        });
    }

    #[rstest::rstest]
    #[case(1)]
    #[case(30)]
    #[case(31)]
    #[case(64)]
    fn barrier_cancellation_releases_handles(
        #[case] count: usize,
        #[values(false, true)] poll: bool,
    ) {
        deterministic::Runner::default().start(|_context| async move {
            // Dropping the barrier future, with or without a first poll, must drop every handle.
            let (senders, handles): (Vec<_>, Vec<_>) = (0..count)
                .map(|_| {
                    let (sender, receiver) = oneshot::channel();
                    (sender, Handle::from_receiver(receiver))
                })
                .unzip();
            let mut durable = Box::pin(Barrier::from_handles::<TestDb>(handles).durable());
            if poll {
                assert!(durable.as_mut().now_or_never().is_none());
            }
            drop(durable);
            for sender in senders {
                assert!(sender.send(Ok(())).is_err());
            }
        });
    }

    #[test]
    fn barrier_empty_is_durable() {
        deterministic::Runner::default().start(|_context| async move {
            assert!(Barrier::from_handles::<TestDb>([]).durable().await);
        });
    }

    #[test]
    fn barrier_reports_shutdown_as_not_durable() {
        deterministic::Runner::default().start(|_context| async move {
            let barrier = Barrier {
                syncs: vec![
                    ("db0", Some(0), Handle::ready(Ok(()))),
                    ("db1", Some(1), Handle::ready(Err(RuntimeError::Closed))),
                ],
            };
            assert!(!barrier.durable().await);

            let barrier = Barrier {
                syncs: vec![("db0", None, Handle::ready(Err(RuntimeError::Aborted)))],
            };
            assert!(!barrier.durable().await);
        });
    }

    type TestAnchor = Anchor<sha256::Digest>;

    fn anchor(n: u64) -> TestAnchor {
        mock_anchor(n, n as u8)
    }

    #[test]
    fn tip_update_observation_follows_recording() {
        deterministic::Runner::default().start(|_context| async move {
            let (update, mut observed) = TipUpdate::with_observation(anchor(1), 7u64);
            let mut recorded = None;

            update.record(|new_anchor, new_target| {
                assert!((&mut observed).now_or_never().is_none());
                recorded = Some((new_anchor, new_target));
            });

            assert_eq!(recorded, Some((anchor(1), 7)));
            observed.await.expect("recorded update should be observed");
        });
    }

    #[test]
    fn dispatch_queued_tips_keeps_highest_recorded_target() {
        deterministic::Runner::default().start(|_context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let (target_tx, mut target_rx) = ring::channel(NZUsize!(1));
            let (newer_update, newer_observed) = TipUpdate::with_observation(anchor(2), 2u64);
            let (older_update, older_observed) = TipUpdate::with_observation(anchor(1), 1u64);

            let _ = tip_tx.send(newer_update).await;
            let _ = tip_tx.send(older_update).await;

            let mut tip_updates = Some(tip_rx);
            let mut current_anchor = anchor(0);
            let mut current_target = 0u64;
            assert!(dispatch_queued_tips(
                &mut tip_updates,
                &target_tx,
                &mut current_anchor,
                &mut current_target,
            ));

            newer_observed
                .await
                .expect("newer update should be observed");
            older_observed
                .await
                .expect("older update should also be observed");
            assert_eq!(current_anchor, anchor(2));
            assert_eq!(current_target, 2);
            assert_eq!(target_rx.recv().await, Some(2));
            assert!(matches!(
                target_rx.try_recv(),
                Err(ring::TryRecvError::Empty)
            ));
        });
    }

    #[test]
    fn dispatch_queued_tips_advances_anchor_without_duplicate_target() {
        deterministic::Runner::default().start(|_context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());
            let (target_tx, mut target_rx) = ring::channel(NZUsize!(1));
            let (update, observed) = TipUpdate::with_observation(anchor(3), 7u64);

            let _ = tip_tx.send(update).await;

            let mut tip_updates = Some(tip_rx);
            let mut current_anchor = anchor(2);
            let mut current_target = 7u64;
            assert!(dispatch_queued_tips(
                &mut tip_updates,
                &target_tx,
                &mut current_anchor,
                &mut current_target,
            ));

            observed.await.expect("update should be observed");
            assert_eq!(current_anchor, anchor(3));
            assert_eq!(current_target, 7);
            assert!(matches!(
                target_rx.try_recv(),
                Err(ring::TryRecvError::Empty)
            ));
        });
    }

    #[test]
    fn single_state_sync_handles_closed_tip_updates_channel() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());
            let (release, release_rx) = oneshot::channel();
            let sync = <Shared<SlowSyncDb> as StateSyncSet<_, _>>::sync(
                context.child("sync"),
                (),
                release_rx,
                anchor(0),
                0,
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NonZeroU64::new(1).unwrap(),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            );
            pin_mut!(sync);

            // The tip channel closes while the coordinator waits for tips and reports.
            assert!(sync.as_mut().now_or_never().is_none());
            drop(tip_tx);
            assert!(sync.as_mut().now_or_never().is_none());

            // Once released, the database reports the initial target and the sync finishes there.
            release.send(()).unwrap();
            let (_database, converged_anchor) = sync.await.expect("sync should succeed");
            assert_eq!(converged_anchor, anchor(0));
        });
    }

    #[test]
    fn single_state_sync_preserves_db_error_when_target_channel_closes() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());
            let _ = tip_tx.send(TipUpdate::new(anchor(1), 1u64)).await;

            let result =
                <Shared<FailingStateSyncDb> as StateSyncSet<deterministic::Context, ()>>::sync(
                    context,
                    (),
                    (),
                    anchor(0),
                    0,
                    tip_rx,
                    SyncEngineConfig {
                        fetch_batch_size: NonZeroU64::new(1).unwrap(),
                        apply_batch_size: NZU64!(1),
                        max_outstanding_requests: NZUsize!(1),
                    },
                )
                .await;

            assert!(matches!(result, Err(TestSyncError)));
        });
    }

    #[test]
    fn single_state_sync_ignores_backward_tip_updates() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let source = ();

            let sync = context
                .child("single_state_sync_ignores_backward_tip_updates")
                .spawn(move |context| async move {
                    <Shared<ObservedSlowSyncDb> as StateSyncSet<deterministic::Context, ()>>::sync(
                        context,
                        (),
                        source,
                        anchor(0),
                        0,
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NonZeroU64::new(1).unwrap(),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("single state sync should succeed")
                });

            let _ = tip_tx.send(TipUpdate::new(anchor(2), 2)).await;
            let _ = tip_tx.send(TipUpdate::new(anchor(1), 1)).await;
            drop(tip_tx);

            let (database, converged_anchor) = sync.await.expect("sync task should complete");
            let final_target = database.read().await.final_target;
            assert_eq!(
                final_target, 2,
                "single-db sync target must never move backward"
            );
            assert_eq!(
                converged_anchor,
                anchor(2),
                "converged anchor must remain on the highest seen tip"
            );
        });
    }

    #[test]
    fn single_state_sync_advances_anchor_without_duplicate_target_update() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let (release, release_rx) = oneshot::channel();
            let sync = <Shared<RejectDuplicateTargetSyncDb> as StateSyncSet<_, _>>::sync(
                context.child("sync"),
                (),
                release_rx,
                anchor(7),
                7,
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NonZeroU64::new(1).unwrap(),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            );
            pin_mut!(sync);

            // Park the coordinator so the tip reaches its select arm rather than the initial
            // drain. The tip advances the anchor but repeats the target.
            assert!(sync.as_mut().now_or_never().is_none());
            let (update, mut observed) = TipUpdate::with_observation(anchor(9), 7);
            assert!(tip_tx.send_lossy(update));
            assert!(sync.as_mut().now_or_never().is_none());
            assert_eq!(
                (&mut observed).now_or_never(),
                Some(Ok(())),
                "the coordinator records a tip that repeats the target"
            );

            // The database never receives a duplicate target and finishes at its initial one.
            release.send(()).unwrap();
            drop(tip_tx);
            let (database, converged_anchor) = sync.await.expect("sync should succeed");
            assert_eq!(database.read().await.final_target, 7);
            assert_eq!(converged_anchor, anchor(9));
        });
    }

    #[test]
    fn single_state_sync_ignores_stale_reached_after_forwarded_tip() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());

            let sync =
                context
                    .child("single_state_sync_stale_reached")
                    .spawn(move |context| async move {
                        <Shared<StaleReachedSyncDb> as StateSyncSet<
                            deterministic::Context,
                            (),
                        >>::sync(
                            context,
                            (),
                            (),
                            anchor(0),
                            0,
                            tip_rx,
                            SyncEngineConfig {
                                fetch_batch_size: NonZeroU64::new(1).unwrap(),
                                apply_batch_size: NZU64!(1),
                                max_outstanding_requests: NZUsize!(1),
                            },
                        )
                        .await
                        .expect("single state sync should succeed")
                    });

            let _ = tip_tx.send(TipUpdate::new(anchor(2), 2)).await;

            let (database, converged_anchor) = sync.await.expect("sync task should complete");
            let final_target = database.read().await.final_target;
            assert_eq!(
                final_target, 2,
                "single-db sync must not finish on a stale reached target",
            );
            assert_eq!(
                converged_anchor,
                anchor(2),
                "converged anchor must match the target the database reached",
            );
        });
    }

    #[test]
    fn tuple_state_sync_converges_before_finish() {
        deterministic::Runner::default().start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let (slow_release, slow_release_for_sync) = oneshot::channel();
            let (fast_done_for_sync, fast_done) = oneshot::channel();
            let sync = context
                .child("tuple_state_sync")
                .spawn(move |context| async move {
                    <(Shared<SlowSyncDb>, Shared<FastSyncDb>) as StateSyncSet<
                        deterministic::Context,
                        (oneshot::Receiver<()>, oneshot::Sender<()>),
                    >>::sync(
                        context,
                        ((), ()),
                        (slow_release_for_sync, fast_done_for_sync),
                        anchor(0),
                        (0, 0),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NonZeroU64::new(1).unwrap(),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                });

            fast_done.await.expect("the fast database should start");
            let _ = tip_tx.send(TipUpdate::new(anchor(1), (1, 1))).await;
            let _ = tip_tx.send(TipUpdate::new(anchor(2), (2, 2))).await;
            slow_release.send(()).unwrap();
            drop(tip_tx);

            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            let slow_target = synced.0.read().await.final_target;
            let fast_target = synced.1.read().await.final_target;

            assert_eq!(
                slow_target, fast_target,
                "all databases should finish on the same converged target set"
            );
            assert_eq!(
                converged_anchor.height.get(),
                slow_target,
                "returned anchor height should match the converged generation"
            );
        });
    }

    #[test]
    fn tuple_state_sync_ignores_backward_tip_updates() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(8).unwrap());
            let (slow_release, slow_release_for_sync) = oneshot::channel();
            let (fast_done_for_sync, fast_done) = oneshot::channel();
            let sync = context
                .child("tuple_state_sync_ignores_backward_tip_updates")
                .spawn(move |context| async move {
                    <(Shared<SlowSyncDb>, Shared<FastSyncDb>) as StateSyncSet<
                        deterministic::Context,
                        (oneshot::Receiver<()>, oneshot::Sender<()>),
                    >>::sync(
                        context,
                        ((), ()),
                        (slow_release_for_sync, fast_done_for_sync),
                        anchor(0),
                        (0, 0),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NonZeroU64::new(1).unwrap(),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                });

            fast_done.await.expect("the fast database should start");

            let _ = tip_tx.send(TipUpdate::new(anchor(2), (2, 2))).await;
            let (backward, observed) = TipUpdate::with_observation(anchor(1), (1, 1));
            let _ = tip_tx.send(backward).await;
            drop(tip_tx);
            observed
                .await
                .expect("the coordinator records the backward tip");
            slow_release.send(()).unwrap();

            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            let slow_target = synced.0.read().await.final_target;
            let fast_target = synced.1.read().await.final_target;
            assert_eq!(
                slow_target, 2,
                "slow database target must never move backward"
            );
            assert_eq!(
                fast_target, 2,
                "fast database target must never move backward"
            );
            assert_eq!(
                converged_anchor,
                anchor(2),
                "converged anchor must remain on the highest seen tip"
            );
        });
    }

    #[test]
    fn tuple_state_sync_rejects_database_target_mismatch() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (_tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());
            let (fast_done, _fast_started) = oneshot::channel();

            let result = <(Shared<MismatchedTargetSyncDb>, Shared<FastSyncDb>) as StateSyncSet<
                deterministic::Context,
                ((), oneshot::Sender<()>),
            >>::sync(
                context,
                ((), ()),
                ((), fast_done),
                anchor(7),
                (7, 7),
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NonZeroU64::new(1).unwrap(),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            )
            .await;

            let err = match result {
                Ok(_) => panic!("tuple state sync should reject a mismatched database target"),
                Err(err) => err,
            };
            assert!(
                err.contains("database targets do not match"),
                "error should identify the target mismatch, got: {err}"
            );
        });
    }

    #[test]
    fn tuple_state_sync_returns_db_error_instead_of_panicking_when_anchor_missing() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (_tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());

            let result =
                <(Shared<ImmediateStateSyncDb>, Shared<FailingStateSyncDb>) as StateSyncSet<
                    deterministic::Context,
                    ((), ()),
                >>::sync(
                    context,
                    ((), ()),
                    ((), ()),
                    anchor(0),
                    (0, 0),
                    tip_rx,
                    SyncEngineConfig {
                        fetch_batch_size: NonZeroU64::new(1).unwrap(),
                        apply_batch_size: NZU64!(1),
                        max_outstanding_requests: NZUsize!(1),
                    },
                )
                .await;

            let err = match result {
                Ok(_) => panic!("tuple state sync should return the database sync error"),
                Err(err) => err,
            };
            assert!(
                err.contains("state sync failed (index 1, db"),
                "error should include failing database index: {err}"
            );
            assert!(
                err.contains("FailingStateSyncDb"),
                "error should include failing database type: {err}"
            );
        });
    }

    #[test]
    fn tuple_state_sync_returns_db_error_when_other_database_waits_for_finish() {
        deterministic::Runner::timed(Duration::from_secs(1)).start(|context| async move {
            let (_tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());
            let (release_tx, release) = oneshot::channel();
            release_tx.send(()).unwrap();

            let result = <(Shared<SlowSyncDb>, Shared<FailingStateSyncDb>) as StateSyncSet<
                deterministic::Context,
                (oneshot::Receiver<()>, ()),
            >>::sync(
                context,
                ((), ()),
                (release, ()),
                anchor(0),
                (0, 0),
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NonZeroU64::new(1).unwrap(),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            )
            .await;

            let err = match result {
                Ok(_) => panic!("tuple state sync should return the database sync error"),
                Err(err) => err,
            };
            assert!(
                err.contains("state sync failed (index 1, db"),
                "error should include failing database index: {err}"
            );
            assert!(
                err.contains("FailingStateSyncDb"),
                "error should include failing database type: {err}"
            );
        });
    }

    #[test]
    fn tuple_state_sync_preserves_original_failure_when_peer_finish_channel_closes() {
        deterministic::Runner::timed(Duration::from_secs(1)).start(|context| async move {
            let (_tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(1).unwrap());

            let result =
                <(Shared<FinishClosedSyncDb>, Shared<FailingStateSyncDb>) as StateSyncSet<
                    deterministic::Context,
                    ((), ()),
                >>::sync(
                    context,
                    ((), ()),
                    ((), ()),
                    anchor(0),
                    (0, 0),
                    tip_rx,
                    SyncEngineConfig {
                        fetch_batch_size: NonZeroU64::new(1).unwrap(),
                        apply_batch_size: NZU64!(1),
                        max_outstanding_requests: NZUsize!(1),
                    },
                )
                .await;

            let err = match result {
                Ok(_) => panic!("tuple state sync should return the database sync error"),
                Err(err) => err,
            };
            assert!(
                err.contains("state sync failed (index 1, db"),
                "error should include failing database index, got: {err}",
            );
            assert!(
                err.contains("FailingStateSyncDb"),
                "error should include failing database type, got: {err}",
            );
        });
    }

    #[test]
    fn coordinator_rejects_stale_reached_event_from_older_generation() {
        let mut state = CoordinatorState::new(2, anchor(0), (0u64, 0u64));

        state.record_tip_update(anchor(1), (1, 1));
        match state.next_action() {
            CoordinatorAction::Dispatch {
                generation,
                targets: (left, right),
            } => {
                assert_eq!(generation, 1, "coordinator should dispatch generation 1");
                assert_eq!((left, right), (1, 1));
            }
            CoordinatorAction::Wait => panic!("coordinator should dispatch the newer tip"),
            CoordinatorAction::Converged { anchor, .. } => {
                panic!("coordinator converged too early at {anchor:?}")
            }
        }

        // This reached event belongs to generation 0 but arrives after the
        // coordinator has already advanced the database to generation 1.
        state.record_reached(1, 0);

        // Only database 0 has actually reached generation 1 so far.
        state.record_reached(0, 1);

        match state.next_action() {
            CoordinatorAction::Wait => {}
            CoordinatorAction::Dispatch { targets, .. } => {
                panic!(
                    "coordinator should wait for a fresh reached event, got dispatch {targets:?}"
                )
            }
            CoordinatorAction::Converged { anchor, .. } => {
                panic!("stale reached event must not allow convergence at {anchor:?}")
            }
        }
    }

    #[test]
    fn coordinator_dispatches_pending_tip_before_converging() {
        let mut state = CoordinatorState::new(2, anchor(0), (0u64, 0u64));

        state.record_tip_update(anchor(1), (1, 1));
        match state.next_action() {
            CoordinatorAction::Dispatch {
                generation,
                targets: (left, right),
            } => {
                assert_eq!(generation, 1, "coordinator should dispatch generation 1");
                assert_eq!((left, right), (1, 1));
            }
            CoordinatorAction::Wait => panic!("coordinator should dispatch the newer tip"),
            CoordinatorAction::Converged { anchor, .. } => {
                panic!("coordinator converged too early at {anchor:?}")
            }
        }

        state.record_reached(0, 1);
        state.record_reached(1, 1);
        state.record_tip_update(anchor(2), (2, 2));

        match state.next_action() {
            CoordinatorAction::Dispatch {
                generation,
                targets: (left, right),
            } => {
                assert_eq!(generation, 2, "coordinator should advance to generation 2");
                assert_eq!((left, right), (2, 2));
            }
            CoordinatorAction::Wait => panic!("coordinator should dispatch the pending tip"),
            CoordinatorAction::Converged { anchor, .. } => {
                panic!("coordinator should not converge with a pending tip: {anchor:?}")
            }
        }
    }

    #[test]
    fn tuple_state_sync_stops_updates_after_reached_until_regroup() {
        deterministic::Runner::default().start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(32).unwrap());
            let (fast_ready_tx, fast_ready) = oneshot::channel();
            let fast_update_count = Arc::new(AtomicUsize::new(0));

            let slow_source = ();
            let fast_source = FastSyncObserver {
                ready: fast_ready_tx,
                update_count: fast_update_count.clone(),
            };
            let sync = context.child("tuple_state_sync_algorithm").spawn(
                move |context| async move {
                    <(
                        Shared<ObservedSlowSyncDb>,
                        Shared<ObservedFastSyncDb>,
                    ) as StateSyncSet<
                        deterministic::Context,
                        ((), FastSyncObserver),
                    >>::sync(
                        context,
                        ((), ()),
                        (slow_source, fast_source),
                        anchor(0),
                        (0, 0),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NonZeroU64::new(1).unwrap(),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                },
            );

            fast_ready.await.expect("the fast database should start");

            for target in 1..=16u64 {
                let _ = tip_tx.send(TipUpdate::new(anchor(target), (target, target))).await;
            }
            drop(tip_tx);

            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            let slow_target = synced.0.read().await.final_target;
            let fast_target = synced.1.read().await.final_target;

            assert_eq!(
                slow_target, fast_target,
                "all databases should finish on the same converged target set"
            );
            assert_eq!(
                converged_anchor.height.get(), slow_target,
                "returned anchor height should match the converged generation"
            );
            assert_eq!(
                fast_update_count.load(Ordering::SeqCst),
                1,
                "a reached database must not receive tip updates before regroup; only regroup retarget should be observed"
            );
        });
    }

    #[test]
    fn tuple_state_sync_allows_noop_database_while_other_catches_up() {
        deterministic::Runner::default().start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let (slow_release, slow_source) = oneshot::channel();
            let (fast_ready_tx, fast_ready) = oneshot::channel();
            let fast_update_count = Arc::new(AtomicUsize::new(0));
            let target = 7u64;

            let sync = context.child("tuple_state_sync_noop").spawn({
                let fast_source = FastSyncObserver {
                    ready: fast_ready_tx,
                    update_count: fast_update_count.clone(),
                };
                move |context| async move {
                    <(Shared<SlowSyncDb>, Shared<ObservedFastSyncDb>) as StateSyncSet<
                        deterministic::Context,
                        (oneshot::Receiver<()>, FastSyncObserver),
                    >>::sync(
                        context,
                        ((), ()),
                        (slow_source, fast_source),
                        anchor(target),
                        (target, target),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NonZeroU64::new(1).unwrap(),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                }
            });

            fast_ready.await.expect("the fast database should start");

            drop(tip_tx);
            slow_release.send(()).unwrap();

            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            let slow_target = synced.0.read().await.final_target;
            let fast_target = synced.1.read().await.final_target;

            assert_eq!(slow_target, target);
            assert_eq!(fast_target, target);
            assert_eq!(converged_anchor, anchor(target));
            assert_eq!(
                fast_update_count.load(Ordering::SeqCst),
                0,
                "already-at-target database should not receive tip updates"
            );
        });
    }

    #[test]
    fn tuple_state_sync_regroup_completes_when_database_target_is_unchanged() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tip_tx, tip_rx) = ring::channel(NonZeroUsize::new(4).unwrap());
            let (slow_release, slow_source) = oneshot::channel();
            let (fast_ready_tx, fast_ready) = oneshot::channel();
            let fast_update_count = Arc::new(AtomicUsize::new(0));

            let sync = context
                .child("tuple_state_sync_regroup_unchanged_target")
                .spawn({
                    let fast_source = FastSyncObserver {
                        ready: fast_ready_tx,
                        update_count: fast_update_count.clone(),
                    };
                    move |context| async move {
                        <(Shared<SlowSyncDb>, Shared<DistinctObservedFastSyncDb>) as StateSyncSet<
                            deterministic::Context,
                            (oneshot::Receiver<()>, FastSyncObserver),
                        >>::sync(
                            context,
                            ((), ()),
                            (slow_source, fast_source),
                            anchor(0),
                            (0, 7),
                            tip_rx,
                            SyncEngineConfig {
                                fetch_batch_size: NonZeroU64::new(1).unwrap(),
                                apply_batch_size: NZU64!(1),
                                max_outstanding_requests: NZUsize!(1),
                            },
                        )
                        .await
                        .expect("tuple state sync should succeed")
                    }
                });

            fast_ready.await.expect("the fast database should start");

            let (update, observed) = TipUpdate::with_observation(anchor(9), (9, 7));
            let _ = tip_tx.send(update).await;
            observed.await.expect("the coordinator records the tip");
            slow_release.send(()).unwrap();
            drop(tip_tx);

            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            let slow_target = synced.0.read().await.final_target;
            let fast_target = synced.1.read().await.final_target;

            assert_eq!(slow_target, 9);
            assert_eq!(fast_target, 7);
            assert_eq!(converged_anchor, anchor(9));
            assert_eq!(
                fast_update_count.load(Ordering::SeqCst),
                0,
                "the unchanged-target database should not receive duplicate target updates",
            );
        });
    }

    #[derive(Default)]
    struct AttachDb1;

    #[derive(Default)]
    struct AttachDb2;

    #[derive(Clone)]
    struct RecordingResolver {
        id: &'static str,
        log: Arc<commonware_utils::sync::Mutex<Vec<&'static str>>>,
    }

    impl RecordingResolver {
        fn new(
            id: &'static str,
            log: Arc<commonware_utils::sync::Mutex<Vec<&'static str>>>,
        ) -> Self {
            Self { id, log }
        }
    }

    impl<DB: Send + Sync + 'static> AttachableResolver<DB> for RecordingResolver {
        async fn attach_database(&self, _db: Shared<DB>) {
            self.log.lock().push(self.id);
        }
    }

    #[test]
    fn single_db_attach_calls_single_resolver() {
        deterministic::Runner::default().start(|_| async move {
            let log = Arc::new(commonware_utils::sync::Mutex::new(Vec::new()));
            let resolver = RecordingResolver::new("db1", log.clone());
            let db = Shared::new("test", AttachDb1);

            resolver.attach_databases(db).await;
            assert_eq!(&*log.lock(), &["db1"]);
        });
    }

    #[test]
    fn tuple_attach_is_index_stable() {
        deterministic::Runner::default().start(|_| async move {
            let log = Arc::new(commonware_utils::sync::Mutex::new(Vec::new()));
            let resolvers = (
                RecordingResolver::new("resolver_0", log.clone()),
                RecordingResolver::new("resolver_1", log.clone()),
            );
            let databases = (
                Shared::new("test", AttachDb1),
                Shared::new("test", AttachDb2),
            );

            resolvers.attach_databases(databases).await;
            assert_eq!(&*log.lock(), &["resolver_0", "resolver_1"]);
        });
    }

    #[test]
    fn heterogeneous_tuple_attach_compiles() {
        deterministic::Runner::default().start(|_| async move {
            let log = Arc::new(commonware_utils::sync::Mutex::new(Vec::new()));
            let resolvers = (
                RecordingResolver::new("db1", log.clone()),
                RecordingResolver::new("db2", log.clone()),
            );
            let databases = (
                Shared::new("test", AttachDb1),
                Shared::new("test", AttachDb2),
            );

            resolvers.attach_databases(databases).await;
            assert_eq!(&*log.lock(), &["db1", "db2"]);
        });
    }

    #[test]
    fn single_state_sync_records_tips_while_database_is_held() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (mut tips, tip_rx) = ring::channel(NZUsize!(4));
            let (release, release_rx) = oneshot::channel();
            let sync = <Shared<SlowSyncDb> as StateSyncSet<_, _>>::sync(
                context.child("sync"),
                (),
                release_rx,
                anchor(0),
                0,
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NZU64!(1),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            );
            pin_mut!(sync);

            // The coordinator records every tip while the database takes no targets. A coordinator
            // that waited for target capacity would stop before recording the third.
            for target in [1, 2, 3] {
                let (update, mut observed) = TipUpdate::with_observation(anchor(target), target);
                tips.send(update).await.unwrap();
                assert!(sync.as_mut().now_or_never().is_none());
                assert_eq!((&mut observed).now_or_never(), Some(Ok(())));
            }

            // Once released, the database takes the newest target and the sync converges there.
            release.send(()).unwrap();
            let (database, converged) = sync.await.unwrap();
            assert_eq!(converged, anchor(3));
            assert_eq!(database.read().await.final_target, 3);
        });
    }

    #[test]
    fn reached_events_keep_the_newest_event_per_database() {
        deterministic::Runner::default().start(|context| async move {
            let (sender, mut receiver) =
                mailbox::new::<ReachedEvent>(context.child("reached_events"), NZUsize!(2));

            // Fill the ready queue, then overflow with events for both databases. Database 0's
            // newer overflow event replaces its older one.
            for (index, generation) in [(0, 0), (1, 0), (0, 1), (1, 1), (0, 2)] {
                assert!(
                    sender
                        .enqueue(ReachedEvent { index, generation })
                        .accepted()
                );
            }
            drop(sender);

            // The ready events arrive first, then each database's newest overflow event.
            let mut received = Vec::new();
            while let Some(ReachedEvent { index, generation }) = receiver.recv().await {
                received.push((index, generation));
            }
            assert_eq!(received, [(0, 0), (1, 0), (1, 1), (0, 2)]);
        });
    }

    /// Gates for [`LaggingReportSyncDb`]: `started` resolves once the database is syncing, and each
    /// message on `report` makes it adopt the newest queued target and report every target it reached.
    struct ReportGates {
        started: oneshot::Sender<()>,
        report: mpsc::Receiver<()>,
    }

    /// Adopts targets and reports them only when told to, so the coordinator can queue several
    /// targets before one is taken and can replace a reported target before the report arrives.
    struct LaggingReportSyncDb {
        final_target: u64,
    }

    impl<E: Send> ManagedDb<E> for LaggingReportSyncDb {
        type Unmerkleized = TestUnmerkleized;
        type Merkleized = TestMerkleized;
        type Error = Infallible;
        type Config = ();
        type SyncTarget = u64;

        fn initial_sync_target() -> Self::SyncTarget {
            unreachable!("LaggingReportSyncDb is only constructed through state sync in tests")
        }

        async fn init(
            _context: E,
            _config: Self::Config,
            _expected: Option<Self::SyncTarget>,
        ) -> Result<Self, InitError<Self::Error, Self::SyncTarget>> {
            unreachable!("LaggingReportSyncDb is only constructed through state sync in tests")
        }

        fn new_batch(_database: BatchContext<'_, Self>) -> Self::Unmerkleized {
            TestUnmerkleized
        }

        fn matches_sync_target(_batch: &Self::Merkleized, _target: &Self::SyncTarget) -> bool {
            true
        }

        ready_apply!();

        fn sync_target(&self) -> Self::SyncTarget {
            self.final_target
        }
    }

    impl<E: Send> StateSyncDb<E, ReportGates> for LaggingReportSyncDb {
        type SyncError = Infallible;

        async fn sync_db(
            _context: E,
            _config: Self::Config,
            gates: ReportGates,
            target: Self::SyncTarget,
            mut tip_updates: ring::Receiver<Self::SyncTarget>,
            finish: Option<mpsc::Receiver<()>>,
            reached_target: Option<ring::Sender<Self::SyncTarget>>,
            _sync_config: SyncEngineConfig,
        ) -> Result<Self, Self::SyncError> {
            let ReportGates {
                started,
                mut report,
            } = gates;
            let mut finish = finish.expect("state sync passes a finish channel");
            let reached_target = reached_target.expect("state sync passes a report channel");
            let mut current = target;

            // Every target reached and not yet reported, in the order reached.
            let mut unreported = vec![target];
            let _ = started.send(());
            loop {
                select! {
                    // Like the engine, a requested finish completes at the current target without
                    // adopting a queued update.
                    _ = finish.recv() => break,
                    trigger = report.recv() => {
                        if trigger.is_none() {
                            break;
                        }
                        while let Ok(update) = tip_updates.try_recv() {
                            if update > current {
                                current = update;
                                unreported.push(update);
                            }
                        }
                        for reached in unreported.drain(..) {
                            let _ = reached_target.send_lossy(reached);
                        }
                    },
                }
            }
            Ok(Self {
                final_target: current,
            })
        }
    }

    /// Starts a single-database sync of [`LaggingReportSyncDb`] and returns once it is running, so
    /// the coordinator is parked before the test sends anything.
    async fn start_lagging_sync(
        context: deterministic::Context,
        tip_rx: ring::Receiver<TipUpdate<sha256::Digest, u64>>,
        report: mpsc::Receiver<()>,
    ) -> Handle<(Shared<LaggingReportSyncDb>, Anchor<sha256::Digest>)> {
        let (started_tx, started_rx) = oneshot::channel();
        let sync = context.child("sync").spawn(move |context| async move {
            <Shared<LaggingReportSyncDb> as StateSyncSet<_, _>>::sync(
                context,
                (),
                ReportGates {
                    started: started_tx,
                    report,
                },
                anchor(0),
                0,
                tip_rx,
                SyncEngineConfig {
                    fetch_batch_size: NZU64!(1),
                    apply_batch_size: NZU64!(1),
                    max_outstanding_requests: NZUsize!(1),
                },
            )
            .await
            .expect("single state sync should succeed")
        });
        started_rx.await.expect("sync should start");
        sync
    }

    /// Sends a burst of tips without yielding, then waits until the coordinator has recorded the
    /// last one, which it dispatches in the same step.
    async fn send_tip_burst(
        tip_tx: &ring::Sender<TipUpdate<sha256::Digest, u64>>,
        heights: RangeInclusive<u64>,
    ) {
        let mut observed = None;
        for height in heights {
            let (update, observed_rx) = TipUpdate::with_observation(anchor(height), height);
            assert!(tip_tx.send_lossy(update));
            observed = Some(observed_rx);
        }
        observed
            .expect("a burst holds at least one tip")
            .await
            .expect("the coordinator records the last tip of a burst");
    }

    #[test]
    fn single_state_sync_converges_with_late_reports_after_evicted_targets() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NZUsize!(2));
            let (report_tx, report_rx) = mpsc::channel(1);
            let sync = start_lagging_sync(context, tip_rx, report_rx).await;

            // Each burst overflows the two-slot tip ring, and the two dispatches overflow the
            // one-slot target ring before the database takes a target.
            send_tip_burst(&tip_tx, 1..=3).await;
            send_tip_burst(&tip_tx, 4..=6).await;

            // A third burst and a report request are queued without yielding, and the fourth burst
            // joins them before the coordinator runs, so it evicts the third. The coordinator reads
            // the report of its current target first but adopts the queued tips before checking
            // for a finish, so the sync continues to the newest target.
            for height in 7..=9u64 {
                assert!(tip_tx.send_lossy(TipUpdate::new(anchor(height), height)));
            }
            report_tx.try_send(()).unwrap();
            send_tip_burst(&tip_tx, 10..=12).await;

            // The database adopts the newest target and reports it, which finishes the sync there.
            report_tx.try_send(()).unwrap();
            drop(tip_tx);
            let (database, converged) = sync.await.expect("sync task should complete");
            assert_eq!(database.read().await.final_target, 12);
            assert_eq!(converged, anchor(12));
        });
    }

    #[test]
    fn single_state_sync_adopts_queued_tip_before_finishing() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NZUsize!(1));
            let (report_tx, report_rx) = mpsc::channel(1);
            let sync = start_lagging_sync(context, tip_rx, report_rx).await;

            // A report of the initial target and a newer tip reach the coordinator in the same
            // step. The report matches the target it last sent, but the queued tip is adopted
            // first, so the sync must not finish.
            let (update, observed) = TipUpdate::with_observation(anchor(1), 1);
            assert!(tip_tx.send_lossy(update));
            drop(tip_tx);
            report_tx.try_send(()).unwrap();
            observed
                .await
                .expect("the coordinator records the queued tip instead of finishing");

            // The database takes the tip's target and reports it, which finishes the sync there.
            report_tx.try_send(()).unwrap();
            let (database, converged) = sync.await.expect("sync task should complete");
            assert_eq!(database.read().await.final_target, 1);
            assert_eq!(converged, anchor(1));
        });
    }

    #[test]
    fn tuple_state_sync_converges_after_evicted_targets() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NZUsize!(4));
            let (slow_release, slow_source) = oneshot::channel();
            let (fast_ready_tx, fast_ready) = oneshot::channel();
            let fast_update_count = Arc::new(AtomicUsize::new(0));

            let sync = context.child("sync").spawn({
                let fast_source = FastSyncObserver {
                    ready: fast_ready_tx,
                    update_count: fast_update_count.clone(),
                };
                move |context| async move {
                    <(Shared<SlowSyncDb>, Shared<ObservedFastSyncDb>) as StateSyncSet<
                        deterministic::Context,
                        (oneshot::Receiver<()>, FastSyncObserver),
                    >>::sync(
                        context,
                        ((), ()),
                        (slow_source, fast_source),
                        anchor(0),
                        (0, 0),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NZU64!(1),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                }
            });

            // The fast database reaches the initial target at once and freezes there.
            fast_ready.await.expect("the fast database should start");

            // Each tip is recorded and dispatched as its own generation while the slow database
            // takes no targets, so its one-slot target ring keeps only the newest target.
            for target in 1..=3u64 {
                let (update, observed) =
                    TipUpdate::with_observation(anchor(target), (target, target));
                assert!(tip_tx.send_lossy(update));
                observed.await.unwrap();
            }
            drop(tip_tx);

            // Once released, the slow database takes only the newest target. The frozen database
            // is regrouped to it once, and the set converges at the newest generation.
            slow_release.send(()).unwrap();
            let (synced, converged_anchor) = sync.await.expect("sync task should complete");
            assert_eq!(synced.0.read().await.final_target, 3);
            assert_eq!(synced.1.read().await.final_target, 3);
            assert_eq!(converged_anchor, anchor(3));
            assert_eq!(
                fast_update_count.load(Ordering::SeqCst),
                1,
                "the reached database is retargeted once, at regroup"
            );
        });
    }

    #[test]
    fn generation_drain_reports_only_the_newest_assignment() {
        deterministic::Runner::default().start(|context| async move {
            let (sender, mut receiver) =
                mailbox::new::<ReachedEvent>(context.child("reached_events"), NZUsize!(2));

            // Two assignments arrive before the forwarder reads them, so the one-slot ring keeps
            // only the newest, whose target the database has already reached.
            let (generation_tx, generation_rx) = ring::channel(NZUsize!(1));
            assert!(generation_tx.send_lossy((1, 1u64)));
            assert!(generation_tx.send_lossy((2, 2u64)));
            let mut generation_rx = Some(generation_rx);
            let mut current_generation = 0;
            let mut current_target = 0u64;
            let mut last_reported_generation = None;
            drain_generation_updates(
                &mut generation_rx,
                &mut current_generation,
                &mut current_target,
                &Some(2u64),
                &mut last_reported_generation,
                &sender,
                7,
            );

            // The skipped assignment is never reported; the newest one is reported once.
            assert_eq!((current_generation, current_target), (2, 2));
            assert_eq!(last_reported_generation, Some(2));
            let ReachedEvent { index, generation } = receiver.recv().await.unwrap();
            assert_eq!((index, generation), (7, 2));
            assert!(matches!(
                receiver.try_recv(),
                Err(std::sync::mpsc::TryRecvError::Empty)
            ));
        });
    }

    #[test]
    fn tuple_state_sync_ignores_stale_reached_after_forwarded_tip() {
        deterministic::Runner::timed(Duration::from_secs(5)).start(|context| async move {
            let (tip_tx, tip_rx) = ring::channel(NZUsize!(4));
            let (fast_ready_tx, fast_ready) = oneshot::channel();
            let fast_update_count = Arc::new(AtomicUsize::new(0));

            let sync = context.child("sync").spawn({
                let fast_source = FastSyncObserver {
                    ready: fast_ready_tx,
                    update_count: fast_update_count.clone(),
                };
                move |context| async move {
                    <(Shared<StaleReachedSyncDb>, Shared<ObservedFastSyncDb>) as StateSyncSet<
                        deterministic::Context,
                        ((), FastSyncObserver),
                    >>::sync(
                        context,
                        ((), ()),
                        ((), fast_source),
                        anchor(0),
                        (0, 0),
                        tip_rx,
                        SyncEngineConfig {
                            fetch_batch_size: NZU64!(1),
                            apply_batch_size: NZU64!(1),
                            max_outstanding_requests: NZUsize!(1),
                        },
                    )
                    .await
                    .expect("tuple state sync should succeed")
                }
            });

            // The stale database reports the initial target only after it takes the forwarded
            // tip, then holds at the initial target longer than a regroup takes. A forwarder that
            // accepted the stale report would converge the set and finish it at mismatched
            // targets.
            fast_ready.await.expect("the fast database should start");
            assert!(tip_tx.send_lossy(TipUpdate::new(anchor(1), (1, 1))));
            drop(tip_tx);

            let (synced, converged) = sync.await.expect("sync task should complete");
            assert_eq!(synced.0.read().await.final_target, 1);
            assert_eq!(synced.1.read().await.final_target, 1);
            assert_eq!(converged, anchor(1));
            assert_eq!(
                fast_update_count.load(Ordering::SeqCst),
                1,
                "the reached database is retargeted once, at regroup"
            );
        });
    }
}