light_client/

state.rs

//! Client-side state used to implement light client fetching and verification.

use std::{
    collections::{BTreeMap, HashMap},
    future::Future,
    ops::Range,
    sync::Arc,
};

use anyhow::{Context, Result, bail, ensure};
use async_lock::RwLock;
use committable::Committable;
use espresso_types::{
    Certificate2, ChainId, DECAF_CHAIN_ID, DrbAndHeaderUpgradeVersion, Header, Leaf2, NamespaceId,
    PubKey, SeqTypes, StakeTableState, Transaction, ValidatorSet,
};
use futures::future::try_join;
use hotshot_query_service_types::{
    HeightIndexed,
    availability::{BlockQueryData, LeafId, LeafQueryData, PayloadQueryData, VidCommonQueryData},
    node::BlockId,
};
use hotshot_types::{data::EpochNumber, stake_table::StakeTableEntry, utils::root_block_in_epoch};
use serde::{Deserialize, Serialize};
use vbs::version::{StaticVersionType, Version};

use crate::{
    client::Client,
    consensus::{
        leaf::LeafProofHint,
        quorum::{Quorum, StakeTable, StakeTablePair, StakeTableQuorum},
    },
    storage::{LeafRequest, Storage},
};

/// Initial state for a [`LightClient`].
///
/// This [`Genesis`] forms the root of trust for a light client. It defines the initial stake table
/// which is used to verify blocks before the first epoch, which in turn allows it to verify
/// transitions to subsequent stake tables. Thus, this genesis must be configured correctly (i.e.
/// matching the genesis state of honest HotShot nodes) or else the light client may not operate
/// correctly.
#[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
#[cfg_attr(
    feature = "rlp",
    derive(alloy_rlp::RlpEncodable, alloy_rlp::RlpDecodable)
)]
pub struct Genesis {
    /// The number of blocks in an epoch.
    pub epoch_height: u64,

    /// The first epoch where the stake table came from the contract, rather than the genesis stake
    /// table.
    pub first_epoch_with_dynamic_stake_table: EpochNumber,

    /// The fixed stake table used before epochs begin.
    pub stake_table: Vec<StakeTableEntry<PubKey>>,

    /// Genesis chain id, used to derive Decaf-specific behavior.
    pub chain_id: ChainId,
}

/// Maximum distance between a requested leaf and the `finalized` hint sent to the server. Servers
/// reject or ignore distant hints; beyond this we omit the hint and verify against a quorum.
const MAX_FINALIZED_HINT_DISTANCE: u64 = 500;

/// Whether a range fetch writes the leaves it chains to a proven leaf into the cache.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum CacheLeaves {
    Yes,
    No,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
#[cfg_attr(feature = "clap", derive(clap::Parser))]
pub struct LightClientOptions {
    /// Maximum number of stake tables to cache in memory at any given time.
    ///
    /// The light client database keeps every stake table, so one evicted from memory is reloaded
    /// from disk rather than rebuilt from peers.
    #[cfg_attr(
        feature = "clap",
        clap(
            long = "light-client-num-stake-tables",
            env = "LIGHT_CLIENT_NUM_STAKE_TABLES",
            default_value = "100",
        )
    )]
    pub num_stake_tables_in_memory: usize,
}

impl Default for LightClientOptions {
    fn default() -> Self {
        Self {
            num_stake_tables_in_memory: 100,
        }
    }
}

/// Client-side state required to implement the light client interface.
///
/// A [`LightClient`] can always be created [from scratch](Self::from_genesis), with no state, since
/// ultimately all data is fetched and verified from external query nodes. However, having some
/// persistent state can make it more efficient to use a [`LightClient`] over a long period of time,
/// as important artifacts can be cached locally, avoiding the need to frequently re-fetch and
/// verify them.
#[derive(Debug)]
pub struct LightClient<P, S> {
    db: P,
    server: S,
    opt: LightClientOptions,

    chain_id: ChainId,
    epoch_height: u64,
    first_epoch_with_dynamic_stake_table: EpochNumber,
    genesis_stake_table: Arc<StakeTable>,

    // We cache stake tables in memory since they are large and expensive to load from the database.
    stake_tables: RwLock<BTreeMap<EpochNumber, Arc<StakeTable>>>,
}

impl<P, S> LightClient<P, S> {
    pub fn into_storage(self) -> P {
        self.db
    }

    pub fn storage(&self) -> &P {
        &self.db
    }

    pub fn into_inner(self) -> S {
        self.server
    }

    pub fn inner(&self) -> &S {
        &self.server
    }
}

impl<P, S> LightClient<P, S>
where
    P: Storage,
    S: Client,
{
    /// Create a light client from scratch, with no state.
    ///
    /// State will automatically be populated as queries are made. The provided genesis becomes the
    /// root of trust for verifying all state that is subsequently loaded by the light client. If
    /// the genesis is not correct (i.e. matching the genesis used by honest HotShot nodes) the
    /// light client may verify incorrect data, or fail to verify correct data.
    pub fn from_genesis(db: P, server: S, genesis: Genesis) -> Self {
        Self::from_genesis_with_options(db, server, genesis, Default::default())
    }

    /// Create a light client from scratch, with no state, using the given options.
    ///
    /// State will automatically be populated as queries are made. The provided genesis becomes the
    /// root of trust for verifying all state that is subsequently loaded by the light client. If
    /// the genesis is not correct (i.e. matching the genesis used by honest HotShot nodes) the
    /// light client may verify incorrect data, or fail to verify correct data.
    pub fn from_genesis_with_options(
        db: P,
        server: S,
        genesis: Genesis,
        opt: LightClientOptions,
    ) -> Self {
        Self {
            db,
            server,
            opt,
            chain_id: genesis.chain_id,
            epoch_height: genesis.epoch_height,
            genesis_stake_table: Arc::new(genesis.stake_table.into()),
            first_epoch_with_dynamic_stake_table: genesis.first_epoch_with_dynamic_stake_table,
            stake_tables: Default::default(),
        }
    }

    pub fn genesis(&self) -> Genesis {
        Genesis {
            epoch_height: self.epoch_height,
            first_epoch_with_dynamic_stake_table: self.first_epoch_with_dynamic_stake_table,
            stake_table: (*self.genesis_stake_table).clone().into(),
            chain_id: self.chain_id,
        }
    }

    /// Get the number of known blocks in the chain.
    ///
    /// This is equivalent to one more than the block number of the latest known block. The latest
    /// known block may come either from the local light client database or from an untrusted query
    /// service (in which case the corresponding leaf is fetched and verified to ensure there is in
    /// fact such a block). Note however that it is always possible that neither this light client
    /// nor the connected server is aware of the true latest block, and so in rare cases the result
    /// of this method may be an underestimate.
    pub async fn block_height(&self) -> Result<u64> {
        let latest_known = self.db.block_height().await?;
        let latest_from_server = self.server.block_height().await?;
        if latest_from_server > latest_known {
            // The server claims there is a newer block than we previously knew about. Verify that
            // this is the case by requesting a finality proof for the corresponding leaf.
            if let Err(err) = self
                .fetch_leaf(LeafId::Number(latest_from_server as usize - 1))
                .await
            {
                tracing::warn!(
                    latest_known,
                    latest_from_server,
                    "failed to verify block height claimed by server: {err:#}"
                );
                return Ok(latest_known);
            } else {
                return Ok(latest_from_server);
            }
        }
        Ok(latest_known)
    }

    /// Fetch and verify the requested leaf.
    pub async fn fetch_leaf(&self, id: LeafId<SeqTypes>) -> Result<LeafQueryData<SeqTypes>> {
        self.fetch_leaf_with_quorum(id, |epoch| {
            StakeTableQuorum::new((epoch, self), self.epoch_height)
        })
        .await
    }

    async fn fetch_leaf_with_quorum<Q>(
        &self,
        id: LeafId<SeqTypes>,
        quorum: impl Send + FnOnce(EpochNumber) -> Q,
    ) -> Result<LeafQueryData<SeqTypes>>
    where
        Q: Send + Quorum,
    {
        let upper_bound = self.db.leaf_upper_bound(id).await?;
        let known_finalized = if let Some(upper_bound) = upper_bound {
            if leaf_matches_id(&upper_bound, id) {
                return Ok(upper_bound);
            }

            Some(upper_bound)
        } else {
            None
        };
        let known_finalized = known_finalized.filter(|anchor| match id {
            LeafId::Number(n) => {
                anchor.height().saturating_sub(n as u64) <= MAX_FINALIZED_HINT_DISTANCE
            },
            // Hash lookups only ever return the requested leaf itself, handled above.
            LeafId::Hash(_) => true,
        });
        let known_finalized = known_finalized.as_ref().map(LeafQueryData::leaf);
        self.fetch_leaf_from_server(id, known_finalized, quorum)
            .await
    }

    /// Fetches leaves in range [start_height, end_height)
    pub async fn fetch_leaves_in_range(
        &self,
        start_height: usize,
        end_height: usize,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.fetch_leaves_in_range_with(start_height, end_height, CacheLeaves::Yes)
            .await
    }

    /// [`fetch_leaves_in_range`](Self::fetch_leaves_in_range) for a caller that needs each leaf
    /// once, such as the state loop replaying history.
    ///
    /// Only the proven last leaf goes into the cache. Writing the others costs one transaction per
    /// leaf, more than fetching the whole range, and a range longer than the cache would evict
    /// every leaf other callers keep there as anchors.
    pub async fn fetch_leaves_in_range_uncached(
        &self,
        start_height: usize,
        end_height: usize,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        self.fetch_leaves_in_range_with(start_height, end_height, CacheLeaves::No)
            .await
    }

    async fn fetch_leaves_in_range_with(
        &self,
        start_height: usize,
        end_height: usize,
        cache: CacheLeaves,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        ensure!(
            start_height < end_height,
            "invalid range: start must be < end"
        );

        // first try to fetch all the leaves from the local database
        let leaves = self
            .db
            .get_leaves_in_range(start_height as u32, end_height as u32)
            .await?;

        if leaves.len() == end_height - start_height {
            // we have all the leaves in the range
            return Ok(leaves);
        }

        // Fetch the last leaf in the range as our known finalized anchor point
        let known_end_leaf = self.fetch_leaf(LeafId::Number(end_height - 1)).await?;
        if end_height - start_height == 1 {
            return Ok(vec![known_end_leaf]);
        }

        // at this point, we know the end leaf is valid and is finalized
        // now we need to fetch all leaves from start to end - 1 from the server
        let leaves = self
            .fetch_leaves_in_range_from_server(start_height, end_height - 1, &known_end_leaf, cache)
            .await
            // add the known end leaf to the result
            .map(|mut leaves| {
                leaves.push(known_end_leaf);
                leaves
            })?;
        Ok(leaves)
    }

    /// Fetches headers in range [start_height, end_height)
    pub async fn fetch_headers_in_range(
        &self,
        start_height: usize,
        end_height: usize,
    ) -> Result<Vec<Header>> {
        ensure!(
            start_height < end_height,
            "invalid range: start must be < end"
        );

        // Reuse the verified leaf path to guarantee header correctness.
        let leaves = self.fetch_leaves_in_range(start_height, end_height).await?;
        Ok(leaves
            .into_iter()
            .map(|leaf| leaf.header().clone())
            .collect())
    }

    /// Fetch and verify the [`Certificate2`] at the given block height.
    ///
    /// Returns `Ok(None)` if the server reports that no cert2 exists at this height
    pub async fn fetch_certificate2(&self, height: u64) -> Result<Option<Certificate2<SeqTypes>>> {
        let Some(cert2) = self.server.cert2(height).await? else {
            return Ok(None);
        };
        let header = self
            .fetch_header(BlockId::Number(height as usize))
            .await
            .context("fetching header to determine cert2 version")?;
        self.verify_certificate2(cert2, &header).await.map(Some)
    }

    /// [`fetch_certificate2`](Self::fetch_certificate2) for a caller that already holds the
    /// verified header of the block, which saves fetching it again.
    pub async fn fetch_certificate2_for_header(
        &self,
        header: &Header,
    ) -> Result<Option<Certificate2<SeqTypes>>> {
        let Some(cert2) = self.server.cert2(header.height()).await? else {
            return Ok(None);
        };
        self.verify_certificate2(cert2, header).await.map(Some)
    }

    async fn verify_certificate2(
        &self,
        cert2: Certificate2<SeqTypes>,
        header: &Header,
    ) -> Result<Certificate2<SeqTypes>> {
        ensure!(
            cert2.data.block_number == header.height(),
            "cert2 block number {} does not match requested height {}",
            cert2.data.block_number,
            header.height(),
        );
        let quorum = StakeTableQuorum::new((cert2.data.epoch, self), self.epoch_height);
        quorum
            .verify_cert2(&cert2, header.version())
            .await
            .context("verifying cert2 signature")?;
        Ok(cert2)
    }

    /// Fetches leaves from the server in range [start_height, end_height) and verifies them by
    /// walking backwards from the known finalized leaf, ensuring each leaf's hash matches the
    /// parent commitment of the subsequent leaf.
    async fn fetch_leaves_in_range_from_server(
        &self,
        start_height: usize,
        end_height: usize,
        known_finalized: &LeafQueryData<SeqTypes>,
        cache: CacheLeaves,
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        // we will fetch all the leaves till the known finalized leaf
        let leaves = self
            .server
            // `get_leaves_in_range` is exclusive of the end height
            // which we dont need because we already know the end leaf
            .get_leaves_in_range(start_height, end_height)
            .await?;

        ensure!(
            leaves.len() == end_height.saturating_sub(start_height),
            "server returned {} leaves for range [{}, {})",
            leaves.len(),
            start_height,
            end_height
        );

        self.verify_leaf_chain(&leaves, known_finalized, cache)
            .await?;
        Ok(leaves)
    }

    /// Verify unproven leaves by chaining them to a leaf already known to be finalized.
    ///
    /// `leaves` must be consecutive and end just below `known_finalized`, which the caller has
    /// verified. Each leaf is then pinned by the parent commitment of the one above it, so the
    /// whole run inherits the anchor's finality without a proof of its own.
    async fn verify_leaf_chain(
        &self,
        leaves: &[LeafQueryData<SeqTypes>],
        known_finalized: &LeafQueryData<SeqTypes>,
        cache: CacheLeaves,
    ) -> Result<()> {
        let mut expected_parent = known_finalized.leaf().parent_commitment();
        for leaf in leaves.iter().rev() {
            let leaf_hash = leaf.hash();
            ensure!(
                leaf_hash == expected_parent,
                "leaf hash mismatch: expected parent hash {:?}, got leaf hash {:?}",
                expected_parent,
                leaf_hash
            );
            expected_parent = leaf.leaf().parent_commitment();
        }

        if cache == CacheLeaves::No {
            return Ok(());
        }
        // Cache the verified leaves, but do not fail the fetch if caching does.
        for leaf in leaves {
            if let Err(err) = self.db.insert_leaf(leaf.clone()).await {
                tracing::warn!(
                    "failed to cache leaf at height {}: {:#?}",
                    leaf.height(),
                    err
                )
            }
        }

        Ok(())
    }

    /// Fetch and verify the leaves in a set of height ranges.
    ///
    /// The bulk of the leaves come from one ranges request, which is what a fragmented set of
    /// heights is worth batching for. Verification is still per contiguous run: each run needs one
    /// leaf whose finality is proven, and the rest of the run chains to it. A run of a single
    /// height is therefore just a proof fetch, with nothing left for the ranges request to carry.
    pub async fn fetch_leaves_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> Result<Vec<LeafQueryData<SeqTypes>>> {
        ensure!(
            ranges.iter().all(|range| range.start < range.end)
                && ranges.is_sorted_by(|a, b| a.end <= b.start),
            "each range must be non-empty, and the ranges ascending and disjoint"
        );

        // Fetching the whole span needs one finality proof, where fetching the runs separately
        // needs one each, and a proof costs about what a hundred leaves cost. So take the span
        // when the gaps are small: at most a few leaves wasted per leaf wanted, and no wider than
        // the leaf range endpoint serves in one request.
        if let (Some(lo), Some(hi)) = (
            ranges.iter().map(|range| range.start).min(),
            ranges.iter().map(|range| range.end).max(),
        ) {
            let wanted: u64 = ranges.iter().map(|range| range.end - range.start).sum();
            let span = hi - lo;
            // hotshot-query-service's default small-object range limit, which the zkVM build
            // cannot import; test_fetch_leaves_for_ranges pins the two together. Should they
            // drift anyway, the server rejects the span and the fetch falls back to per-run.
            let limit = 500;
            if span <= limit && span <= wanted * 4 {
                match self.fetch_leaves_in_range(lo as usize, hi as usize).await {
                    Ok(leaves) => {
                        return Ok(leaves
                            .into_iter()
                            .filter(|leaf| {
                                ranges.iter().any(|range| range.contains(&leaf.height()))
                            })
                            .collect());
                    },
                    // The span is an optimization over the runs, not the only way to get them,
                    // so a failed span costs the shortcut, never the call: the per-run path below
                    // fetches and verifies the same heights independently.
                    Err(err) => {
                        tracing::info!(
                            err = %format_args!("{err:#}"),
                            lo,
                            hi,
                            "span fetch failed, fetching each run instead"
                        );
                    },
                }
            }
        }

        // Only runs with something below their anchor have bulk to fetch; the anchors themselves
        // are fetched with a proof.
        let bulk = ranges
            .iter()
            .filter(|range| range.end - range.start > 1)
            .map(|range| range.start..range.end - 1)
            .collect::<Vec<_>>();
        let mut fetched: HashMap<u64, LeafQueryData<SeqTypes>> = HashMap::new();
        if !bulk.is_empty() {
            match self.server.get_leaves_for_ranges(&bulk).await {
                Ok(leaves) => {
                    for leaf in leaves {
                        fetched.insert(leaf.height(), leaf);
                    }
                },
                Err(err) => {
                    // A peer that predates the ranges endpoint answers it with an error, so
                    // propagating here would wedge catchup until an upstream is upgraded. The
                    // range fetch uses only endpoints every release serves.
                    tracing::info!(
                        err = %format_args!("{err:#}"),
                        "leaf ranges fetch failed, fetching each range instead"
                    );
                    let mut leaves = vec![];
                    for range in ranges {
                        leaves.extend(
                            self.fetch_leaves_in_range(range.start as usize, range.end as usize)
                                .await?,
                        );
                    }
                    return Ok(leaves);
                },
            }
        }

        let mut leaves = vec![];
        for range in ranges {
            // The anchor is proven finalized on its own, and pins the rest of the run.
            let anchor = self
                .fetch_leaf(LeafId::Number(range.end as usize - 1))
                .await?;

            let mut run = vec![];
            for height in range.start..range.end - 1 {
                let leaf = match fetched.remove(&height) {
                    Some(leaf) => leaf,
                    None => self.fetch_leaf(LeafId::Number(height as usize)).await?,
                };
                run.push(leaf);
            }
            self.verify_leaf_chain(&run, &anchor, CacheLeaves::Yes)
                .await?;
            leaves.extend(run);
            leaves.push(anchor);
        }

        Ok(leaves)
    }

    /// Fetch and verify the requested header.
    pub async fn fetch_header(&self, id: BlockId<SeqTypes>) -> Result<Header> {
        self.fetch_header_with_quorum(id, |epoch| {
            StakeTableQuorum::new((epoch, self), self.epoch_height)
        })
        .await
    }

    async fn fetch_header_with_quorum<Q>(
        &self,
        id: BlockId<SeqTypes>,
        quorum: impl Send + FnOnce(EpochNumber) -> Q,
    ) -> Result<Header>
    where
        Q: Send + Quorum,
    {
        let Some(upper_bound) = self.db.leaf_upper_bound(id).await? else {
            // We have neither the requested header nor an upper bound for it. All we can do is fetch
            // the corresponding leaf from the server (verifying a leaf proof) and then extract the
            // header from there.
            let leaf = self.fetch_leaf_from_server(id, None, quorum).await?;
            return Ok(leaf.header().clone());
        };

        // If we have the leaf for the requested header in our database already, we can just
        // extract the header.
        if leaf_matches_id(&upper_bound, id) {
            return Ok(upper_bound.header().clone());
        }

        // Otherwise, if we have a leaf that is known to be greater than the requested
        // header, we can ask the server for an inclusion proof for the requested header
        // relative to the Merkle root in the upper bound leaf.
        let proof = match self.server.header_proof(upper_bound.height(), id).await {
            Ok(proof) => proof,
            // Archive nodes prune old merklized state, which a leaf proof doesn't need.
            Err(err) => {
                tracing::info!(
                    %id,
                    root = upper_bound.height(),
                    err = %format_args!("{err:#}"),
                    "header proof fetch failed, fetching a leaf proof instead"
                );
                let known_finalized = match id {
                    BlockId::Number(n) => {
                        upper_bound.height().saturating_sub(n as u64) <= MAX_FINALIZED_HINT_DISTANCE
                    },
                    // Hash lookups only ever return the requested header itself, handled above.
                    BlockId::Hash(_) | BlockId::PayloadHash(_) => true,
                }
                .then(|| upper_bound.leaf());
                let leaf = self
                    .fetch_leaf_from_server(id, known_finalized, quorum)
                    .await
                    .with_context(|| format!("header proof also failed: {err:#}"))?;
                return Ok(leaf.header().clone());
            },
        };
        let header = proof
            .verify(upper_bound.header().block_merkle_tree_root())
            .context("invalid header proof")?;

        // The server has given us a header and correctly proved it finalized, but we still
        // need to verify that it actually gave us the header we requested.
        ensure!(
            header_matches_id(&header, id),
            "server returned a valid header proof for the wrong header (requested header {id}, \
             got header {} with hash {})",
            header.height(),
            header.commit(),
        );

        Ok(header)
    }

    /// Fetch and verify the requested payload.
    pub async fn fetch_payload(&self, id: BlockId<SeqTypes>) -> Result<PayloadQueryData<SeqTypes>> {
        Ok(self.fetch_block(id).await?.into())
    }

    /// Fetch and verify the full payload with the requested header.
    pub async fn fetch_payload_for_header(
        &self,
        header: Header,
    ) -> Result<PayloadQueryData<SeqTypes>> {
        Ok(self.fetch_block_for_header(header).await?.into())
    }

    /// Fetch and verify the requested block.
    pub async fn fetch_block(&self, id: BlockId<SeqTypes>) -> Result<BlockQueryData<SeqTypes>> {
        Ok(self.fetch_block_and_vid_common(id).await?.0)
    }

    /// Fetch and verify the full block with the requested header.
    pub async fn fetch_block_for_header(&self, header: Header) -> Result<BlockQueryData<SeqTypes>> {
        Ok(self.fetch_block_and_vid_common_for_header(header).await?.0)
    }

    /// Fetch and verify the VID common data for the requested block.
    pub async fn fetch_vid_common(
        &self,
        id: BlockId<SeqTypes>,
    ) -> Result<VidCommonQueryData<SeqTypes>> {
        Ok(self.fetch_block_and_vid_common(id).await?.1)
    }

    /// Fetch and verify the VID common data for the requested header.
    pub async fn fetch_vid_common_for_header(
        &self,
        header: Header,
    ) -> Result<VidCommonQueryData<SeqTypes>> {
        Ok(self.fetch_block_and_vid_common_for_header(header).await?.1)
    }

    /// Fetch and verify the requested payload and the associated VID common data.
    pub async fn fetch_block_and_vid_common(
        &self,
        id: BlockId<SeqTypes>,
    ) -> Result<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>)> {
        let header = self.fetch_header(id).await?;
        self.fetch_block_and_vid_common_for_header(header).await
    }

    /// Fetch and verify the payload corresponding to `header`, and the associated VID common data.
    pub async fn fetch_block_and_vid_common_for_header(
        &self,
        header: Header,
    ) -> Result<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>)> {
        let proof = self.server.payload_proof(header.height()).await?;
        let (payload, vid_common) = proof.verify_with_vid_common(&header)?;
        Ok((
            BlockQueryData::new(header.clone(), payload),
            VidCommonQueryData::new(header, vid_common),
        ))
    }

    /// Fetch and verify the blocks in the height range `[start, end)`.
    pub async fn fetch_blocks_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> Result<Vec<BlockQueryData<SeqTypes>>> {
        let res = self
            .fetch_blocks_and_vid_common_in_range(start, end)
            .await?;
        Ok(res.into_iter().map(|(block, _)| block).collect())
    }

    /// Fetch and verify the VID common data for blocks in the height range `[start, end)`.
    pub async fn fetch_vid_common_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> Result<Vec<VidCommonQueryData<SeqTypes>>> {
        let res = self
            .fetch_blocks_and_vid_common_in_range(start, end)
            .await?;
        Ok(res.into_iter().map(|(_, vid)| vid).collect())
    }

    /// Fetch and verify the blocks and VID common data in the height range `[start, end)`.
    pub async fn fetch_blocks_and_vid_common_in_range(
        &self,
        start: usize,
        end: usize,
    ) -> Result<Vec<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>)>> {
        let (headers, proofs) = try_join(
            self.fetch_headers_in_range(start, end),
            self.server
                .payload_proofs_in_range(start as u64, end as u64),
        )
        .await?;
        ensure!(
            headers.len() == proofs.len(),
            "server returned wrong number of payload proofs for range {start}..{end} ({})",
            proofs.len(),
        );
        headers
            .into_iter()
            .zip(proofs)
            .map(|(header, proof)| {
                let (payload, vid_common) = proof.verify_with_vid_common(&header)?;
                Ok((
                    BlockQueryData::new(header.clone(), payload),
                    VidCommonQueryData::new(header, vid_common),
                ))
            })
            .collect()
    }

    /// Fetch and verify the blocks and VID common data for a set of height ranges.
    ///
    /// One request for the proofs covering every range, rather than one per range. Each proof is
    /// still verified against its own header, so batching changes only how they arrive.
    pub async fn fetch_blocks_and_vid_common_for_ranges(
        &self,
        ranges: &[Range<u64>],
    ) -> Result<Vec<(BlockQueryData<SeqTypes>, VidCommonQueryData<SeqTypes>)>> {
        let proofs = async {
            match self.server.payload_proofs_for_ranges(ranges).await {
                Ok(proofs) => Ok(proofs),
                Err(err) => {
                    // Same fallback as the leaf ranges request, on the range endpoint every release serves.
                    tracing::info!(
                        err = %format_args!("{err:#}"),
                        "payload proof ranges fetch failed, fetching each range instead"
                    );
                    let mut proofs = vec![];
                    for range in ranges {
                        proofs.extend(
                            self.server
                                .payload_proofs_in_range(range.start, range.end)
                                .await?,
                        );
                    }
                    Ok(proofs)
                },
            }
        };
        let (leaves, proofs) = try_join(self.fetch_leaves_for_ranges(ranges), proofs).await?;
        ensure!(
            leaves.len() == proofs.len(),
            "server returned {} payload proofs for {} heights",
            proofs.len(),
            leaves.len(),
        );

        leaves
            .into_iter()
            .zip(proofs)
            .map(|(leaf, proof)| {
                let header = leaf.header().clone();
                let (payload, vid_common) = proof.verify_with_vid_common(&header)?;
                Ok((
                    BlockQueryData::new(header.clone(), payload),
                    VidCommonQueryData::new(header, vid_common),
                ))
            })
            .collect()
    }

    /// Fetch and verify the transactions in the given namespace of the requested block.
    pub async fn fetch_namespace(
        &self,
        id: BlockId<SeqTypes>,
        namespace: NamespaceId,
    ) -> Result<Vec<Transaction>> {
        let header = self.fetch_header(id).await?;
        self.fetch_namespace_for_header(&header, namespace).await
    }

    /// Fetch and verify the transactions in the given namespace of the requested header.
    pub async fn fetch_namespace_for_header(
        &self,
        header: &Header,
        namespace: NamespaceId,
    ) -> Result<Vec<Transaction>> {
        let proof = self
            .server
            .namespace_proof(header.height(), namespace)
            .await?;
        proof.verify(header, namespace)
    }

    /// Fetch and verify the transactions in the given namespace of blocks in the range
    /// `[start_height, end_height)`.
    pub async fn fetch_namespaces_in_range(
        &self,
        start_height: usize,
        end_height: usize,
        namespace: NamespaceId,
    ) -> Result<Vec<Vec<Transaction>>> {
        let headers = self
            .fetch_headers_in_range(start_height, end_height)
            .await?;
        let proofs = self
            .server
            .namespace_proofs_in_range(start_height as u64, end_height as u64, namespace)
            .await?;
        ensure!(
            proofs.len() == headers.len(),
            "server returned wrong number of namespace proofs (expected {}, got {})",
            headers.len(),
            proofs.len()
        );
        proofs
            .into_iter()
            .zip(&headers)
            .map(|(proof, header)| proof.verify(header, namespace))
            .collect()
    }

    /// Fetch and verify the transactions in the given namespaces of blocks in the range
    /// `[start_height, end_height)`.
    ///
    /// For each block, the result maps a namespace ID to its verified transactions, including only
    /// those requested namespaces that are actually present in that block.
    pub async fn fetch_all_namespaces_in_range(
        &self,
        start_height: usize,
        end_height: usize,
        namespaces: &[NamespaceId],
    ) -> Result<Vec<HashMap<NamespaceId, Vec<Transaction>>>> {
        let headers = self
            .fetch_headers_in_range(start_height, end_height)
            .await?;
        let proofs = self
            .server
            .namespaces_proofs_in_range(start_height as u64, end_height as u64, namespaces)
            .await?;
        ensure!(
            proofs.len() == headers.len(),
            "server returned wrong number of namespace proofs (expected {}, got {})",
            headers.len(),
            proofs.len()
        );
        proofs
            .into_iter()
            .zip(&headers)
            .map(|(proofs, header)| {
                proofs
                    .into_iter()
                    .map(|(namespace, proof)| Ok((namespace, proof.verify(header, namespace)?)))
                    .collect()
            })
            .collect()
    }

    /// Fetch and verify the stake table for the requested epoch.
    pub async fn quorum_for_epoch(&self, epoch: EpochNumber) -> Result<Arc<StakeTable>> {
        if epoch < self.first_epoch_with_dynamic_stake_table {
            return Ok(self.genesis_stake_table.clone());
        }

        // Check cache for the desired stake table.
        {
            let cache = self.stake_tables.read().await;
            if let Some(stake_table) = cache.get(&epoch) {
                tracing::debug!(%epoch, "found stake table in cache");
                return Ok(stake_table.clone());
            }
        }

        // Look up an earlier stake table we can catch up from, in cache or DB.
        //
        // `epoch_root_protocol_version` is the version of the snapshot root header (root in epoch
        // `k-2` for the epoch `k` we are computing). On DB load we seed it from the stored
        // `next_snapshot_version` of `lower_bound`, which is the snapshot for `lower_bound + 1`
        // (= iter 1's target). On genesis we fetch the root at `first_dynamic - 2` (signed by
        // the genesis stake table) and use its version.
        let (lower_bound, mut stake_table, mut prev_quorum, mut epoch_root_protocol_version) =
            if let Some((lower_bound, stake_table, stored_version, next_stored_version)) =
                self.db.stake_table_lower_bound(epoch).await?
            {
                if lower_bound == epoch {
                    tracing::debug!(%epoch, "found stake table in database");
                    let quorum = stake_table_state_to_quorum(&stake_table, stored_version)?;
                    return Ok(self.cache_stake_table(epoch, Arc::new(quorum)).await);
                }

                let quorum = stake_table_state_to_quorum(&stake_table, stored_version)?;
                (
                    lower_bound,
                    stake_table,
                    Arc::new(quorum),
                    next_stored_version,
                )
            } else {
                // We don't have any stake table earlier than `epoch` as a starting point, so we must
                // start from the genesis state.
                let bootstrap_epoch = (*self.first_epoch_with_dynamic_stake_table)
                    .checked_sub(2)
                    .context(
                        "first_epoch_with_dynamic_stake_table must be >= 2 for genesis catchup",
                    )?;
                let bootstrap_height = root_block_in_epoch(bootstrap_epoch, self.epoch_height);
                let bootstrap_quorum = (
                    self.genesis_stake_table.clone(),
                    self.genesis_stake_table.clone(),
                );
                let bootstrap_root = self
                    .fetch_header_with_quorum(
                        BlockId::Number(bootstrap_height as usize),
                        move |_epoch| StakeTableQuorum::new(bootstrap_quorum, self.epoch_height),
                    )
                    .await
                    .context("fetching snapshot root header for genesis catchup")?;
                (
                    self.first_epoch_with_dynamic_stake_table - 1,
                    StakeTableState::default(),
                    self.genesis_stake_table.clone(),
                    bootstrap_root.version(),
                )
            };
        tracing::info!(from = %lower_bound, to = %epoch, "performing stake table catchup");

        // Replay one epoch at a time from the lower bound stake table to the requested epoch.
        for epoch in *lower_bound + 1..=*epoch {
            let events = self
                .server
                .stake_table_events(EpochNumber::new(epoch))
                .await?;
            let num_events = events.len();
            tracing::debug!(epoch, num_events, "reconstruct stake table");
            for event in events {
                tracing::debug!(epoch, ?event, "replay event");
                if let Err(err) = stake_table.apply_event(event).context("applying event")? {
                    tracing::warn!("allowed error in event: {err:#}");
                }
            }

            let next_quorum = Arc::new(stake_table_state_to_quorum(
                &stake_table,
                epoch_root_protocol_version,
            )?);

            let root_height = root_block_in_epoch(epoch - 1, self.epoch_height);
            let root = {
                let prev_quorum = prev_quorum.clone();
                let next_quorum = next_quorum.clone();
                self.fetch_header_with_quorum(
                    BlockId::Number(root_height as usize),
                    move |_epoch| {
                        StakeTableQuorum::new((prev_quorum, next_quorum), self.epoch_height)
                    },
                )
                .await
                .context(format!("fetching epoch root for {epoch}"))?
            };
            if let Some(hash) = root.next_stake_table_hash() {
                ensure!(
                    hash == stake_table.commit(),
                    "epoch {epoch} root {root_height} stake table hash {hash} does not match \
                     reconstructed hash {}",
                    stake_table.commit(),
                );
            } else if self.decaf() && root.version() < DrbAndHeaderUpgradeVersion::VERSION {
                // Decaf briefly ran a version of PoS where epoch root headers did not contain the
                // hash of the expected stake table, and so we can not verify that we computed the
                // correct stake table. Since this applies only to Decaf, which is a testnet, it is
                // acceptable to trust that the server supplied the correct events in this case.
                tracing::debug!(
                    ?epoch,
                    root_height,
                    computed_hash = %stake_table.commit(),
                    "trusting stake table prior to DRB upgrade on Decaf",
                );
            } else {
                bail!("epoch {epoch} root {root_height} does not have next stake table hash");
            }

            // Cache the reconstructed stake table in the database. We store
            // `epoch_root_protocol_version` (snapshot for this epoch, used by cache hits) and
            // `root.version()` (snapshot for `epoch + 1`, used to seed iter 1 of a future
            // catchup that resumes from this row).
            if let Err(err) = self
                .db
                .insert_stake_table(
                    EpochNumber::new(epoch),
                    &stake_table,
                    epoch_root_protocol_version,
                    root.version(),
                )
                .await
            {
                // If this fails, we can continue with the stake table that we have in memory right
                // now, so this is just a warning.
                tracing::warn!(epoch, "failed to cache stake table: {err:#}");
            }

            tracing::info!(epoch, num_events, "finished stake table catchup for epoch");
            prev_quorum = next_quorum;
            epoch_root_protocol_version = root.version();
        }

        Ok(self.cache_stake_table(epoch, prev_quorum).await)
    }

    fn fetch_leaf_from_server<'a, 'b, Q>(
        &'a self,
        id: impl Send + Into<LeafRequest> + 'a,
        known_finalized: Option<&'b Leaf2>,
        make_quorum: impl 'a + Send + FnOnce(EpochNumber) -> Q,
    ) -> impl 'b + Send + Future<Output = Result<LeafQueryData<SeqTypes>>>
    where
        'a: 'b,
        Q: Send + Quorum,
    {
        async move {
            let id = id.into();
            let proof = self
                .server
                .leaf_proof(id, known_finalized.map(Leaf2::height))
                .await?;
            let quorum;
            let hint = match proof.epoch(self.epoch_height)? {
                Some(epoch) => {
                    quorum = make_quorum(epoch);
                    LeafProofHint::Quorum(&quorum)
                },
                None => LeafProofHint::Assumption(known_finalized.context(
                    "server returned proof with assumption, but we have no finalized upper bound \
                     to verify assumption",
                )?),
            };
            let leaf = proof.verify(hint).await?;

            // The server has given us a leaf and correctly proved it finalized, but we still need to
            // verify that it actually gave us the leaf we requested.
            ensure!(
                leaf_matches_id(&leaf, id),
                "server returned a valid leaf proof for the wrong leaf (requested leaf {id}, got \
                 leaf {} with hash {})",
                leaf.height(),
                leaf.hash(),
            );

            // Having fetched and verified the leaf from the server, we can now cache it locally to
            // improve future requests.
            if let Err(err) = self.db.insert_leaf(leaf.clone()).await {
                // If this fails, we can still successfully return the leaf that we have in memory right
                // now, so this is just a warning.
                tracing::warn!("failed to cache fetched leaf: {err:#}");
            }

            Ok(leaf)
        }
    }

    async fn cache_stake_table(
        &self,
        epoch: EpochNumber,
        stake_table: Arc<StakeTable>,
    ) -> Arc<StakeTable> {
        let mut cache = self.stake_tables.write().await;

        // If inserting the new stake table would cause the cache to exceed its maximum size, first
        // delete an old stake table.
        if cache.len() >= self.opt.num_stake_tables_in_memory {
            // Always delete the _second oldest_ stake table. We want to keep the oldest around
            // because it is the hardest to catch up for if we need it again (we would have to go
            // all the way back to genesis). The second oldest is the least likely to be used again
            // after the oldest, while still being easy to replay if we do need it (because we can
            // just replay from the cached oldest).
            if let Some(&second_oldest_epoch) = cache.keys().nth(1) {
                cache.remove(&second_oldest_epoch);
            }
        }

        cache.entry(epoch).insert_entry(stake_table).get().clone()
    }

    fn decaf(&self) -> bool {
        self.chain_id == DECAF_CHAIN_ID
    }
}

fn stake_table_state_to_quorum(
    state: &StakeTableState,
    protocol_version: Version,
) -> Result<StakeTable> {
    let vs = ValidatorSet::from_state(state, protocol_version)?;
    Ok(vs
        .active_validators()
        .values()
        .map(|validator| StakeTableEntry {
            stake_key: *validator.stake_table_key(),
            stake_amount: validator.stake,
        })
        .collect())
}

impl<P, S> StakeTablePair for (EpochNumber, &LightClient<P, S>)
where
    P: Storage,
    S: Client,
{
    async fn stake_table(&self) -> Result<Arc<StakeTable>> {
        self.1.quorum_for_epoch(self.0).await
    }

    async fn next_epoch_stake_table(&self) -> Result<Arc<StakeTable>> {
        self.1.quorum_for_epoch(self.0 + 1).await
    }
}

fn leaf_matches_id(leaf: &LeafQueryData<SeqTypes>, id: impl Into<LeafRequest>) -> bool {
    match id.into() {
        LeafRequest::Leaf(LeafId::Number(h)) | LeafRequest::Header(BlockId::Number(h)) => {
            (h as u64) == leaf.height()
        },
        LeafRequest::Leaf(LeafId::Hash(h)) => h == leaf.hash(),
        LeafRequest::Header(BlockId::Hash(h)) => h == leaf.block_hash(),
        LeafRequest::Header(BlockId::PayloadHash(h)) => h == leaf.payload_hash(),
    }
}

fn header_matches_id(header: &Header, id: BlockId<SeqTypes>) -> bool {
    match id {
        BlockId::Number(n) => header.height() == (n as u64),
        BlockId::Hash(h) => header.commit() == h,
        BlockId::PayloadHash(h) => header.payload_commitment() == h,
    }
}

#[cfg(test)]
mod test {
    use std::collections::HashSet;

    use espresso_types::{
        NsIndex, RegisteredValidatorMap, StakeTableState, ValidatorSet, v0_3::RegisteredValidator,
    };
    use hotshot_query_service_types::availability::TransactionIndex;
    use hotshot_types::{addr::NetAddr, x25519};
    use itertools::izip;
    use pretty_assertions::assert_eq;
    use versions::{DRB_AND_HEADER_UPGRADE_VERSION, EPOCH_VERSION, NEW_PROTOCOL_VERSION};

    use super::*;
    use crate::{
        storage::SqlStorage,
        testing::{TestClient, leaf_chain, random_validator},
    };

    #[tokio::test]
    #[test_log::test]
    async fn test_block_height() {
        let client = TestClient::default();
        let db = SqlStorage::default().await.unwrap();

        let lc = LightClient::from_genesis(db.clone(), client.clone(), client.genesis().await);

        // Test empty block height.
        assert_eq!(lc.block_height().await.unwrap(), 0);

        // Local block height greater than server.
        let leaf = leaf_chain(1..2, DRB_AND_HEADER_UPGRADE_VERSION)
            .await
            .remove(0);
        db.insert_leaf(leaf).await.unwrap();
        assert_eq!(lc.block_height().await.unwrap(), 2);

        // Server block height greater than local.
        client.leaf(2).await;
        assert_eq!(lc.block_height().await.unwrap(), 3);
        // In the process of verifying the block height, we should have fetched and stored the
        // latest leaf.
        assert_eq!(db.block_height().await.unwrap(), 3);

        // Server lies about the block height.
        client.mock_block_height(10).await;
        client.forget_leaf(9).await;
        assert_eq!(lc.block_height().await.unwrap(), 3);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaf_twice() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // Fetch the leaf for the first time. We will need to get it from the server.
        let leaf = client.remember_leaf(1).await;
        assert_eq!(lc.fetch_leaf(LeafId::Number(1)).await.unwrap(), leaf);

        // Fetching the leaf again hits the cache.
        client.forget_leaf(1).await;
        assert_eq!(lc.fetch_leaf(LeafId::Number(1)).await.unwrap(), leaf);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaf_upper_bound() {
        let client = TestClient::default();

        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(client.leaf(2).await).await.unwrap();

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        assert_eq!(
            lc.fetch_leaf(LeafId::Number(1)).await.unwrap(),
            client.leaf(1).await,
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaf_distant_upper_bound() {
        let client = TestClient::default();
        client.reject_distant_finalized_hints(16).await;

        // A cached leaf far above the requested one must not be sent as the finalized hint; the
        // fetch should use a quorum-verified proof instead.
        let distant = MAX_FINALIZED_HINT_DISTANCE + 2;
        let distant_leaf = leaf_chain(distant..=distant, DRB_AND_HEADER_UPGRADE_VERSION)
            .await
            .remove(0);
        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(distant_leaf).await.unwrap();

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        assert_eq!(
            lc.fetch_leaf(LeafId::Number(1)).await.unwrap(),
            client.leaf(1).await,
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaf_invalid_proof() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_invalid_proof(1).await;
        lc.fetch_leaf(LeafId::Number(1)).await.unwrap_err();
        lc.fetch_leaf(LeafId::Hash(client.leaf(1).await.hash()))
            .await
            .unwrap_err();
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaf_wrong_leaf() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_wrong_leaf(1, 2).await;
        lc.fetch_leaf(LeafId::Number(1)).await.unwrap_err();
        lc.fetch_leaf(LeafId::Hash(client.leaf(1).await.hash()))
            .await
            .unwrap_err();
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_twice() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // Fetch the header for the first time. We will need to get it from the server.
        let leaf = client.remember_leaf(1).await;
        assert_eq!(
            lc.fetch_header(BlockId::Number(1)).await.unwrap(),
            *leaf.header()
        );

        // Fetching the header again hits the cache.
        client.forget_leaf(1).await;
        assert_eq!(
            lc.fetch_header(BlockId::Number(1)).await.unwrap(),
            *leaf.header()
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_upper_bound() {
        let client = TestClient::default();

        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(client.leaf(2).await).await.unwrap();

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        assert_eq!(
            lc.fetch_header(BlockId::Number(1)).await.unwrap(),
            *client.leaf(1).await.header(),
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_falls_back_to_leaf_proof_when_header_proof_fails() {
        let client = TestClient::default();

        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(client.leaf(2).await).await.unwrap();
        client.fail_header_proofs().await;

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        assert_eq!(
            lc.fetch_header(BlockId::Number(1)).await.unwrap(),
            *client.leaf(1).await.header(),
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_fallback_anchors_on_cached_successor_without_cert2() {
        let client = TestClient::default();
        client.set_upgrade(1, NEW_PROTOCOL_VERSION).await;

        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(client.leaf(2).await).await.unwrap();
        client.fail_header_proofs().await;
        client.forget_cert2(1).await;

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        assert_eq!(
            lc.fetch_header(BlockId::Number(1)).await.unwrap(),
            *client.leaf(1).await.header(),
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_reports_header_proof_error_when_fallback_fails() {
        let client = TestClient::default();

        let db = SqlStorage::default().await.unwrap();
        db.insert_leaf(client.leaf(2).await).await.unwrap();
        client.fail_header_proofs().await;
        client.forget_leaf(1).await;

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        let err = lc.fetch_header(BlockId::Number(1)).await.unwrap_err();
        assert!(
            format!("{err:#}").contains("header proof also failed: no merklized state at root 2"),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_invalid_proof() {
        let client = TestClient::default();
        let db = SqlStorage::default().await.unwrap();

        // Start with an upper bound, so that `fetch_header` goes through the `client.header_proof`
        // path.
        db.insert_leaf(client.leaf(2).await).await.unwrap();

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        client.return_invalid_proof(1).await;

        let err = lc.fetch_header(BlockId::Number(1)).await.unwrap_err();
        assert!(err.to_string().contains("invalid header proof"), "{err:#}");
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_header_wrong_header() {
        let client = TestClient::default();
        let db = SqlStorage::default().await.unwrap();

        // Start with an upper bound, so that `fetch_header` goes through the `client.header_proof`
        // path.
        db.insert_leaf(client.leaf(2).await).await.unwrap();

        let lc = LightClient::from_genesis(db, client.clone(), client.genesis().await);
        client.return_wrong_leaf(1, 0).await;

        let err = lc.fetch_header(BlockId::Number(1)).await.unwrap_err();
        assert!(
            err.to_string()
                .contains("server returned a valid header proof for the wrong header"),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_in_range() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // Fetch leaves in range [1,3) for the first time. We will need to get them from the server.
        let leaf1 = client.remember_leaf(1).await;
        let leaf2 = client.remember_leaf(2).await;
        client.remember_leaf(3).await;

        let leaves = lc.fetch_leaves_in_range(1, 3).await.unwrap();

        assert_eq!(leaves, vec![leaf1.clone(), leaf2.clone()]);

        // now remove from server and this time it should be able to fetch from local db
        client.forget_leaf(1).await;
        client.forget_leaf(2).await;
        let leaves = lc.fetch_leaves_in_range(1, 3).await.unwrap();
        assert_eq!(leaves, vec![leaf1, leaf2]);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_in_range_uncached() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        let leaf1 = client.remember_leaf(1).await;
        let leaf2 = client.remember_leaf(2).await;
        let leaf3 = client.remember_leaf(3).await;

        let leaves = lc.fetch_leaves_in_range_uncached(1, 4).await.unwrap();
        assert_eq!(leaves, vec![leaf1, leaf2, leaf3.clone()]);

        // Only the proven last leaf is cached; the ones chained to it are not.
        assert_eq!(lc.db.get_leaves_in_range(1, 4).await.unwrap(), vec![leaf3]);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_headers_in_range() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // Fetch headers in range [1,3) for the first time. We will need to get them from the server.
        let leaf1 = client.remember_leaf(1).await;
        let leaf2 = client.remember_leaf(2).await;
        client.remember_leaf(3).await;

        let headers = lc.fetch_headers_in_range(1, 3).await.unwrap();

        assert_eq!(
            headers,
            vec![leaf1.header().clone(), leaf2.header().clone()]
        );

        // now remove from server and this time it should be able to fetch from local db
        client.forget_leaf(1).await;
        client.forget_leaf(2).await;
        let headers = lc.fetch_headers_in_range(1, 3).await.unwrap();
        assert_eq!(
            headers,
            vec![leaf1.header().clone(), leaf2.header().clone()]
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_in_range_invalid_proof() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_invalid_proof(2).await;
        let err = lc.fetch_leaves_in_range(1, 3).await.unwrap_err();
        assert!(
            err.to_string().contains(
                "server returned proof with assumption, but we have no finalized upper bound to \
                 verify assumption"
            ),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_headers_in_range_invalid_proof() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_invalid_proof(2).await;
        let err = lc.fetch_headers_in_range(1, 3).await.unwrap_err();
        assert!(
            err.to_string().contains(
                "server returned proof with assumption, but we have no finalized upper bound to \
                 verify assumption"
            ),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_in_range_wrong_leaf() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_wrong_leaf(1, 2).await;
        let err = lc.fetch_leaves_in_range(1, 4).await.unwrap_err();
        assert!(err.to_string().contains("leaf hash mismatch"), "{err:#}");
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_for_ranges() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        let mut expected = vec![];
        for height in [1usize, 2, 5, 6, 7] {
            expected.push(client.leaf(height).await);
        }
        let fetched = lc.fetch_leaves_for_ranges(&[1..3, 5..8]).await.unwrap();
        assert_eq!(fetched, expected);
    }

    // Pins the coalescing cap written out in fetch_leaves_for_ranges to the default it copies.
    #[test]
    fn coalescing_cap_matches_query_service_default() {
        assert_eq!(
            hotshot_query_service::availability::Options::default().small_object_range_limit,
            500
        );
    }

    // Touching ranges are what the scanner sends when a run straddles a chunk boundary, so they
    // must pass; unordered or overlapping ones must not, or the positional pairing downstream
    // would silently misalign.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_for_ranges_ordering() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        let mut expected = vec![];
        for height in 1..5usize {
            expected.push(client.leaf(height).await);
        }
        assert_eq!(
            lc.fetch_leaves_for_ranges(&[1..3, 3..5]).await.unwrap(),
            expected
        );
        for ranges in [[5..8, 1..3], [1..4, 3..6]] {
            let err = lc.fetch_leaves_for_ranges(&ranges).await.unwrap_err();
            assert!(err.to_string().contains("ascending"), "{err:#}");
        }
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_for_ranges_wrong_leaf() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // A substituted leaf fails the span's chain walk. The per-run path that failure falls
        // through to must reject it too: the leaf self-reports its real height, so it lands in the
        // bulk map under the wrong key and height 1 goes to an individually proven fetch.
        client.return_wrong_leaf(1, 2).await;
        #[allow(clippy::single_range_in_vec_init)]
        let err = lc.fetch_leaves_for_ranges(&[1..4]).await.unwrap_err();
        assert!(err.to_string().contains("wrong leaf"), "{err:#}");
    }

    // Gaps too wide to coalesce into one span: the bulk of the leaves come from the ranges request
    // endpoint and each run is anchored by its own proof.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_for_ranges_wide_span() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        let mut expected = vec![];
        for height in [1usize, 2, 60, 61, 62] {
            expected.push(client.leaf(height).await);
        }
        let fetched = lc.fetch_leaves_for_ranges(&[1..3, 60..63]).await.unwrap();
        assert_eq!(fetched, expected);
    }

    // The gaps are too wide to coalesce, so this reaches the ranges endpoint and exercises the
    // fallback when it fails; a coalesced span never asks the ranges endpoint at all.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_leaves_for_ranges_without_ranges_endpoint() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.fail_leaf_ranges().await;

        let mut expected = vec![];
        for height in [1usize, 2, 60, 61, 62] {
            expected.push(client.leaf(height).await);
        }
        let fetched = lc.fetch_leaves_for_ranges(&[1..3, 60..63]).await.unwrap();
        assert_eq!(fetched, expected);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_headers_in_range_wrong_leaf() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.return_wrong_leaf(1, 2).await;
        let err = lc.fetch_headers_in_range(1, 4).await.unwrap_err();
        assert!(err.to_string().contains("leaf hash mismatch"), "{err:#}");
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_twice_cache() {
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            genesis.clone(),
        );

        // Fetch a dynamic stake table for the first time. We will need to get it from the server.
        let epoch = genesis.first_epoch_with_dynamic_stake_table + 5;
        let expected = client.quorum_for_epoch(epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);

        // Fetching the stake table again hits the cache.
        client.forget_quorum(epoch).await;
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_twice_storage() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        // Fetch a dynamic stake table for the first time. We will need to get it from the server.
        let epoch = genesis.first_epoch_with_dynamic_stake_table + 5;
        let expected = client.quorum_for_epoch(epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);

        // Even if the in-memory cache is cleared, we can still get this stake table without hitting
        // the server, but fetching from the database.
        client.forget_quorum(epoch).await;
        let lc = LightClient::from_genesis(db, client.clone(), genesis);
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_from_lower_bound() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        // Fetch a stake table, causing it to be stored in the stake table, making it usable as a
        // lower bound for later catchup.
        let epoch = genesis.first_epoch_with_dynamic_stake_table + 5;
        lc.quorum_for_epoch(epoch).await.unwrap();

        // Cause the server to forget older epochs' stake table events, so that we can only catch up
        // successfully if we start from the saved lower bound.
        for epoch in 0..*epoch - 1 {
            client.forget_quorum(EpochNumber::new(epoch)).await;
        }

        // Fetch a future stake table, catching up from the saved lower bound.
        let expected = client.quorum_for_epoch(epoch + 5).await.into();
        assert_eq!(*lc.quorum_for_epoch(epoch + 5).await.unwrap(), expected);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_without_header_proofs() {
        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let epoch = genesis.first_epoch_with_dynamic_stake_table + 5;

        let db = SqlStorage::default().await.unwrap();
        let above_every_root = root_block_in_epoch(*epoch + 1, 10) as usize;
        db.insert_leaf(client.leaf(above_every_root).await)
            .await
            .unwrap();
        client.fail_header_proofs().await;

        let lc = LightClient::from_genesis(db, client.clone(), genesis);
        let expected = client.quorum_for_epoch(epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_cache_removal() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis_with_options(
            db.clone(),
            client.clone(),
            genesis.clone(),
            LightClientOptions {
                num_stake_tables_in_memory: 2,
            },
        );

        // Fetch two stake tables, causing the cache to be filled up.
        for i in 0..2 {
            let epoch = genesis.first_epoch_with_dynamic_stake_table + 5 + i;
            assert_eq!(
                *lc.quorum_for_epoch(epoch).await.unwrap(),
                client.quorum_for_epoch(epoch).await.into(),
            );
        }
        assert_eq!(lc.stake_tables.read().await.len(), 2);

        // Fetch a third stake table, causing the second-oldest stake table to be deleted.
        assert_eq!(
            *lc.quorum_for_epoch(genesis.first_epoch_with_dynamic_stake_table + 10)
                .await
                .unwrap(),
            client
                .quorum_for_epoch(genesis.first_epoch_with_dynamic_stake_table + 10)
                .await
                .into(),
        );
        assert_eq!(lc.stake_tables.read().await.len(), 2);

        // Even if the server forgets all earlier stake tables, we can still catch up because we
        // didn't remove the lower bound stake table.
        let epoch = genesis.first_epoch_with_dynamic_stake_table + 6;
        for i in 0..*epoch {
            client.forget_quorum(EpochNumber::new(i)).await;
        }
        assert_eq!(
            *lc.quorum_for_epoch(epoch).await.unwrap(),
            client.quorum_for_epoch(epoch).await.into(),
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_invalid() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let epoch = genesis.first_epoch_with_dynamic_stake_table + 5;
        client.return_invalid_quorum(epoch).await;
        let err = lc.quorum_for_epoch(epoch).await.unwrap_err();
        assert!(
            err.to_string()
                .contains("does not match reconstructed hash"),
            "{err:#}"
        );
    }

    /// Regression: epoch-root leaf proofs must locate a 2-chain when views immediately after
    /// the root are non-contiguous. Alternates failed view (gap) and ok view past the root.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_with_view_gaps_after_root() {
        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            genesis.clone(),
        );

        let target_epoch = genesis.first_epoch_with_dynamic_stake_table + 1;
        let root_height = root_block_in_epoch(*target_epoch - 1, 10) as usize;
        for i in [0, 2, 4] {
            client.add_view_gap_after(root_height + i).await;
        }

        let expected = client.quorum_for_epoch(target_epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(target_epoch).await.unwrap(), expected);
    }

    /// Regression: epoch-transition QC verification must actually check `next_epoch_qc`
    /// against the next epoch's quorum.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_rejects_tampered_next_epoch_qc() {
        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            genesis.clone(),
        );

        let target_epoch = genesis.first_epoch_with_dynamic_stake_table + 1;
        let root_height = root_block_in_epoch(*target_epoch - 1, 10) as usize;
        client.add_view_gap_after(root_height).await;
        client.corrupt_next_epoch_qc().await;

        let err = lc.quorum_for_epoch(target_epoch).await.unwrap_err();
        assert!(
            err.to_string().contains("verifying next epoch QC")
                || err
                    .chain()
                    .any(|e| e.to_string().contains("verifying next epoch QC")),
            "{err:#}"
        );
    }

    /// Regression: from-genesis catchup must seed the snapshot version by fetching the root at
    /// `first_dynamic - 2` (verified by the genesis stake table), not by reusing the version of
    /// the root in epoch `first_dynamic - 1`. Upgrades the root at `first_dynamic - 1` to
    /// CLIQUENET so it differs from the bootstrap root at `first_dynamic - 2`.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_genesis_bootstrap_version() {
        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let db = SqlStorage::default().await.unwrap();
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let epoch = genesis.first_epoch_with_dynamic_stake_table;
        let root_height = root_block_in_epoch(*epoch - 1, 10);
        client.set_upgrade(root_height, NEW_PROTOCOL_VERSION).await;

        lc.quorum_for_epoch(epoch).await.unwrap();

        let (cached_epoch, _, cached_version, _) =
            db.stake_table_lower_bound(epoch).await.unwrap().unwrap();
        assert_eq!(cached_epoch, epoch);
        assert_eq!(cached_version, DRB_AND_HEADER_UPGRADE_VERSION);
    }

    /// Regression: iter 1 of DB-load catchup must seed from `next_stored_version`, not
    /// `stored_version`.
    ///
    /// - insert stake table for `lower_bound_epoch` with `stored_version` = V4 and
    ///   `next_stored_version` = CLIQUENET
    /// - activate the upgrade at `target_epoch_root`
    /// - request quorum for `target_epoch`
    /// - fix: seed = `next_stored_version` = CLIQUENET, filter drops all validators, errors
    /// - previous bug: seed = `stored_version` = V4, no filter, catchup wrongly succeeds
    ///
    /// Brittle: asserts that an error happens, not the filter version. Avoids x25519/p2p
    /// plumbing in the shared test client.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_db_load_advances_snapshot_version() {
        use crate::client::Client;

        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let db = SqlStorage::default().await.unwrap();
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let lower_bound_epoch = genesis.first_epoch_with_dynamic_stake_table;
        let target_epoch = lower_bound_epoch + 1;
        let target_epoch_root = root_block_in_epoch(*target_epoch - 2, 10);

        let mut prev_state = StakeTableState::default();
        for event in client.stake_table_events(lower_bound_epoch).await.unwrap() {
            prev_state.apply_event(event).unwrap().unwrap();
        }
        db.insert_stake_table(
            lower_bound_epoch,
            &prev_state,
            DRB_AND_HEADER_UPGRADE_VERSION,
            NEW_PROTOCOL_VERSION,
        )
        .await
        .unwrap();

        client
            .set_upgrade(target_epoch_root, NEW_PROTOCOL_VERSION)
            .await;

        let err = lc.quorum_for_epoch(target_epoch).await.unwrap_err();
        assert!(
            err.to_string()
                .contains("No validators met the minimum criteria")
                || err.chain().any(|e| e
                    .to_string()
                    .contains("No validators met the minimum criteria")),
            "{err:#}"
        );
    }

    /// Regression: catchup filtered the candidate active set at `root.version()` (root in
    /// epoch `e-1`) instead of the snapshot root's version (epoch `e-2`). When those versions
    /// differ across CLIQUENET, the LC diverges from the espresso node.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_catchup_uses_snapshot_root_version() {
        use crate::client::Client;

        let client = TestClient::with_epoch_height(10);
        let genesis = client.genesis().await;
        let db = SqlStorage::default().await.unwrap();
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let prev_epoch = genesis.first_epoch_with_dynamic_stake_table;
        let target_epoch = prev_epoch + 1;
        let root_height = root_block_in_epoch(*target_epoch - 1, 10);

        let mut prev_state = StakeTableState::default();
        for event in client.stake_table_events(prev_epoch).await.unwrap() {
            prev_state.apply_event(event).unwrap().unwrap();
        }
        db.insert_stake_table(
            prev_epoch,
            &prev_state,
            DRB_AND_HEADER_UPGRADE_VERSION,
            DRB_AND_HEADER_UPGRADE_VERSION,
        )
        .await
        .unwrap();

        client.set_upgrade(root_height, NEW_PROTOCOL_VERSION).await;

        let expected = client.quorum_for_epoch(target_epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(target_epoch).await.unwrap(), expected);
    }

    /// Regression: CLIQUENET active-set filter must run on the cache-hit path using the stored
    /// version, dropping validators missing x25519/p2p.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_cliquenet_filter() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let mut complete: RegisteredValidator<PubKey> = random_validator().into();
        complete.x25519_key = Some(x25519::PublicKey::try_from([42u8; 32].as_slice()).unwrap());
        complete.p2p_addr = Some("127.0.0.1:9000".parse::<NetAddr>().unwrap());

        let incomplete: RegisteredValidator<PubKey> = random_validator().into();
        assert!(incomplete.x25519_key.is_none());
        assert!(incomplete.p2p_addr.is_none());

        let mut validators = RegisteredValidatorMap::default();
        validators.insert(complete.account, complete.clone());
        validators.insert(incomplete.account, incomplete.clone());

        let complete_bls = complete.stake_table_key.expect("authenticated validator");
        let incomplete_bls = incomplete.stake_table_key.expect("authenticated validator");
        let mut used_bls = HashSet::default();
        used_bls.insert(complete_bls);
        used_bls.insert(incomplete_bls);

        let mut used_schnorr = HashSet::default();
        used_schnorr.insert(
            complete
                .state_ver_key
                .clone()
                .expect("random_validator has valid schnorr key"),
        );
        used_schnorr.insert(
            incomplete
                .state_ver_key
                .clone()
                .expect("random_validator has valid schnorr key"),
        );

        let mut used_x25519 = HashSet::default();
        used_x25519.insert(complete.x25519_key.unwrap());

        let state = StakeTableState::new(
            validators,
            Default::default(),
            used_bls,
            used_schnorr,
            used_x25519,
        );

        let epoch = genesis.first_epoch_with_dynamic_stake_table + 1;
        db.insert_stake_table(epoch, &state, NEW_PROTOCOL_VERSION, NEW_PROTOCOL_VERSION)
            .await
            .unwrap();

        let quorum = lc.quorum_for_epoch(epoch).await.unwrap();
        let expected: StakeTable = vec![StakeTableEntry {
            stake_key: complete_bls,
            stake_amount: complete.stake,
        }]
        .into();
        assert_eq!(*quorum, expected);

        // Cross-check: the same state under EPOCH_VERSION (pre-cliquenet) keeps both. This
        // guards against the filter being unconditionally on.
        let pre_cliquenet = ValidatorSet::from_state(&state, EPOCH_VERSION).unwrap();
        assert_eq!(pre_cliquenet.active_validators().len(), 2);
    }

    /// Regression: catchup must thread the snapshot version end-to-end through reconstruction
    /// and cache it with each replayed epoch.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_stake_table_cliquenet_filter_reconstruction() {
        let db = SqlStorage::default().await.unwrap();
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(db.clone(), client.clone(), genesis.clone());

        let epoch = genesis.first_epoch_with_dynamic_stake_table + 2;
        assert!(db.stake_table_lower_bound(epoch).await.unwrap().is_none());

        let expected = client.quorum_for_epoch(epoch).await.into();
        assert_eq!(*lc.quorum_for_epoch(epoch).await.unwrap(), expected);

        let (cached_epoch, _, cached_version, _) =
            db.stake_table_lower_bound(epoch).await.unwrap().unwrap();
        assert_eq!(cached_epoch, epoch);
        assert_eq!(cached_version, DRB_AND_HEADER_UPGRADE_VERSION);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_payload() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        let mut hashes = vec![];
        for i in 1..10 {
            let payload = client.payload(i).await;
            let res = lc.fetch_payload(BlockId::Number(i)).await.unwrap();
            assert_eq!(res.data(), &payload);
            assert_eq!(res.height(), i as u64);
            assert_eq!(res.block_hash(), client.leaf(i).await.block_hash());
            assert_eq!(res.hash(), client.leaf(i).await.payload_hash());
            hashes.push(res.hash());
        }

        // Test ranged fetching.
        let blocks = lc.fetch_blocks_in_range(1, 10).await.unwrap();
        let vid = lc.fetch_vid_common_in_range(1, 10).await.unwrap();
        assert_eq!(blocks.len(), 9);
        assert_eq!(vid.len(), 9);
        for (i, (hash, block, vid)) in izip!(hashes, blocks, vid).enumerate() {
            assert_eq!(block.height() as usize, i + 1);
            assert_eq!(vid.height() as usize, i + 1);
            assert_eq!(block.payload_hash(), hash);
            assert_eq!(vid.payload_hash(), hash);
        }
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_payload_invalid() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        client.return_invalid_payload(1).await;
        let err = lc.fetch_payload(BlockId::Number(1)).await.unwrap_err();
        assert!(
            err.to_string()
                .contains("commitment of payload does not match commitment in header"),
            "{err:#}"
        );
    }

    // The leaf ranges request succeeds while the payload one fails, so this exercises the payload
    // fallback on its own.
    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_blocks_for_ranges_without_ranges_endpoint() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.fail_payload_proof_ranges().await;
        client.payload(62).await;

        let fetched = lc
            .fetch_blocks_and_vid_common_for_ranges(&[1..3, 60..63])
            .await
            .unwrap();
        assert_eq!(
            fetched
                .iter()
                .map(|(block, _)| block.height())
                .collect::<Vec<_>>(),
            [1, 2, 60, 61, 62]
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_blocks_for_ranges_invalid_proof() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );
        client.payload(7).await;

        client.return_invalid_payload(6).await;
        let err = lc
            .fetch_blocks_and_vid_common_for_ranges(&[1..3, 5..8])
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("commitment of payload does not match commitment in header"),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_namespace() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        for i in 1..10 {
            let leaf = client.leaf(i).await;
            let payload = client.payload(i).await;

            for id in [
                BlockId::Number(i),
                BlockId::Hash(leaf.block_hash()),
                BlockId::PayloadHash(leaf.payload_hash()),
            ] {
                // Request a non-empty namespace.
                let ns_index = NsIndex::from(0);
                let tx = payload
                    .transaction(&TransactionIndex {
                        ns_index,
                        position: 0,
                    })
                    .unwrap();
                let txs = lc.fetch_namespace(id, tx.namespace()).await.unwrap();
                assert_eq!(txs, std::slice::from_ref(&tx));

                // Request an empty namespace.
                let txs = lc
                    .fetch_namespace(id, NamespaceId::from(u64::from(tx.namespace()) + 1))
                    .await
                    .unwrap();
                assert_eq!(txs, []);
            }
        }

        // Fetch by range.
        let ns = client
            .payload(1)
            .await
            .transaction(&TransactionIndex {
                ns_index: 0.into(),
                position: 0,
            })
            .unwrap()
            .namespace();
        let namespaces = lc.fetch_namespaces_in_range(1, 10, ns).await.unwrap();
        assert_eq!(namespaces.len(), 9);
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_all_namespaces_in_range() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        // Request the namespaces from the first few blocks (which may differ per block), plus one
        // that is absent everywhere.
        let mut requested = vec![];
        for height in 1..4 {
            requested.push(
                client
                    .payload(height)
                    .await
                    .transaction(&TransactionIndex {
                        ns_index: 0.into(),
                        position: 0,
                    })
                    .unwrap()
                    .namespace(),
            );
        }
        let absent =
            NamespaceId::from(requested.iter().map(|ns| u64::from(*ns)).max().unwrap() + 1);
        requested.push(absent);

        let (start, end) = (1, 10);
        let blocks = lc
            .fetch_all_namespaces_in_range(start, end, &requested)
            .await
            .unwrap();
        assert_eq!(blocks.len(), end - start);

        for (i, block) in blocks.iter().enumerate() {
            let height = i + 1;
            let header = client.leaf(height).await.header().clone();

            // The map contains exactly the requested namespaces present in this block, and the
            // absent namespace is never included.
            assert!(!block.contains_key(&absent), "block {height}");
            for &ns in &requested {
                let expected = lc
                    .fetch_namespace(BlockId::Number(height), ns)
                    .await
                    .unwrap();
                if header.ns_table().find_ns_id(&ns).is_some() {
                    assert_eq!(block.get(&ns), Some(&expected), "block {height}, ns {ns}");
                } else {
                    assert!(!block.contains_key(&ns), "block {height}, ns {ns}");
                }
            }
        }
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_fetch_namespace_invalid() {
        let client = TestClient::default();
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            client.genesis().await,
        );

        let payload = client.payload(1).await;
        let ns_index = NsIndex::from(0);
        let tx = payload
            .transaction(&TransactionIndex {
                ns_index,
                position: 0,
            })
            .unwrap();

        client.return_invalid_payload(1).await;
        let err = lc
            .fetch_namespace(BlockId::Number(1), tx.namespace())
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("invalid namespace proof"),
            "{err:#}"
        );
    }

    #[tokio::test]
    #[test_log::test]
    async fn test_genesis_round_trip() {
        let client = TestClient::default();
        let genesis = client.genesis().await;
        let lc = LightClient::from_genesis(
            SqlStorage::default().await.unwrap(),
            client.clone(),
            genesis.clone(),
        );
        assert_eq!(lc.genesis(), genesis);
    }
}

#[cfg(all(test, feature = "rlp"))]
mod rlp_test {
    use alloy::primitives::U256;
    use alloy_rlp::{Decodable, Encodable};
    use hotshot_types::traits::signature_key::SignatureKey;

    use super::*;

    #[test_log::test]
    fn rlp_genesis_round_trip() {
        let genesis = Genesis {
            epoch_height: 1000,
            first_epoch_with_dynamic_stake_table: EpochNumber::new(3),
            stake_table: vec![StakeTableEntry {
                stake_key: PubKey::generated_from_seed_indexed(
                    Default::default(),
                    Default::default(),
                )
                .0,
                stake_amount: U256::MAX,
            }],
            chain_id: DECAF_CHAIN_ID,
        };

        let mut buf = vec![];
        genesis.encode(&mut buf);

        let mut buf = buf.as_slice();
        assert_eq!(genesis, Genesis::decode(&mut buf).unwrap());
        assert!(buf.is_empty());
    }
}